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工具委托的执行单元，独立运行、完成具体任务后返回结果。SubAgent（TypedDict）是声明式配置描述字典，框架自动编译为可运行对象；CompiledSubAgent（TypedDict）是你自己编译好的可运行对象封装，框架直接使用。_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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_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( model=os.getenv("TAOTOKEN_MODEL", "gpt-4o-mini"), api_key=os.getenv("TAOTOKEN_API_KEY"), base_url="https://taotoken.net/api", temperature=0, ) resp = llm.invoke("用一句话说明什么是多 Agent 协作") print(resp.content).env文件内容：
TAOTOKEN_API_KEY=sk-你的key TAOTOKEN_MODEL=gpt-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依赖清单：
langchain>=0.3.0 langchain-openai>=0.2.0 langgraph>=0.2.0 deepagents>=0.1.0 python-dotenv>=1.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( model=os.getenv("TAOTOKEN_MODEL", "gpt-4o-mini"), api_key=os.getenv("TAOTOKEN_API_KEY"), base_url="https://taotoken.net/api", temperature=0, ) agent = create_deep_agent( model=model, 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，就得自己先编译一个 Runnable，state 必须含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/api，Key 填控制台生成的，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，再遍历用户subagents，SubAgent走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。这个动作比读十遍源码都直观。
