1. 从一次多智能体协作翻车说起
Deep Agents 是 LangChain 生态里专门用来做多智能体协作的框架,核心思路是让一个 Supervisor Agent 把复杂任务拆开,交给多个 SubAgent 分头处理。SubAgent 负责串行、有依赖的子任务,Async SubAgent 负责并行、互不依赖的子任务。这套机制适合谁?适合已经写过单 Agent、现在被"一个 Agent 干太多事导致 prompt 爆炸"折磨的开发者。
我最早接触 Deep Agents 是想做一个技术文档分析流水线:先搜索内部知识库,再总结,最后生成报告。一开始我把所有工具塞进一个 Agent,结果模型在"该搜索还是该总结"之间反复横跳,工具调用顺序完全失控。后来拆成 SubAgent 才理顺。但拆完之后又遇到新问题——多个 SubAgent 各自持有独立的模型客户端配置,Key 管理、Base URL、模型 ID 散落在各处,调试时根本不知道是哪个子代理报的错。
这篇文章就按真实落地顺序走一遍:先讲清楚 SubAgent 和 Async SubAgent 的职责边界,再解决统一接入通道的问题,然后给出可复制的配置片段,最后用实际请求验证结果,并把几个高频报错逐个拆掉。全程围绕 Deep Agents 多智能体协作这个场景,不跑题。
需要提前说明的是,SubAgent 的模型调用可以走任意兼容 OpenAI 协议的通道。我这边为了统一管理 Key 和排查日志,用的是 TaoToken 的 API 通道,后面配置片段里会体现。你完全可以替换成自己的通道,结构是一样的。
2. SubAgent 与 Async SubAgent 的职责边界与 TaoToken 接入前置
先把概念钉死,不然后面配置容易混。
SubAgent 本质是一个独立 Agent 实例,有自己的 system_prompt、工具集、模型参数。Supervisor 调用它时是串行等待的:发起 → 执行 → 返回结果 → Supervisor 拿到结果再决定下一步。它适合有依赖关系的任务链,比如"先清洗数据,再基于清洗结果做统计"。
Async SubAgent 是异步版本,Supervisor 可以同时发起多个,它们并行跑,全部完成后统一聚合结果。适合互不依赖的任务,比如"同时采集三个数据源"。
两者的关键差异不在 API 名字,而在调度语义:串行保证中间状态可见,并行追求吞吐。混合模式就是先串行做前置处理,再并行做分支计算。
接下来说接入前置。Deep Agents 里每个 SubAgent 都要实例化一个 LLM 客户端。如果每个 SubAgent 都写一遍 api_key、base_url、model,会有三个问题:Key 泄露面变大、模型切换要改多处、日志无法按通道聚合。我的做法是统一走一个兼容 OpenAI 协议的入口,把 base_url 指向 TaoToken 的 API 地址,Key 用同一个,模型 ID 按 SubAgent 职责分配。
TaoToken 在这里的角色就是统一 Key 与 API 通道:你拿到一个 Key,配置一个 Base URL,就能在多个 SubAgent 里复用,不用为每个子代理单独申请凭证。官网入口在 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,API 地址是 https://taotoken.net/api ,注意 API 地址后面不加任何 UTM 参数,直接用于代码里的 base_url。
这里有个容易踩的坑:很多人把官网地址当成 API 地址填进 base_url,结果请求 404。记住分工——官网用来注册和拿 Key,API 地址才是代码里填的。
前置准备清单:
第一,注册并拿到 API Key,在控制台的 API Keys 页面创建,地址是 https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite 。
第二,确认你要用的模型 ID。Deep Agents 里不同 SubAgent 可以用不同模型,比如审查类用低 temperature 的推理模型,撰写类用高 temperature 的生成模型。模型 ID 在模型对话页面可以试跑确认,地址 https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite 。
第三,本地环境变量准备好。我习惯用.env管理,不把 Key 写进代码。
# .env TAOTOKEN_API_KEY=sk-你的实际key TAOTOKEN_BASE_URL=https://taotoken.net/api第四,安装依赖。Deep Agents 依赖 LangChain 核心包,版本要对齐,否则 SubAgent 的导入路径会变。
pip install langchain langchain-openai langchain-deepagents python-dotenv装完之后先别急着写多智能体,先用一个最小脚本验证通道通不通。这一步能省掉后面 80% 的排查时间,因为如果通道本身有问题,SubAgent 报的错会非常迷惑。
import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() llm = ChatOpenAI( model="gpt-4o-mini", api_key=os.getenv("TAOTOKEN_API_KEY"), base_url=os.getenv("TAOTOKEN_BASE_URL"), temperature=0.2, ) resp = llm.invoke("用一句话说明什么是子代理") print(resp.content)这段跑通,说明 Key、Base URL、模型 ID 三件套正确。跑不通就先解决通道问题,别往下走。三件套的对应关系是:Base URL 填https://taotoken.net/api,Key 填控制台创建的凭证,Model ID 填你确认可用的模型名。这三者缺一不可,任何一个错都会导致 401 或 404。
3. 可复制的 SubAgent 与 Async SubAgent 配置
这一节给完整可复制的配置。我把它拆成三层:模型工厂、SubAgent 定义、Supervisor 编排。模型工厂的作用是让所有 SubAgent 共享同一个通道配置,只改模型 ID 和温度。
先写模型工厂:
import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() def build_llm(model_id: str, temperature: float = 0.2) -> ChatOpenAI: return ChatOpenAI( model=model_id, api_key=os.getenv("TAOTOKEN_API_KEY"), base_url=os.getenv("TAOTOKEN_BASE_URL"), temperature=temperature, timeout=60, max_retries=2, )注意base_url直接读环境变量,值是https://taotoken.net/api。max_retries=2是给 Async SubAgent 用的,并行时偶发限流能自动重试。
接着定义工具。SubAgent 的能力边界由工具决定,工具要职责单一:
from langchain_core.tools import tool @tool def search_knowledge_base(query: str) -> str: """搜索内部知识库,返回相关文档片段""" return f"命中与 '{query}' 相关的文档 3 篇,摘要如下:..." @tool def summarize_text(text: str) -> str: """对长文本做摘要""" return f"摘要:{text[:80]}..." @tool def lint_code(code: str) -> str: """静态检查代码质量问题""" return "发现 2 处命名不规范,1 处未处理异常" @tool def render_markdown(sections: str) -> str: """把结构化内容渲染成 Markdown""" return f"# 报告\n\n{sections}"然后定义 SubAgent。串行链路上的子代理用 SubAgent:
from langchain_deepagents import SubAgent cleaner = SubAgent( name="数据清洗员", llm=build_llm("gpt-4o-mini", 0.1), tools=[search_knowledge_base], system_prompt="你负责清洗和验证原始数据,只做清洗,不做分析。", ) analyzer = SubAgent( name="文档分析员", llm=build_llm("gpt-4o-mini", 0.2), tools=[search_knowledge_base, summarize_text], system_prompt="你负责搜索并总结技术文档,输出结构化要点。", )并行分支上的子代理用 AsyncSubAgent:
from langchain_deepagents import AsyncSubAgent stat_agent = AsyncSubAgent( name="统计分析员", llm=build_llm("gpt-4o-mini", 0.2), tools=[summarize_text], system_prompt="你负责对清洗后的数据做统计分析。", ) viz_agent = AsyncSubAgent( name="可视化专家", llm=build_llm("gpt-4o-mini", 0.3), tools=[render_markdown], system_prompt="你负责把分析结果转成可视化描述和 Markdown 报告。", )最后组装 Supervisor:
from langchain_deepagents import DeepAgent main_agent = DeepAgent( llm=build_llm("gpt-4o-mini", 0.2), sub_agents=[cleaner, analyzer], async_sub_agents=[stat_agent, viz_agent], )如果你更习惯用配置文件管理,可以把通道参数抽成 JSON,避免硬编码:
{ "provider": "openai-compatible", "base_url": "https://taotoken.net/api", "api_key_env": "TAOTOKEN_API_KEY", "default_model": "gpt-4o-mini", "sub_agents": { "cleaner": { "model": "gpt-4o-mini", "temperature": 0.1 }, "analyzer": { "model": "gpt-4o-mini", "temperature": 0.2 }, "stat_agent": { "model": "gpt-4o-mini", "temperature": 0.2 }, "viz_agent": { "model": "gpt-4o-mini", "temperature": 0.3 } } }读取时用json.load把base_url和api_key_env注入模型工厂即可。这样切换通道只改一个文件,SubAgent 定义完全不动。
这里强调一个配置原则:SubAgent 的 system_prompt 要写"只做什么",不要写"顺便做什么"。职责越窄,Supervisor 调度越准。我见过把搜索、总结、写报告塞进一个 SubAgent 的写法,结果它在该返回结果的时候又去调工具,链路直接卡死。
4. 发起请求验证 SubAgent 与 Async SubAgent 执行结果
配置写完必须验证,而且要分层验证:先验证单个 SubAgent 能跑,再验证 Supervisor 能调度,最后验证 Async 并行确实生效。
第一步,单独跑一个 SubAgent 的底层模型调用,确认通道和工具绑定没问题:
result = analyzer.invoke("请分析最新的 API 文档变更") print(result)如果这一步报错,问题在 SubAgent 自身或通道,跟 Supervisor 无关。
第二步,跑 Supervisor 的串行调度:
result = main_agent.invoke("请先清洗这批数据,再分析其中的技术文档") print(result)观察输出里是否出现了"数据清洗员"先执行、"文档分析员"后执行的痕迹。Deep Agents 的调度日志会体现调用顺序。
第三步,验证 Async SubAgent 并行。注意这里要用异步入口:
import asyncio async def run_parallel(): result = await main_agent.ainvoke( "请对清洗后的数据同时做统计分析和可视化" ) print(result) asyncio.run(run_parallel())判断并行是否真的生效,看两个信号:一是总耗时是否明显小于两个子任务串行之和;二是日志里两个 Async SubAgent 的启动时间戳是否接近。如果启动时间戳一前一后差很多,说明调度没并行,通常是误用了同步入口invoke而不是ainvoke。
我实测下来,两个 Async SubAgent 各耗时约 3 秒,并行总耗时在 3.5 秒左右,串行则接近 6 秒。这个差距在子任务变多时会放大。
验证通过后,建议把每次调用的模型 ID、SubAgent 名称、耗时打到日志里,方便后续排查。一个简单的装饰器就够:
import time, logging def log_subagent(name): def deco(fn): def wrapper(*args, **kwargs): start = time.time() out = fn(*args, **kwargs) logging.info(f"[{name}] cost={time.time()-start:.2f}s") return out return wrapper return deco日志里能看到每个 SubAgent 的实际耗时,哪个是瓶颈一目了然。如果某个 Async SubAgent 耗时异常长,先查它的模型 ID 是否可用,再查工具里是否有阻塞式 IO。
5. 高频报错排查:401、local proxy failed、reading choices、OAuth
这一节按真实报错逐个拆。这些错我在接入 Deep Agents 多智能体时基本都遇到过。
401 Unauthorized。最常见,原因是 Key 没读到或读错。检查三点:.env是否被load_dotenv()正确加载;环境变量名是否和代码里os.getenv一致;Key 是否有多余空格或换行。如果 Key 是从控制台复制的,注意别把前后空白带进去。还有一种情况是 Key 被禁用或额度耗尽,去控制台确认状态。
local proxy failed / connection refused。这个报错通常和本地网络环境有关。先确认base_url填的是https://taotoken.net/api而不是官网地址。再确认本机没有残留的代理环境变量干扰,检查HTTP_PROXY、HTTPS_PROXY是否被设置成了不可用的地址。如果公司网络有出口限制,联系网络管理员放行对应域名。不要试图用任何非正规网络工具绕过,合规问题自己承担。
Error reading choices / KeyError 'choices'。这个错说明请求发出去了,但返回体结构不符合预期。常见原因有两个:一是base_url少了/api后缀,请求打到了错误路径,返回的是 HTML 而不是 JSON;二是模型 ID 写错,服务端返回了错误对象,代码却按正常响应解析choices。排查方法:把原始响应打印出来看。
import httpx, os resp = httpx.post( f"{os.getenv('TAOTOKEN_BASE_URL')}/chat/completions", headers={"Authorization": f"Bearer {os.getenv('TAOTOKEN_API_KEY')}"}, json={"model": "gpt-4o-mini", "messages": [{"role": "user", "content": "ping"}]}, timeout=30, ) print(resp.status_code) print(resp.text[:500])如果status_code是 200 但text里没有choices,就是模型 ID 问题;如果是 404,就是路径问题。
OAuth / authentication 相关报错。如果你用的是需要 OAuth 的客户端(比如某些 CLI 工具),报错往往出在 token 过期或 scope 不足。Deep Agents 本身走的是 API Key 模式,不涉及 OAuth。但如果你在周边工具里混用了 OAuth 流程,要单独确认 token 有效期。Codex 类工具如果用auth.json管理凭证,要确保里面的base_url和api_key与 Deep Agents 用的通道一致,否则会出现"CLI 能跑、SubAgent 报 401"的割裂现象。
排查顺序建议固定下来:先跑第 2 节的最小脚本确认三件套,再跑第 4 节的单 SubAgent,最后跑 Supervisor。逐层缩小范围,比一上来就调多智能体高效得多。
6. 把通道固定下来,再谈多智能体扩展
多智能体协作的复杂度不在 SubAgent 数量,而在配置一致性。SubAgent 越多,Key、Base URL、模型 ID 越容易散落。我的做法是:所有 SubAgent 共用模型工厂,工厂只读环境变量,环境变量只指向一个通道。这样新增一个 SubAgent 只需要写它的 prompt 和工具,接入层零改动。
如果你要长期跑编码类或 Agent 类任务,可以考虑用 Coding Plan 把额度固定下来,地址 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite 。接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite ,里面有各语言的完整示例。API Key 管理在 https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite 。模型试跑在 https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite 。
最后留一个实用技巧:给每个 SubAgent 的 system_prompt 末尾加一句"完成后直接返回结果,不要追问",能显著减少 Supervisor 和 SubAgent 之间的无效往返。这个改动很小,但在串行链路上效果立竿见影。