LangGraph 做 Multi-Agent 编排最近一段时间我几乎每天都在跟它打交道。从最初在 LangChain 里用链式调用硬凑多步逻辑到后来切到 LangGraph 的状态图模型最大的感受是Multi-Agent 系统的复杂度不在单个 Agent 的能力而在多个 Agent 之间的协作与状态流转。LangGraph 刚好把这块最难啃的骨头变成了可声明、可控制、可调试的图结构。这篇文章我就用自己实际做的一个客服工单处理系统当案例把 LangGraph 里 Multi-Agent 从架构设计到落地实现的关键细节完整拆一遍。1. 为什么用 LangGraph 做多智能体编排1.1 从 LangChain 链式调用到状态图的本质变化早先我用 LangChain 处理多步骤任务常规做法是Chain | Chain | Chain数据沿着一个固定管道单向流动。这个模式解决单线任务没问题可一旦出现分支判断、循环重试、多 Agent 间共享上下文链式结构就变得非常别扭。你要么在外面包一层大循环要么得自己在代码里维护状态机甚至需要在 Prompt 里塞大量历史记录来模拟记忆。调试的时候更痛苦——执行流程不透明每一步到底发生了什么全靠日志打印靠猜。LangGraph 带来的核心变化是把执行流程从线性管道改成显式的状态图。每个节点是一个处理单元可以是普通函数也可以是带 LLM 调用的 Agent 逻辑节点之间通过边连接边可以带条件判断。执行时一个共享的State对象在节点之间传递和更新任何节点都能读取和写入全局状态。这意味着多个 Agent 之间天然共享工作记忆不再需要自己维护跨 Agent 的上下文容器。这种设计带来一个非常实用的副产品你可以让多个 Agent 并行处理同一份状态的不同部分。比如我那个客服系统里用户提交一个工单后检索 Agent 去查历史知识库质检 Agent 去评估工单完整性这两个节点之间没有依赖关系在 LangGraph 里用一条无条件的边就能让它们并行执行。这在 LangChain 的链式模型里要做到得额外引入并发控制逻辑非常繁琐。1.2 三个核心抽象State、Node、EdgeLangGraph 的体系里最核心的抽象就三个State、Node、Edge。State 是整个图的数据中枢它决定了一个 Agent 能读到什么、能写出什么。LangGraph 对 State 有一个很有意思的设计叫 Reducer。默认情况下节点返回的字典会直接覆盖 State 中同名字段但如果你用add_messages这种 Reducer更新字段时会把新内容追加到旧内容的末尾而不是覆盖。这个特性是给对话历史用的——每个 Agent 产生的消息都应该被追加保存而不是互相顶撞。自定义 Reducer 也很简单就是写一个 Python 函数指定Annotated[字段类型, reducer函数]即可。我见过不少朋友一开始都在这里踩坑定义的 State 没有给字段加上 Reducer结果第二个 Agent 写入消息时把第一个 Agent 的回复覆盖掉了整个对话记录只剩最后一步。排查半天还以为是 Prompt 写错了。所以第一节课就是理解 State 的更新语义需要累积、追加的字段如messages、scratchpad、tool_results必须配 Reducer需要快照、覆盖的字段如user_intent、final_answer保持默认行为就好。Node 就是图里的一个执行单元典型的 Node 函数签名是(state: AgentState) - dict。函数接收当前的 State内部可以调用 LLM、执行工具、查数据库最后返回一个字典告诉 LangGraph 要更新 State 的哪些字段。这个设计非常贴近函数式编程的思路把每个 Agent 的输入输出契约刀切得明明白白。后面我会给一个实际的节点代码到时候你会看到一个 Agent 节点内部其实也就是调 Prompt → 拿结果 → 解析成结构化输出 → 写入 State这四个动作。Edge 是图里的路径定义执行顺序。LangGraph 支持三种边普通边总是走这条路、条件边根据当前 State 的内容决定走哪条路、以及 START 和 END 两个特殊节点。条件边是 Multi-Agent 系统的灵魂。比如路由节点判断这个问题需不需要人工介入返回human_intervention则走人工节点返回auto_reply则走自动应答节点。这个判断能力直接决定了一个 Agent 系统是死板的流水线还是会思考的工作流。在动手写代码之前我强烈建议先在白板上画图有哪些 Agent、它们之间什么关系、什么条件下流转。我在做客服系统的时候光画这个图就画了三版。第一版想堆一个大一统 Agent被否了第二版做成了串行流水线太慢第三版才最终定成路由 并行 条件汇合的结构效率最理想。图的设计直接决定系统上线后的可维护性这一步别省。2. 多智能体架构的核心形态与选型思路2.1 三种主流架构对比LangGraph 本身不限定 Multi-Agent 的组织方式但在真实项目中我看到的架构基本能归纳成三种主流形态。第一种是 Supervisor主管-工人模式。图上有一个 supervisor 节点它不直接干活只负责看情况派活。supervisor 通过 LLM 判断当前任务应该交给哪个 worker然后通过条件边把控制流转到指定节点worker 处理完把结果写回状态控制权再交还 supervisor由 supervisor 决定下一步。这个模式的优点是分工清晰、扩展性好加一个新 worker 只需在 supervisor 的工具列表里多注册一个缺点是 supervisor 的 LLM 调用会成瓶颈而且如果 supervisor 判断不准整个系统的天花板就卡在那里。第二种是 Network网络模式。所有 Agent 两两之间都可以通信没有中心调度者。每个 Agent 有自己的上下文窗口可以主动向其他 Agent 发起协作请求。这种模式最灵活也最难控制——LangGraph 虽然支持用SendAPI 实现动态路由和并行分发但在一个全互联的图里Agent 之间消息互相传递很容易出现重复处理或循环依赖。我目前只在小规模实验里用这种模式生产环境不太敢上。第三种是 Hierarchical层级模式。在 supervisor 之上再套 supervisor形成树状的三级、四级调度结构。这种模式适合超大规模团队协作场景比如一个业务系统里有 20 个以上的 Agent单层主管管不过来。LangGraph 对这种模式支持得也很好因为图之间可以互相嵌套——一个图的节点可以是另一个图的入口。层级多了之后每个子图都是独立的 State 隔离空间调试复杂度会指数上升不建议一开始就上。三种模式的适用场景差异很大我用一张表总结一下我在项目里实际选型的参考维度架构模式适用规模调度风格扩展性可控性调试难度Supervisor2-10 个 Agent中心化好高低Network3-6 个 Agent去中心化差低高Hierarchical10 个 Agent多层中心化最好中最高2.2 客服工单系统为何选择 Supervisor我这个案例是给某企业内部做的客服工单自动处理系统需求一句话概括用户提交工单系统自动判断工单类型能查知识库的回答直接答需要内部操作的走指定流程模棱两可的转人工。这个场景天然适合 Supervisor 模式原因有三。第一工单处理的专业分工很明确——检索知识库、查用户历史订单、质检合规审查这三个职责拆开成独立 Agent每个 Agent 的 Prompt 只需要聚焦自己的领域模型判断质量明显高于一个大全量 Prompt。第二流程变化频繁业务说我们这季度想加一个退款审核 AgentSupervisor 模式加一个 worker 即可完全不用动其他逻辑。第三工单处理需要完全可审计——从用户提交到最后答复每一步是哪个 Agent 处理的、依据了什么数据都必须能回溯。Supervisor 模式的中心化调度天然支持在 State 里留痕审计就好做了。确定 Supervisor 模式之后我的 Agent 划分如下Router Agent负责给工单打标签判断走哪个处理分支。Retrieve Agent负责检索企业知识库找相似历史工单和标准流程文档。Compliance Agent负责质检回复检查合规风险。Coordinator Agent作为 Supervisor 主控调度上面三个 worker并负责对用户最终回复的裁决。这套体系跑起来之后最明显的好处是 Prompt 的可维护性大大提升。以前一个一大坨的 Prompt 要同时处理判断意图 检索 质检 回复生成现在每个 Agent 只需要在自己的专业维度上做到极致单个 Prompt 的长度大幅缩减出问题的概率自然下降。接下来我拿这个案例逐步讲实现。3. 客服工单系统的 Multi-Agent 完整实现3.1 状态设计与节点函数实现第一步是定义 State。我自定义了一个AgentState的 TypedDict核心字段包括messages完整的对话历史用add_messagesReducer 追加。ticket当前工单数据包括工单号、用户描述、时间戳。route路由 Agent 打出的分类标签。retrieved_docs检索 Agent 返回的知识库命中片段。compliance_flags质检 Agent 给出的风险标记。final_answer最终返回给用户的答复文本。对应的 State 定义代码如下这段代码我加了不少注释方便理解from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] ticket: dict route: str retrieved_docs: list[str] compliance_flags: list[str] final_answer: str这里最关键的是messages字段。多个 Agent 都会往消息列表里写内容——Router 写的意图分析、Retrieve 写的检索摘要、Compliance 写的质检意见如果不加add_messagesReducer后写的 Agent 会把前面所有内容覆盖掉系统直接失忆。这个细节我一开始也吃过亏后来每次定义 State 都会先问自己一句这个字段是累加还是替换。接下来实现节点函数。每个节点本质上就是一个入参为AgentState、返回值为字典的普通 Python 函数。以 Route Agent 为例它调用 LLM让模型输出一个固定格式的分类结果然后写回 State 的route字段def route_agent(state: AgentState) - dict: prompt f 你是工单分类专家。根据以下工单内容将其分类为 - order_query订单查询 - refund_request退款申请 - complaint投诉 - other其他 工单内容{state[ticket][description]} 只需输出一个分类标签不要输出其他内容。 response llm.invoke(prompt) route response.content.strip() return {route: route}Retrieve Agent 和 Compliance Agent 也按同样模式实现。Retrieve Agent 拿到route标签后按标签决定检索策略——比如refund_request就优先查退款流程文档complaint就优先查客诉处理规范然后再把命中的文档片段写入retrieved_docs。Compliance Agent 则是把用户原始工单和 Retrieve 给的资料拼在一起让模型检查有没有合规风险比如承诺了超出政策允许的赔偿、泄露了用户隐私等等命中的风险项写入compliance_flags。3.2 条件路由与工具调用的实现细节节点的逻辑实现好之后真正让 Multi-Agent 活起来的是图的组装。我用StateGraph来搭骨架用add_node注册节点用add_edge和add_conditional_edges建立流转关系。Coordinator 是这个图的 supervisor 主控。它的职责是先调用route_agent判断工单类型再根据类型决定派发给谁订单查询派给 Retrieve退款申请也要先 Retrieve 然后 Compliance投诉则要三个 Agent 并行处理。也就是说Coordinator 发起一个路由决策通过条件边把控制权交给对应的 worker。from langgraph.graph import StateGraph, START, END graph StateGraph(AgentState) graph.add_node(route, route_agent) graph.add_node(retrieve, retrieve_agent) graph.add_node(compliance, compliance_agent) graph.add_node(coordinator, coordinator_agent) graph.add_edge(START, route) graph.add_edge(route, retrieve) # 根据工单类型分发任务 graph.add_conditional_edges( retrieve, route_to_worker, { refund_request: compliance, complaint: compliance, }, ) graph.add_edge(compliance, coordinator) graph.add_edge(retrieve, coordinator) graph.add_edge(coordinator, END)这里的route_to_worker是一个条件函数入参是当前 State返回一个字符串LangGraph 会根据返回值匹配边字典中的键决定下一个节点def route_to_worker(state: AgentState) - str: if state[route] in (refund_request, complaint): return compliance return coordinator如果你只需要串行执行add_edge就够了但一旦出现根据状态选下一步的需求就必须用add_conditional_edges。这也是多智能体跟普通工作流的核心区别——执行路径不是写死的而是根据每一步的中间结果动态生长出来的。工具调用方面LangGraph 给 Agent 挂工具的姿势非常顺。把工具函数定义好后在节点里用ToolNode承载工具执行逻辑让 LLM 先决定要不要调工具、调哪个然后把结果注入messages流。在我这个案例里Retrieve Agent 挂了一个search_knowledge_base工具函数内部就是查向量库标注了tool装饰器然后通过create_react_agent或者 Node 内部手动 tool-calling 都能跑起来。挂工具之后节点逻辑就从一次 LLM 调用变成LLM 思考-调用工具-观察结果-再思考的 Agent LoopLangGraph 内部会自动处理这个循环无需自己写 while。3.3 编译与执行一步步跑通工作流图定义完后调用.compile()即可得到一个可执行的CompiledStateGraph。这个对象提供了一个极其好用的功能stream事件流机制。执行时你可以逐节点观察状态变化这在调试 Multi-Agent 系统时的价值无可替代。app graph.compile() config {configurable: {thread_id: ticket-10001}} inputs {ticket: {id: T-10001, description: 用户购买的商品三个月内出现质量问题要求退款。}} final_state app.invoke(inputs, config) print(final_state[final_answer])运行后LangGraph 会依次执行route节点 →retrieve节点 → 按条件决定是否走compliance→ 最后到coordinator汇总输出。整个流程的状态流转在控制台可以看得一清二楚。thread_id这个配置也很关键它把状态持久化到了存储层同一个工单后续的追问会自动携带此前所有上下文不需要我们自己接数据库。在调试期你可能会遇到一个情况模型输出格式不对、解析出错、或者节点间状态更新异常。LangGraph 的interrupt_before/interrupt_after参数可以让你在指定节点前后暂停执行然后手动检查或修整状态再继续——这功能对我来说就是调试神器有问题直接在中间插一脚比反复改代码重启高效太多了。我还强烈建议用.get_state()查看任意时刻的快照。在 Multi-Agent 场景下每一轮对话都可能经过五六个节点状态里到底残留了什么字段、哪些字段被覆盖过一条命令就能看全。实际项目里我有个习惯生产环境出问题时先拉出失败工单的 State 快照看看哪个节点的写入不符合预期再针对性修复。这比看日志猜流程要精确一个量级。4. 调试经验与常见问题排查实录4.1 模型调用失败与结构化输出解析错误Multi-Agent 系统最常见的故障点不在图逻辑而在 LLM 的输出稳定性。节点 A 让模型输出 JSON结果它给你一段带解释文字的段落JSON 解析器直接崩溃节点 B 让模型输出分类标签它给你贴了一个 Prompt 里从未出现过的标签条件边路由时找不到对应键直接报 KeyError。我处理这个问题的第一道防线是用结构化输出约束模型。现在主流模型都支持函数调用或 JSON 模式我在 LangGraph 节点里全部改成走 structured output也就是定义一个 Pydantic 模型让 LLM 严格按这个 schema 输出。比如 Route Agent 输出的就是一个RouteResult模型里面只有一个route: Literal[order_query, refund_request, complaint, other]字段。这样输出解析失败的概率会急剧降低。from pydantic import BaseModel from typing import Literal class RouteResult(BaseModel): route: Literal[order_query, refund_request, complaint, other] def route_agent(state: AgentState) - dict: llm_with_structure llm.with_structured_output(RouteResult) result llm_with_structure.invoke(f工单内容{state[ticket][description]}) return {route: result.route}第二道防线是在条件边函数里做兜底。即使模型给了个意外值也不要直接抛异常。我的route_to_worker函数末尾会加一个else分支默认走向 Coordinator 做兜底处理保证整个图的执行不会因为一个异常标签就中断。生产环境的稳定性不是靠模型不出错而是靠系统对模型的任何输出都有预案。4.2 上下文丢失与状态膨胀问题上下文丢失是另一类高频问题。最常见的情形两个节点先后修改同一个messages字段但 State 里这个字段忘了加add_messagesReducer后一个节点把前面的所有消息覆盖后续节点就像是失忆了。这种 bug 隐蔽性极高因为你只有在完整跑完整个图之后才会发现最终回复逻辑混乱很难定位是哪一步丢的。排查方法也简单——在各节点之间打印 State 快照或者用 LangGraph 的stream模式跟踪消息序列的变化。与之相对的是状态膨胀问题。Multi-Agent 跑得越多messages越长最后几轮之后 token 消耗指数上涨甚至超出模型的上下文窗口。我在处理客服系统时就遇到过一个工单被重复追问了十几轮整个对话历史加上检索文档已经上万 token模型响应质量开始明显下降。解决办法有两个一是给每个 Agent 的 Prompt 里加一个只关注最近 N 条消息的显式指令二是在 State 里单独维护一个scratchpad字段只用来存当前工单的处理摘要避免把全部历史都塞进每一次模型调用。我还尝试过在节点之间定期做摘要压缩——用一个轻量 LLM 把前面的对话总结成一小段摘要然后用摘要替换原始内容。这在长对话场景下效果非常明显。4.3 循环死锁与路由兜底策略条件边虽然灵活但也引入了一个大麻烦循环。如果节点 A 的条件边把任务派给节点 BB 的处理结果又让 A 重复处理图就会陷入无限循环。LangGraph 本身没有内置循环次数上限一旦死循环调用会一直挂着直到你手动杀掉进程。我在实际项目中遇到过两次。第一次是 Coordinator 不断把工单派回 Retrieve因为 Retrieve 返回的结果总被 Coordinator 判定为信息不足第二次是两个 Agent 之间互相质疑对方输出来回不下十轮。LangGraph 的recursion_limit参数就是防这种问题的默认值好像是 25把它调到合理值并在业务层面设置最大处理轮数超了就强制走人工。app graph.compile() config {configurable: {thread_id: ticket-10001}, recursion_limit: 20}此外我在所有条件边函数的末尾都会强制加一个else分支让任何意外情况都汇入一个安全节点比如 Coordinator 兜底或人工介入。这个兜底思维是做 Multi-Agent 系统非常重要的一条经验宁可让系统给出一个保守的回答也不让它卡死或无限循环。生产环境里稳定性永远排在能力前面。我把这阶段遇到的高频问题整理成一个排查速查表方便你对照症状可能原因排查思路某个节点执行后其他节点看不到它写的数据State 字段被覆盖或字段名拼写不一致用 get_state 查看各节点写入检查 Reducer图一直执行不停条件边形成循环检查 recursion_limit给条件边加兜底模型输出无法解析结构化输出约束不足改用 with_structured_output校验 schema对话history越来越长导致超 token状态膨胀增加摘要压缩节点限制历史长度同一个工单多次追问上下文错乱thread_id 未固定使用 thread_id 持久化状态5. 实战心得与避坑建议5.1 先画图再写代码流程设计永远优先我在这个项目里最深的体会是Multi-Agent 系统的成败在图的设计阶段就决定了。代码写得再干净如果节点关系理不清后面就是无尽的调试地狱。我的建议是拿到需求先别急着写 State、写节点用白板或纸把以下问题想清楚这个系统需要几个 Agent每个 Agent 的职责边界是否清晰哪些 Agent 之间是串行依赖哪些可以并行用户的状态在哪一个节点发生了本质变化什么情况下需要人工兜底兜底节点怎么布置画完图再动手写代码你会发现 LangGraph 的实现过程本身非常快真正花时间的是把逻辑理清楚。我自己大概有 60% 的时间花在白板设计上30% 花在调试 Prompt 和输出稳定性只有 10% 是写图代码。5.2 日志与可观测性必须一开始就搭好Multi-Agent 系统一旦跑起来两三个 Agent 之间的交互还看得清五六个 Agent 交错行动时没有完善可观测性非常容易失控。我强烈建议从第一个版本开始就统一用stream事件流记录每个节点的输入输出摘要、延迟、token 消耗。LangGraph 提供了事件回调机制可以自定义回调函数在每个节点执行后自动记录信息。把这个日志系统搭好后期问题排查会省非常多时间。我在客服系统上线后维护体验很好很大程度就要归功于这个从第一天起就维护的事件日志。具体实现倒不难注册一个回调函数在节点执行结束后把node_name、state_snapshot的关键字段、elapsed_time写入结构化日志配合现有监控体系查询就行。后续分析可以逐步做到看日志就能重建一次工单的完整处理轨迹。5.3 思维转变从一个模型干所有事到多个模型各干一件事最后说说认知层面的收获。做 Multi-Agent 之前我总想让一个模型的 Prompt 尽量覆盖所有情况结果 Prompt 越写越长输出质量越来越差调试越来越难。切到 LangGraph 做多 Agent 编排之后我最大的心态转变是接受每个模型只需要做一件小事——Router 只负责分类Retrieve 只负责检索Compliance 只负责质检。每一件小事都容易做好组合起来的效果远好于一个大模型硬扛所有事。当然代价也不是没有。多 Agent 系统的整体延迟比单个 LLM 调用高因为多次串行调用必然带来时间成本token 消耗也翻倍了因为每次调用都要携带上下文。所以我的原则是能单 Agent 解决的事不要强行拆多 Agent只有确实存在专业化分工、审核流程、多步骤工具调用等需求时才值得上 Multi-Agent 架构。这个判断标准我每次开启新项目都会拿出来用一遍能省不少不必要的复杂度开销。LangGraph 把多 Agent 编排变成了一个可设计、可调试、可维护的工程问题这是它跟用 Prompt 硬撑所有逻辑最本质的区别。如果你正准备做类似系统我建议你从自己的实际业务场景出发先画图再动手跑通之后再逐步加复杂度。这套路我走下来妥的。