DeepAgent 多子代理协作：中断授权与 Agent 间通讯机制

本文整理 DeepAgent 在多子代理场景下的两个关键工程问题：

1. 子 Agent 发起中断授权时，主 Agent 和其他子 Agent 怎么执行；
2. 多个子 Agent 之间想共享数据、互相协作时，应该怎么设计。

适合场景：支付审批、发邮件审批、文件写入审批、多 Agent 并行研究、代码审查、任务分派系统。

1. 先给结论

DeepAgent 的子代理协作不是“多个 Agent 互相聊天”的模式。

更准确的模型是：

主 DeepAgent -> 通过 task 工具调用子 Agent -> 子 Agent 独立执行任务 -> 子 Agent 返回最终结果 -> 主 DeepAgent 汇总和继续决策

子 Agent 之间默认不能直接通讯。它们的数据通常通过以下方式传递：

• 主 Agent 中转；
• 共享文件系统；
• 数据库 / 对象存储 / Redis / 业务 API；
• 外层 LangGraph state；
• 异步子代理任务 API。

中断授权也不是“全局暂停所有 Agent”。准确说：

同一个thread_id下，某个子 Agent 触发中断后，当前这次主 DeepAgent run 会暂停；主 Agent 不会继续下一轮推理，直到用同一个thread_id恢复执行。其他独立 thread 不受影响。

2. 没写 Graph，也仍然有 Graph

很多人会说：

我没有写 LangGraph，只是 create_deep_agent。

但要注意：

agent=create_deep_agent(...)

返回的本身就是一个 LangGraph runnable。

所以即使你没有手写外层业务 Graph，DeepAgent 内部仍然有：

• 状态；
• messages；
• tool call；
• checkpoint；
• interrupt；
• resume；
• subagent 调用。

可以理解成：

你没有自己画 Graph，但 create_deep_agent 已经帮你封装了一个执行图。

3. 子 Agent 是怎么被调用的

DeepAgent 的SubAgentMiddleware会给主 Agent 注入一个task工具。

主 Agent 不是直接“进入”子 Agent，而是调用：

task(subagent_type="researcher", description="去搜索某主题论文")

执行流程：

主 Agent -> 发起 task 工具调用 -> SubAgentMiddleware 找到对应子 Agent -> 子 Agent 使用独立 messages 执行 -> 子 Agent 完成后返回最终消息 -> 这个最终消息变成主 Agent 看到的 ToolMessage

默认情况下，子 Agent 的中间工具调用、搜索过程、内部 messages 不会全量回传给主 Agent。

主 Agent 通常只拿到：

子 Agent 的最终报告 / 最终结构化结果

4. 子 Agent 中断授权是怎么实现的

DeepAgent 的中断授权主要有两种方式。

4.1 用interrupt_on拦截工具调用

这是最常见方式。

例如支付工具：

fromlangchain.toolsimporttool@tooldefcreate_payment(order_id:str,amount:float,payee:str)->str:"""执行支付动作；该工具只应在人工审批通过后被调用。"""returnf"支付成功：订单={order_id}，金额={amount}，收款方={payee}"

配置子 Agent：

payment_subagent={"name":"payment-agent","description":"负责支付前检查和支付执行。","system_prompt":("你是支付子代理。你只能在订单信息完整时调用 create_payment。""支付动作必须等待人工审批，不得绕过审批。"),"tools":[create_payment],"interrupt_on":{"create_payment":{"allowed_decisions":["approve","reject"],}},}

这表示：

当 payment-agent 想调用 create_payment 时，先暂停，等待人审批。 approve：继续执行工具 reject：拒绝执行工具

4.2 在工具内部调用interrupt()

如果审批逻辑更业务化，也可以在工具内部中断。

fromlangchain.toolsimporttoolfromlanggraph.typesimportinterrupt@tooldefpay_with_manual_approval(order_id:str,amount:float,payee:str)->str:"""先请求人工支付授权，审批通过后再执行支付。"""approval=interrupt({"type":"payment_approval","order_id":order_id,"amount":amount,"payee":payee,"message":f"是否批准支付订单{order_id}，金额{amount}，收款方{payee}？",})ifnotapproval.get("approved"):returnf"支付被拒绝：{approval.get('reason','未提供原因')}"returnf"支付成功：订单={order_id}，金额={amount}，收款方={payee}"

这种适合：

• 支付；
• 发邮件；
• 发短信；
• 改生产配置；
• 写生产数据库；
• 删除文件；
• 高成本 API 调用。

5. 多个子 Agent 并行时，一个触发中断会怎样

假设主 Agent 同一轮发出 3 个task：

task(agent_a, "处理 A") task(agent_b, "处理 B，需要支付") task(agent_c, "处理 C")

底层可以并行跑：

主 DeepAgent -> 子 Agent A -> 子 Agent B -> 子 Agent C

如果子 Agent B触发支付中断：

子 Agent B -> create_payment -> interrupt

那么当前主 run 的状态是：

当前 thread_id 的主 DeepAgent run 暂停 等待人工 approve / reject / respond

5.1 会不会暂停所有子 Agent？

不会全局暂停所有 Agent。

更准确：

暂停的是当前 thread_id 对应的这一次执行链。

也就是：

• 当前主 Agent 不会继续下一轮推理；
• 当前主 Agent 不会提前汇总 A/C 的结果；
• 当前这批 task 要等中断恢复后才能完整返回；
• 其他用户请求、其他 thread_id、其他独立 Agent run 不受影响。

5.2 已经完成的并行子任务怎么办

如果 A/C 已经完成，它们的结果可能已经在当前 run 的内部状态或 checkpoint 中。

但是主 Agent 不会继续消费这些结果，直到 B 的中断被恢复。

重要的是：

中断不是事务，已经执行过的外部副作用不会自动回滚。

例如：

Agent A 已经写库 Agent C 已经发消息 Agent B 等待支付审批

如果 B 最后被拒绝，A/C 已经发生的副作用不会自动撤销。

所以生产设计里，危险副作用不要和审批无序并行。

推荐流程：

并行：查询、分析、风控、生成计划 串行：人工审批 审批通过后：支付、写库、发通知

6. 多个中断同时发生怎么办

如果并行分支里多个子 Agent 都触发中断，运行时可能返回多个 interrupt。

恢复时要为每个中断提供对应结果。

工具调用审批场景通常是：

fromlanggraph.typesimportCommanddefresume_multiple_tool_interrupts(agent,config):"""恢复多个工具调用审批中断。"""returnagent.invoke(Command(resume={"decisions":[{"type":"approve"},{"type":"reject"},]}),config=config,version="v2",)

如果是并行分支里的多个interrupt()，更推荐用 interrupt id 映射恢复：

fromlanggraph.typesimportCommanddefresume_interrupts_by_id(agent,config,interrupt_a_id:str,interrupt_b_id:str):"""根据 interrupt id 恢复多个并行中断。"""returnagent.invoke(Command(resume={interrupt_a_id:{"approved":True},interrupt_b_id:{"approved":False,"reason":"金额异常"},}),config=config,)

这样能避免多个并行中断靠顺序匹配导致混乱。

7. 完整 Demo：主 Agent + 多个子 Agent + 支付审批

下面示例重点展示结构，不绑定某个具体模型供应商。

fromlangchain.toolsimporttoolfromlanggraph.checkpoint.memoryimportMemorySaverfromlanggraph.typesimportCommandfromdeepagentsimportcreate_deep_agent@tooldefquery_order(order_id:str)->dict:"""查询订单信息，返回订单金额、收款方和状态。"""return{"order_id":order_id,"amount":199.0,"payee":"merchant_a","status":"pending",}@tooldefcreate_payment(order_id:str,amount:float,payee:str)->str:"""执行支付动作；人工审批通过后才允许真实执行。"""returnf"支付成功：订单={order_id}，金额={amount}，收款方={payee}"@tooldefwrite_audit_log(order_id:str,message:str)->str:"""写入审计日志，用于记录审批前后的关键事件。"""returnf"审计日志已写入：{order_id}-{message}"order_subagent={"name":"order-agent","description":"负责查询和整理订单信息。","system_prompt":"你是订单子代理，只负责查询订单信息，不执行支付。","tools":[query_order],}payment_subagent={"name":"payment-agent","description":"负责执行支付动作，支付前必须等待人工审批。","system_prompt":("你是支付子代理。你只能根据主 Agent 提供的订单信息调用 create_payment。""你不得修改金额或收款方。"),"tools":[create_payment],"interrupt_on":{"create_payment":{"allowed_decisions":["approve","reject"],}},}audit_subagent={"name":"audit-agent","description":"负责写审计日志。","system_prompt":"你是审计子代理，只负责记录日志，不执行支付。","tools":[write_audit_log],}checkpointer=MemorySaver()agent=create_deep_agent(model="openai:gpt-5.5",tools=[],subagents=[order_subagent,payment_subagent,audit_subagent],checkpointer=checkpointer,system_prompt=("你是主 Agent，负责支付流程编排。\n""规则：\n""1. 查询订单必须调用 order-agent。\n""2. 支付必须调用 payment-agent。\n""3. 审计日志必须调用 audit-agent。\n""4. 你不得亲自执行支付。\n""5. 支付前必须等待人工审批。"),)config={"configurable":{"thread_id":"pay-order-10001"}}result=agent.invoke({"messages":[{"role":"user","content":"请处理订单 order-10001 的支付流程。",}]},config=config,version="v2",)ifresult.interrupts:approval_result=agent.invoke(Command(resume={"decisions":[{"type":"approve"},]}),config=config,version="v2",)

关键点：

• checkpointer必须有；
• thread_id必须一致；
• 支付工具只给payment-agent；
• 主 Agent 不直接持有create_payment；
• 审批通过前，create_payment不应该真正执行；
• 支付工具内部仍要做幂等和权限校验。

8. 子 Agent 之间怎么通讯

默认情况下，子 Agent 之间不直接通讯。

不要想象成：

researcher <-> critic <-> writer

更像：

researcher -> 主 Agent -> critic -> 主 Agent -> writer

9. 通讯方式一：主 Agent 中转

这是最简单、最推荐的方式。

主 Agent -> task(researcher, "搜索论文") <- researcher 返回论文列表 主 Agent -> task(critic, "基于下面论文列表评分：...") <- critic 返回评分结果 主 Agent -> 写最终报告

优点：

• 最清晰；
• 最容易审计；
• 容易控制流程；
• 不容易出现子 Agent 私下乱调用。

缺点：

• 中间数据太大时会占用主 Agent 上下文；
• 数据结构要在 prompt 里约束清楚。

适合：

• 小到中等规模数据；
• 结构化结果；
• 审计要求高的流程。

10. 通讯方式二：共享文件系统

如果数据比较大，可以让一个子 Agent 写文件，另一个子 Agent 读文件。

researcher -> write_file("/work/pay-order-10001/research.json") critic -> read_file("/work/pay-order-10001/research.json")

主 Agent 只负责传路径：

请读取 /work/pay-order-10001/research.json，并基于其中内容评分。

推荐路径设计：

/work/{thread_id}/{subagent_name}/output.json /work/{thread_id}/shared/research_result.json /work/{thread_id}/shared/payment_plan.json

优点：

• 不把大文本塞回 messages；
• 可以保留中间产物；
• 适合大文件、大搜索结果、大报告草稿。

缺点：

• 要做好路径隔离；
• 并行写文件可能冲突；
• 默认StateBackend是虚拟文件系统，不一定是本机真实磁盘；
• 生产环境要配置权限和持久化策略。

11. 通讯方式三：共享数据库 / 对象存储 / 业务 API

生产系统更推荐这种。

例如：

fromlangchain.toolsimporttool@tooldefsave_agent_result(task_id:str,role:str,data:dict)->str:"""保存子代理结果，供主 Agent 或其他子 Agent 后续读取。"""# 生产环境中可写入数据库、Redis、对象存储或企业内部 APIreturnf"saved:{task_id}:{role}"@tooldefload_agent_result(task_id:str,role:str)->dict:"""读取指定子代理之前保存的结果。"""# 生产环境中可从数据库、Redis、对象存储或企业内部 API 读取return{"task_id":task_id,"role":role,"data":{},}

推荐数据流：

researcher -> save_agent_result(task_id, "researcher", data) critic -> load_agent_result(task_id, "researcher") critic -> save_agent_result(task_id, "critic", scores) 主 Agent -> load_agent_result(task_id, "critic")

优点：

• 可审计；
• 可恢复；
• 可跨进程；
• 可支持异步任务；
• 可接权限系统。

缺点：

• 工程复杂度更高；
• 要设计 schema、幂等、权限、过期时间；
• Agent 需要知道 task_id 和数据契约。

12. 通讯方式四：外层 LangGraph 管 state

如果流程很复杂，建议不要让一个主 DeepAgent 管全部。

更稳的是：

LangGraph 做流程骨架 DeepAgent 做复杂节点

例如：

Graph state: order_info risk_result payment_approval payment_result audit_log Node 1: order DeepAgent Node 2: risk DeepAgent Node 3: human approval Node 4: payment tool Node 5: audit DeepAgent

这种方式适合：

• 强流程；
• 支付；
• 审批；
• 风控；
• 多阶段业务；
• 需要恢复和审计的系统。

DeepAgent 负责“智能任务”，Graph 负责“确定性流程”。

13. 不推荐：子 Agent 私下互相调用

不推荐这样设计：

researcher -> task(critic) critic -> task(writer) writer -> task(researcher)

原因：

• 调用链难审计；
• 容易递归；
• 成本不可控；
• 中断授权难管理；
• 主 Agent 失去编排权；
• 并行和恢复语义更复杂。

生产里应该让主 Agent 或外层 Graph 负责调度。

14. 支付审批场景的最佳实践

支付 / 扣款 / 转账 / 下单这类场景，不要只靠 prompt。

推荐同时做：

1. 支付工具只给 payment-agent，不给主 Agent。 2. payment-agent 配 interrupt_on。 3. allowed_decisions 建议只开放 approve / reject。 4. 不建议开放 edit，避免审批人直接改金额或收款方。 5. 支付工具内部二次校验订单金额、收款方、用户权限。 6. 使用 idempotency_key 防止重复扣款。 7. 审批记录落库：审批人、时间、原始参数、审批结果。 8. 审批通过前不要调用真实支付网关。 9. 不要让支付动作和其他副作用无序并行。 10. 用同一个 thread_id 恢复中断。

支付流程建议：

查询订单 -> 风控检查 -> 生成支付计划 -> 人工审批 -> 真正支付 -> 写审计日志 -> 通知用户

不建议：

支付、写库、发通知、写审计全部并行跑

15. 常见坑点

15.1 以为中断会暂停整个系统

不会。

中断暂停的是当前thread_id对应的执行链。

其他请求、其他用户、其他 thread 不受影响。

15.2 以为中断会自动回滚其他子任务

不会。

如果并行子 Agent 已经执行了外部副作用，比如写库、发消息、调用支付网关，不会因为另一个子 Agent 中断而自动撤销。

15.3 忘记配置 checkpointer

Human-in-the-loop 需要 checkpoint。

没有 checkpointer，就不能可靠恢复到中断点。

15.4 resume 时换了 thread_id

恢复必须使用同一个thread_id。

否则运行时找不到之前暂停的状态。

15.5 compiled 子 Agent 没配自己的 interrupt

声明式 subagent 默认可以继承主 Agent 的interrupt_on。

但如果你传的是已经编译好的 child runnable，例如：

child_agent=create_deep_agent(...)

再作为 compiled subagent 传入，主 Agent 的interrupt_on不会自动注入它。

这种情况下，要在child_agent自己创建时配置interrupt_on和checkpointer。

15.6 子 Agent 通讯全靠自然语言

如果靠自然语言描述传数据，很容易出现：

• 字段丢失；
• 数字写错；
• JSON 格式不稳定；
• 上下文过长；
• 审计困难。

生产中建议用结构化数据、文件、数据库或外部 state。

15.7 多子 Agent 并行写同一路径

如果多个子 Agent 都写：

/work/result.json

就容易互相覆盖。

应该写：

/work/{thread_id}/{subagent_name}/result.json

16. 选择建议

16.1 数据怎么传

16.2 中断怎么做

17. 总结

DeepAgent 的多子代理协作可以用一句话理解：

主 Agent 负责调度，子 Agent 负责执行；中断暂停当前 thread 的执行链，通讯通过主 Agent 或共享存储完成。

对于中断授权：

• 子 Agent 可以通过interrupt_on或interrupt()暂停；
• 同一个thread_id中，一个子 Agent 中断会暂停当前主 run；
• 不会全局暂停其他 thread；
• 已发生的副作用不会自动回滚；
• 支付类动作必须做幂等、审计和权限校验。

对于 Agent 间通讯：

• 默认子 Agent 不直接互相通讯；
• 小数据走主 Agent 中转；
• 大数据走共享文件；
• 生产数据走数据库 / 对象存储 / 业务 API；
• 强流程走外层 LangGraph state。

最稳的生产架构是：

Graph 管流程 主 DeepAgent 管调度 子 DeepAgent 管复杂任务 数据库 / 文件系统 管数据交换 Human-in-the-loop 管高风险动作

参考资料

• Deep Agents Human-in-the-loop
• Deep Agents Subagents
• Deep Agents Async subagents
• Deep Agents Permissions
• Deep Agents Context engineering
• LangChain Human-in-the-loop
• LangGraph Interrupts
• LangGraph Persistence
• LangGraph Graph API