做了几年的智能体开发,陆陆续续把单智能体、顺序编排、监督者模式都落地到生产环境之后,你会发现一个很尴尬的瓶颈:当任务本身需要多个角色来回协商、互相移交、甚至彼此辩论时,传统的 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(content=result)]}, graph="technical_agent" ) # 否则自己继续服务 return Command( update={"messages": [AIMessage(content=result)]}, graph="sales_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 节点把控制权转给 B,B 节点在推理时发生异常,那整个图就可能卡死在 B 上;或者 C 节点返回了一个根本不存在的目标节点名,那图执行器会直接报错退出。
应对这种问题,我总结了一套组合拳:
第一,网络里的每条动态边都要有“默认兜底”。写条件路由时不要只在某个条件下转移,而是要写一个 else 分支,让无法匹配的情况回归到某个安全节点。
第二,给节点加上超时机制。LangGraph 里可以为节点指定 timeout,超时会触发失败回调,再从检查点恢复。
node_builder = create_agent(llm, tools=[...], timeout=30)第三,坚持使用 checkpointer。网络架构状态复杂,一旦某个节点出了不可恢复的错误,不要自己写状态恢复逻辑,直接回滚到上一个稳定的 checkpoint 重放。
第四,设计幂等节点。网络架构里同一个消息可能被重放、重试,如果节点业务有副作用(比如写数据库、发通知),一定要做幂等约束。
3. 实操过程:用网络架构构建一个电商客服团队
3.1 需求确定与拓扑选型
我这里用一个很多场景都会遇到的例子——构建一个电商售后与咨询客服团队。团队成员包括四个智能体:
- 售前咨询顾问:负责商品信息、优惠活动、推荐选品。
- 订单专家:负责查询订单、修改地址、催发货。
- 售后专员:负责退换货、投诉处理。
- 退款仲裁人:当售后无法在线解决,或者客户情绪激烈时,由它做最终裁决。
这四个角色之间天然存在横向协同。比如客户先咨询售前,聊着聊着订单遇到问题了,售前应该直接把会话转给订单专家。订单专家发现商品需要走售后流程,又转给售后专员。售后专员处理不了的高危投诉,再转给退款仲裁人。这个链路如果用监督者模式,每一步都要经过 Supervisor 中转,不仅慢,而且容易丢上下文。用网络架构直接节点间跳转,干净利落。
拓扑选型上,我采用的是 Mesh+Hub 混合:售前、订单、售后、仲裁四个核心节点之间相互可达(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(content=resp)], "current_intent": "order"}, graph="order_node" ) return Command( update={"messages": [AIMessage(content=resp)], "current_intent": "sales"}, graph="sales_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(content=resp)], "final_response": resp}, graph=END )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(content=f"人工处理意见:{human_note}")]}, graph=END ) ...这个能力在生产环境有多重要,只有真正做过智能客服系统的人才有体会。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(content=f"[已压缩历史] {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 模式,我建议大胆入手本篇文章的网络架构模式,这是多智能体系统从“编流程”走向“组件协作”的分水岭。在设计阶段多画图、多推演路由,别急着写代码。对于网络架构来说,设计图的合理性决定系统成败,代码只是把设计图如实地翻译出来而已。