LangChain DeepAgent 多 Agent 架构原理学习：从 SubAgent 到 Middleware 的协作链路拆解

1. 从一次“任务全塞给一个 Agent”的翻车说起LangChain DeepAgent 多 Agent 架构说白了就是让一个主 Agent 当“项目经理”把活拆给若干 SubAgent 去干中间用 Middleware 做拦截和状态隔离。它适合谁适合已经会用 LangChain 写单 Agent、但一遇到“调研 写代码 审查”这种复合任务就上下文爆炸的工程师。我试过把三个任务塞进一个 Agent结果 messages 越滚越长模型开始胡编工具调用最后连自己上一步干了啥都记不住。DeepAgent 的解法很直接主 Agent 只保留 task 工具和最终汇总职责具体执行交给独立的 SubAgent每个 SubAgent 拿到的是干净上下文——只有一条HumanMessage(description)跑完只回传最后一条消息的文本。这背后有三条主线SubAgent 分工谁干什么、Middleware 拦截在哪插手、消息流转状态怎么隔离和回传。把这三条线拆清楚你就能从“会调 API”升级到“看得懂框架设计”。这篇不堆概念直接给你可复制的目录结构、依赖清单、配置片段再附上启动后观察 Agent 调用顺序和中间件触发时机的验证动作。核心检索词先摆出来LangChain DeepAgent 多 Agent 架构、SubAgent 分工、Middleware 拦截、消息流转。读完你应该能自己画出create_deep_agent()到task工具注入的完整链路。先明确几个术语后面代码里会反复出现。主 Agent 由create_deep_agent()构建持有task工具负责任务分发与结果汇总。SubAgent 是主 Agent 通过task工具委托的执行单元独立运行、完成具体任务后返回结果。SubAgentTypedDict是声明式配置描述字典框架自动编译为可运行对象CompiledSubAgentTypedDict是你自己编译好的可运行对象封装框架直接使用。_SubagentSpec是框架内部统一产物包含name、description、runnable三个字段。task tool注入到主 Agent 的StructuredTool是 LLM 调用子 Agent 的唯一入口。SubAgentMiddleware是核心中间件负责构建子 Agent、注册路由表、生成 task tool 并注入系统提示。_EXCLUDED_STATE_KEYS是父子 Agent 间隔离的 state 字段集合防止状态污染。Command是 LangGraph 原语携带 update 字典直接修改图的 state子 Agent 用它回传结果。2. TaoToken 前置把模型入口先跑通在拆架构之前得先有个能稳定调用的模型入口否则你连create_deep_agent()都跑不起来。这里用 TaoToken 做统一接入它的 API 地址是https://taotoken.net/api兼容 OpenAI 风格的调用方式LangChain 的ChatOpenAI可以直接指过去。官网在https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content注册后在控制台生成 Key 即可。为什么先讲这个因为 DeepAgent 的多 Agent 协作会放大调用次数——主 Agent 一轮可能并发触发多个task每个 SubAgent 内部又有多轮工具调用。如果模型入口不稳定你根本分不清是架构问题还是网络问题。把入口固定下来后面排障才有基准。操作路径很清晰进控制台创建 API Key然后在项目里用环境变量注入。我习惯把 Key 放在.env里代码里用os.getenv读取避免硬编码。模型 ID 按你控制台里可用的填比如gpt-4o-mini这类通用对话模型就够跑通最小示例。如果你后面要做长期编码或 Agent 编排可以看下 Coding Plan 页面它针对高频调用场景做了额度规划单纯验证模型行为的话模型对话页面就能直接试。这里要提醒一句TaoToken 是模型接入层不是编辑器替代品也不做任何灰色中转。你把它当成一个标准的 OpenAI 兼容端点用就行。配置片段如下先确保这段能跑通再往下拆架构import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() llm ChatOpenAI( modelos.getenv(TAOTOKEN_MODEL, gpt-4o-mini), api_keyos.getenv(TAOTOKEN_API_KEY), base_urlhttps://taotoken.net/api, temperature0, ) resp llm.invoke(用一句话说明什么是多 Agent 协作) print(resp.content).env文件内容TAOTOKEN_API_KEYsk-你的key TAOTOKEN_MODELgpt-4o-mini跑通这段你会看到模型正常返回。如果这里就报 401先别往下走去 API Keys 页面确认 Key 是否复制完整、有没有多余空格。这一步是整个多 Agent 架构的地基地基不稳后面全是玄学。3. 可复制配置目录结构、依赖与最小多 Agent 示例这一节给你一套能直接跑的工程骨架。目录结构如下deepagent-demo/ ├── .env ├── requirements.txt ├── main.py ├── agents/ │ ├── __init__.py │ ├── researcher.py │ └── code_reviewer.py └── configs/ └── subagents.pyrequirements.txt依赖清单langchain0.3.0 langchain-openai0.2.0 langgraph0.2.0 deepagents0.1.0 python-dotenv1.0.0configs/subagents.py里定义两个 SubAgent 的声明式配置。注意SubAgent是 TypedDict字段包括name、description、system_prompt、tools、model、middleware、skills。description是 LLM 路由决策的唯一依据必须写清楚能力边界from deepagents import SubAgent RESEARCHER_SUBAGENT: SubAgent { name: researcher, description: 专注文献检索与资料汇总。当任务需要查找背景信息、整理多来源观点时使用。, system_prompt: 你是一个严谨的研究员输出结构化摘要标注信息来源。, } CODE_REVIEWER_SUBAGENT: SubAgent { name: code-reviewer, description: 专注代码审查。当任务需要检查代码质量、发现潜在 bug、给出改进建议时使用。, system_prompt: 你是一个资深代码审查员逐行检查逻辑与边界条件输出问题清单。, }main.py里用create_deep_agent()组装主 Agent。注意subagents参数接收的是列表框架会自动把general-purpose内置子 Agent 拼在首位import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI from deepagents import create_deep_agent from configs.subagents import RESEARCHER_SUBAGENT, CODE_REVIEWER_SUBAGENT load_dotenv() model ChatOpenAI( modelos.getenv(TAOTOKEN_MODEL, gpt-4o-mini), api_keyos.getenv(TAOTOKEN_API_KEY), base_urlhttps://taotoken.net/api, temperature0, ) agent create_deep_agent( modelmodel, tools[], subagents[RESEARCHER_SUBAGENT, CODE_REVIEWER_SUBAGENT], system_prompt你是主协调 Agent负责拆解任务并委托给合适的子 Agent。, ) if __name__ __main__: result agent.invoke({ messages: [{role: user, content: 帮我调研 LangGraph 的 Command 机制并审查下面这段代码的边界问题def f(x): return 1/x}] }) for msg in result[messages]: print(type(msg).__name__, -, getattr(msg, content, )[:120])如果你要用CompiledSubAgent就得自己先编译一个 Runnablestate 必须含messageskey。配置片段如下from langgraph.graph import StateGraph, MessagesState from deepagents import CompiledSubAgent def build_custom_graph(): g StateGraph(MessagesState) # 这里省略节点与边定义重点是 compile 后作为 runnable return g.compile() CUSTOM_RAG: CompiledSubAgent { name: custom-rag, description: 自定义 RAG 检索子 Agent当需要访问内部知识库时使用。, runnable: build_custom_graph(), }把CUSTOM_RAG加进subagents列表即可。框架检测到 spec 里有runnable字段就直接透传不注入任何标准 middleware也不调用create_agent()。这是SubAgent和CompiledSubAgent的核心区别前者框架帮你编译并注入 TodoList、Filesystem、Summarization、PromptCaching、PatchToolCalls 这套标准中间件栈后者完全由你控制。4. 验证请求观察调用顺序与中间件触发时机配置跑起来后关键动作是验证。你要观察三件事主 Agent 是否先推理再决定调task、task的subagent_type是否命中注册表、子 Agent 返回后主 Agent 是否继续推理。在main.py里加一段回调打印每次 LLM 调用前的 system message 尾部就能看到SubAgentMiddleware的wrap_model_call是否把TASK_SYSTEM_PROMPT和可用子 Agent 列表追加进去了from langchain_core.callbacks import BaseCallbackHandler class TraceHandler(BaseCallbackHandler): def on_llm_start(self, serialized, prompts, **kwargs): for p in prompts: if Available subagent types in p: print([Middleware 注入] 检测到子 Agent 列表已追加到 system message) agent.invoke( {messages: [{role: user, content: 调研 LangGraph Command 并审查代码}]}, config{callbacks: [TraceHandler()]}, )预期输出顺序是这样的先打印[Middleware 注入]说明wrap_model_call在 LLM 调用前生效然后主 Agent 返回一个带tasktool call 的AIMessage接着task()闭包函数被触发_validate_and_prepare_state()从父 state 复制并过滤_EXCLUDED_STATE_KEYS重置messages为单条HumanMessage(description)子 Agent 独立跑完_return_command_with_state_update()取result[messages][-1].text.rstrip()包成ToolMessage通过Command(update...)写回父 state最后主 Agent 收到ToolMessage继续推理输出汇总。如果你在一轮里看到多个tasktool call说明主 Agent 在做并行委托。每个子 Agent 是独立invoke互不共享 messages只回传最终结论。这正是_EXCLUDED_STATE_KEYS的设计意图messages、todos、structured_response、skills_metadata、memory_contents都不传给子 Agent子 Agent 启动时上下文干净只有一条明确的任务指令。验证成功的结果长这样控制台先出现中间件注入日志再出现AIMessage带task调用然后ToolMessage回填最后AIMessage给出综合回复。如果subagent_type写错你会看到错误字符串提示合法类型列表而不是抛异常——这是task()里先查注册表再校验tool_call_id的顺序决定的。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth排障这块按真实报错来。第一个高频错是 401通常出现在ChatOpenAI初始化或首次invoke时。原因无非三种Key 没读到、Key 复制带了空格、base_url写错。检查.env是否被load_dotenv()正确加载base_url必须是https://taotoken.net/api不要多加/v1或结尾斜杠。修复后重跑第 2 节的验证片段即可。第二个是local proxy failed。这个报错说明请求根本没出去多半是本机网络层配置干扰。检查你的 shell 里有没有残留的HTTP_PROXY、HTTPS_PROXY环境变量有就清掉再跑。注意这里说的是清理本机环境变量不是让你去配什么代理工具方向别搞反。第三个是reading choices这类 KeyError。典型场景是模型返回体结构和你预期不符LangChain 解析response[choices]时炸了。常见诱因是base_url指向了一个不兼容 OpenAI 格式的端点或者模型 ID 填错导致返回了错误对象。把base_url固定为https://taotoken.net/api模型 ID 用控制台里确认可用的值基本能消掉。第四个是 OAuth 相关报错。如果你在 Claude Code 或 Codex 这类工具里接入报 OAuth 失败通常是因为认证方式选错了——这些工具默认走 OAuth 流程而你要用的是 API Key 模式。以 Codex 的auth.json为例三件套必须写全Base URL、Key、Model ID。配置片段如下{ base_url: https://taotoken.net/api, api_key: sk-你的key, model: gpt-4o-mini }如果你用 CC Switch 或 Cline MCP 接入同样记住三件套Base URL 填https://taotoken.net/apiKey 填控制台生成的Model ID 填可用模型。少任何一个都会在握手阶段失败。Cline MCP 场景下还要确认 MCP server 的启动命令没有把环境变量覆盖掉。还有一个容易忽略的错CompiledSubAgent must return a state containing a messages key。这是_return_command_with_state_update()里的显式校验。你自定义的 Runnable 如果 compile 出来的 state 不含messages就会命中这个 ValueError。修复方式是确保你的StateGraph用MessagesState或手动在 state schema 里加上messages字段。6. 语义一致 CTA把架构认知落到可复用入口拆到这里你应该能画出完整链路了create_deep_agent()构建阶段先造general-purpose内置子 Agent再遍历用户subagentsSubAgent走create_agent()编译并注入标准中间件栈CompiledSubAgent直接透传然后_build_task_tool()把{name: runnable}注册表闭包进task工具tool description 里塞进所有子 Agent 的 name description 供 LLM 软路由最后SubAgentMiddleware把 task 工具和系统提示注入主 Agent。运行阶段wrap_model_call每次 LLM 调用前追加子 Agent 列表task()做状态隔离与结果回传Command把ToolMessage写回父 state。要把这套跑顺模型入口得先稳。API Keys 页面生成 Key接入文档里有各语言的完整示例想先验证模型行为模型对话页面直接试长期做编码或 Agent 编排Coding Plan 的额度规划更合适。Claude Code 场景看 ClaudeCodeAnthropic 接入说明控制台在 console 页面。最后留个实用技巧调试多 Agent 时先把subagents列表缩到一个确认单条task链路通了再加第二个观察并行行为。_EXCLUDED_STATE_KEYS的隔离效果你可以通过在子 Agent 的system_prompt里让它打印收到的 messages 长度来验证——正常情况下永远是 1。这个动作比读十遍源码都直观。