73-LangGraph多Agent状态机-Agent握手交接-并行汇聚与故障恢复

【73.Python+AI】用LangGraph实现多Agent状态机：Agent间的握手、交接与故障恢复

📖文章简介：本文讲解如何用LangGraph构建生产级的多Agent状态机。在第62篇单Agent状态图的基础上，本文聚焦多Agent场景特有的三个工程难题：Agent间的状态传递（一个Agent的产物如何干净地交给下一个）、并行执行与汇聚（多个Agent同时干活再合并结果）、以及故障恢复机制（某个Agent执行失败后如何重试/降级/人工接管）。通过一个"调研+写作+翻译"三Agent并行协作的完整实现，配上Mermaid有向图展示状态流转与汇聚节点设计，适合已经用过LangGraph单Agent流程、需要向多Agent编排进阶的开发者。

🎬 个人主页：源码骑士

❄专栏传送门：《Android开发基础》《python基础课程》

⭐️热衷从源码视角拆解技术底层原理，将复杂架构讲得通俗易懂

🎬 源码骑士的简介：
5年Android Framework系统开发经验，曾主导多项系统级性能优化专项
技术栈覆盖Android系统全链路（Binder/Handler/AMS/WMS/启动流程）及Java后端全家桶（Spring + MyBatis + Redis + Oracle）
累计产出原创技术文章100+篇，文章以流程图为特色，被读者评价为"看一篇胜过啃一周源码"

导入语

用LangGraph跑通一个单Agent审批流之后，你自然会想：能不能让两个Agent并行干活？比如一个Agent在调研资料的同时，另一个Agent在写报告框架，最后第三个Agent把两者合并？

想法很美好，实操全是坑：并行的结果怎么合并？一个Agent挂了整个流程怎么办？状态在Agent之间传来传去，怎么保证不丢字段？这篇文章就用一个完整的案例把多Agent状态机的三大难题一次解决。

1 ~> 多Agent状态机总览

1.1 三大工程难题

2 ~> State设计：多Agent共享的"交接单"

2.1 字段按Agent归属分区

fromtypingimportTypedDict,OptionalclassTeamState(TypedDict):# 公共输入topic:str# 调研Agent的产物research_notes:Optional[str]research_status:str# pending/done/failed# 写作Agent的产物draft:Optional[str]draft_status:str# 翻译Agent的产物translated:Optional[str]# 全局控制error:Optional[str]

设计要点：每个Agent只写自己的字段，读别人的字段。这比所有Agent共用一个result字段清晰得多——出了问题一眼就能看出是哪个Agent的产物缺失。

2.2 Agent节点：只负责自己的一亩三分地

defresearcher_node(state:TeamState)->dict:"""调研Agent：只更新自己的字段"""try:notes=search_and_summarize(state["topic"])return{"research_notes":notes,"research_status":"done"}exceptExceptionase:return{"research_status":"failed","error":str(e)}defwriter_node(state:TeamState)->dict:"""写作Agent：可以先写框架，不强依赖调研"""draft=llm.invoke(f"为主题'{state['topic']}'写一份报告初稿").contentreturn{"draft":draft,"draft_status":"done"}defmerge_node(state:TeamState)->dict:"""汇聚节点：把调研结果融入初稿"""ifstate["research_status"]=="done":final=llm.invoke(f"初稿：{state['draft']}\n调研资料：{state['research_notes']}\n"f"请用调研资料充实初稿").contentelse:# 调研失败的降级：直接用初稿final=state["draft"]return{"draft":final}

3 ~> 并行执行与汇聚

3.1 fan-out / fan-in 的实现

LangGraph中，多条出边指向不同节点就是并行分发，多条入边汇入同一节点就是汇聚：

fromlanggraph.graphimportStateGraph,END workflow=StateGraph(TeamState)workflow.add_node("dispatch",lambdas:{})# 分发节点workflow.add_node("researcher",researcher_node)workflow.add_node("writer",writer_node)workflow.add_node("merge",merge_node)workflow.add_node("translator",translator_node)workflow.set_entry_point("dispatch")# fan-out：dispatch同时触发两个并行分支workflow.add_edge("dispatch","researcher")workflow.add_edge("dispatch","writer")# fan-in：两个分支都完成后才进入mergeworkflow.add_edge("researcher","merge")workflow.add_edge("writer","merge")workflow.add_edge("merge","translator")workflow.add_edge("translator",END)app=workflow.compile()

关键认知：LangGraph的汇聚是自动的——merge节点会等所有指向它的上游分支都完成才执行。你不需要自己写计数器或锁。

3.2 并行vs串行的收益

串行：调研(8s)→ 写作(6s)→ 翻译(4s)=18秒 并行：max(调研8s, 写作6s)→ 汇聚(2s)→ 翻译(4s)=14秒 ↓ 节省22%

任务越独立、单个任务越耗时，并行收益越大。

4 ~> 故障恢复机制

4.1 三级降级策略

defresilient_researcher(state:TeamState,max_retry=2)->dict:forattemptinrange(max_retry+1):try:notes=search_and_summarize(state["topic"])return{"research_notes":notes,"research_status":"done"}exceptRateLimitError:ifattempt<max_retry:time.sleep(2**attempt)# 指数退避continuereturn{"research_status":"failed","error":"API限流，重试耗尽"}exceptExceptionase:return{"research_status":"failed","error":str(e)}

4.2 用条件边做降级路由

defroute_after_research(state:TeamState)->str:"""调研后决定走哪条路"""return"merge"# 无论成败都进merge，由merge内部决定降级策略# 更激进的方案：失败直接走fallback节点workflow.add_conditional_edges("researcher",lambdas:"merge"ifs["research_status"]=="done"else"fallback")

4.3 故障恢复决策表

5 ~> 多Agent状态机的调试

# 开启流式输出，观察每个节点的执行foreventinapp.stream({"topic":"异步框架对比","research_status":"pending","draft_status":"pending"}):fornode_name,outputinevent.items():print(f"[{node_name}] 输出字段:{list(output.keys())}")

多Agent流程的调试口诀：盯状态字段，不盯打印日志。每个节点只更新自己负责的字段，任何字段异常都能直接定位到责任Agent。

思考 && 总结

1. 多Agent状态机的核心是"分区状态"设计：每个Agent一个专属字段区，读公共区、写私有区——职责边界清晰了，协作才不会乱。
2. LangGraph的fan-in是自动汇聚：多条入边的节点天然等待所有上游完成，这是它比手写asyncio编排优雅的地方。
3. 故障恢复要分层：节点内重试（抗抖动）→ 条件边降级（抗单点失败）→ merge内兜底（保证最终有输出），三层缺一不可。
4. 并行的前提是任务独立：如果写作Agent必须等调研结果才能动笔，强行并行只会得到两个互相等待的节点。

从单Agent流程到多Agent状态机，本质是从"程序设计"走向"系统设计"——你考虑的不再只是逻辑对不对，而是分工、容错和协作效率。

结尾

各位小伙伴，本文的内容到这里就全部结束了，源码骑士在这里再次感谢您的阅读！

源码骑士 — Android Framework & 全栈开发

👀关注：跟博主一起从源码视角深耕底层原理，见证每一次成长

❤️点赞：让优质内容被更多人看见，让知识传递更有力量

⭐收藏：把核心知识点存好，在需要时随时查、随时用

💬评论：分享你的经验或疑问，评论区一起交流避坑

🔄一键四连：不要忘记给博主"一键四连"哦！

🗡️寄语：技术之路难免有困惑，但同行的人会让前进更有方向

结语：多Agent协作不是把Agent数量堆上去，而是把分工、交接、容错这三件事设计明白。画好状态图再写代码，你的多Agent系统就成功了一半。不要忘记给博主"一键四连"哦！