作为一个 39 岁的技术人我最近在啃 DeepAgents 这个框架。前面学完了同步子 Agent但在实际使用中发现一个问题当子任务要花几分钟甚至几十分钟时主 Agent 在用户面前就成了死机状态——既无法继续聊也无法插话调整方向。本章就来拆解 DeepAgents 0.5.0 的预览特性Async Subagent异步子 Agent看看它是如何让主 Agent 立即拿到任务 ID 就返回子 Agent 在后台继续跑用户可以随时问进度、追加要求甚至中途取消的。一、为什么要有异步子智能体1.1 同步子 Agent 的瓶颈回顾一下上一章的多子 Agent 协作模式# 主 Agent 调用同步子 Agentresulttask(nameresearcher,task深入调研 LangGraph 生态)# 此时主 Agent 在等待——可能要等 60 秒、120 秒甚至更久# 用户只能盯着对话框转圈同步子 Agent 在两类场景下会让用户体验非常糟糕长程任务如深度调研、大规模代码迁移、批量数据处理子 Agent 工作时间从分钟级到小时级可交互任务用户在子 Agent 跑到一半时发现需要补充约束“换个数据源再来一次”、“加上 2024 年的数据”但同步模式下根本插不进去更糟的是——同步子 Agent 在被主 Agenttask()调用期间主 Agent 自己也被阻塞。这意味着在子 Agent 完成之前用户无法和主 Agent 继续聊别的话题。1.2 异步子 Agent 解决的两个核心问题异步子 Agent 解决的就是这两件事不阻塞主线对话主 Agent 启动子任务后立即返回任务 ID用户可以继续和主 Agent 对话支持中途控制用户可以随时查询进度、追加指令、甚至取消任务1.3 同步 vs 异步六个维度的对比维度同步子 Agent异步子 Agent执行模型阻塞——主 Agent 等到完成才能继续非阻塞——立即返回任务 ID并发性可并行触发但主 Agent 仍被整批阻塞完全并行主 Agent 全程不阻塞中途追加指令❌ 不支持✅update_async_task注入新指令取消❌ 不支持✅cancel_async_task请求取消任务状态性无状态——每次调用相互独立有状态——子 Agent 拥有自己的会话thread会话历史持续累积典型场景一问一答、毫秒级到秒级的快速委派几分钟以上的研究、编码、迁移等长程任务需要在对话中互动管理简单的判定法则子任务能在 5 秒内完成用同步子任务可能跑数分钟以上、且过程需要可交互上异步。二、异步子智能体 vs 同步子智能体2.1 定义对比同步子 Agent的定义fromdeepagentsimportSubAgent sync_subagents[SubAgent(nameresearcher,description深度网络调研需要多次搜索 信息综合时使用,graphcreate_researcher_graph(),# 直接传入编译好的图),]异步子 Agent的定义fromdeepagentsimportAsyncSubAgent async_subagents[AsyncSubAgent(nameresearcher,description深度网络调研需要多次搜索 信息综合时使用。适合需要 3 分钟以上的研究任务。,graph_idresearcher,# 必须与 langgraph.json 中注册的 graph 名称一致# 不传 url → ASGI 传输同部署),]2.2 关键区别为什么异步子 Agent 要以服务形式存在这是很多人第一次接触异步子 Agent 时最大的疑惑为什么不能像同步那样直接传入一个编译好的图而是要用graph_id引用一个远程服务原因有三层原因 1进程隔离同步子 Agent 和主 Agent 在同一个 Python 进程中运行共享内存和事件循环。而异步子 Agent 需要在独立的进程/容器中运行这样才能真正的并行执行不被主 Agent 的 GIL 限制独立的资源隔离CPU、内存、网络故障隔离子 Agent 崩溃不影响主 Agent原因 2生命周期管理异步子 Agent 有自己的**会话thread和运行run**生命周期。服务端需要创建独立的 thread 保存消息和状态启动 run 执行子 Agent 的逻辑维护任务状态pending → running → success/failed/cancelled支持中途查询、更新、取消这些都是服务级别的职责不是一个异步函数能承担的。原因 3传输协议异步子 Agent 通过Agent Protocol进行通信这是一套调用 Agent 的 API 规范约定如何创建会话POST /threads启动运行POST /threads/{thread_id}/runs查询状态GET /threads/{thread_id}/runs/{run_id}取消任务POST /threads/{thread_id}/runs/{run_id}/cancel所以异步子 Agent 必须以**服务Agent Server**的形式存在而不是一个函数。三、异步子智能体的协议规范3.1 核心概念分层在使用异步子 Agent 之前需要分清几个不同层次的概念名称职责Agent Protocol一套调用 Agent 的 API 规范约定如何创建会话、启动运行、查询状态和取消任务等Agent Server实现这些接口的运行服务加载 Agent 代码调度执行并管理会话状态与结果LangSmith Deployment部署和运行 Agent Server 的平台能力也可以使用自托管的兼容服务LangSmith Observability采集和查看 trace帮助分析模型调用、工具执行、耗时与错误主 Agent 负责决定任务怎么拆、交给谁、如何整合结果Agent Server 负责承接和执行任务。3.2 主 Agent 的 5 把遥控器LangChain 的AsyncSubAgentMiddleware中间件会自动给主 Agent 注入 5 个工具就像给主 Agent 配了一个遥控器工具名作用底层做了什么start_async_task启动后台异步任务通过 LangGraph SDK 创建子任务的 thread 和 run返回 task_idcheck_async_task查询任务状态通过 SDK 查询指定 run 的状态pending/running/success/failed/cancelled和最新输出update_async_task追加新指令向正在运行的 run 发送新的用户消息子 Agent 会继续处理cancel_async_task取消任务向服务端发送取消请求服务端停止该 run 的执行list_async_tasks列出所有任务查询当前会话下所有异步任务的状态汇总3.3 任务元数据为何要单开一个 channel这是一个很巧妙的设计。在 LangGraph 的 State 中任务元数据存在独立的async_taskschannel 中与消息历史解耦。这样做的好处是即便上下文被压缩summarization任务 ID 永不丢失。想象一下如果任务 ID 只存在消息历史里当对话太长触发摘要时ID 可能被压缩掉后续就无法 check 或 cancel 了。单开 channel 确保了任务状态的持久性和可恢复性。四、异步子智能体的示例及代码解读4.1 项目结构我们先搭建一个最小可运行的项目async_subagent_demo/ ├── langgraph.json # 注册主 Agent 和子 Agent ├── .env # 环境变量 ├── src/ │ ├── agent.py # 主 Agent (Supervisor) │ └── researcher.py # 研究者子 Agent └── run_demo.py # 验证脚本4.2 第 1 步安装依赖pipinstalllanggraph langgraph-sdk langchain-openai deepagents0.5.0⚠️ 注意deepagents 需要 Python 3.11Python 3.8/3.10 无法安装。4.3 第 2 步准备环境变量创建.env文件OPENAI_API_KEYyour-api-key-here LANGSMITH_API_KEYlsv2_your-key-here LANGSMITH_TRACINGtrue如果你使用阿里云 DashScope还需要设置OPENAI_API_BASEhttps://dashscope.aliyuncs.com/compatible-mode/v14.4 第 3 步编写 langgraph.json这是整个异步架构的注册中心{graphs:{supervisor:./src/agent.py:graph,researcher:./src/researcher.py:graph},env:.env}关键字段说明graphs注册所有可用的 Agent 图key 是 graph_idvalue 是导入路径supervisor主 Agent 的 graph_idresearcher子 Agent 的 graph_id必须与 AsyncSubAgent 中的graph_id一致4.5 第 4 步编写一个故意运行很慢的 Subagent# src/researcher.py 研究者子 Agent —— 一个故意运行很慢的异步子 Agent 用来演示异步子 Agent 的核心价值主 Agent 不被阻塞用户可以中途追加指令 importtimefromtypingimportTypedDict,Annotatedimportoperatorfromlanggraph.graphimportStateGraph,START,ENDclassResearchState(TypedDict):messages:Annotated[list,operator.add]research_topic:strfindings:strdefslow_researcher_node(state:ResearchState)-dict:研究节点 —— 模拟一个很慢的研究任务topicstate.get(research_topic,AI Agent)print(f\n[Researcher] 开始研究任务{topic})# 模拟真实研究中需要花费时间的操作steps[f正在搜索关于 {topic} 的最新资料...,正在阅读 10 篇相关论文...,正在整理关键发现和引用...,正在综合不同观点形成结论...,正在生成最终报告...,]forstepinsteps:print(f [Researcher]{step})time.sleep(1.5)# 模拟每个步骤耗时findings(f【研究报告】关于 {topic}\n\nf1. 核心概念{topic}是当前 AI Agent 领域的热门方向\nf2. 主要发现异步处理可以显著提升用户体验\nf3. 最佳实践建议从单部署 ASGI 开始按需拆分\nf4. 注意事项Worker Pool 要调大描述要具体\nf5. 参考资料LangGraph 官方文档、async-deep-agents 仓库)return{findings:findings,messages:[{role:assistant,content:findings}]}# 构建研究子 Agent 的图builderStateGraph(ResearchState)builder.add_node(research,slow_researcher_node)builder.add_edge(START,research)builder.add_edge(research,END)graphbuilder.compile()这段代码的关键点使用time.sleep(1.5)模拟每个研究步骤耗时总共约 7.5 秒构建了一个简单的 StateGraph包含一个研究节点最终编译为graph变量供langgraph.json注册使用4.6 第 5 步创建 Supervisor主 Agent# src/agent.py 主 Agent (Supervisor) —— 管理异步子 Agent 演示如何声明异步子 Agent、启动任务、查询进度、更新指令 importosfromtypingimportTypedDict,Annotatedimportoperatorfromlangchain_openaiimportChatOpenAIfromdeepagentsimportcreate_deep_agent,AsyncSubAgentfromdeepagents.middlewareimportAsyncSubAgentMiddlewarefromlanggraph.graphimportStateGraph,START,END# 声明异步子 Agent async_subagents[AsyncSubAgent(nameresearcher,description深度网络调研需要多次搜索 信息综合时使用。适合需要 3 分钟以上的研究任务。,graph_idresearcher,# 必须与 langgraph.json 中注册的 graph 名称一致# 不传 url → ASGI 传输同部署),]classAgentState(TypedDict):messages:Annotated[list,operator.add]defcreate_supervisor():创建带有异步子 Agent 中间件的主 AgentmodelChatOpenAI(modelqwen-plus,openai_api_basehttps://dashscope.aliyuncs.com/compatible-mode/v1,openai_api_keyos.getenv(OPENAI_API_KEY),)system_prompt你是一个研究主管负责管理一个研究团队。 你可以将研究任务委派给 researcher 异步处理。 重要规则 1. 派出异步子 Agent 之后必须立刻把控制权交还给用户 2. 不要在没有用户提问的情况下主动 check_async_task 3. 回答任务进度前必须先调用 check_async_task 获取最新状态 4. 始终使用完整的 task_id不要截断、不要缩写、不要改写 # 创建带有异步子 Agent 的主 Agentagentcreate_deep_agent(modelmodel,system_promptsystem_prompt,subagentsasync_subagents,)returnagent# 构建 LangGraph 图supervisorcreate_supervisor()# 用 StateGraph 包装供 langgraph.json 注册builderStateGraph(AgentState)builder.add_node(agent,supervisor)builder.add_edge(START,agent)builder.add_edge(agent,END)graphbuilder.compile()关键点解读AsyncSubAgent只需要name、description、graph_id三个必填字段不传url参数时默认使用 ASGI 传输同部署零延迟create_deep_agent的subagents参数接受同步和异步子 Agent 的混合列表system_prompt 中必须强调派出异步子 Agent 后立刻交还控制权否则模型可能会退化成伪同步4.7 第 6 步启动本地 Agent Serverlanggraph dev --n-jobs-per-worker10这条命令会读取langgraph.json配置加载所有注册的 graph启动本地 ASGI 服务默认http://127.0.0.1:2024设置 worker pool 大小为 10支持最多 10 个并发运行⚠️Worker Pool 很重要每个活跃的运行会占用一个 Worker 槽位。一个主 Agent 同时跑 3 个子 Agent至少需要 4 个槽位1 主 3 子。槽位不够时新启动的任务会排队。4.8 第 7 步使用 SDK 验证异步行为# run_demo.py 验证脚本使用 LangGraph SDK 验证异步行为 演示完整的异步子 Agent 生命周期 importasynciofromlanggraph_sdkimportget_clientfrompprintimportpprintasyncdefmain():# 连接到本地 Agent Serverclientget_client(urlhttp://127.0.0.1:2024)assistant_idsupervisor# 创建会话线程threadawaitclient.threads.create()thread_idthread[thread_id]print(fthread_id {thread_id})# 第 1 次交互启动异步任务 firstawaitclient.runs.wait(thread_id,assistant_id,input{messages:[{role:user,content:(请把这个任务交给 researcher 异步处理用后台任务总结 async subagent 的关键行为。),}]},)print(\n first response )pprint(first)# 应该很快返回里面有一个后台任务 ID而不是卡 8 秒等 researcher 完成# 第 2 次交互查询进度 secondawaitclient.runs.wait(thread_id,assistant_id,input{messages:[{role:user,content:刚才那个后台任务现在进展如何,}]},)print(\n second response )pprint(second)# 大概率会看到 running或者已经拿到阶段性状态信息# 第 3 次交互追加指令 thirdawaitclient.runs.wait(thread_id,assistant_id,input{messages:[{role:user,content:补充约束完成时请把答案写成 3 条 bullet。,}]},)print(\n third response )pprint(third)# 不会要求你重开任务而是会尝试更新已有后台任务# 第 4 次交互再次查询应该完成了 fourthawaitclient.runs.wait(thread_id,assistant_id,input{messages:[{role:user,content:现在任务完成了吗给我最终结果。,}]},)print(\n fourth response )pprint(fourth)# 状态会从 running 变成 success并带上 researcher 的最终结果if__name____main__:asyncio.run(main())运行它python run_demo.py4.9 你应该看到什么只要出现下面这组现象就说明这条本地 ASGI 路径已经打通了first response很快返回里面有一个后台任务 ID而不是卡 8 秒等 researcher 完成second response大概率会看到running或者已经拿到阶段性状态信息third response不会要求你重开任务而是会尝试更新已有后台任务过几秒后再次问进度时状态会从running变成success并带上 researcher 的最终结果这个示例的目标不是做真实研究而是稳定验证异步机制本身。一旦这套最小示例跑通你再把 researcher 替换成真正的深度智能体、搜索工具或远程 HTTP 子智能体排障成本会低很多。五、部署方式和最佳实践5.1 三种部署拓扑拓扑形态推荐场景单部署Single所有 Agent 同部署全部用 ASGI绝大多数项目的起点一台服务好运维、零网络延迟拆分部署Split主 Agent 一台子 Agent 一台全用 HTTP子 Agent 资源画像或扩缩容策略与主 Agent 显著不同混合Hybrid一部分子 Agent 走 ASGI同部署另一部分走 HTTP远程大多数子 Agent 同部署省事少数特殊子 Agent 单独扩混合形态长这样async_subagents[AsyncSubAgent(nameresearcher,description研究 Agent,graph_idresearcher,# 不传 url → ASGI同部署),AsyncSubAgent(namecoder,description编码 Agent,graph_idcoder,urlhttps://coder-deployment.langsmith.dev,# 传了 url → HTTP远程),]起手式建议先用单部署 ASGI等遇到具体的扩缩容/团队边界问题再拆。5.2 最佳实践1. 本地开发要把 Worker Pool 调大每个活跃的运行会占用一个 Worker 槽位。一个主 Agent 同时跑 3 个子 Agent至少需要 4 个槽位1 主 3 子。槽位不够时新启动的任务会排队。常见表现包括start_async_task长时间不返回或虽然拿到了任务 ID但后续check_async_task长时间看不到实质进展。langgraph dev --n-jobs-per-worker102. 描述要具体行为导向主 Agent 靠description决定派给谁。两个对照# ✅ 好AsyncSubAgent(nameresearcher,description深度网络调研需要多次搜索 信息综合时使用,graph_idresearcher,)# ❌ 差AsyncSubAgent(namehelper,description帮你处理事情,graph_idhelper,)3. 用 Thread ID 串联追踪LangGraph 部署里每次异步子 Agent 运行都是一次普通的 LangGraph run。配置并启用 LangSmith 追踪后可以在 Observability 中查看这些运行。主 Agent 的 trace 会显示 launch / check / update / cancel / list 这些工具调用每个子 Agent 的运行是另一条 trace通过子任务的 thread ID也就是 task ID就能把两边对上。5.3 部署排查清单现象优先检查处理方式start_async_task报找不到 Agentgraph_id是否与langgraph.json注册名一致确认主 Agent 使用的graph_id和部署配置完全一致尤其注意大小写和下划线远程 HTTP 子 Agent 调用失败url、headers、LANGSMITH_API_KEY/LANGGRAPH_API_KEYLangSmith 部署优先依赖环境变量自托管服务则把鉴权头显式放进headers本地同部署能跑远程拆分后失败子 Agent 服务是否兼容 Agent Protocol先用 SDK 直接访问远程服务创建 thread / run再接回AsyncSubAgent任务一直runningworker 数、外部工具超时、子 Agent 是否卡在长工具调用提高--n-jobs-per-worker给外部 API / 搜索 / 代码执行设置超时避免后台 run 永久占住 workercancel_async_task后状态不立刻变化服务端取消是异步生效cancel 后再调用一次check_async_task或list_async_tasks确认最终状态不要只依赖本地旧消息主 Agent 查不到之前的任务thread / checkpoint 是否持久化确保主 Agent 配置 checkpointer任务元数据依赖async_taskschannel进程重启后需要可恢复状态LangSmith 中 trace 对不上task ID、thread ID、run ID 是否记录完整保留完整 task ID用 thread ID 串联主 Agent 的 launch 工具调用和子 Agent 的实际 run六、小结本章我们解锁了 DeepAgents 0.5.0 的预览特性 Async Subagent核心动机突破同步子 Agent 的两个瓶颈——主 Agent 不再被阻塞任务可以中途追加指令或取消5 把遥控器start/check/update/cancel/list主 Agent 像调普通工具一样用它们操控后台子任务状态独立通道任务元数据存在async_taskschannel 中与消息历史解耦——即便上下文被压缩任务 ID 永不丢失两种传输默认 ASGI同部署、零延迟按需切换 HTTP远程、可独立扩缩容三种拓扑单部署 / 拆分部署 / 混合起手用单部署按工程需要再拆避坑要点worker pool 要足够、描述要具体、永远基于实时 check 而非对话历史报告状态参考实现LangChain 官方提供了一个完整可跑的示例仓库 async-deep-agentsPython 与 TypeScript 双版本演示了一个主 Agent researcher coder 子 Agent 的部署形态。强烈建议 clone 下来跑一遍亲眼看看主 Agent 派活之后立刻能继续聊的实际效果。我是华仔一个 39 岁的技术人。这篇文章是我啃透 DeepAgents 异步子 Agent 后的学习总结希望能帮到同样在探索 AI Agent 架构的你。如果你觉得有帮助欢迎点赞收藏我们下一篇见