资讯动态

LangGraph多智能体网络架构:从监督者到动态路由的进阶实践

发布时间:2026/10/21 8:18:31 来源:尧图企业网站定制
做了几年的智能体开发陆陆续续把单智能体、顺序编排、监督者模式都落地到生产环境之后你会发现一个很尴尬的瓶颈当任务本身需要多个角色来回协商、互相移交、甚至彼此辩论时传统的 Supervisor监督者模式会显得特别死板。监督者相当于一个中央调度员所有消息都要先汇到它那里再分发出去一旦这个调度员判断错了方向整个流程就卡住了。于是就有了这个系列的第四篇也是我个人觉得最接近“真正智能体协作”的一种设计思路——LangGraph 多智能体网络架构。网络架构说白了就是让多个智能体不再通过唯一的中央协调者沟通而是直接基于图的拓扑关系自由连接、动态路由、相互调用。每个智能体都是图中一个节点节点与节点之间可以形成网状连接消息走到哪个节点、接下来该交给谁都由当前节点的推理结果动态决定。它解决的是监督者模式中“协调者单点决策”的瓶颈适合那些角色分工明确、但又需要频繁横向协同的任务场景比如售后工单流转、多人协作的创意策划、多角色谈判模拟等。这个系列前面几篇讲过的内容如果你还没有接触过建议先快速过一遍单智能体、顺序流和监督者模式因为这些是组件基础。这篇文章会从网络架构的设计动机开始讲逐步拆解它在 LangGraph 中的核心机制最后用一个完整的电商客服多智能体团队示例把实现细节、路由逻辑、状态收敛和容错机制全部串起来。读完后你应该能独立搭建一个具备动态路由能力的多智能体网络系统。1. 多智能体网络架构的设计思路拆解1.1 为什么说监督者模式不够用了先回顾一下监督者模式的典型结构一个顶层的 Supervisor 智能体负责接收用户请求、规划任务、分发给各个 Worker 智能体然后接收 Worker 的返回结果决定下一步动作。这种模式在任务边界清晰、流程相对固定时非常好用比如“先查库存再计算物流最后生成报价单”。但一旦任务呈现出强交互、强反馈、结果需要多方博弈才能确定的特征比如多个智能体分别扮演销售、技术、售后角色去处理一个复杂客户投诉监督者模式的问题就暴露了。监督者必须理解所有子任务的领域这让它自身成为智力瓶颈。所有消息都经过监督者中转延迟高上下文容易被截断。如果监督者一次调度的判断错了后续环节很难自行纠偏整个流程容易走入死胡同。网络架构的思路恰好绕开这些问题。它不设唯一的中央调度员而是让每个智能体个体之间自己协商、自己传递控制权。理论上各个节点知道自己的能力边界也知道“当遇到某种情况时应该找谁”这样就形成了一种去中心化的协作网络。1.2 网络架构的几种典型拓扑形态有人会把网络架构等同于“全互联广播”这是误区。真正的网络架构在 LangGraph 里依然是一个有向图只是边的连接方式、路由策略更加灵活。我通常会把网络拓扑归纳为这么几类拓扑类型结构特征适用场景优点风险全互联 Mesh每个节点都与其它节点连接小规模3~5个智能体角色协商路由最灵活任意节点间直达连线爆炸状态收敛困难星型 Hub一个中心节点与多个子节点直连子节点间不直连信息汇聚、共享内存场景状态集中易于监控中心失效则整体不可用层级网络网络内部嵌入 Supervisor 子图子图间再由动态边连接中大规模团队、分组协作兼顾纵向管理与横向协同结构复杂调试难度陡增动态自治集群节点间边不预定义通过共享消息总线自行寻找下一跳开放复杂决策灵活度最高大概率出现死循环、消息丢失在实际生产中我见过不少团队把 Mesh 架构当成“银弹”结果一上规模就疯狂烧 token智能体们来回踢皮球。真正能落地的网络架构通常是 Mesh 和 Hub 的混合体核心决策节点之间用 Mesh外围执行节点以 Hub 方式挂载到某个核心节点下。1.3 网络架构的执行模型不是并行是动态路由第一次听说“网络架构”的人会直觉以为所有智能体同时并行工作这其实是误解。LangGraph 的执行模型从来都是图执行器按拓扑顺序调度节点网络模式也不例外。它的本质区别在于普通图模式的边是静态写死的网络模式的边是运行时由智能体自身的输出动态决定的。打个比方传统工作流像火车轨道铺在哪火车就走哪网络架构更像出租车乘客当前智能体决定下一站开去哪。执行顺序依然是一条时间线上的串行过程但路径本身不是预先固定的。这就是为什么 LangGraph 特别适合做网络架构。它对动态边、条件路由、状态叠加有天然支持不需要像传统 Agent 框架那样自己造一套消息路由轮子。2. 网络架构的核心机制与设计细节2.1 动态路由Command 与 send API网络架构的生命线是“节点之间如何传递控制权”。在 LangGraph 中主要有两把武器Command 和 send。Command 更多用于“单跳转移控制权”。一个节点执行完自己的逻辑可以在返回值里指定“接下来谁接管整个对话”。这种模式下接收节点能看到全部上游共享状态不需要额外拼接上下文。自定义节点里可以这样写from langgraph.types import Command def sales_agent(state): # 当前销售智能体判断这个客户需要技术团队介入 result llm.chat(state[messages]) if 需要技术支持 in result: return Command( update{messages: [AIMessage(contentresult)]}, graphtechnical_agent ) # 否则自己继续服务 return Command( update{messages: [AIMessage(contentresult)]}, graphsales_agent )而 send 则是“扇出”的操作适合网络架构中的并行分支。比如网络中有多个候选专家当前节点希望同时把小组件任务发给多个节点去各算各的最后再汇总。send 的作用是给目标节点分别提交独立的状态副本去执行。from langgraph.constants import Send def route_to_experts(state): candidates state[candidates] return [ Send(expert_a, {query: state[query], specialty: 前端}), Send(expert_b, {query: state[query], specialty: 后端}), Send(expert_c, {query: state[query], specialty: 数据库}), ]这里有个容易搞混的细节send 传的是传给子任务的独立状态片段不是全局共享状态。如果子任务结果需要合并回主流程通常需要你再定义一个聚合节点把这些子任务发送回的状态统一做 reduce 或 merge。2.2 共享状态与消息收敛网络架构的命门网络架构最大的坑在于状态发散。因为每个节点都会向公共状态写入新内容如果设计不好 reducer整个 state 会像一团乱麻一样膨胀。LangGraph 的 State 定义里可以使用 Annotated 类型加上 reducer 函数。常见的做法是给 messages 字段加 add_messages reducer让它对历史消息做自动追加给某些数值字段加统计型 reducer给自定义字段加“覆盖”型 reducer。网络架构下我强烈建议消息历史类字段全部走 add_messages避免手动拼接列表。from typing import Annotated, TypedDict from langgraph.graph.message import add_messages from operator import add class NetState(TypedDict): messages: Annotated[list, add_messages] active_agents: Annotated[set, add_routes] turn_count: int final_answer: str消息收敛是网络架构的另一个命门。节点之间互相转移控制权本质上是在消耗上下文长度和 token而明明有些问题只需要两三个节点来回就可以结束。我的经验是在网络层加一个“仲裁节点”或“汇总节点”来做收敛。比如最大轮次设为 6 轮超过之后所有节点不再转移而是把当前 messages 扔给一个专门的 Finalizer 智能体由它统一生成最终答案。这个 Finalizer 的设计非常重要它相当于跑团游戏里的主持人平时不掺和玩家的自由行动但玩家拖延太久时负责推进度。用 LangGraph 实现就是在图的顶层条件边里判断 turn_count。2.3 广播、定向与汇集三种通信范式网络架构中的节点间通信可以总结为三种范式对应不同场景。广播式通信节点把消息发到公共通道public channel所有其它节点可见。LangGraph 中最接近的做法是更新共享 messages 列表但由于执行是串行的广播效率并不高只适合小规模团队。定向式通信节点在 Command 里明确指定下一个处理节点。这是网络模式中使用最频繁的方式也是实现动态路由的基石。定向式通信的每个节点必须知道“自己能力边界之外的事情应该找谁”所以每个节点的 system prompt 里需要写一份“团队通讯录”。汇集式通信多个节点都把处理结果写到同一个共享汇总字段里由一个最终的 Arranger 节点从头到尾阅读这些记录再做综合决策。它通常配合 send 扇出一起使用。比如多个专家节点各自返回专项分析最后的汇总节点拿到所有结果后输出集成结论。我个人在实现网络架构时推荐优先把核心业务流转用定向式通信把可并行计算的部分用汇集式通信广播模式能不用就不用。全互联广播会带来非常高的 token 消耗和多轮重复讨论这对生产环境来说成本极高。2.4 容错与恢复网络架构如何防止雪崩网络架构比其它模式更怕节点异常。想象一个场景A 节点把控制权转给 BB 节点在推理时发生异常那整个图就可能卡死在 B 上或者 C 节点返回了一个根本不存在的目标节点名那图执行器会直接报错退出。应对这种问题我总结了一套组合拳第一网络里的每条动态边都要有“默认兜底”。写条件路由时不要只在某个条件下转移而是要写一个 else 分支让无法匹配的情况回归到某个安全节点。第二给节点加上超时机制。LangGraph 里可以为节点指定 timeout超时会触发失败回调再从检查点恢复。node_builder create_agent(llm, tools[...], timeout30)第三坚持使用 checkpointer。网络架构状态复杂一旦某个节点出了不可恢复的错误不要自己写状态恢复逻辑直接回滚到上一个稳定的 checkpoint 重放。第四设计幂等节点。网络架构里同一个消息可能被重放、重试如果节点业务有副作用比如写数据库、发通知一定要做幂等约束。3. 实操过程用网络架构构建一个电商客服团队3.1 需求确定与拓扑选型我这里用一个很多场景都会遇到的例子——构建一个电商售后与咨询客服团队。团队成员包括四个智能体售前咨询顾问负责商品信息、优惠活动、推荐选品。订单专家负责查询订单、修改地址、催发货。售后专员负责退换货、投诉处理。退款仲裁人当售后无法在线解决或者客户情绪激烈时由它做最终裁决。这四个角色之间天然存在横向协同。比如客户先咨询售前聊着聊着订单遇到问题了售前应该直接把会话转给订单专家。订单专家发现商品需要走售后流程又转给售后专员。售后专员处理不了的高危投诉再转给退款仲裁人。这个链路如果用监督者模式每一步都要经过 Supervisor 中转不仅慢而且容易丢上下文。用网络架构直接节点间跳转干净利落。拓扑选型上我采用的是 MeshHub 混合售前、订单、售后、仲裁四个核心节点之间相互可达Mesh同时每个节点各自携带若干工具Tool工具层相当于 Hub 形态。3.2 定义共享状态与节点实现先定义整个客服团队的共享状态结构。核心思路是消息全量共享专用字段按节点覆盖。from typing import Annotated, TypedDict from langgraph.graph import add_messages from langchain_core.messages import AnyMessage class CustomerServiceState(TypedDict): messages: Annotated[list[AnyMessage], add_messages] current_intent: str # 当前用户意图标签 order_id: str # 订单上下文 escalate: bool # 是否升级 turn_count: int # 轮次计数器 final_response: str节点实现上我建议每个节点尽量保持简单职责集中。每个节点做的事情就是三件看当前共享状态决定自己有没有能力处理如果有就生成回复如果没有就指明下一个节点。from langchain_core.messages import AIMessage from langgraph.types import Command def sales_node(state: CustomerServiceState): messages state[messages] resp llm.chat(messages, system你是售前顾问只负责咨询、推荐、活动解答。) if 地址修改 in resp or 订单 in resp and 咨询 not in resp: return Command( update{messages: [AIMessage(contentresp)], current_intent: order}, graphorder_node ) return Command( update{messages: [AIMessage(contentresp)], current_intent: sales}, graphsales_node )这里的 llm.chat 是我在实际项目中封装的一个底层模型调用方法你完全可以替换成自己项目的模型客户端或 LangChain 的任何模型封装类核心不在这而在于 Command 的路由逻辑。同理实现 order_node、after_sales_node、arbitration_node。仲裁节点比较特殊它是网络里的“终点节点”处理完之后必须返回 END让整个图结束。def arbitration_node(state: CustomerServiceState): resp llm.chat(state[messages], system你是最终仲裁人给出不可驳回的最终结论。) return Command( update{messages: [AIMessage(contentresp)], final_response: resp}, graphEND )3.3 图组装与条件路由编排节点的网络关系需要在一个 StateGraph 里最终落实。为了避免节点之间乱跳导致无限循环我会在图构建层面再加一道“轮次守卫”的条件判断。from langgraph.graph import StateGraph, START, END def should_continue(state: CustomerServiceState): if state[turn_count] 8 or state.get(final_response): return finalizer return router def router(state: CustomerServiceState): intent state.get(current_intent) if intent order: return order_node if intent after_sales: return after_sales_node if intent arbitration: return arbitration_node return sales_node graph StateGraph(CustomerServiceState) graph.add_node(sales_node, sales_node) graph.add_node(order_node, order_node) graph.add_node(after_sales_node, after_sales_node) graph.add_node(arbitration_node, arbitration_node) graph.add_node(finalizer, finalizer_node) graph.add_edge(START, sales_node) graph.add_conditional_edges(sales_node, should_continue, {router: router, finalizer: finalizer}) graph.add_conditional_edges(order_node, should_continue, {router: router, finalizer: finalizer}) graph.add_conditional_edges(after_sales_node, should_continue, {router: router, finalizer: finalizer}) graph.add_conditional_edges(arbitration_node, lambda s: finalizer, {finalizer: finalizer}) graph.add_conditional_edges(router, router, {...})这里我刻意把路由判断拆成了两层第一层判断对话轮次是否耗尽should_continue第二层判断具体去向哪个节点router。这就是网络架构的安全带没有这条安全带网络很容易进入死循环。从字面上看节点之间似乎是在自由互联但最外层仍然有一个“监考老师”时刻注意着轮次上限。3.4 网络架构的人机协作与中断介入如果你做过真实业务系统你会知道智能体团队不可能完全无人值守。我强烈建议在网络架构里留一个人工兜底入口。LangGraph 的 interrupt 机制可以用来暂停图的执行等待人工输入后继续。看这个思路当仲裁节点判断客户情绪激烈无法用固定话术解决时它可以触发 interrupt() 挂起整个网络同时由人工客服介入处理。from langgraph.types import interrupt def arbitration_node(state: CustomerServiceState): if state.get(escalate): # 触发人工介入等待外部传回决策 human_note interrupt({query: state[messages]}) return Command( update{messages: [AIMessage(contentf人工处理意见{human_note})]}, graphEND ) ...这个能力在生产环境有多重要只有真正做过智能客服系统的人才有体会。Auto 模式在可控范围内让智能体自行解决一旦检测到高风险、高投诉倾向立即挂起等待人工决策。这种“自动为主、人工兜底”的混合模式也是网络架构落地最常见的最佳实践。4. 常见问题与排查技巧实录4.1 消息不收敛智能体之间反复横跳这是我做网络架构踩过最多的坑没有之一。两个节点互相觉得“应该由对方来处理”结果在 LangGraph 里出现 A - B - A - B 的无限循环。第一次踩这个坑时token 消耗几乎失控日志看得人头皮发麻。排查思路先关闭动态路由把整个网络图转换成只有固定边观察问题是否消失。如果消失基本可以确定是路由策略里的条件判断存在模糊地带。解决方法在 system prompt 里明确每个节点的“能力边界”和“不可处理的场景”减少踢皮球现象。对每个节点路由条件加白名单约束只有满足明确关键词/意图才做转移。加轮次上限 finalizer宁可机器人生硬收尾也不让它死循环烧钱。4.2 状态膨胀上下文越来越长网络架构会把每个节点的输出都写入共享消息列表。聊到第 10 轮、第 20 轮时messages 可能会累积几千个 token然后模型输入越来越慢推理越来越贵。我最推荐的办法是“结构裁剪加摘要压缩”在轮次达到一定数量后触发压缩节点把前面的历史对话交给一个 LLM 生成摘要然后用摘要代替历史消息同时保留最后两轮完整消息。def compress_state(state: CustomerServiceState): summary llm.chat(请压缩以下对话为要点..., state[messages][:-4]) return {messages: [AIMessage(contentf[已压缩历史] {summary})] state[messages][-4:]}这个压缩节点可以像 finalizer 一样挂在 should_continue 的分支里在每轮结束前检查 token 数量超标就调用。4.3 send 扇出后的状态同步陷阱网络架构中我经常用 send 把一个问题同时并发交给多个专家节点处理这时候很容易踩的一个坑是子图的输出不会自动合并回父图状态。你期待的是三个专家答案汇聚到主流程但实际上主流程状态里啥也没有加进去。解决办法是在 State 的某个汇总字段上加 reducer或者定义一个集合节点把所有 send 出来的子结果收集起来再做一次汇总from operator import add class ExpertState(TypedDict): expert_results: Annotated[list, add]子节点返回时必须显式执行类似下面的操作return {expert_results: [{expert: expert_a, answer: resp}]}只有显式写入带 reducer 的字段主图才能正确收集到结果。这个细节官方文档只给了很短的说明但实际踩坑的人非常多。4.4 调试与可视化日志里看不到路由路径怎么办网络架构的调试难度远高于其它模式。普通静态图可以一眼看出路径动态路由你却不知道哪个节点的哪一步推理触发了转移。我的建议是给每个节点加一层 trace 包装在节点入口和出口输出结构化日志。def traced_node(node_name, node_func): def wrapper(state): print(f[{node_name}] 收到 state, intent{state.get(current_intent)}) result node_func(state) print(f[{node_name}] 返回 Command, 下一跳{result.get(graph)}) return result return wrapper生产环境不要用 print换成标准 logging 或对接可观测性系统比如 LangSmith 这类链路追踪。这点在 LangGraph 项目里极其重要因为你不可能靠肉眼从执行引擎的底层状态栈里找出路由路径。5. 网络架构设计的三条经验心得做了好几个 LangGraph 网络架构项目后我总结出三条自己的经验写在这里供你参考。第一个心得是“先分组再互联”。不要一上来就画 All-to-all 全互联图那是灾难的起点。先按业务角色把智能体分成几个小组每个小组内部用小规模网络互联小组之间通过组代表或汇总节点交互。这样既保留了横向协同能力又把单点之间互相乱跳的几率降到最低。第二个心得是“不要贪多能静则静”。网络架构里每个节点的 system prompt 都要明确写好“下一步路由规则”我甚至会在 prompt 里加一句“如果你不是该任务的最终负责人请明确指定最合适的下一个团队角色不要与当前角色反复拉扯”。看起来普通的一句话实际能减少三分之一的无效循环。第三个心得是“永远保留一块责任田”。不管网络多么自由业务责任最终必须落脚到具体节点上。比如这个客服例子里仲裁节点就是最终责任田。没有终点的网络是没办法上线运行的。责任田设计得好即使是全互联网络模式也能稳定收敛输出可靠结果。最后一个实际体会网上关于 LangGraph 的资料越来越多了但真正讲透网络模式的还是不多。如果你已经能熟练使用 Supervisor 模式我建议大胆入手本篇文章的网络架构模式这是多智能体系统从“编流程”走向“组件协作”的分水岭。在设计阶段多画图、多推演路由别急着写代码。对于网络架构来说设计图的合理性决定系统成败代码只是把设计图如实地翻译出来而已。

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价 →
↑