Deep Agents 实战:SubAgent 与 Async SubAgent 的配置与验证

发布时间:2026/10/7 21:49:15
Deep Agents 实战:SubAgent 与 Async SubAgent 的配置与验证 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_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 地址是 https://taotoken.net/api 注意 API 地址后面不加任何 UTM 参数直接用于代码里的 base_url。这里有个容易踩的坑很多人把官网地址当成 API 地址填进 base_url结果请求 404。记住分工——官网用来注册和拿 KeyAPI 地址才是代码里填的。前置准备清单第一注册并拿到 API Key在控制台的 API Keys 页面创建地址是 https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 。第二确认你要用的模型 ID。Deep Agents 里不同 SubAgent 可以用不同模型比如审查类用低 temperature 的推理模型撰写类用高 temperature 的生成模型。模型 ID 在模型对话页面可以试跑确认地址 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 。第三本地环境变量准备好。我习惯用.env管理不把 Key 写进代码。# .env TAOTOKEN_API_KEYsk-你的实际key TAOTOKEN_BASE_URLhttps://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( modelgpt-4o-mini, api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL), temperature0.2, ) resp llm.invoke(用一句话说明什么是子代理) print(resp.content)这段跑通说明 Key、Base URL、模型 ID 三件套正确。跑不通就先解决通道问题别往下走。三件套的对应关系是Base URL 填https://taotoken.net/apiKey 填控制台创建的凭证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( modelmodel_id, api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL), temperaturetemperature, timeout60, max_retries2, )注意base_url直接读环境变量值是https://taotoken.net/api。max_retries2是给 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。串行链路上的子代理用 SubAgentfrom langchain_deepagents import SubAgent cleaner SubAgent( name数据清洗员, llmbuild_llm(gpt-4o-mini, 0.1), tools[search_knowledge_base], system_prompt你负责清洗和验证原始数据只做清洗不做分析。, ) analyzer SubAgent( name文档分析员, llmbuild_llm(gpt-4o-mini, 0.2), tools[search_knowledge_base, summarize_text], system_prompt你负责搜索并总结技术文档输出结构化要点。, )并行分支上的子代理用 AsyncSubAgentfrom langchain_deepagents import AsyncSubAgent stat_agent AsyncSubAgent( name统计分析员, llmbuild_llm(gpt-4o-mini, 0.2), tools[summarize_text], system_prompt你负责对清洗后的数据做统计分析。, ) viz_agent AsyncSubAgent( name可视化专家, llmbuild_llm(gpt-4o-mini, 0.3), tools[render_markdown], system_prompt你负责把分析结果转成可视化描述和 Markdown 报告。, )最后组装 Supervisorfrom langchain_deepagents import DeepAgent main_agent DeepAgent( llmbuild_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: fBearer {os.getenv(TAOTOKEN_API_KEY)}}, json{model: gpt-4o-mini, messages: [{role: user, content: ping}]}, timeout30, ) 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_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里面有各语言的完整示例。API Key 管理在 https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 。模型试跑在 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 。最后留一个实用技巧给每个 SubAgent 的 system_prompt 末尾加一句完成后直接返回结果不要追问能显著减少 Supervisor 和 SubAgent 之间的无效往返。这个改动很小但在串行链路上效果立竿见影。