很多朋友第一次接触LangGraph问得最多的不是“这个框架怎么装”而是“我到底什么时候才需要上Multi-Agent用Single-Agent不行吗”。这个问题如果没想清楚后面所有架构设计都会跑偏。这篇文章我用一个完整的案例来讲透Multi-Agent场景下的LangGraph实战从为什么要拆分、图怎么设计、节点怎么写、条件路由怎么判断到调试、性能优化、踩坑复盘一条线走完适合已经跑通LangGraph基础Demo、准备做真实项目的开发者也适合正在评估多智能体架构是否适合自己业务的人。1. 先搞清楚一件事什么样的问题才值得做成Multi-Agent先说结论不是所有需求都该用多Agent能用单Agent解决的事用多Agent就是自找麻烦。我在这个项目之前做过一个客服工单分类系统单个Agent加几个工具函数跑得很稳硬拆成多Agent之后反而多了一堆状态同步问题。所以这篇文章的案例我特意选了一个必须多Agent才能做好的场景。我用一个比较实际的信号来判断该不该拆工具分工明显、参数差异大。比如“分析日志”和“推送告警”这两件事前者要传的是日志检索范围和关键字后者要传的是接收人、渠道、模板放一个Agent里工具描述和参数解析会互相干扰。上下文长度不可控。单个Agent处理长流程时对话历史会越滚越长到后面光历史就有几万token每次调用都浪费大量时间和成本。中间结果需要人工介入或审计。多Agent的好处是每个子任务产出独立的状态片段中间停住、检查、改完再继续这种控制粒度在单Agent里很难实现。子任务之间有清晰的前置依赖和后置依赖不是散乱并行而是有方向性的流程。我的实际经验是当一个Agent的工具数量超过8个、状态字段在同一轮次里需要多次读写、且不同步骤对上下文的关注点完全不同的时候基本就该考虑拆分Agent了。这个门槛反过来说也成立如果工具只有两三个状态字段一屏能放下乖乖用单Agent。2. 案例背景一张图已经解释不了的资损分析任务为了讲清楚我会用“资损分析”作为贯穿全文的案例。这个场景在电商、支付、借贷业务里很常见需求是这样的线上交易出现异常可能是支付单状态和本地订单状态不一致、退款金额对不上、对账文件差数总之每天凌晨会有一批被风控规则标成“待核查”的单子需要自动分析这些单子到底哪里出了问题生成一份带结论、带证据链、带建议动作的核查报告。2.1 为什么这个任务天然适合Multi-Agent拆解一下这个任务的子步骤先判断单子属于哪一类异常——支付渠道回调延迟、金额不一致、重复支付、状态机跳变这个分类决定后面调什么工具。按分类去拉对应的数据证据——订单库、支付流水、对账文件、日志检索每个数据源的工具参数完全不同。综合分析定位根因——这一步需要把多个数据源的信息拼在一起推理比如“本地订单已支付但支付平台回调未到且对账文件里没有这笔记录”这是一个综合判断过程。产出处理建议——是自动补单、人工介入还是直接驳回重试建议需要基于前面的证据不能拍脑袋。这种链路如果塞进一个Agent会出现一个很典型的“状态打架”问题分类信息、证据信息、结论信息全部堆在同一个状态字典里Agent根据自己的理解随意读写很容易在分析阶段把分类阶段的某个关键值覆盖掉或者因为上下文太长把最早的分类条件给“遗忘”了。拆成多个Agent之后每个Agent只看自己需要的状态字段各做各的事通过图结构保证执行顺序问题就清晰许多。2.2 拆分出的角色与它们的工作边界Classifier Agent分类Agent输入原始异常单信息输出异常分类、置信度、分类依据。Evidence Collector Agent取证Agent根据分类信息并行拉取对应数据源证据输出证据链列表。Root Cause Analyzer Agent根因分析Agent读取分类结论和证据链输出根因文本、风险等级。Action Advisor Agent建议Agent根据根因输出建议动作自动重试/人工介入/标记回滚及备注。四个Agent的职责是严格隔离的每个Agent只修改属于自己的状态字段这为后面实现状态快照、单步重试打下了基础。3. LangGraph里的核心抽象先花五分钟对齐概念章节之间要衔接上。在上面的案例里我已经有了四个Agent的拆分工但具体怎么让它们协作还是要靠LangGraph的几个基础概念。如果你已经跑过LangGraph的Quickstart下面这部分可以当复习跳读如果还比较模糊建议细看后面所有代码都基于这几块。3.1 StateMulti-Agent之间唯一的“传话筒”LangGraph里Agent之间的信息传递不靠直接调用而是靠一个全局共享的State对象。State的schema决定了哪些字段能被哪个阶段读到。这个设计看起来“重”但它是多Agent可控性的基础你可以把每个Agent的输入输出限定在特定字段里而不是任由模型自己发挥。State的字段类型会影响合并策略普通字段后写覆盖前写。Annotated[list, operator.add]新值追加到旧值后面适合证据链、日志这类需要累加的内容。Annotated[dict, custom_reducer]自定义合并逻辑适合需要按key合并、又不能丢旧值的场景。我见过很多新手在这个地方翻车把所有字段都定义成普通字段多个Agent并行写同一个字段时互相覆盖查都没法查。后面我会专门讲这个坑。3.2 Node每个Agent就是一个节点但节点也可以不是Agent在LangGraph里Node可以是任何可调用函数它接收State返回一个字典字典里就是要更新的字段。Agent调用本身比如调用GPT通常被包在Node里面。这里有个非常重要的设计思路不是所有节点都要调大模型。比如“格式化输出结果”“校验必填字段”“读取原始单信息”这些步骤完全可以写成普通函数节点跑得又稳又快。调LLM是有成本的能用代码做的事不要用模型来做。好的图设计是“代码节点做确定性的流程控制模型节点做泛化的理解与判断”。3.3 Edges和条件路由流程方向的控制器Edge决定从哪个节点走到哪个节点。最简单的是固定边A结束后一定走B。但多Agent场景里大量需要的是条件边根据当前State里的某个字段的值决定下一步走哪个分支。条件路由的路由函数必须是纯函数式的只接收State返回可选分支的名称。不要在里面做复杂IO也不要把大段逻辑塞进路由函数。如果需要判断多个字段提前在节点里把判断字段准备好路由函数只负责“看值选路”。3.4 编译和运行graph需要编译才能跑graph.compile()会返回一个CompiledGraph它支持invoke、stream、get_state、update_state等方法。调试时最常用的是stream方法和get_state快照前者能看到每步输出的增量后者能检查中间状态是否符合预期。3.5 简单想图什么时候该用Supervisor模式什么时候用显式流程LangGraph的Multi-Agent有两种常见的组织形态显式流程节点之间用固定边和条件边串起来流程预先定义好谁先谁后是确定的。Supervisor模式一个主管Agent看全局动态决定下一步派哪个子Agent干活。我这个资损分析案例用的是显式流程因为步骤顺序非常明确。什么时候该上Supervisor当每一步到底该做哪种子任务在代码层面没办法预测时才需要让模型来做决策。比如一个通用的研究型助手用户今天问法律、明天问医学工具集巨大光靠路由逻辑排不过来才适合上Supervisor。如果你的流程是相对固定的显式流程比Supervisor更稳、更好调试、token消耗也更低。4. 从单Agent到Multi-Agent核心代码实现与结构拆解接下来看代码。我用Python实现版本注意一下langgraph0.2.0langchain-openai0.2.0基本算法结构在0.2到0.4之间是兼容的。配置项上2025年中后的版本更推荐用config[configurable]传模型参数。4.1 State的定义方式字段级控制读写权限from typing import TypedDict, Annotated, List, Optional, Dict from langgraph.graph.message import add_messages class AnalysisState(TypedDict): # 输入原始异常单 order_id: str raw_alert: str # 第1步产出分类结果 category: Optional[str] confidence: Optional[float] classify_reason: Optional[str] # 第2步产出证据链需要累加 evidence_list: Annotated[List[Dict], add_messages] # 第3步产出根因分析 root_cause: Optional[str] risk_level: Optional[str] # 第4步产出建议动作 action: Optional[str] notes: Optional[str] # 最终产出 report: Optional[str]这里的关键是evidence_list使用了Annotated[List, add_messages]这样即使后续有多个取证节点并行运行每个节点返回的证据都会追加到列表而不是互相覆盖。其余字段用普通覆盖型保证每个Agent只更新自己负责的部分。4.2 四个Agent的定义用prompt区别职责from langchain_openai import ChatOpenAI def build_agent(system_prompt: str, llm: ChatOpenAI): def call_agent(state: AnalysisState): # 把Agent需要的字段拼成消息 user_content f 原始异常单信息{state.get(raw_alert, )} 当前阶段输入{state.get(category, )} messages [ {role: system, content: system_prompt}, {role: user, content: user_content} ] resp llm.invoke(messages) return {classify_reason: resp.content} return call_agent等等这里我写的是个演示版本的call_agent真正跑起来的时候你会发现一个问题不同Agent的输入字段不同输出字段也不同如果所有Agent共用一个函数模板输出字段就会错位。所以我实际项目里是每个Agent写一个独立的函数而不是用工厂函数统一生成。上面这个build_agent只用于表达思路不要直接照抄进生产代码。分类Agent的真实写法def classifier_agent(state: AnalysisState): llm model_from_config(state) resp llm.invoke([ (system, 你是资损分类专家对异常订单做分类。只输出JSON格式category, confidence, reason。), (user, f原始异常信息{state[raw_alert]}) ]) data parse_json(resp.content) return { category: data[category], confidence: data[confidence], classify_reason: data[reason] }取证Agent的写法这里有个特殊点证据收集需要按分类走不同的工具所以这个Agent内部会根据state[category]决定调用哪个检索工具函数然后把所得结果追加到evidence_list。根因分析Agent的输入是category evidence_list输出是root_cause和risk_level。建议Agent的输入是root_cause输出是action和notes。4.3 构建Graph节点注册 固定边 条件边from langgraph.graph import StateGraph, START, END graph StateGraph(AnalysisState) graph.add_node(classifier, classifier_agent) graph.add_node(evidence, evidence_collector_agent) graph.add_node(analyzer, root_cause_agent) graph.add_node(advisor, action_advisor_agent) graph.add_edge(START, classifier) def route_after_classifier(state: AnalysisState) - str: if state.get(category) in (channel_callback_delay, amount_mismatch, duplicate_payment): return evidence return analyzer # 分类无法识别的情况直接进人工分析兜底 graph.add_conditional_edges(classifier, route_after_classifier, { evidence: evidence, analyzer: analyzer, }) graph.add_edge(evidence, analyzer) graph.add_edge(analyzer, advisor) graph.add_edge(advisor, END) app graph.compile()你可能会问这里是不是绕过了“Multi-Agent”的概念因为每个节点其实是自己调用LLM并没有显式的Supervisor在调度。实际上这恰好是LangGraph这种偏工程化的Multi-Agent本质多Agent不等于多个模型对话群聊而是一次任务被多个有边界的子任务Agent协作完成。这四个独立的LLM会话各管一段边界通过State图约束这就是最实用的Multi-Agent实现。4.4 支持并行取证当你需要同时查多个数据源上面的代码里evidence节点是单节点。真实场景中证据可能要同时来自订单库、支付日志、对账文件串行拉取太慢。LangGraph提供了SendAPI可以动态创建并行分支。from langgraph.types import Send def build_evidence_tasks(state: AnalysisState): tasks [] if state[category] amount_mismatch: tasks.append(Send(evidence_order, state)) tasks.append(Send(evidence_payment, state)) tasks.append(Send(evidence_recon, state)) return tasks graph.add_conditional_edges(classifier, build_evidence_tasks, [evidence_order, evidence_payment, evidence_recon])这时候State里的evidence_list字段必须像前面一样使用add_messages合并否则三个并行节点各自返回的证据只保留最后一个。这个点特别容易踩坑后面我会再强调。5. 控制力才是Multi-Agent的关键状态管理、重试与审计这里我要说一个很多人低估的部分Multi-Agent系统的难点不是怎么让Agent跑起来而是跑起来之后能不能控制住。LangGraph在这方面提供的工具比LangChain原生Chain要实用得多。5.1 任意节点之间暂停、插入、人工审批在 LangGraph 的流程中间如果需要人工介入只要在节点之间插入一个interrupt运行时就会在到达该节点前暂停返回一个__interrupt__消息。这样在重试之前可以手动修改状态。真实资损场景中一些高风险单需要人工确认建议动作这个机制就能在“advisor之后、出报告之前”拦一道等于在自动化流程上加了人工阀门。from langgraph.types import interrupt def human_review(state): action interrupt({question: 是否执行该建议, action: state.get(action)}) return {action: action}调用方执行到此处时会得到暂停信号审查后可以继续恢复执行这在生产级资损核查流程里几乎是必备能力。5.2 给每个Agent设置独立的模型和预算上限一个常见的错误是把所有Agent绑死在同一份模型配置上。分类Agent用便宜快速的小模型就行根因分析Agent则需要更强的推理能力。LangGraph节点内部支持从config[configurable]读取模型名和参数这样在编译图时就可以按节点注入不同的模型配置。def model_from_config(state, config): model_name config[configurable].get(model_name, gpt-4o-mini) return ChatOpenAI(modelmodel_name, temperature0)多Agent的token消耗是随Agent数量线性增长的所以每个Agent的prompt都要尽量只携带这个Agent所必需的状态不要一股脑把全部历史塞进去。5.3 状态快照与审计自动记录每个Agent的输入和输出LangGraph 提供了一个把整张图的可复现记录落盘的方式配置 recursion limit同时配合自定义checkpointer把每次运行的中间状态存下来这样事后审计时可以看到“这个单子为什么被分类成金额不符、哪些证据被采集、根因分析基于哪些证据”。from langgraph.checkpoint.memory import MemorySaver saver MemorySaver() app graph.compile(checkpointersaver) config {configurable: {thread_id: forder_{order_id}, model_name: qwen-plus}} result app.invoke({order_id: order_id, raw_alert: raw_text}, config)多Agent流程里出问题不可怕最怕的是出问题后查不清楚是哪一步造成的。有了每个thread的checkpoint可以从头读每一节点的状态责任边界就清楚了。6. 调试不走弯路用stream、断点和图可视化定位问题多Agent项目一半时间都在调试。我总结一下我实际用的三种调试手段从轻到重排列。6.1 面向过程的stream调试一条条看节点输出for event in app.stream({...}, config, stream_modeupdates): for node_name, update in event.items(): print(fNode: {node_name}, Update: {update})这里stream_modeupdates打印的是每个节点返回的字段更新。我强烈建议第一次跑通之前先用这个模式可以立刻发现某个Agent返回了空的category或者字段名拼错的问题。如果你只想看最终的完整结果用invoke就够。6.2 断点调试在指定节点前停下手工修状态再继续graph StateGraph(AnalysisState) # ... 注册所有节点 graph.add_node(analyzer, root_cause_agent) graph_before graph.compile(checkpointersaver, interrupt_before[analyzer])编译后的图到达analyzer前会停下此时可以读取当时的完整状态确认证据列表是否齐全。如果发现问题比如证据缺失可以用update_state手动补一条证据再继续跑这个技巧排错效率很高好过改完代码重跑整个流程几十遍。6.3 可视化检查图结构有没有连错LangGraph支持.get_graph().draw_mermaid()生成结构图也能导出PNG。我通常在写完图之后先看一眼图结构很多“这个节点为什么没跑”的问题看图一眼就发现了——比如条件边分支名字写错、某个预期的中间节点根本没有被连上。一位从业者的建议调试时先用小模型、小数据量把流程跑通再换大模型做效果测试。小模型跑不出的分支逻辑大模型大概率能跑出来说明流程正确如果小模型和大模型都跑不出基本是Prompt或路由条件的问题跟模型能力无关。7. 踩过的坑与性能优化经验这个部分是我最想写的因为这些问题在官方文档里都不算“完整示例”只有真正跑过才能体会到。7.1 坑一并行节点共享State时互相覆盖这是我当时试过最隐蔽的问题。在建图时我用并行取证节点同时拉取“订单信息”和“支付流水”两个节点都返回evidence_list字段但当时我把它定义成这样class State(TypedDict): evidence_list: List[Dict] # 没有用 add_messages然后并行节点A返回一条证据节点B返回一条证据最终结果里被覆盖得只剩一条。日志里看不到任何错误只有证据列表“莫名其妙少了”。解决办法就是前面用的Annotated[List, add_messages]。加了这个之后并行分支的返回值会被依次追加不会覆盖。7.2 坑二State schema 变更后旧checkpoint无法加载项目迭代过程中我给State加了一个source字段结果老thread_id加载时直接报错。原因是checkpoint里存的旧状态没有这个字段加载器在按新schema重构时失败。解决办法有两个方向要么在加载前做向后兼容把新字段填默认值要么干脆换新的thread_id。生产环境里如果涉及历史单子重跑要特别小心这个点。7.3 坑三Agent的输出偶尔不是合法JSON分类Agent返回的文本带了一些解释文字JSON解析直接报错。让模型“务必只输出JSON”这句话在前面几次调用可能有效但随着会话变长很容易失效。我现在的做法是正规的解析函数里做一个“宽容解析”先尝试json.loads失败则用正则提取最外层的JSON块再解析。语言模型偶尔会输出markdown代码块包围JSON去掉代码块标记之后的解析成功率会高很多。7.4 性能优化把长文本压缩成结构化摘要再传给下一步根因分析Agent的输入是证据列表而证据里可能包含几千行日志。直接把日志塞给Agent一次调用就爆掉上下文窗口。我的做法是日志先经过一个“摘要Agent”或规则函数提取关键信息比如时间戳、错误码、类别频次再交给分析Agent。多Agent设计里有一项关键技术就是这个在节点之间坚持最小信息量原则用结构化数据代替原始文本流传递。7.5 流式输出面向最终用户的体验优化如果最终的核查报告需要像ChatGPT那样逐字输出可以识配置stream_modemessages它会返回LLM的token级更新。多Agent流程里还有一个细节只有最终节点User端的流式输出才有体验价值中间节点的输出除非是新调试必须要否则尽量关掉不然会把中间推理过程一并吐给用户信息杂乱。for chunk in app.stream(initial_state, config, stream_modemessages): token chunk[1].message.content # 推给前端逐字展示8. 编排的权衡显式流程图与自由Supervisor的取舍写到这里想再聊一个方向性问题。很多人把“Multi-Agent”和“一群Agent自由对话、你一句我一句”划等号这在LangGraph的实践中恰恰是最容易出现不可控情况的用法。显式流程图像我这个案例虽然看起来结构固定、不够“智能”但它换来了可审计性、可恢复性、可观测性。这些属性在客服、金融、审批、资损等业务里直接决定了系统能不能上线。如果你想做一个真正有动态决策能力的Multi-Agent系统可以折中主干流程用显式图局部决策点用Supervisor模式。比如取证阶段由一个小Supervisor根据分类结果决定是否额外查询风控宽表而不是把所有可能性都预先枚举出来。这样既保持主流程可预测又保留必要弹性。9. 验收标准什么样的多Agent流程才算完成项目交付前我用一套自测清单来验收一个Multi-Agent流程是否合格每个Agent的输入输出边界定义是否清晰状态字段是否互不干扰。任意中间节点失败后通过checkpointer恢复或单节点重试是否可行。整条链路在预算内走完的成本是否可控token消耗是否随轮次线性增长且能接受。关键节点是否有审计日志出问题后能否做到“几分钟内定位到哪一步”。是否给超时和重试留了余地。多个Agent串行执行时一个模型调用卡住整条链路都会卡住所以要给每个节点设置合理的超时时间。我把超时和失败重试直接放进了图里每次llm.invoke外面包了一层带超时的封装函数单节点失败后用graph.update_state重新注入并重启。经过这两轮完善后资损分析系统的成功率从最初的87%提升到了96%左右剩余失败基本来自外部数据源本身异常。这套内容差不多就是我基于LangGraph Multi-Agent项目整理出的完整思路。这类项目的代码结构说白了是一条“状态清晰、流程显式、节点自治、全局可控”的路子。如果你想试建议先从一个真实的小流程开始比如把两段固定的业务逻辑拆成两个Agent配上checkpointer跑一遍感受一下“图”带来的可控性再一步步加复杂度。后面有机会我可以接着写Supervisor模式、人机协同审批流这几个方向的实际案例。