企业官网建设流程全解析

作为一个 39 岁的技术人，我最近在啃 DeepAgents 这个框架。前面学完了同步子 Agent，但在实际使用中发现一个问题：当子任务要花几分钟甚至几十分钟时，主 Agent 在用户面前就成了"死机状态"——既无法继续聊，也无法插话调整方向。

本章就来拆解 DeepAgents 0.5.0+ 的预览特性Async Subagent（异步子 Agent），看看它是如何让主 Agent 立即拿到任务 ID 就返回，子 Agent 在后台继续跑，用户可以随时问进度、追加要求，甚至中途取消的。

一、为什么要有异步子智能体？

1.1 同步子 Agent 的瓶颈

回顾一下上一章的多子 Agent 协作模式：

# 主 Agent 调用同步子 Agentresult=task(name="researcher",task="深入调研 LangGraph 生态")# 此时主 Agent 在等待——可能要等 60 秒、120 秒，甚至更久# 用户只能盯着对话框转圈

同步子 Agent 在两类场景下会让用户体验非常糟糕：

1. 长程任务：如深度调研、大规模代码迁移、批量数据处理，子 Agent 工作时间从分钟级到小时级
2. 可交互任务：用户在子 Agent 跑到一半时，发现需要补充约束（“换个数据源再来一次”、“加上 2024 年的数据”），但同步模式下根本插不进去

更糟的是——同步子 Agent 在被主 Agenttask()调用期间，主 Agent 自己也被阻塞。这意味着：在子 Agent 完成之前，用户无法和主 Agent 继续聊别的话题。

1.2 异步子 Agent 解决的两个核心问题

异步子 Agent 解决的就是这两件事：

• 不阻塞主线对话：主 Agent 启动子任务后立即返回任务 ID，用户可以继续和主 Agent 对话
• 支持中途控制：用户可以随时查询进度、追加指令、甚至取消任务

1.3 同步 vs 异步：六个维度的对比

简单的判定法则：子任务能在 5 秒内完成，用同步；子任务可能跑数分钟以上、且过程需要可交互，上异步。

二、异步子智能体 vs 同步子智能体

2.1 定义对比

同步子 Agent的定义：

fromdeepagentsimportSubAgent sync_subagents=[SubAgent(name="researcher",description="深度网络调研，需要多次搜索 + 信息综合时使用",graph=create_researcher_graph(),# 直接传入编译好的图),]

异步子 Agent的定义：

fromdeepagentsimportAsyncSubAgent async_subagents=[AsyncSubAgent(name="researcher",description="深度网络调研，需要多次搜索 + 信息综合时使用。适合需要 3 分钟以上的研究任务。",graph_id="researcher",# 必须与 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 负责决定任务怎么拆、交给谁、如何整合结果；Agent Server 负责承接和执行任务。

3.2 主 Agent 的 5 把"遥控器"

LangChain 的AsyncSubAgentMiddleware中间件会自动给主 Agent 注入 5 个工具，就像给主 Agent 配了一个遥控器：

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 deepagents>=0.5.0

⚠️ 注意：deepagents 需要 Python 3.11+，Python 3.8/3.10 无法安装。

4.3 第 2 步：准备环境变量

创建.env文件：

OPENAI_API_KEY=your-api-key-here LANGSMITH_API_KEY=lsv2_your-key-here LANGSMITH_TRACING=true

如果你使用阿里云 DashScope，还需要设置：

OPENAI_API_BASE=https://dashscope.aliyuncs.com/compatible-mode/v1

4.4 第 3 步：编写 langgraph.json

这是整个异步架构的注册中心：

{"graphs":{"supervisor":"./src/agent.py:graph","researcher":"./src/researcher.py:graph"},"env":".env"}

关键字段说明：

• graphs：注册所有可用的 Agent 图，key 是 graph_id，value 是导入路径
• supervisor：主 Agent 的 graph_id
• researcher：子 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:"""研究节点 —— 模拟一个很慢的研究任务"""topic=state.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\n"f"1. 核心概念：{topic}是当前 AI Agent 领域的热门方向\n"f"2. 主要发现：异步处理可以显著提升用户体验\n"f"3. 最佳实践：建议从单部署 + ASGI 开始，按需拆分\n"f"4. 注意事项：Worker Pool 要调大，描述要具体\n"f"5. 参考资料：LangGraph 官方文档、async-deep-agents 仓库")return{"findings":findings,"messages":[{"role":"assistant","content":findings}]}# 构建研究子 Agent 的图builder=StateGraph(ResearchState)builder.add_node("research",slow_researcher_node)builder.add_edge(START,"research")builder.add_edge("research",END)graph=builder.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(name="researcher",description="深度网络调研，需要多次搜索 + 信息综合时使用。适合需要 3 分钟以上的研究任务。",graph_id="researcher",# 必须与 langgraph.json 中注册的 graph 名称一致# 不传 url → ASGI 传输（同部署）),]classAgentState(TypedDict):messages:Annotated[list,operator.add]defcreate_supervisor():"""创建带有异步子 Agent 中间件的主 Agent"""model=ChatOpenAI(model="qwen-plus",openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1",openai_api_key=os.getenv("OPENAI_API_KEY"),)system_prompt="""你是一个研究主管，负责管理一个研究团队。 你可以将研究任务委派给 researcher 异步处理。 重要规则： 1. 派出异步子 Agent 之后，必须立刻把控制权交还给用户 2. 不要在没有用户提问的情况下主动 check_async_task 3. 回答任务进度前，必须先调用 check_async_task 获取最新状态 4. 始终使用完整的 task_id，不要截断、不要缩写、不要改写 """# 创建带有异步子 Agent 的主 Agentagent=create_deep_agent(model=model,system_prompt=system_prompt,subagents=async_subagents,)returnagent# 构建 LangGraph 图supervisor=create_supervisor()# 用 StateGraph 包装，供 langgraph.json 注册builder=StateGraph(AgentState)builder.add_node("agent",supervisor)builder.add_edge(START,"agent")builder.add_edge("agent",END)graph=builder.compile()

关键点解读：

• AsyncSubAgent只需要name、description、graph_id三个必填字段
• 不传url参数时，默认使用 ASGI 传输（同部署，零延迟）
• create_deep_agent的subagents参数接受同步和异步子 Agent 的混合列表
• system_prompt 中必须强调"派出异步子 Agent 后立刻交还控制权"，否则模型可能会退化成伪同步

4.7 第 6 步：启动本地 Agent Server

langgraph dev --n-jobs-per-worker10

这条命令会：

1. 读取langgraph.json配置
2. 加载所有注册的 graph
3. 启动本地 ASGI 服务（默认http://127.0.0.1:2024）
4. 设置 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 Serverclient=get_client(url="http://127.0.0.1:2024")assistant_id="supervisor"# 创建会话线程thread=awaitclient.threads.create()thread_id=thread["thread_id"]print(f"thread_id ={thread_id}")# ===== 第 1 次交互：启动异步任务 =====first=awaitclient.runs.wait(thread_id,assistant_id,input={"messages":[{"role":"user","content":("请把这个任务交给 researcher 异步处理：""用后台任务总结 async subagent 的关键行为。"),}]},)print("\n=== first response ===")pprint(first)# 应该很快返回，里面有一个后台任务 ID，而不是卡 8 秒等 researcher 完成# ===== 第 2 次交互：查询进度 =====second=awaitclient.runs.wait(thread_id,assistant_id,input={"messages":[{"role":"user","content":"刚才那个后台任务现在进展如何？",}]},)print("\n=== second response ===")pprint(second)# 大概率会看到 running，或者已经拿到阶段性状态信息# ===== 第 3 次交互：追加指令 =====third=awaitclient.runs.wait(thread_id,assistant_id,input={"messages":[{"role":"user","content":"补充约束：完成时请把答案写成 3 条 bullet。",}]},)print("\n=== third response ===")pprint(third)# 不会要求你重开任务，而是会尝试更新已有后台任务# ===== 第 4 次交互：再次查询（应该完成了） =====fourth=awaitclient.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.py

4.9 你应该看到什么

只要出现下面这组现象，就说明这条本地 ASGI 路径已经打通了：

1. first response很快返回，里面有一个后台任务 ID，而不是卡 8 秒等 researcher 完成
2. second response大概率会看到running，或者已经拿到阶段性状态信息
3. third response不会要求你重开任务，而是会尝试更新已有后台任务
4. 过几秒后再次问进度时，状态会从running变成success，并带上 researcher 的最终结果

这个示例的目标不是做真实研究，而是稳定验证异步机制本身。一旦这套最小示例跑通，你再把 researcher 替换成真正的深度智能体、搜索工具或远程 HTTP 子智能体，排障成本会低很多。

五、部署方式和最佳实践

5.1 三种部署拓扑

混合形态长这样：

async_subagents=[AsyncSubAgent(name="researcher",description="研究 Agent",graph_id="researcher",# 不传 url → ASGI（同部署）),AsyncSubAgent(name="coder",description="编码 Agent",graph_id="coder",url="https://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-worker10

2. 描述要具体，行为导向

主 Agent 靠description决定派给谁。两个对照：

# ✅ 好AsyncSubAgent(name="researcher",description="深度网络调研，需要多次搜索 + 信息综合时使用",graph_id="researcher",)# ❌ 差AsyncSubAgent(name="helper",description="帮你处理事情",graph_id="helper",)

3. 用 Thread ID 串联追踪

LangGraph 部署里，每次异步子 Agent 运行都是一次普通的 LangGraph run。配置并启用 LangSmith 追踪后，可以在 Observability 中查看这些运行。主 Agent 的 trace 会显示 launch / check / update / cancel / list 这些工具调用；每个子 Agent 的运行是另一条 trace，通过子任务的 thread ID（也就是 task ID）就能把两边对上。

5.3 部署排查清单

六、小结

本章我们解锁了 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-agents，Python 与 TypeScript 双版本，演示了一个主 Agent + researcher + coder 子 Agent 的部署形态。强烈建议 clone 下来跑一遍，亲眼看看主 Agent "派活之后立刻能继续聊"的实际效果。

我是华仔，一个 39 岁的技术人。这篇文章是我啃透 DeepAgents 异步子 Agent 后的学习总结，希望能帮到同样在探索 AI Agent 架构的你。如果你觉得有帮助，欢迎点赞收藏，我们下一篇见！