尧图网页设计 
1. 为什么同步 Subagents 会把主代理拖死AsyncSubAgent 要解决的真实问题如果你用 LangChain 的 Deep Agents 搭过多代理系统大概率踩过这个坑主代理Supervisor把任务派给子代理之后整个流程就卡住了。子代理在那边跑搜索、跑代码生成、跑数据清洗主代理只能干等用户发来的新消息也进不来。等子代理终于返回主代理才继续往下走。任务一多体验直接崩。这就是同步子代理的硬伤——阻塞。它适合那种「必须等结果才能继续」的场景比如先查天气再决定穿什么。但现实里大量任务是长耗时的一份行业调研要跑十几轮搜索一段数据分析代码要反复调试。让主代理全程阻塞等于把整个对话系统变成单线程。LangChain Deep Agents 里的AsyncSubAgent就是冲着这个问题来的。它让主代理启动后台任务后立即拿到一个 task_id 并返回控制权子代理在自己的线程上并发执行。主代理可以继续和用户聊天随时用check_async_task查进度、用update_async_task追加指令、用cancel_async_task取消任务。任务状态存在主代理图的专用状态通道async_tasks里和消息历史分开所以哪怕上下文窗口被压缩任务 ID 也不会丢。这篇文章面向三类人正在用create_deep_agent搭多代理协作的开发者、被同步阻塞坑过的 LangChain 用户、以及想把异步调度做成可观测系统的人。我会从配置片段讲到并发验证再到超时和失败重试的排查全程给可复制的代码。模型调用这一层我用 TaoToken 统一 Key 和 API 通道省得在多个 provider 之间来回切环境变量。需要先说明一点异步子代理在 deepagents 0.5.0 里还是预览功能API 可能变。所以下面的配置我会标注版本你升级后如果报错先回来看字段有没有改。先看同步和异步的核心差异这张表决定了你该选哪种模式维度同步子代理异步子代理执行模型主代理阻塞直到子代理完成立即返回 task_id主代理继续并发性并行但阻塞并行且非阻塞任务中更新不可能通过update_async_task发送后续指令取消不可能通过cancel_async_task取消运行中任务状态性无状态调用间无持久状态有状态在自己的线程上维护跨交互状态适用场景代理应等结果再继续的任务聊天中交互式管理的长耗时复杂任务看懂这张表你就明白为什么异步模式更适合「可观测的调度」——因为任务有独立线程、有状态、有 ID你才能追踪它、干预它、聚合它。同步模式下任务跑完就没了你连中间状态都看不到。2. 用 TaoToken 统一模型通道AsyncSubAgent 接入前的环境准备在写create_deep_agent之前先把模型调用这层理顺。Deep Agents 支持多种模型 providermodel字段可以写google_genai:gemini-3.1-pro-preview、openai:gpt-4o这类格式。但如果你同时用多个 providerKey 管理、Base URL 切换、额度监控会变成一堆散落的.env变量调试时很难定位问题。我的做法是用 TaoToken 做统一入口。它是一个兼容 OpenAI 接口规范的模型调用通道把不同模型的 Key 和 API 地址收敛成一套配置。对 Deep Agents 来说你只需要把 Base URL 指向 TaoToken 的 API 地址Key 用 TaoToken 生成的令牌模型 ID 按它的命名规则填就行。具体操作路径是这样的先到官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册账号然后在控制台 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 里创建 API Key。创建完去 API Keys 页面 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 复制令牌注意它只显示一次复制后存到安全的地方。拿到 Key 之后环境变量这样配。我习惯用.env文件加python-dotenv避免把 Key 硬编码进代码# .env TAOTOKEN_API_KEYsk-你的taotoken令牌 TAOTOKEN_BASE_URLhttps://taotoken.net/api然后在 Python 里加载。Deep Agents 底层走的是 LangChain 的模型接口所以你可以用ChatOpenAI指向 TaoToken 的 Base URL再传给create_deep_agentimport os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() llm ChatOpenAI( modelgemini-3.1-pro-preview, # 按 TaoToken 支持的模型 ID 填 api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.environ[TAOTOKEN_BASE_URL], temperature0.2, )这里有个容易踩的坑base_url末尾不要带/v1还是不带取决于 TaoToken 的接口约定。我实测下来https://taotoken.net/api这个地址直接可用SDK 会自动补全路径。如果你填成https://taotoken.net/api/v1反而可能 404。拿不准的时候先用模型对话页面 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite 发一条消息验证通道通不通再去写代码。为什么要在 AsyncSubAgent 场景下强调统一通道因为异步子代理会并发发起多个模型调用。如果每个子代理用不同 provider 的 Key一旦某个 Key 额度耗尽或限流你很难判断是哪个子代理挂了。统一到 TaoToken 之后所有调用走同一个通道日志里的请求 ID 能直接对应到具体任务排查效率高很多。另外异步任务的生命周期可能很长跨几十分钟甚至几小时。TaoToken 的 Key 如果中途失效正在跑的子代理会直接报错。所以建议在启动长任务前先用一个轻量请求探活def health_check(llm): try: resp llm.invoke(ping) return True except Exception as e: print(f通道探活失败: {e}) return False if not health_check(llm): raise RuntimeError(模型通道不可用检查 TaoToken Key 和 Base URL)这一步花两秒能省掉后面半小时的无效调试。环境准备好之后就可以进入create_deep_agent的配置了。3. 可复制的 create_deep_agent 配置AsyncSubAgent 注册与 langgraph.json 对齐这一节是全文的核心我给你一份能直接跑的配置。先定义异步子代理列表每个AsyncSubAgent指向一个 Agent Protocol 服务器上的图。关键字段有三个name是主代理派活时用的标识description决定主代理把任务分给谁graph_id必须和langgraph.json里注册的图名一致。from deepagents import AsyncSubAgent, create_deep_agent async_subagents [ AsyncSubAgent( nameresearcher, descriptionConducts in-depth research using web search. Use for questions requiring multiple searches and synthesis., graph_idresearcher, # 无 url → ASGI 传输同部署协同 ), AsyncSubAgent( namecoder, descriptionGenerates and reviews Python code for data analysis and visualization tasks., graph_idcoder, # urlhttps://coder-deployment.example.com # 有 url → HTTP 传输远程 ), ] agent create_deep_agent( modelllm, # 上一节配好的 TaoToken 通道 system_promptYou are a supervisor agent coordinating research and coding tasks. After launching an async subagent, ALWAYS return control to the user. Never call check_async_task immediately after launch. Always show the full task_id, never truncate or abbreviate it., subagentsasync_subagents, )字段说明我拆开讲。name必填唯一标识符主代理启动任务时用它。description必填主代理靠这段描述决定委派给哪个代理所以要写得具体、以行动为导向。对比一下好坏# 好具体、说明何时使用 AsyncSubAgent( nameresearcher, descriptionConducts in-depth research using web search. Use for questions requiring multiple searches and synthesis., graph_idresearcher, ) # 差模糊主代理无法判断 AsyncSubAgent( namehelper, descriptionhelps with stuff, graph_idhelper, )graph_id必填对应 Agent Protocol 服务器上的图 ID。如果你用 LangGraph 部署这个值必须和langgraph.json里注册的图名匹配。url可选省略时用 ASGI 传输进程内调用设置时用 HTTP 传输到远程服务器。headers可选给远程服务器加自定义认证头。langgraph.json的配置长这样所有图注册在同一个文件里{ graphs: { supervisor: ./src/supervisor.py:graph, researcher: ./src/researcher.py:graph, coder: ./src/coder.py:graph }, env: .env }注意supervisor是主代理的图researcher和coder是子代理的图。三个图在同一个langgraph.json里注册ASGI 传输才能通过进程内函数调用找到它们不需要走 HTTP 路由也就没有网络延迟和额外认证。传输方式的选择直接影响部署拓扑。我整理成对照表传输方式触发条件延迟认证适用场景ASGI省略url进程内无网络延迟无需额外配置协同部署推荐默认HTTP设置url走网络LangGraph SDK 用LANGSMITH_API_KEY子代理独立扩展、不同团队维护混合部署也支持一部分子代理 ASGI 协同一部分 HTTP 远程async_subagents [ AsyncSubAgent( nameresearcher, descriptionResearch agent for information gathering., graph_idresearcher, # 无 url → ASGI ), AsyncSubAgent( namecoder, descriptionCoding agent for code generation., graph_idcoder, urlhttps://coder-deployment.example.com, # 有 url → HTTP ), ]配置写完之后主代理的 LLM 会通过AsyncSubAgentMiddleware拿到五个工具start_async_task、check_async_task、update_async_task、cancel_async_task、list_async_tasks。中间件自动处理线程创建、运行管理和状态持久化你不需要手动管线程。这里有个部署时的关键点本地用langgraph dev跑的时候工作池大小要调够。每个活动运行占一个工作槽主代理加 3 个并发子代理需要 4 个槽。默认池子不够会导致子代理启动排队看起来像卡住了。启动命令加参数langgraph dev --n-jobs-per-worker 10配置到这一步AsyncSubAgent 的骨架就搭好了。下一节我们验证它是不是真的并发执行。4. 验证并发执行、超时与结果聚合日志与回调的实操动作配置写完不代表异步真的生效了。我见过不少人以为配了AsyncSubAgent就是异步结果主代理还是在启动后立刻轮询check_async_task把异步硬生生用成了阻塞。所以这一节专门讲怎么验证。先看生命周期。一次典型的异步交互是这样的主代理调用start_async_task服务器创建新线程以任务描述为输入启动运行返回线程 ID 作为 task_id。主代理向用户报告这个 ID不轮询完成状态。用户过一会儿说「查一下进度」主代理才调check_async_task获取当前运行状态。如果运行成功检索线程状态提取子代理最终输出如果还在跑就报告进行中。验证并发最直接的办法是看时间戳。启动两个子代理记录各自的created_at如果它们的时间差在毫秒级说明是并发启动的import time from datetime import datetime # 启动两个任务记录时间 t1 datetime.now() task_a agent.invoke({input: research AI agent trends}) t2 datetime.now() task_b agent.invoke({input: code a data visualization script}) t3 datetime.now() print(f任务A启动耗时: {(t2 - t1).total_seconds():.3f}s) print(f任务B启动耗时: {(t3 - t2).total_seconds():.3f}s) # 如果两个都在 1s 内返回说明是非阻塞启动如果start_async_task返回很快通常几百毫秒而子代理实际执行要几十秒那就对了。反过来如果启动调用本身卡了十几秒说明它退化成同步了。再看状态回传。任务元数据存在主代理图的async_tasks状态通道里每条记录包含 task_id、代理名、线程 ID、运行 ID、状态和时间戳created_at、last_checked_at、last_updated_at。你可以直接读这个通道来验证# 假设 agent 已经跑过几轮 state agent.get_state() async_tasks state.values.get(async_tasks, {}) for task_id, meta in async_tasks.items(): print(f任务 {task_id}: 代理{meta[agent_name]}, 状态{meta[status]}, f创建于{meta[created_at]}, 最后检查{meta[last_checked_at]})这个通道和消息历史是分开的这点很关键。Deep Agents 在上下文窗口填满时会压缩消息历史如果 task_id 只存在工具消息里压缩后就丢了。专用通道保证主代理始终能通过list_async_tasks回忆任务。超时验证要主动构造。给子代理设一个短超时看它是否正确报错而不是无限挂起# 在子代理图里配置超时示意具体参数看你的 LangGraph 版本 config {configurable: {timeout: 30}} # 30 秒超时 result agent.invoke({input: research something slow}, configconfig)超时后任务状态应该变成error而不是一直停在running。如果它一直 running说明超时没生效检查子代理图的运行配置。结果聚合是异步调度的最后一环。主代理拿到多个子代理的结果后需要合并成一份输出。用list_async_tasks遍历所有任务对非终态任务并行拉取实时状态终态任务success、error、cancelled从缓存返回def aggregate_results(agent): state agent.get_state() tasks state.values.get(async_tasks, {}) results {} for task_id, meta in tasks.items(): if meta[status] success: results[meta[agent_name]] meta.get(result, ) elif meta[status] error: results[meta[agent_name]] f[失败] {meta.get(error, 未知错误)} else: results[meta[agent_name]] f[进行中] {meta[status]} return results可观测性这块LangSmith 是标配。每个异步子代理运行都是标准的 LangGraph 运行在 LangSmith 里完全可见。主代理的追踪显示 launch、check、update、cancel、list 的工具调用每个子代理运行作为单独追踪出现通过线程 ID 链接。用线程 ID就是 task_id关联主代理编排追踪和子代理执行追踪一眼就能看出哪个子代理慢、哪个失败了。如果你不想依赖 LangSmith也可以在 TaoToken 的调用日志里看请求分布。因为所有模型调用走统一通道每个子代理的请求都会带上时间戳和模型 ID并发时能看到请求交错出现而不是串行排队。这算是统一通道带来的额外观测点。验证通过的标准很简单启动快、状态可查、超时能报错、结果能聚合。四条都满足异步调度就算跑通了。5. 常见报错排查401、local proxy failed、reading choices、OAuth 对照异步子代理跑起来之后报错会集中在几个地方。我把真实遇到过的错误和排查路径列出来你对照着看。401 Unauthorized。这个最常见八成是 TaoToken Key 或 Base URL 的问题。先确认.env里的TAOTOKEN_API_KEY没有多余空格TAOTOKEN_BASE_URL是https://taotoken.net/api而不是带/v1的变体。如果 Key 是对的还报 401去 API Keys 页面 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 确认令牌没过期、没被删。还有一种情况是子代理用了 HTTP 传输到远程服务器远程服务器的认证头没配这时候检查AsyncSubAgent的headers字段。local proxy failed。这个错误通常出现在网络层不是模型层。如果你在本地跑langgraph devASGI 传输是进程内调用理论上不该有代理问题。出现这个报错先检查是不是环境里设了HTTP_PROXY或HTTPS_PROXY变量把它们清掉再试。另外确认langgraph.json里所有图都注册了缺图会导致 SDK 找不到目标报出类似代理失败的错。reading choices 相关报错。这类错误一般是模型返回格式不符合预期。Deep Agents 期望模型返回结构化的工具调用如果模型输出被截断或格式错乱解析就会失败。排查方向一是确认model字段填的模型 ID 在 TaoToken 通道里可用去模型对话页面 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite 试一下同一个模型二是检查temperature是不是太高异步任务建议 0.2 以下三是看是不是上下文太长导致输出被截断适当精简 system_prompt。OAuth 相关报错。如果你用 HTTP 传输到远程 LangGraph 部署认证走的是LANGSMITH_API_KEY或LANGGRAPH_API_KEY环境变量。报 OAuth 错误通常是这两个变量没设或设错。自托管 Agent Protocol 服务器可能用不同的认证机制这时候要在AsyncSubAgent的headers里手动加认证头AsyncSubAgent( namecoder, descriptionCoding agent., graph_idcoder, urlhttps://coder-deployment.example.com, headers{Authorization: Bearer 你的远程服务令牌}, )除了这些还有几个异步特有的坑。主代理启动后立即轮询表现是启动任务后马上循环调check把异步变成阻塞。中间件会注入系统提示规则来防止如果还发生在system_prompt里强化「After launching an async subagent, ALWAYS return control to the user. Never call check_async_task immediately after launch.」主代理报告过时状态它引用对话历史里早期的任务状态而不是重新 check。解决办法是在系统提示里加「对话历史中的任务状态总是过时的」并要求报告前必须调check或list。任务 ID 查找失败主代理截断或重新格式化 task_id导致 check 或 cancel 失败。这通常是模型特定问题在系统提示里加「始终显示完整 task_id绝不截断或缩写」或者换个模型试试。子代理启动排队启动挂起或很久才开始。这是工作池耗尽用langgraph dev --n-jobs-per-worker 10加大池子。记住主代理加 N 个并发子代理需要 N1 个槽。排查的时候有个通用技巧先看 TaoToken 的调用日志确认请求有没有发出去。如果日志里没有对应请求问题在客户端配置如果有请求但报错问题在模型或参数。这一步能快速定位是网络层还是模型层。6. 把异步调度用起来从单机验证到长期 Coding Plan到这里AsyncSubAgent 的完整链路你应该能跑通了TaoToken 统一通道 →create_deep_agent配置 → 并发验证 → 报错排查。最后说几个实战里的经验。异步子代理最适合的场景是「长耗时 需要中途干预」。比如一份深度调研子代理跑十分钟用户中途想加个方向用update_async_task发新指令之前的运行被中断子代理带着完整对话历史加新指令重启task_id 不变。这种交互同步模式根本做不到。如果你要把这套东西用在长期编码或 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里面有各语言的完整示例。如果你用 Claude Code 这类工具Anthropic 兼容接入的说明在 https://taotoken.net/claudecode-anthropic?utm_sourcetaotoken_aicg_blog_endutm_contentclaudecode-anthropicutm_campaignrewrite。一个我踩过的坑异步任务的状态通道async_tasks会随着任务增多而膨胀。如果你跑几百个任务不清理主代理的状态会越来越大。建议在任务进入终态success、error、cancelled后定期归档或清理只保留最近 N 条。清理逻辑可以挂在主代理的收尾节点上。还有description字段值得反复打磨。主代理选子代理全靠它写得越具体委派越准。我一般会写清楚「什么时候用」和「能做什么」比如「Use for questions requiring multiple searches and synthesis」比「research agent」有用得多。最后异步不等于不管。启动任务后返回控制权是对的但用户问进度时你得真的去 check不能拿历史状态糊弄。中间件提示里那句「对话历史中的任务状态总是过时的」要刻进脑子里。做到这一点你的多代理系统才算真正可观测、可干预、可聚合。 Deep Agents 异步 Subagents 实战：用 create_deep_agent 构建可观测的 AsyncSubAgent 调度
1. 为什么同步 Subagents 会把主代理拖死AsyncSubAgent 要解决的真实问题如果你用 LangChain 的 Deep Agents 搭过多代理系统大概率踩过这个坑主代理Supervisor把任务派给子代理之后整个流程就卡住了。子代理在那边跑搜索、跑代码生成、跑数据清洗主代理只能干等用户发来的新消息也进不来。等子代理终于返回主代理才继续往下走。任务一多体验直接崩。这就是同步子代理的硬伤——阻塞。它适合那种「必须等结果才能继续」的场景比如先查天气再决定穿什么。但现实里大量任务是长耗时的一份行业调研要跑十几轮搜索一段数据分析代码要反复调试。让主代理全程阻塞等于把整个对话系统变成单线程。LangChain Deep Agents 里的AsyncSubAgent就是冲着这个问题来的。它让主代理启动后台任务后立即拿到一个 task_id 并返回控制权子代理在自己的线程上并发执行。主代理可以继续和用户聊天随时用check_async_task查进度、用update_async_task追加指令、用cancel_async_task取消任务。任务状态存在主代理图的专用状态通道async_tasks里和消息历史分开所以哪怕上下文窗口被压缩任务 ID 也不会丢。这篇文章面向三类人正在用create_deep_agent搭多代理协作的开发者、被同步阻塞坑过的 LangChain 用户、以及想把异步调度做成可观测系统的人。我会从配置片段讲到并发验证再到超时和失败重试的排查全程给可复制的代码。模型调用这一层我用 TaoToken 统一 Key 和 API 通道省得在多个 provider 之间来回切环境变量。需要先说明一点异步子代理在 deepagents 0.5.0 里还是预览功能API 可能变。所以下面的配置我会标注版本你升级后如果报错先回来看字段有没有改。先看同步和异步的核心差异这张表决定了你该选哪种模式维度同步子代理异步子代理执行模型主代理阻塞直到子代理完成立即返回 task_id主代理继续并发性并行但阻塞并行且非阻塞任务中更新不可能通过update_async_task发送后续指令取消不可能通过cancel_async_task取消运行中任务状态性无状态调用间无持久状态有状态在自己的线程上维护跨交互状态适用场景代理应等结果再继续的任务聊天中交互式管理的长耗时复杂任务看懂这张表你就明白为什么异步模式更适合「可观测的调度」——因为任务有独立线程、有状态、有 ID你才能追踪它、干预它、聚合它。同步模式下任务跑完就没了你连中间状态都看不到。2. 用 TaoToken 统一模型通道AsyncSubAgent 接入前的环境准备在写create_deep_agent之前先把模型调用这层理顺。Deep Agents 支持多种模型 providermodel字段可以写google_genai:gemini-3.1-pro-preview、openai:gpt-4o这类格式。但如果你同时用多个 providerKey 管理、Base URL 切换、额度监控会变成一堆散落的.env变量调试时很难定位问题。我的做法是用 TaoToken 做统一入口。它是一个兼容 OpenAI 接口规范的模型调用通道把不同模型的 Key 和 API 地址收敛成一套配置。对 Deep Agents 来说你只需要把 Base URL 指向 TaoToken 的 API 地址Key 用 TaoToken 生成的令牌模型 ID 按它的命名规则填就行。具体操作路径是这样的先到官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册账号然后在控制台 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 里创建 API Key。创建完去 API Keys 页面 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 复制令牌注意它只显示一次复制后存到安全的地方。拿到 Key 之后环境变量这样配。我习惯用.env文件加python-dotenv避免把 Key 硬编码进代码# .env TAOTOKEN_API_KEYsk-你的taotoken令牌 TAOTOKEN_BASE_URLhttps://taotoken.net/api然后在 Python 里加载。Deep Agents 底层走的是 LangChain 的模型接口所以你可以用ChatOpenAI指向 TaoToken 的 Base URL再传给create_deep_agentimport os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() llm ChatOpenAI( modelgemini-3.1-pro-preview, # 按 TaoToken 支持的模型 ID 填 api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.environ[TAOTOKEN_BASE_URL], temperature0.2, )这里有个容易踩的坑base_url末尾不要带/v1还是不带取决于 TaoToken 的接口约定。我实测下来https://taotoken.net/api这个地址直接可用SDK 会自动补全路径。如果你填成https://taotoken.net/api/v1反而可能 404。拿不准的时候先用模型对话页面 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite 发一条消息验证通道通不通再去写代码。为什么要在 AsyncSubAgent 场景下强调统一通道因为异步子代理会并发发起多个模型调用。如果每个子代理用不同 provider 的 Key一旦某个 Key 额度耗尽或限流你很难判断是哪个子代理挂了。统一到 TaoToken 之后所有调用走同一个通道日志里的请求 ID 能直接对应到具体任务排查效率高很多。另外异步任务的生命周期可能很长跨几十分钟甚至几小时。TaoToken 的 Key 如果中途失效正在跑的子代理会直接报错。所以建议在启动长任务前先用一个轻量请求探活def health_check(llm): try: resp llm.invoke(ping) return True except Exception as e: print(f通道探活失败: {e}) return False if not health_check(llm): raise RuntimeError(模型通道不可用检查 TaoToken Key 和 Base URL)这一步花两秒能省掉后面半小时的无效调试。环境准备好之后就可以进入create_deep_agent的配置了。3. 可复制的 create_deep_agent 配置AsyncSubAgent 注册与 langgraph.json 对齐这一节是全文的核心我给你一份能直接跑的配置。先定义异步子代理列表每个AsyncSubAgent指向一个 Agent Protocol 服务器上的图。关键字段有三个name是主代理派活时用的标识description决定主代理把任务分给谁graph_id必须和langgraph.json里注册的图名一致。from deepagents import AsyncSubAgent, create_deep_agent async_subagents [ AsyncSubAgent( nameresearcher, descriptionConducts in-depth research using web search. Use for questions requiring multiple searches and synthesis., graph_idresearcher, # 无 url → ASGI 传输同部署协同 ), AsyncSubAgent( namecoder, descriptionGenerates and reviews Python code for data analysis and visualization tasks., graph_idcoder, # urlhttps://coder-deployment.example.com # 有 url → HTTP 传输远程 ), ] agent create_deep_agent( modelllm, # 上一节配好的 TaoToken 通道 system_promptYou are a supervisor agent coordinating research and coding tasks. After launching an async subagent, ALWAYS return control to the user. Never call check_async_task immediately after launch. Always show the full task_id, never truncate or abbreviate it., subagentsasync_subagents, )字段说明我拆开讲。name必填唯一标识符主代理启动任务时用它。description必填主代理靠这段描述决定委派给哪个代理所以要写得具体、以行动为导向。对比一下好坏# 好具体、说明何时使用 AsyncSubAgent( nameresearcher, descriptionConducts in-depth research using web search. Use for questions requiring multiple searches and synthesis., graph_idresearcher, ) # 差模糊主代理无法判断 AsyncSubAgent( namehelper, descriptionhelps with stuff, graph_idhelper, )graph_id必填对应 Agent Protocol 服务器上的图 ID。如果你用 LangGraph 部署这个值必须和langgraph.json里注册的图名匹配。url可选省略时用 ASGI 传输进程内调用设置时用 HTTP 传输到远程服务器。headers可选给远程服务器加自定义认证头。langgraph.json的配置长这样所有图注册在同一个文件里{ graphs: { supervisor: ./src/supervisor.py:graph, researcher: ./src/researcher.py:graph, coder: ./src/coder.py:graph }, env: .env }注意supervisor是主代理的图researcher和coder是子代理的图。三个图在同一个langgraph.json里注册ASGI 传输才能通过进程内函数调用找到它们不需要走 HTTP 路由也就没有网络延迟和额外认证。传输方式的选择直接影响部署拓扑。我整理成对照表传输方式触发条件延迟认证适用场景ASGI省略url进程内无网络延迟无需额外配置协同部署推荐默认HTTP设置url走网络LangGraph SDK 用LANGSMITH_API_KEY子代理独立扩展、不同团队维护混合部署也支持一部分子代理 ASGI 协同一部分 HTTP 远程async_subagents [ AsyncSubAgent( nameresearcher, descriptionResearch agent for information gathering., graph_idresearcher, # 无 url → ASGI ), AsyncSubAgent( namecoder, descriptionCoding agent for code generation., graph_idcoder, urlhttps://coder-deployment.example.com, # 有 url → HTTP ), ]配置写完之后主代理的 LLM 会通过AsyncSubAgentMiddleware拿到五个工具start_async_task、check_async_task、update_async_task、cancel_async_task、list_async_tasks。中间件自动处理线程创建、运行管理和状态持久化你不需要手动管线程。这里有个部署时的关键点本地用langgraph dev跑的时候工作池大小要调够。每个活动运行占一个工作槽主代理加 3 个并发子代理需要 4 个槽。默认池子不够会导致子代理启动排队看起来像卡住了。启动命令加参数langgraph dev --n-jobs-per-worker 10配置到这一步AsyncSubAgent 的骨架就搭好了。下一节我们验证它是不是真的并发执行。4. 验证并发执行、超时与结果聚合日志与回调的实操动作配置写完不代表异步真的生效了。我见过不少人以为配了AsyncSubAgent就是异步结果主代理还是在启动后立刻轮询check_async_task把异步硬生生用成了阻塞。所以这一节专门讲怎么验证。先看生命周期。一次典型的异步交互是这样的主代理调用start_async_task服务器创建新线程以任务描述为输入启动运行返回线程 ID 作为 task_id。主代理向用户报告这个 ID不轮询完成状态。用户过一会儿说「查一下进度」主代理才调check_async_task获取当前运行状态。如果运行成功检索线程状态提取子代理最终输出如果还在跑就报告进行中。验证并发最直接的办法是看时间戳。启动两个子代理记录各自的created_at如果它们的时间差在毫秒级说明是并发启动的import time from datetime import datetime # 启动两个任务记录时间 t1 datetime.now() task_a agent.invoke({input: research AI agent trends}) t2 datetime.now() task_b agent.invoke({input: code a data visualization script}) t3 datetime.now() print(f任务A启动耗时: {(t2 - t1).total_seconds():.3f}s) print(f任务B启动耗时: {(t3 - t2).total_seconds():.3f}s) # 如果两个都在 1s 内返回说明是非阻塞启动如果start_async_task返回很快通常几百毫秒而子代理实际执行要几十秒那就对了。反过来如果启动调用本身卡了十几秒说明它退化成同步了。再看状态回传。任务元数据存在主代理图的async_tasks状态通道里每条记录包含 task_id、代理名、线程 ID、运行 ID、状态和时间戳created_at、last_checked_at、last_updated_at。你可以直接读这个通道来验证# 假设 agent 已经跑过几轮 state agent.get_state() async_tasks state.values.get(async_tasks, {}) for task_id, meta in async_tasks.items(): print(f任务 {task_id}: 代理{meta[agent_name]}, 状态{meta[status]}, f创建于{meta[created_at]}, 最后检查{meta[last_checked_at]})这个通道和消息历史是分开的这点很关键。Deep Agents 在上下文窗口填满时会压缩消息历史如果 task_id 只存在工具消息里压缩后就丢了。专用通道保证主代理始终能通过list_async_tasks回忆任务。超时验证要主动构造。给子代理设一个短超时看它是否正确报错而不是无限挂起# 在子代理图里配置超时示意具体参数看你的 LangGraph 版本 config {configurable: {timeout: 30}} # 30 秒超时 result agent.invoke({input: research something slow}, configconfig)超时后任务状态应该变成error而不是一直停在running。如果它一直 running说明超时没生效检查子代理图的运行配置。结果聚合是异步调度的最后一环。主代理拿到多个子代理的结果后需要合并成一份输出。用list_async_tasks遍历所有任务对非终态任务并行拉取实时状态终态任务success、error、cancelled从缓存返回def aggregate_results(agent): state agent.get_state() tasks state.values.get(async_tasks, {}) results {} for task_id, meta in tasks.items(): if meta[status] success: results[meta[agent_name]] meta.get(result, ) elif meta[status] error: results[meta[agent_name]] f[失败] {meta.get(error, 未知错误)} else: results[meta[agent_name]] f[进行中] {meta[status]} return results可观测性这块LangSmith 是标配。每个异步子代理运行都是标准的 LangGraph 运行在 LangSmith 里完全可见。主代理的追踪显示 launch、check、update、cancel、list 的工具调用每个子代理运行作为单独追踪出现通过线程 ID 链接。用线程 ID就是 task_id关联主代理编排追踪和子代理执行追踪一眼就能看出哪个子代理慢、哪个失败了。如果你不想依赖 LangSmith也可以在 TaoToken 的调用日志里看请求分布。因为所有模型调用走统一通道每个子代理的请求都会带上时间戳和模型 ID并发时能看到请求交错出现而不是串行排队。这算是统一通道带来的额外观测点。验证通过的标准很简单启动快、状态可查、超时能报错、结果能聚合。四条都满足异步调度就算跑通了。5. 常见报错排查401、local proxy failed、reading choices、OAuth 对照异步子代理跑起来之后报错会集中在几个地方。我把真实遇到过的错误和排查路径列出来你对照着看。401 Unauthorized。这个最常见八成是 TaoToken Key 或 Base URL 的问题。先确认.env里的TAOTOKEN_API_KEY没有多余空格TAOTOKEN_BASE_URL是https://taotoken.net/api而不是带/v1的变体。如果 Key 是对的还报 401去 API Keys 页面 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 确认令牌没过期、没被删。还有一种情况是子代理用了 HTTP 传输到远程服务器远程服务器的认证头没配这时候检查AsyncSubAgent的headers字段。local proxy failed。这个错误通常出现在网络层不是模型层。如果你在本地跑langgraph devASGI 传输是进程内调用理论上不该有代理问题。出现这个报错先检查是不是环境里设了HTTP_PROXY或HTTPS_PROXY变量把它们清掉再试。另外确认langgraph.json里所有图都注册了缺图会导致 SDK 找不到目标报出类似代理失败的错。reading choices 相关报错。这类错误一般是模型返回格式不符合预期。Deep Agents 期望模型返回结构化的工具调用如果模型输出被截断或格式错乱解析就会失败。排查方向一是确认model字段填的模型 ID 在 TaoToken 通道里可用去模型对话页面 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite 试一下同一个模型二是检查temperature是不是太高异步任务建议 0.2 以下三是看是不是上下文太长导致输出被截断适当精简 system_prompt。OAuth 相关报错。如果你用 HTTP 传输到远程 LangGraph 部署认证走的是LANGSMITH_API_KEY或LANGGRAPH_API_KEY环境变量。报 OAuth 错误通常是这两个变量没设或设错。自托管 Agent Protocol 服务器可能用不同的认证机制这时候要在AsyncSubAgent的headers里手动加认证头AsyncSubAgent( namecoder, descriptionCoding agent., graph_idcoder, urlhttps://coder-deployment.example.com, headers{Authorization: Bearer 你的远程服务令牌}, )除了这些还有几个异步特有的坑。主代理启动后立即轮询表现是启动任务后马上循环调check把异步变成阻塞。中间件会注入系统提示规则来防止如果还发生在system_prompt里强化「After launching an async subagent, ALWAYS return control to the user. Never call check_async_task immediately after launch.」主代理报告过时状态它引用对话历史里早期的任务状态而不是重新 check。解决办法是在系统提示里加「对话历史中的任务状态总是过时的」并要求报告前必须调check或list。任务 ID 查找失败主代理截断或重新格式化 task_id导致 check 或 cancel 失败。这通常是模型特定问题在系统提示里加「始终显示完整 task_id绝不截断或缩写」或者换个模型试试。子代理启动排队启动挂起或很久才开始。这是工作池耗尽用langgraph dev --n-jobs-per-worker 10加大池子。记住主代理加 N 个并发子代理需要 N1 个槽。排查的时候有个通用技巧先看 TaoToken 的调用日志确认请求有没有发出去。如果日志里没有对应请求问题在客户端配置如果有请求但报错问题在模型或参数。这一步能快速定位是网络层还是模型层。6. 把异步调度用起来从单机验证到长期 Coding Plan到这里AsyncSubAgent 的完整链路你应该能跑通了TaoToken 统一通道 →create_deep_agent配置 → 并发验证 → 报错排查。最后说几个实战里的经验。异步子代理最适合的场景是「长耗时 需要中途干预」。比如一份深度调研子代理跑十分钟用户中途想加个方向用update_async_task发新指令之前的运行被中断子代理带着完整对话历史加新指令重启task_id 不变。这种交互同步模式根本做不到。如果你要把这套东西用在长期编码或 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里面有各语言的完整示例。如果你用 Claude Code 这类工具Anthropic 兼容接入的说明在 https://taotoken.net/claudecode-anthropic?utm_sourcetaotoken_aicg_blog_endutm_contentclaudecode-anthropicutm_campaignrewrite。一个我踩过的坑异步任务的状态通道async_tasks会随着任务增多而膨胀。如果你跑几百个任务不清理主代理的状态会越来越大。建议在任务进入终态success、error、cancelled后定期归档或清理只保留最近 N 条。清理逻辑可以挂在主代理的收尾节点上。还有description字段值得反复打磨。主代理选子代理全靠它写得越具体委派越准。我一般会写清楚「什么时候用」和「能做什么」比如「Use for questions requiring multiple searches and synthesis」比「research agent」有用得多。最后异步不等于不管。启动任务后返回控制权是对的但用户问进度时你得真的去 check不能拿历史状态糊弄。中间件提示里那句「对话历史中的任务状态总是过时的」要刻进脑子里。做到这一点你的多代理系统才算真正可观测、可干预、可聚合。