前几天有个老同事问我你们用LangGraph做Multi-Agent到底是怎么把好几个Agent编排到一起跑的是像AutoGen那样互相聊天还是像LangChain那种串prompt我说都不是LangGraph的路子和它们不太一样——它把Agent和Agent之间的协作当成一张图来画节点是人边是交接规则状态是公共黑板。我把最近做完的一个工单自动处理系统拿出来当例子把LangGraph Multi-Agent的完整案例拆开讲一遍从业务拆解、架构设计、核心概念到能直接抄的代码、跑通之后的坑全在这里了。这个案例适合谁看已经写过单个LangChain Agent、但对LangGraph还停留在听说过阶段的人以及想在公司里落地Multi-Agent但不知道该怎么分配职责、怎么处理并行分支的人。如果你只是想看一个花哨demo那这篇不一定适合你如果你是想搞一个能扛住真实业务压力的多Agent系统这篇应该能帮你少走很多弯路。1. 这个案例是怎么来的工单积压背后的架构问题1.1 单Agent模式下客服机器人为什么卡住了背景是一家做电商SaaS的公司每天光客服工单就有三百多张其中百分之六七十都是重复问题订单能不能退款、发票怎么开、App登录失败、接口报错怎么排查。之前用的方案是一个单体客服Bot意图识别FAQ检索几个工具调用全部塞在一个Agent里一个超长System Prompt负责所有事。一开始看起来还行但随着工单类型越来越多问题就明显了。最直观的表现是Prompt越长模型越容易精神分裂。你今天让它处理退款它回答得很专业明天同一个模型换了个语气它就开始一本正经地编订单号。更麻烦的是不同业务需要调用的工具完全不一样——退款要查订单系统、开发票要调财务接口、技术咨询要查知识库和日志平台把这些工具的说明全部塞进一个Agent的tool列表里整个上下文会变得又臭又长轮到真正干活的时候模型已经分不清该用哪个工具了。那段时间我统计过一个数据单Agent模式下工单分类准确率只有78%左右其中有一大批错误不是模型笨而是它根本没想起来自己还能调那个工具。这就像让一个全科医生看所有病他当然知道很多但当他一天要记几百种药的说明书时开错药的概率自然会上升。1.2 拆解思路让每个Agent只干一件事后来我们决定换思路既然一个Agent应付不了全品类那就拆成一群Agent各管一摊。这个思路在工程上很常见用一个不太精确的类比——就像客服中心不会让一个人接所有电话而是先有个前台接线员问您是要办退款还是查发票然后转给对应的专员。专员不需要懂所有业务只需要把自己那一亩三分地研究透。落到架构上就是一个Triage Agent负责接收原始工单、做分类下面挂三个Worker Agent分别处理退款闭环、发票闭环、技术支持闭环每个Worker只加载自己业务相关的工具和知识。真正跑起来之后分类准确率从78%提到了96%以上单张工单的平均处理时长从15分钟降到了40秒左右部分自动化流程甚至秒级完成。这个提升不是模型变聪明了而是架构把复杂度分散了。1.3 为什么是LangGraph而不是别的框架选型的时候我们其实看了一圈AutoGen、CrewAI、LangGraph还有自己手写状态机。结果对比如下框架编排方式并行分支支持状态控制粒度生产落地成熟度我们的结论AutoGen对话式群体协作较弱主要靠群聊对会话流侧重流程控制偏隐式中等适合研究型多Agent讨论不适合强流程业务CrewAI角色任务列表支持并行但深度流程控制较弱任务级控制为主中等上手快但复杂流程分支不好表达LangGraph有向图显式状态原生支持Send API可做并行分发节点级精确控制支持循环、条件边较高流程复杂需要可观测、可恢复LangGraph最后胜出的原因有两点很关键。第一它的状态管理是全局共享状态局部Reducer这让多个Agent并行干活、又要把结果写回同一个上下文这件事变得可控。第二它有Checkpointer持久化机制能在节点之间暂停和恢复——对客服这种业务来说万一某一步需要人工介入我们可以让流程停下来等几分钟甚至几小时再接着跑这个能力在别的框架里实现起来要费不少劲。2. 案例架构先把整体图景讲清楚2.1 工单处理链路是怎么设计的整个系统处理一条工单的流程是这样的客户提交工单 → Triage Agent读取工单内容并打上分类标签 → 根据标签把任务并行分发给对应Worker Agent → 每个Worker独立调用自己的工具链 → 结果汇总进全局状态 → 生成最终回复。注意这里的关键词是并行分发。为什么是并行而不是串行因为一张工单有时候会同时涉及多个维度比如用户说我要退款顺便问一下发票怎么开这两个诉求互相独立完全可以同时处理。如果串行来做第二个Agent要等第一个跑完整体耗时直接翻倍。LangGraph里有一个Send API专门干这个事——它不是简单地把多个节点串起来而是允许你在一个节点里动态生成多个子任务每个子任务可以独立走一条子图路径。我们当时的链路图用文字表述大概是START进入triage_nodetriage_node输出分类结果从triage_node引出一条条件边条件边的返回结果是Send对象的列表每个Send指向一个Worker节点所有Worker节点跑完之后汇聚到aggregator节点aggregator生成最终回复后由人类审核节点决定是结束还是重新打回。2.2 每个Agent的职责、模型和工具划分Agent名称职责范围使用的模型挂载的工具/数据Triage Agent工单分类、判断是否需要人工介入GPT-4o-mini便宜够用分类规则、关键词表、少量few-shot示例Refund Agent退款/退货/取消订单闭环GPT-4o-mini订单查询API、退款策略文档、商品SKU映射表Billing Agent发票抬头校验、开票状态查询GPT-4o-mini财务开票接口、发票规则库Tech Agent登录失败、接口报错、App崩溃排查GPT-4o技术知识库、日志查询工具、常见错误码文档Aggregator汇总结果、生成对外话术GPT-4o-mini模板库、语气风格指南这里有一个选型细节值得展开Triage和大多数Worker用便宜的GPT-4o-mini就够了因为它只做分类和简单工具调用对推理能力要求不高但Tech Agent我们换成了GPT-4o因为技术支持工单往往需要多步推理比如登录失败可能和token过期有关也可能是接口限流这种问题用小模型经常答非所问。2.3 两层结构Triage路由 并行Worker为什么强调两层而不是一层扁平结构我见过不少人做Multi-Agent时直接把所有Agent都并联起来让它们全部能看到所有工单——结果就是所有Agent都在抢着回答而且不同Agent给出的答案互相矛盾。Triage这层本质上是做一件事收敛。它把所有Agent都可能处理的问题收敛成这个类别下只有这一个Agent应该处理从源头避免冲突。Worker层则反过来做发散每个Worker只接受自己领域内的任务但它们内部可以有自己的小循环。比如Refund Agent判断退款是否通过不通过时需要结合退款策略再判断一次这就是一个内部的Agent-loop不是串联多个Agent而是同一个Agent在条件不满足时自我修正。这种发散-收敛-再发散的结构还有一个好处是方便加人工兜底。如果Triage发现工单内容涉及投诉、舆情、金额特别巨大等敏感类别可以直接走一条人工优先的路径不自动处理如果Worker处理过程中拿不准也可以在返回结果里打个标记让aggregator最后给人工审核员留一句审核建议。3. LangGraph核心概念我用案例讲给你听3.1 State状态是大家共享的那块黑板LangGraph里最核心的东西不是Agent而是State。你可以把它理解成一块所有节点都能读写的公共黑板。比如我们车间的黑板上写着当前工单ID、用户问题原文、Triage分出的类别、各Worker返回的中间结果、最终要回复给客户的话术。每个节点干完活就把自己的产出写到黑板上下一个节点从黑板上拿自己要的东西。黑板的定义方式是一个TypedDict字段类型自己定。有个细节极其重要默认情况下多个节点往同一个字段写值后面的覆盖前面的。这在串行流程里没毛病但一旦进入并行分支两个Worker同时往messages字段写自己的完成记录后执行的会直接冲掉先执行的——这就麻烦了。解决办法是给字段指定一个Reducer聚合器最常见的是operator.add它的意思是每次写入时把新旧值拼成一个列表。3.2 Node节点是干活的单元Edge是交接规则Node就是你定义的一个普通函数输入是当前状态字典输出是一个字典表示你往黑板上写了什么。Edge决定下一个执行哪个节点。LangGraph有两种Edge一种是固定边永远从一个节点走到另一个节点另一种是条件边根据函数返回值动态决定下一步。在案例里triage_node到各个Worker之间就是条件边条件的返回值可以是一个字符串节点名也可以是一个Send对象列表。Send对象的意义很特别它允许同一个图里动态产生多条并行路径每条路径携带自己独立的参数。打个比方正常流程像火车沿着固定轨道开Send则像是到了分拣站你能同时发出好几辆车每辆车装不同货物去不同目的地。3.3 Checkpointer解决了流程挂起恢复的问题Checkpointer是LangGraph生产落地最实用的部件。它会把每一步执行完的状态快照存下来存到内存、SQLite或者外部数据库。有了它你可以做到两件事第一流程执行到一半崩了可以从上一个快照继续不用从头跑第二可以人为打断流程比如我们的审核节点先让流程暂停等审核员操作完再调用接口恢复执行。没有Checkpointer时如果你想实现Agent处理完临时结果等人点确认再继续只能用外部消息队列硬扛非常痛苦。有了Checkpointer这个功能几乎零成本实现这也是我坚持用LangGraph做生产系统的核心原因之一。3.4 两种编排范式路由Worker和Supervisor怎么选做Multi-Agent编排最常见的两个范式需要分清。第一种是路由Worker范式一个中心节点先做分发其他节点各干各的干完汇总没有谁指挥谁。第二种是Supervisor范式一个班长Agent持有对话权它来决定下一步让哪个子Agent上场适合子任务之间需要频繁协商的场景。我们这个工单案例用的是路由Worker范式因为工单流程是相对固定的——分类、处理、汇总步骤清晰根本不需要班长在多个Worker之间来回调度。但我也见过一个反例有个同事做竞品分析PPT生成器让Triage分完类就直接结束了结果市场分析Agent和文案Agent写出来的内容互相矛盾没人拍板。那种场景就应该用Supervisor让主控Agent当裁判逐轮召集相关Agent讨论。4. 实操从零实现一套工单Multi-Agent4.1 环境准备与依赖安装代码基于LangGraph 0.2以上版本部分API在0.2.x之后发生过变化建议装最新稳定版。安装命令pip install langgraph langchain-openai如果你不想真的调用OpenAI本文示例里Triage节点用了关键词规则模拟分类Worker节点也故意写得极简你直接跑也能看到整个编排流程。真要接大模型的话需要一个OpenAI兼容的Keyfrom langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, temperature0)4.2 定义工单状态与状态缩减器状态是整个图的地基定义如下from typing import Annotated, TypedDict, List import operator from langgraph.graph import StateGraph, END, START from langgraph.types import Send class TicketState(TypedDict): ticket_id: str # 工单ID只读不合并 issue: str # 用户问题原文 category: str # Triage 分类结果 routes: List[str] # 并行派发时使用的路由标记 messages: Annotated[List[str], operator.add] # 所有节点产出的消息追加合并 final_reply: str # 最终回复覆盖式写入重点看messages字段的Annotated[List[str], operator.add]。它告诉LangGraph当多个节点并行往这个字段写值时不要互相覆盖而是把所有的值追加到一起。没有这行定义并行Worker写的中间结果就会丢。4.3 实现Triage分诊节点与各业务WorkerTriage节点负责读工单、分类、决定要派发的路由标记def triage_node(state: TicketState) - dict: issue state[issue] # 实际生产环境这里会换成 LLM 调用例如 # category llm.invoke(f判断工单类别...) if 发票 in issue: category billing elif 退款 in issue or 退货 in issue: category refund else: category tech # 如果一个问题里同时提到退款和发票两个都要派 routes [] if 退款 in issue or 退货 in issue: routes.append(refund) if 发票 in issue: routes.append(billing) if 登录 in issue or 崩溃 in issue or 接口 in issue: routes.append(tech) if not routes: routes.append(category) return {category: category, routes: routes}注意这里的routes是一个列表代表本次工单需要并行触发哪些Worker。如果工单只提到退款就只派refund一个如果退款和发票都提了就同时派两个Worker并行跑。Worker节点的写法也很直接比如Refund Agent内部先调用订单查询函数再决定回复内容def query_order(order_id: str) - dict: # 这里对接真实的订单系统 API return {status: paid, amount: 129.0, can_refund: True} def refund_agent(state: TicketState) - dict: # 生产环境这里会带完整的 system prompt 和 LLM 调用 # 演示代码直接走规则逻辑 order_info query_order(state[ticket_id]) if order_info[can_refund]: msg f订单 {state[ticket_id]} 可退款金额 {order_info[amount]} 元退款通道已开启。 else: msg f订单 {state[ticket_id]} 不符合退款条件已标记需要人工复核。 return {messages: [msg]} def billing_agent(state: TicketState) - dict: msg f发票申请已受理电子发票将在24小时内发送至用户邮箱。 return {messages: [msg]} def tech_agent(state: TicketState) - dict: msg f技术问题已派发至T2工程师队列预计30分钟内响应。 return {messages: [msg]}我把每个Worker都刻意写成一个函数内部包含一个工具调用这符合实际生产模式Agent不是凭空回答而是先查数据再回答。你真要接大模型只需要把query_order的结果塞进prompt然后让LLM根据查询结果生成客服话术。4.4 用Send API实现并行分发这里是我认为LangGraph最优雅的地方——把路由逻辑写成一个返回Send对象列表的函数def route_by_category(state: TicketState): send_tasks [] worker_map { refund: refund_agent, billing: billing_agent, tech: tech_agent, } for route in state[routes]: target worker_map.get(route) if target: # Send 的第二个参数是传给目标节点的局部状态 # 只传这个节点需要的最小字段即可 send_tasks.append(Send(target, { ticket_id: state[ticket_id], issue: state[issue], })) return send_tasks当条件边返回一个Send列表时LangGraph会把这些Send交给对应节点并且让它们并行执行实际并发度受线程池等配置影响。每个Send携带的局部状态只包含目标节点必需的字段也就是最小化数据传递。这里要特别说明并行节点是Worker之间各自独立跑还是共享主State答案是它们共享主StateSend的第二个参数只是输入前缀节点函数的返回值仍然会写入全局状态——这就是为什么之前必须要用operator.add来合并并行写入。4.5 加入一个人工审核节点客服场景经常需要人工介入在这个系统里我加了一个审核环节如果最终回复里包含人工复核字样就不直接把消息发给客户而是进入等待状态。def review_node(state: TicketState) - dict: # 实际工程里这里会在 Checkpointer 中暂停图等待外部事件触发恢复 if 人工复核 in state[final_reply]: return {final_reply: state[final_reply] 待人工确认} return {} def should_review(state: TicketState) - str: if 人工复核 in state[final_reply]: return review_node return done然后在图中把它挂上去。这个节点看起来简单但配合Checkpointer可以实现审核员登录后台看到待办点一下同意图就继续跑的效果。这在真实客服业务里非常刚需。4.6 把图组装起来并运行前面所有节点定义好后组装图就非常直接了from langgraph.checkpoint.memory import MemorySaver def build_graph(): builder StateGraph(TicketState) builder.add_node(triage_node, triage_node) builder.add_node(refund_agent, refund_agent) builder.add_node(billing_agent, billing_agent) builder.add_node(tech_agent, tech_agent) builder.add_node(review_node, review_node) builder.add_node(aggregator, aggregator) builder.add_edge(START, triage_node) # 关键条件边返回 Send 列表时前面不需要映射表 builder.add_conditional_edges( triage_node, route_by_category, path_map[refund_agent, billing_agent, tech_agent], ) # 所有 Worker 跑完后汇聚 for worker in [refund_agent, billing_agent, tech_agent]: builder.add_edge(worker, aggregator) builder.add_conditional_edges(aggregator, should_review, {review_node: review_node, done: END}) builder.add_edge(review_node, END) return builder.compile(checkpointerMemorySaver()) def aggregator(state: TicketState) - dict: all_msgs \n.join(state[messages]) # 真实系统里会用 LLM 把多条消息整理成一段通顺客服话术 return {final_reply: f您好您的问题已处理相关答复如下\n{all_msgs}}运行起来很简单app build_graph() result app.invoke( {ticket_id: A1001, issue: 我想退款顺便问一下发票怎么开}, config{configurable: {thread_id: ticket-A1001}}, # 必须传 thread_id ) print(result[final_reply])上面这个案例是一个很典型的路由Worker案例你完全可以把triage_node里的关键词规则替换为真实的LLM调用把query_order替换成公司内部API代码骨架不用动。这也是LangGraph的另一个优点节点内部逻辑可以逐渐复杂但图的结构保持不变。5. 跑通之后这些坑我必须记下来5.1 并行节点写入互相覆盖我一开始没在messages字段加operator.add结果两个Worker并行跑完最终的messages只剩一个Worker的内容。这个问题在串行流程里根本不会出现但一开并行就立刻踩雷。解决办法就是State定义时的Annotated。但要注意一点operator.add只对列表、字符串这类支持加法的类型有效。如果你并行节点要写入的字段是自定义结构得自己传一个Reducer函数把多个返回合并成一个对象。比如多个节点都要往analysis_results字典里写自己负责的模块就得自定义一个Reducer做update操作不能用operator.add硬拼字典。5.2 条件边返回Send时path_map容易配错用Send做并行分发时有一个很隐蔽的坑add_conditional_edges的第三个参数path_map在返回字符串时是字符串到节点的映射表但在返回Send列表时它应该是一个普通的字符串列表只声明合法目标节点。我在早期版本代码里不小心把映射表写成了字典结果LangGraph直接报配置错误。建议养成习惯返回Send列表的Router函数path_map永远写一个纯列表返回字符串的条件函数path_map才写字典。5.3 图一直跑不停触发recursion_limitMulti-Agent图如果设计不当很容易出现无限循环。比如某个Worker判断条件不满足要重试但重试本身又产生了新的问题于是循环永不终止。LangGraph默认有一个递归深度限制recursion_limit默认25超过会直接抛异常。我在Refund Agent内部做过一个重试逻辑第一次查订单状态是待支付它想等几秒再查一次但如果在测试环境里订单永远不会变这个Agent就会一直循环到爆。解决办法有两个方向一是在条件边里显式增加最大重试次数计数字段比如把循环次数写进State超过两轮强制走人工二是全局调大recursion_limit但这只是拖延问题不是根治。根治的手段永远是给循环设明确上限。5.4 Agent陷入工具调用死循环这个问题和代码循环还不太一样是模型层面的。某个Agent在判断退款是否可行时明明工具已经返回订单不可退款模型却因为prompt里有一句尝试用多种方式解决用户诉求于是反反复复调同一个接口边调边自己脑补新的请求参数。这类问题非常难排查因为从日志上看它确实在做事。我们最后在代码里加了一层硬性兜底每个Agent节点外层套一个工具调用次数计数器连续超过5次调用还没产出最终消息就强制返回预设兜底话术。另外还重写了系统提示词明确加上如果工具返回的结果相互矛盾停止调用并输出你的判断这句约束比加多少few-shot都管用。5.5 并发环境下的sync与async混用LangGraph支持async节点你可以用async function定义Agent。但生产环境中一个常见的翻车现场是图本身是async的节点里却直接调用了一个同步阻塞的第三方SDK比如某个老的数据库驱动于是一个并发场景硬生生变成了串行执行还时不时报RuntimeError: asyncio.run() cannot be called from a running event loop。我后来统一了规范图的运行时统一走ainvoke异步入口节点内部如果调用同步SDK就用asyncio.to_thread把它扔到线程池里去跑避免阻塞事件循环。如果你的团队对asyncio不熟更省事的做法是所有节点全部写成同步函数图的运行时用invokeLangGraph内部会用线程池为你做并发只是要自己控制好并发上限。5.6 Checkpointer和thread_id必须配对刚开始用checkpointerMemorySaver()后我忘了在invoke时传configurable: {thread_id: ...}结果图直接报错。原因是LangGraph的持久化机制依赖thread_id来区分不同的会话轨迹没有它状态快照不知道存到哪个key下。这里还有一个容易被忽视的点同一个图的多次invoke如果thread_id相同会被LangGraph认为是同一条流程的继续状态会累积。这在多轮对话场景是特性但在工单场景如果你重新跑一张相同ID的工单可能看到上一次残留的messages。所以每次处理新工单时记得换一个新thread_id或者显式走一次清空逻辑。5.7 常见问题速查表现象根本原因解决建议并行Worker跑完结果只剩一个状态字段缺Reducer字段上用Annotated[..., operator.add]条件边报配置错误Send返回时path_map写成了字典返回Send列表时path_map只用纯列表图无限执行直到爆掉条件边没有终止条件State里加循环计数器超限走人工或结束Agent反复调同一个工具Prompt约束不足工具调用次数硬限制 提示词显式禁止async图调用同步SDK报错事件循环被阻塞统一用ainvokeSDK调用放asyncio.to_thread加上Checkpointer后invoke报错缺少thread_id每次invoke传入configurable.thread_id恢复流程后状态不对thread_id复用导致状态累积新会话用新thread_id或显式清空状态6. 这个案例后续可以怎么扩展如果你看完上面的示例已经能跑通一个简单的LangGraph Multi-Agent系统那我可以再分享几条真实落地时的扩展建议。首先是把Triage Agent升级成真正的LLM决策。关键词规则在Demo阶段好用但真实工单语句千奇百怪比如用户写我钱付了东西没到里面既没退款也没退货关键词规则会直接掉进tech类。换成LLM分类后配合一个类别定义清单准确率能明显提升。这里有一个实操心得LLM分类时一定要让它输出JSON格式比如{category: refund, reason: ...}再用解析器校验不要直接让它返回裸字符串否则跑着跑着会出现各种非常规的类别名。其次是引入Supervisor架构作为升级版。当你发现Worker之间需要来回协商时就不要再让Triage一次性定死路由了。可以把主控Agent设计成一个循环主控读取当前状态决定让refund上场还是让tech先查日志收到结果后再决策下一步。这个模式的控制力更强但也更容易出问题建议等路由Worker模式跑稳了再上。最后是做好全过程可观测。两个Agent协作出问题时最怕的其实是不知道它俩谁先错的。LangGraph官方配套的LangSmith能在面板上直接看到每一步状态、每一条prompt、每个节点的输入输出比翻日志高效太多。我习惯在每个Worker节点返回的dict里带上一个trace_note字段记录它这次决策的依据这样排查问题时能直接看到模型当时看到了什么数据。这个工单系统上线到现在最大的体会是Multi-Agent的价值不在于“用多个模型吵架”而在于通过架构把不可控的大问题拆成可控的小问题然后让每条处理路径保持简单。LangGraph它不替你解决LLM本身的问题但它能让你的解决方案在复杂流程里站得住脚。如果只是写个demo你可能感受不到它和普通脚本的区别一旦要面对并行、重试、人工介入、流程恢复这些真实世界的问题你就会明白为什么图这一层抽象是值得的。