文章目录【73.PythonAI】用LangGraph实现多Agent状态机Agent间的握手、交接与故障恢复导入语1 ~ 多Agent状态机总览1.1 三大工程难题2 ~ State设计多Agent共享的交接单2.1 字段按Agent归属分区2.2 Agent节点只负责自己的一亩三分地3 ~ 并行执行与汇聚3.1 fan-out / fan-in 的实现3.2 并行vs串行的收益4 ~ 故障恢复机制4.1 三级降级策略4.2 用条件边做降级路由4.3 故障恢复决策表5 ~ 多Agent状态机的调试思考 总结结尾【73.PythonAI】用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状态机总览失败重试成功重试失败开始任务分发节点调研Agent写作Agent汇聚节点等待两个分支完成翻译Agent结束重试/降级降级路径跳过调研直接写1.1 三大工程难题难题本质问题本文方案状态传递下游Agent需要上游的哪些字段显式State schema 增量更新并行汇聚并行分支的结果怎么合并LangGraph的fan-out/fan-in故障恢复单点失败拖垮整个流程try-except包装 条件边降级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:notessearch_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可以先写框架不强依赖调研draftllm.invoke(f为主题{state[topic]}写一份报告初稿).contentreturn{draft:draft,draft_status:done}defmerge_node(state:TeamState)-dict:汇聚节点把调研结果融入初稿ifstate[research_status]done:finalllm.invoke(f初稿{state[draft]}\n调研资料{state[research_notes]}\nf请用调研资料充实初稿).contentelse:# 调研失败的降级直接用初稿finalstate[draft]return{draft:final}3 ~ 并行执行与汇聚3.1 fan-out / fan-in 的实现LangGraph中多条出边指向不同节点就是并行分发多条入边汇入同一节点就是汇聚fromlanggraph.graphimportStateGraph,END workflowStateGraph(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-outdispatch同时触发两个并行分支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)appworkflow.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_retry2)-dict:forattemptinrange(max_retry1):try:notessearch_and_summarize(state[topic])return{research_notes:notes,research_status:done}exceptRateLimitError:ifattemptmax_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:调研后决定走哪条路returnmerge# 无论成败都进merge由merge内部决定降级策略# 更激进的方案失败直接走fallback节点workflow.add_conditional_edges(researcher,lambdas:mergeifs[research_status]doneelsefallback)4.3 故障恢复决策表故障类型策略实现API限流/超时重试指数退避最多2~3次工具不可用降级跳过该Agent下游用默认输入LLM输出格式错误重试兜底重新生成仍失败则返回占位文本状态字段缺失校验merge节点检查上游status字段5 ~ 多Agent状态机的调试# 开启流式输出观察每个节点的执行foreventinapp.stream({topic:异步框架对比,research_status:pending,draft_status:pending}):fornode_name,outputinevent.items():print(f[{node_name}] 输出字段:{list(output.keys())})多Agent流程的调试口诀盯状态字段不盯打印日志。每个节点只更新自己负责的字段任何字段异常都能直接定位到责任Agent。思考 总结多Agent状态机的核心是分区状态设计每个Agent一个专属字段区读公共区、写私有区——职责边界清晰了协作才不会乱。LangGraph的fan-in是自动汇聚多条入边的节点天然等待所有上游完成这是它比手写asyncio编排优雅的地方。故障恢复要分层节点内重试抗抖动→ 条件边降级抗单点失败→ merge内兜底保证最终有输出三层缺一不可。并行的前提是任务独立如果写作Agent必须等调研结果才能动笔强行并行只会得到两个互相等待的节点。从单Agent流程到多Agent状态机本质是从程序设计走向系统设计——你考虑的不再只是逻辑对不对而是分工、容错和协作效率。结尾各位小伙伴本文的内容到这里就全部结束了源码骑士在这里再次感谢您的阅读源码骑士 — Android Framework 全栈开发关注跟博主一起从源码视角深耕底层原理见证每一次成长❤️点赞让优质内容被更多人看见让知识传递更有力量⭐收藏把核心知识点存好在需要时随时查、随时用评论分享你的经验或疑问评论区一起交流避坑一键四连不要忘记给博主一键四连哦️寄语技术之路难免有困惑但同行的人会让前进更有方向结语多Agent协作不是把Agent数量堆上去而是把分工、交接、容错这三件事设计明白。画好状态图再写代码你的多Agent系统就成功了一半。不要忘记给博主一键四连哦
