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

DeepAgent 多子代理协作中断授权与 Agent 间通讯机制本文整理 DeepAgent 在多子代理场景下的两个关键工程问题子 Agent 发起中断授权时主 Agent 和其他子 Agent 怎么执行多个子 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。但要注意agentcreate_deep_agent(...)返回的本身就是一个 LangGraph runnable。所以即使你没有手写外层业务 GraphDeepAgent 内部仍然有状态messagestool callcheckpointinterruptresumesubagent 调用。可以理解成你没有自己画 Graph但 create_deep_agent 已经帮你封装了一个执行图。3. 子 Agent 是怎么被调用的DeepAgent 的SubAgentMiddleware会给主 Agent 注入一个task工具。主 Agent 不是直接“进入”子 Agent而是调用task(subagent_typeresearcher, description去搜索某主题论文)执行流程主 Agent - 发起 task 工具调用 - SubAgentMiddleware 找到对应子 Agent - 子 Agent 使用独立 messages 执行 - 子 Agent 完成后返回最终消息 - 这个最终消息变成主 Agent 看到的 ToolMessage默认情况下子 Agent 的中间工具调用、搜索过程、内部 messages 不会全量回传给主 Agent。主 Agent 通常只拿到子 Agent 的最终报告 / 最终结构化结果4. 子 Agent 中断授权是怎么实现的DeepAgent 的中断授权主要有两种方式。4.1 用interrupt_on拦截工具调用这是最常见方式。例如支付工具fromlangchain.toolsimporttooltooldefcreate_payment(order_id:str,amount:float,payee:str)-str:执行支付动作该工具只应在人工审批通过后被调用。returnf支付成功订单{order_id}金额{amount}收款方{payee}配置子 Agentpayment_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.typesimportinterrupttooldefpay_with_manual_approval(order_id:str,amount:float,payee:str)-str:先请求人工支付授权审批通过后再执行支付。approvalinterrupt({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 个tasktask(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 / respond5.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},]}),configconfig,versionv2,)如果是并行分支里的多个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:金额异常},}),configconfig,)这样能避免多个并行中断靠顺序匹配导致混乱。7. 完整 Demo主 Agent 多个子 Agent 支付审批下面示例重点展示结构不绑定某个具体模型供应商。fromlangchain.toolsimporttoolfromlanggraph.checkpoint.memoryimportMemorySaverfromlanggraph.typesimportCommandfromdeepagentsimportcreate_deep_agenttooldefquery_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],}checkpointerMemorySaver()agentcreate_deep_agent(modelopenai:gpt-5.5,tools[],subagents[order_subagent,payment_subagent,audit_subagent],checkpointercheckpointer,system_prompt(你是主 Agent负责支付流程编排。\n规则\n1. 查询订单必须调用 order-agent。\n2. 支付必须调用 payment-agent。\n3. 审计日志必须调用 audit-agent。\n4. 你不得亲自执行支付。\n5. 支付前必须等待人工审批。),)config{configurable:{thread_id:pay-order-10001}}resultagent.invoke({messages:[{role:user,content:请处理订单 order-10001 的支付流程。,}]},configconfig,versionv2,)ifresult.interrupts:approval_resultagent.invoke(Command(resume{decisions:[{type:approve},]}),configconfig,versionv2,)关键点checkpointer必须有thread_id必须一致支付工具只给payment-agent主 Agent 不直接持有create_payment审批通过前create_payment不应该真正执行支付工具内部仍要做幂等和权限校验。8. 子 Agent 之间怎么通讯默认情况下子 Agent 之间不直接通讯。不要想象成researcher - critic - writer更像researcher - 主 Agent - critic - 主 Agent - writer9. 通讯方式一主 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.toolsimporttooltooldefsave_agent_result(task_id:str,role:str,data:dict)-str:保存子代理结果供主 Agent 或其他子 Agent 后续读取。# 生产环境中可写入数据库、Redis、对象存储或企业内部 APIreturnfsaved:{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 忘记配置 checkpointerHuman-in-the-loop 需要 checkpoint。没有 checkpointer就不能可靠恢复到中断点。15.4 resume 时换了 thread_id恢复必须使用同一个thread_id。否则运行时找不到之前暂停的状态。15.5 compiled 子 Agent 没配自己的 interrupt声明式 subagent 默认可以继承主 Agent 的interrupt_on。但如果你传的是已经编译好的 child runnable例如child_agentcreate_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.json16. 选择建议16.1 数据怎么传数据规模推荐方式很小主 Agent 直接中转中等结构化数据主 Agent 中转 JSON大文本 / 大文件共享文件系统生产业务数据数据库 / 对象存储 / 业务 API强流程、多阶段审批外层 LangGraph state16.2 中断怎么做场景推荐方式普通工具审批interrupt_on文件读写审批permissions的modeinterrupt支付 / 转账 / 复杂审批工具内部interrupt() 业务校验多个并行中断interrupt id 映射恢复远程异步子 Agent在远程子 Agent 自己配置审批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-loopDeep Agents SubagentsDeep Agents Async subagentsDeep Agents PermissionsDeep Agents Context engineeringLangChain Human-in-the-loopLangGraph InterruptsLangGraph PersistenceLangGraph Graph API