过去半年我前后参与了三个基于 LangGraph 的多智能体项目落地一个是电网调度场景下的多智能体协同评估一个是医疗知识问答方向的多角色拆解助手还有一个是团队内部流程自动化的小工具。三个项目都上了生产也都踩了不少坑。说实话LangGraph 的官方文档写得不差但能跑通 demo和能在生产环境稳住之间隔着一条河河里流的全是工程问题状态怎么管、智能体之间怎么通信、失败了怎么恢复、延迟怎么压、token 怎么省。这篇就把我在几个落地项目里反复验证过的工程实践整理出来适合那些已经玩过 LangGraph 示例、正准备把自己的多智能体系统推向生产的团队。1. 动手之前想清楚为什么需要多智能体1.1 先算一笔账多智能体到底贵在哪里很多团队上多智能体是被概念推着走的不是被问题推着走的。这里必须先把成本账算清楚不然等项目跑起来发现账单爆了才想起来做预算已经晚了。一次多智能体协作的成本不是各个智能体成本相加而是近似相乘。原因在于每个智能体都要读取共享状态而共享状态里带着前面所有智能体的推理历史和中间结果prompt 一次比一次长。我见过一个三智能体的项目单次请求的 token 消耗是单智能体方案的 7 倍以上延迟从 3 秒飙到 15 秒。这是物理层面跑不掉的——图里有多少个节点就有多少次 LLM 调用而且后一个节点的输入往往包含前一个节点的完整输出。你用 4 个模型实例跑一个用户问题账就是这么堆出来的。还有一层隐性成本调试成本。单智能体出了问题翻一个会话的日志就能定位多智能体出了问题你得知道是哪个节点、哪条边、哪一步决策错了。这个成本在开发期往往比 token 成本还高因为它消耗的是人的时间而且不可压缩。我在第一个项目里就吃过这个亏上线第一周一半的时间花在复现问题-定位节点-修正状态上比写代码本身还累。1.2 什么场景真的需要多智能体结合我手上的几个项目真正适合多智能体的场景大概分三类。第一类领域知识边界清晰的任务。比如电网可靠性评估里面涉及拓扑分析、负荷预测、故障研判三个完全不同的专业方向。一个通用大模型很难在同一个上下文里同时做好三件事因为每个方向需要不同的工具集、不同的判断逻辑和不同的输出格式。拆成三个智能体之后每个智能体只绑定自己领域的工具模型的指令遵循能力明显提升幻觉也少了。第二类天然有角色划分的任务。比如流程审批、合同谈判、客服多轮转接这些场景本身就有角色的概念——审批人、谈判方、客服专员。用多智能体去模拟角色比让单智能体假装切换角色更可控因为每个角色的状态、权限、可用工具是物理隔离的不会串。第三类带循环和分支的复杂工作流。比如数据分析先理解问题、再查数据、再可视化、最后写结论每一步的产出是下一步的输入而且中间可能需要回退重来。这种流程用线性的 Chain 很难表达图是更天然的载体。LangGraph 的价值恰恰在这里它不是让你画流程图而是让你把一个不确定的流程变成一个可执行、可恢复的状态机。1.3 我的判断标准三步决策法我现在判断一个需求要不要上多智能体就问三个问题任何一个不满足就劝退这个任务能不能用一个 prompt 加若干工具描述清楚如果能就不要拆。拆开之后每个子任务是否真的需要独立的记忆状态如果不需要就合回去。子任务之间的交互是顺序流水还是动态路由顺序流水用 Chain 就够只有需要按中间结果决定下一步方向的场景才值得上 LangGraph。这三个问题过完你大概率会发现一半以上的多智能体需求其实用单智能体加工具调用就能解决。这不是坏事是省钱省事。我那三个项目里有一个最初设计了五个智能体过完这三关之后砍成了两个跑出来的效果反而更稳——节点少了出错点就少了。2. 把 LangGraph 当作状态机理解2.1 图是程序节点是函数LangGraph 最容易被误解的地方是很多人把它当成流程图画布想怎么连就怎么连。实际上 LangGraph 的核心是一台状态机你先定义 StateSchema状态结构然后定义节点普通函数最后定义节点之间的边。真正的执行逻辑全在节点函数里图本身只描述了状态如何流转。一个节点的函数签名大致长这样def my_node(state: MyState) - dict: # 读取 state做处理 result do_something(state[input]) # 返回部分状态更新 return {output: result}注意几个重点节点函数是纯函数式的它的全部输入来自 state全部输出是对 state 的部分更新。这意味着同一个节点可以被多次触发比如在循环里只要 state 正确结果就是确定的。这个特性是可排查、可重放的基础——生产环境出了错你可以用同一个状态重新跑一遍同一个节点结果应该完全一致这就是可复现性。我在团队里要求每个人先写清楚一张表每个节点消费哪些 state 字段、产出哪些 state 字段、可能走哪些边。表格理清楚了再写代码比边写边想靠谱得多。这个习惯救了我很多次尤其在多智能体场景里节点之间的依赖关系一旦混乱排查成本指数级上升。2.2 Reducer 的规则共享状态怎么改State 是全局共享的但多个节点可能并发更新同一个字段也可能在一个循环里反复往同一个字段追加内容。这时候就需要 reducer——它是 LangGraph 控制状态更新方式的钩子决定了当节点返回一个字段的新值时到底怎么处理。最常见的 reducer 是消息列表的追加from typing import Annotated, TypedDict from langgraph.graph import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] current_owner: str artifacts: dictAnnotated[list, add_messages]的意思是当任何节点返回新的消息时不是覆盖而是追加到已有列表末尾。这个设计解决了一个核心问题在多智能体协作里每个智能体产生的中间结果都不应该丢它们全部累积在共享状态里后续节点可以随时回溯。Reducer 写错是重灾区。我第二个项目就栽过一回两个 worker 同时更新 messages 字段因为 reducer 配错成覆盖模式后完成的 worker 把先完成的覆盖掉了最终答案少了一半。查了半天才发现是 reducer 的问题而不是模型的问题。所以我的建议是凡是列表类、消息类字段一律用add_messages或自定义追加式 reducer只有那些当前负责人任务状态这类单值字段才考虑覆盖语义。2.3 条件边和循环把业务流程搬进图里LangGraph 支持条件边也就是根据当前状态决定走哪条路。这是多智能体和单链路的本质区别静态的 Chain 只有一条路走到底图可以分岔、可以回退、可以循环。一个典型的路由函数def route_after_analysis(state: AgentState) - str: if state.get(needs_more_data): return fetch_data if state.get(analysis_complete): return report_writer return human_review路由函数返回的是下一个节点的名字LangGraph 会根据返回值走向对应的边。循环在这里特别重要——ReAct 模式模型决定调工具、工具返回结果、模型再看结果继续决策的本质就是一个循环。在 LangGraph 里这个循环可以直接用图结构表达不需要你在代码里写 while 来包着一次次 LLM 调用。我建议团队每个成员先忘掉 LangChain 里 Chain 的思维惯性把整个业务当成一张状态图来画。每个节点问自己我消费什么状态、产出什么状态每条边问自己什么条件下走这条路。图设计清楚了代码只是翻译。反过来如果你发现图画完是个大直线那说明你根本不需要图一个 Chain 就够。3. 三种可落地的多智能体协作模式3.1 Supervisor 模式一个大脑多个专家最稳妥、也是我推荐大多数团队先试的模式一个 supervisor 节点负责理解用户请求、分派任务多个 worker 节点各自负责一个领域最后由 supervisor 或单独的聚合节点汇总产出。在 LangGraph 里supervisor 本质上也是一个 LLM 节点它通过路由函数或工具调用来决定下一个执行哪个 worker。关键设计原则是supervisor 不亲自干活只做决策——该问谁就问谁、该收手就收手。worker 之间不直接对话也不看彼此的完整上下文只拿到自己任务需要的那部分状态。这个模式的好处很实在每个 worker 的 prompt 和工具集可以做得很专指令遵循压力小输出更稳定supervisor 可以控制预算比如限制最多调几个 worker、最多几轮防止智能体陷入无限折腾新增一个领域只需要加一个 worker 节点和一条路由规则不改其他节点扩展成本低。我那个电网评估项目用的就是这个模式supervisor 先解析问题然后路由给拓扑分析、负荷预测、故障研判三个 worker最后聚合节点统一生成评估报告。每个 worker 只绑自己的数据查询工具效果比单智能体好很多。3.2 Handoff 模式谁接手谁负责Handoff 模式更适合接力棒式的任务典型场景是客服流程接待 → 意图识别 → 转接给售后、技术或退款等专门智能体。和 Supervisor 模式最大的区别是控制权完全转移前一个节点不再参与后续决策。LangGraph 里可以用Command实现这种转移from langgraph.types import Command def handoff_to_billing(state: AgentState) - Command: return Command( update{ current_owner: billing_agent, handoff_reason: 用户咨询退款政策 }, gotobilling_agent )Command同时更新了状态并指定下一个节点比先 return 再走条件边更直接语义也更清晰。我在实践里的一个体会是Handoff 模式必须显式记录当前谁在负责和为什么转交这两个字段否则追上下文时会非常痛苦。你看到一个用户会话中途换了智能体但不知道为什么要换、换了之后前一个智能体还记不记得这回事调试就像盲人摸象。3.3 工具即接口用工具调用打通智能体边界第三种模式经常被忽略把另一个智能体封装成一个工具当前智能体通过 tool call 去调用它。这是按需召唤专家的典型做法。from langchain_core.tools import tool tool def expert_advisor(query: str) - str: 当用户问题涉及专业领域时调用领域专家智能体。 result expert_graph.invoke( {messages: [{role: user, content: query}]}, config{thread_id: fsub-{uuid4()}} ) return result[messages][-1].content这个模式特别适合主智能体保持轻量的场景平时它只用少量通用工具处理常规问题遇到专业问题才动态拉起一个子图。注意子图必须用独立的 thread_id避免 checkpointer 状态和主图串在一起——这是我在工具即接口模式里踩过最大的坑子图状态混进主图之后整个会话的get_state_history都乱了。实际项目里我通常是三种模式混合用外层一个 supervisor 做全局调度中间用 handoff 做明显的角色转交某些节点内部再嵌套工具形式的子智能体。模式本身不冲突关键是每层之间的状态接口要干净——谁写什么字段、谁读什么字段必须白纸黑字定下来。4. 生产环境落地FastAPI 集成与状态持久化4.1 从 Notebook 到服务的改造要点demo 里graph.invoke()一把梭很爽生产环境面对的却是另一套问题并发请求、状态隔离、超时控制、失败重试、流式输出。这些东西在 Notebook 里统统不会出现一上服务全都来了。我的标准做法是用 FastAPI 包一层把 graph 编译成单例接口只暴露必要的入参from fastapi import FastAPI from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver app FastAPI() async def init_graph(): async with AsyncSqliteSaver.from_conn_string(checkpoints.db) as saver: graph builder.compile(checkpointersaver) return graph GRAPH await init_graph() app.post(/chat) async def chat(request: ChatRequest): config {configurable: {thread_id: request.session_id}} result await GRAPH.ainvoke( {messages: [{role: user, content: request.query}]}, configconfig ) return {answer: result[messages][-1].content}几个容易踩的坑逐个说不要把 graph 放在每个请求里新建。编译一次、复用实例状态隔离交给 checkpointer 做否则每个请求都在重复编译开销并发一上来直接崩。checkpointer 必须选持久化的。默认的 MemorySaver 重启即丢生产至少上 SQLiteQPS 再高就换 Postgres。异步 SQLite 写入在高并发下有瓶颈如果你的服务要扛住几十路并发建议直接上AsyncPostgresSaver连接池配好别在小文件数据库上死磕。FastAPI 本身是异步的LangGraph 也支持ainvoke和astream所以整个链路天然是异步的。但要注意 LLM 调用本身是 IO 密集操作如果某个第三方 SDK 是同步阻塞的最好放进线程池或者用asyncio.to_thread包一层别卡住事件循环。4.2 Checkpointer 与人工介入LangGraph 的 checkpointer 不只是为断点续传设计的它给多智能体项目带来了两个特别重要的能力人机协同和故障重放。人机协同用interrupt()实现from langgraph.types import interrupt def review_node(state: AgentState) - dict: user_decision interrupt( {draft: state[draft], question: 这份方案是否可以发布} ) if user_decision approve: return {status: approved} return {status: rejected, reason: user_decision}这个函数执行到interrupt时会暂停把当前状态持久化到 checkpointer然后返回给调用方一个待恢复的句柄。后续通过Command(resume...)或者再次 invoke 带回用户的决定图会从暂停的位置继续执行。整个过程中其他节点不会空跑非常适合智能体出草稿、人类做审批这类半自动流程。我在内部流程自动化那个项目里审批环节就这么做的效果非常好——机器做不了决定的时候停下来问人而不是硬着头皮编。故障重放就更实用。生产环境出了错拿着thread_id用同一个图实例重新 invoke能把每一次节点执行的状态都翻出来精确到哪个节点改了哪个字段。这个能力让多智能体系统的排错成本下降了一个量级我是真的依赖它。4.3 流式输出与并发控制多智能体系统延迟高用户体验全靠流式输出兜底。LangGraph 支持多种流式模式最常用的是astreamasync for chunk in GRAPH.astream( {messages: [{role: user, content: query}]}, configconfig, stream_modemessages ): if chunk[0].content: yield fdata: {chunk[0].content}\n\nstream_modemessages会逐 token 地输出 LLM 生成内容前端用 SSE 接住就行。但我的经验是多智能体场景下只吐最终答案不够最好把当前在哪个环节也一起吐出来。比如用stream_modeupdates捕获每个节点完成时的状态变化前端显示正在调用数据分析工具…正在生成评估报告…这类进度提示。用户知道系统还在干活耐心会显著提升——这一点在延迟动辄十秒以上的多智能体系统里几乎决定产品口碑。并发控制方面多个用户共享同一个 graph 实例加持久化 checkpointer 是安全的因为状态隔离在 thread_id 上互不干扰。真正的瓶颈在 LLM API 的 rate limit 和数据库连接池。建议给所有 LLM 调用加统一的限流中间件给 checkpointer 配连接池并给每个请求设置总超时上限比如 60 秒超时就返回一个友好的降级提示而不是让用户无限等待。5. 可观测性与调试多智能体系统排错的手段5.1 状态快照排查法多智能体系统排错第一步永远是拿到这个请求完整跑过的状态快照。LangGraph 的 checkpointer 天然支持这一点不用额外做埋点config {configurable: {thread_id: demo-001}} states list(GRAPH.get_state_history(config)) for state in states: print(state.values.get(messages)[-1].content[:100]) print(--- next:, state.next)get_state_history返回从最新到最早的所有状态版本每个版本包含当时的完整状态值和下一步要执行的节点。这几乎等于给每个请求装了行车记录仪任何一个中间结果出了问题都能精确地从历史里找到那一帧。我处理线上问题的标准流程是先复制 thread_id拉状态历史看最后一条消息是谁写的、当时图停在哪个节点然后决定是重放、修正状态、还是直接恢复。这个流程我大概用了两个月就形成了肌肉记忆效率提升非常明显。团队里其他人也慢慢养成了这个习惯——遇到问题先拉状态而不是瞎猜模型。5.2 结构化追踪与日志光有状态快照还不够多智能体系统的日志必须结构化。我维护一份统一格式的运行日志每条记录包含这些字段字段说明run_id整次请求的追踪 ID前端透传进来node_name当前节点名agent_role该节点所属的智能体角色model_name用到的模型input_tokens / output_tokenstoken 消耗latency_ms节点耗时status成功、失败或超时用 Python 的 logging 打全这些字段或者直接发到 OpenTelemetry 都行。关键是每次 LLM 调用前后都把日志打全这样后续做成本分析、延迟分析、错误归因都有数据可查。LangSmith 如果预算允许也可以接它能把图中的每次调用串成 trace可视化程度很高。但我的经验是别只依赖它。第三方平台可能因为网络问题丢 trace自己维护的结构化日志才是排错的第一手资料。5.3 两个典型的线上故障说两个我实际遇到过的故障给大家提个醒。第一个是死循环。某个 agent 节点在条件边里判断是否还有未处理的问题LLM 连续判定还有于是反复调用自己直到触发超时。排查时拉状态历史发现同一个节点执行了 40 多次。解决方法是给循环加计数器字段——每次进入循环加一超过阈值就直接走出口边不让 LLM 无限自我延伸。这是多智能体系统的必修课任何一个循环都必须有硬上限。第二个是结果污染。两个 worker 同时更新 messages 字段因为 reducer 写错成覆盖模式后完成的 worker 把先完成的覆盖掉了。排查时靠状态快照一眼就看出 messages 少了内容修复 reducer 就解决了。这类问题在单智能体里根本不存在多智能体里却很容易犯——并发更新共享状态reducer 语义不正确数据就悄悄丢了。6. 性能与成本优化低延迟不是玄学6.1 串行转并行多智能体延迟高的最大来源是串行执行。很多流程里两个子任务其实互不依赖但默认写法是一个接一个跑白白浪费时间。LangGraph 支持并行节点——只要两个节点之间没有依赖边它们会在同一个 super-step 里并发执行from langgraph.graph import StateGraph, START builder StateGraph(AgentState) builder.add_node(agent_a, agent_a) builder.add_node(agent_b, agent_b) builder.add_node(merge, merge_results) builder.add_edge(START, agent_a) builder.add_edge(START, agent_b) builder.add_edge(agent_a, merge) builder.add_edge(agent_b, merge)agent_a和agent_b并行执行merge等两个都完成后再跑。对于可以拆分的任务这一步能把总延迟从两个节点耗时相加变成取最大值。我在数据分析项目里把一轮请求从 12 秒压到 7 秒靠的就是这个改动一行 prompt 没动。改之前总觉得慢是模型的问题改完之后才意识到是架构的问题。当然并行不是银弹。两个节点写同一个字段时就别并行要么用 reducer 合并要么串行。并行之前先确认状态依赖关系别为了快把正确性搭进去。6.2 模型分工大模型做决策小模型干杂活另一个降低延迟和成本的关键是别让所有节点都用同一个旗舰模型。我的原则路由、分类、信息抽取这类简单任务用快的小模型比如 7B 级别的开源模型或者 API 的低档位型号复杂推理、生成最终方案用最强的大模型工具调用相关的模型至少用中档以上因为工具调用对指令遵循要求高。LangGraph 的节点函数里可以自由选择模型所以在图设计阶段就要把每个节点的模型选定。我算过一笔账一个五节点的流程把其中三个节点的模型从旗舰换成中档成本降了六成以上整体质量几乎不受影响。因为那些节点本来就是干脏活累活的大模型反而容易想太多输出不稳定。6.3 token 节省的几个具体手段最后说几个 token 节省的具体手段都是我在项目里验证过有效的状态瘦身。别把所有历史消息都塞进共享状态。定义 state 时区分短期工作记忆和长期记忆工作记忆在节点完成后可以裁剪掉只保留必要的结果摘要。压缩摘要。如果确实需要保留长对话历史定期用小模型把前面的消息压成摘要只保留最近 N 轮原文。缓存工具结果。查询型工具查数据库、查文档在同一个 session 内按参数哈希做缓存避免同一份数据被多个节点重复拉取。这个优化在工具调用密集的智能体上效果特别明显。精简工具描述。每多一个工具描述模型每轮调用都要多读一遍token 成本是累计的。只暴露当前节点真正需要的工具别把所有工具全部绑给每个智能体。7. 踩坑实录与我的几点建议7.1 五个常见的坑整理一下我见过、踩过最多的五个坑新团队照着避雷就行把 LangChain 的 Chain 思维带进 LangGraph。想到串联就连一条线最后画出一个巨大的线性链分支和循环能力完全没用上图的优势荡然无存。共享状态里塞太多东西。什么都往 state 里堆模型每次请求都要读一整套无关上下文token 飙升、响应变慢。状态设计要克制只放真正跨节点流转的数据。忽略 checkpointer 配置。demo 用 MemorySaver 没问题上了生产忘了换持久化重启丢状态所有用户会话全部断掉。这个问题我在生产环境见过不止一次。不给循环设上限。agent 卡在我还能再做一步的幻觉里直到超时才停下来。所有循环必须有最大轮次这是铁律。所有节点用同一个模型。又贵又慢简单任务用大模型反而容易想太多、输出格式不稳定。模型分工要趁早做。7.2 让多智能体系统可回归多智能体系统最让人头疼的是不确定性同样的输入可能跑出不同的结果。这对测试是个大挑战我的做法是给测试加确定性单元层测试节点函数。节点是普通函数直接 mock LLM 返回值验证状态更新逻辑和路由逻辑不依赖真实模型。集成层录制回放。用 VCR 风格的录制库把 LLM 和工具的响应录下来测试时回放保证流程可复现。这个库很多语言都有Python 里有现成的。回归基线。准备一组标准 query每次改动后跑一遍对比关键指标答案是否包含核心要点、是否调用了预期工具、有没有死循环、延迟有没有劣化。线上灰度。任何图的修改先切 10% 流量对比 token 消耗、延迟、用户反馈确认没问题再全量放开。这几件事做完多智能体系统才敢说可维护。否则每次改 prompt 都像重新开一个项目——这不是夸张是真实体感。7.3 最后几句真心话LangGraph 这套框架的价值不在于它让我们用上了多智能体这个词而在于它逼着你把业务流程画成一张状态图逼着你考虑状态、边界、恢复这些工程问题。这些素养对任何复杂系统都是通用的不亏。如果你正准备开始我的建议是先拿一个月只做一个小而真的场景——一个 supervisor 加两个 worker生产部署、流式输出、结构化日志一样不落地跑起来比在 Notebook 里搭十个花架子 demo 有用得多。等这个最小闭环稳了再往里面加模式、加节点、加优化。真实流量的反馈永远比精心设计的 demo 案例更能暴露问题。
LangGraph多智能体生产落地：状态管理、性能优化与工程实践
过去半年我前后参与了三个基于 LangGraph 的多智能体项目落地一个是电网调度场景下的多智能体协同评估一个是医疗知识问答方向的多角色拆解助手还有一个是团队内部流程自动化的小工具。三个项目都上了生产也都踩了不少坑。说实话LangGraph 的官方文档写得不差但能跑通 demo和能在生产环境稳住之间隔着一条河河里流的全是工程问题状态怎么管、智能体之间怎么通信、失败了怎么恢复、延迟怎么压、token 怎么省。这篇就把我在几个落地项目里反复验证过的工程实践整理出来适合那些已经玩过 LangGraph 示例、正准备把自己的多智能体系统推向生产的团队。1. 动手之前想清楚为什么需要多智能体1.1 先算一笔账多智能体到底贵在哪里很多团队上多智能体是被概念推着走的不是被问题推着走的。这里必须先把成本账算清楚不然等项目跑起来发现账单爆了才想起来做预算已经晚了。一次多智能体协作的成本不是各个智能体成本相加而是近似相乘。原因在于每个智能体都要读取共享状态而共享状态里带着前面所有智能体的推理历史和中间结果prompt 一次比一次长。我见过一个三智能体的项目单次请求的 token 消耗是单智能体方案的 7 倍以上延迟从 3 秒飙到 15 秒。这是物理层面跑不掉的——图里有多少个节点就有多少次 LLM 调用而且后一个节点的输入往往包含前一个节点的完整输出。你用 4 个模型实例跑一个用户问题账就是这么堆出来的。还有一层隐性成本调试成本。单智能体出了问题翻一个会话的日志就能定位多智能体出了问题你得知道是哪个节点、哪条边、哪一步决策错了。这个成本在开发期往往比 token 成本还高因为它消耗的是人的时间而且不可压缩。我在第一个项目里就吃过这个亏上线第一周一半的时间花在复现问题-定位节点-修正状态上比写代码本身还累。1.2 什么场景真的需要多智能体结合我手上的几个项目真正适合多智能体的场景大概分三类。第一类领域知识边界清晰的任务。比如电网可靠性评估里面涉及拓扑分析、负荷预测、故障研判三个完全不同的专业方向。一个通用大模型很难在同一个上下文里同时做好三件事因为每个方向需要不同的工具集、不同的判断逻辑和不同的输出格式。拆成三个智能体之后每个智能体只绑定自己领域的工具模型的指令遵循能力明显提升幻觉也少了。第二类天然有角色划分的任务。比如流程审批、合同谈判、客服多轮转接这些场景本身就有角色的概念——审批人、谈判方、客服专员。用多智能体去模拟角色比让单智能体假装切换角色更可控因为每个角色的状态、权限、可用工具是物理隔离的不会串。第三类带循环和分支的复杂工作流。比如数据分析先理解问题、再查数据、再可视化、最后写结论每一步的产出是下一步的输入而且中间可能需要回退重来。这种流程用线性的 Chain 很难表达图是更天然的载体。LangGraph 的价值恰恰在这里它不是让你画流程图而是让你把一个不确定的流程变成一个可执行、可恢复的状态机。1.3 我的判断标准三步决策法我现在判断一个需求要不要上多智能体就问三个问题任何一个不满足就劝退这个任务能不能用一个 prompt 加若干工具描述清楚如果能就不要拆。拆开之后每个子任务是否真的需要独立的记忆状态如果不需要就合回去。子任务之间的交互是顺序流水还是动态路由顺序流水用 Chain 就够只有需要按中间结果决定下一步方向的场景才值得上 LangGraph。这三个问题过完你大概率会发现一半以上的多智能体需求其实用单智能体加工具调用就能解决。这不是坏事是省钱省事。我那三个项目里有一个最初设计了五个智能体过完这三关之后砍成了两个跑出来的效果反而更稳——节点少了出错点就少了。2. 把 LangGraph 当作状态机理解2.1 图是程序节点是函数LangGraph 最容易被误解的地方是很多人把它当成流程图画布想怎么连就怎么连。实际上 LangGraph 的核心是一台状态机你先定义 StateSchema状态结构然后定义节点普通函数最后定义节点之间的边。真正的执行逻辑全在节点函数里图本身只描述了状态如何流转。一个节点的函数签名大致长这样def my_node(state: MyState) - dict: # 读取 state做处理 result do_something(state[input]) # 返回部分状态更新 return {output: result}注意几个重点节点函数是纯函数式的它的全部输入来自 state全部输出是对 state 的部分更新。这意味着同一个节点可以被多次触发比如在循环里只要 state 正确结果就是确定的。这个特性是可排查、可重放的基础——生产环境出了错你可以用同一个状态重新跑一遍同一个节点结果应该完全一致这就是可复现性。我在团队里要求每个人先写清楚一张表每个节点消费哪些 state 字段、产出哪些 state 字段、可能走哪些边。表格理清楚了再写代码比边写边想靠谱得多。这个习惯救了我很多次尤其在多智能体场景里节点之间的依赖关系一旦混乱排查成本指数级上升。2.2 Reducer 的规则共享状态怎么改State 是全局共享的但多个节点可能并发更新同一个字段也可能在一个循环里反复往同一个字段追加内容。这时候就需要 reducer——它是 LangGraph 控制状态更新方式的钩子决定了当节点返回一个字段的新值时到底怎么处理。最常见的 reducer 是消息列表的追加from typing import Annotated, TypedDict from langgraph.graph import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] current_owner: str artifacts: dictAnnotated[list, add_messages]的意思是当任何节点返回新的消息时不是覆盖而是追加到已有列表末尾。这个设计解决了一个核心问题在多智能体协作里每个智能体产生的中间结果都不应该丢它们全部累积在共享状态里后续节点可以随时回溯。Reducer 写错是重灾区。我第二个项目就栽过一回两个 worker 同时更新 messages 字段因为 reducer 配错成覆盖模式后完成的 worker 把先完成的覆盖掉了最终答案少了一半。查了半天才发现是 reducer 的问题而不是模型的问题。所以我的建议是凡是列表类、消息类字段一律用add_messages或自定义追加式 reducer只有那些当前负责人任务状态这类单值字段才考虑覆盖语义。2.3 条件边和循环把业务流程搬进图里LangGraph 支持条件边也就是根据当前状态决定走哪条路。这是多智能体和单链路的本质区别静态的 Chain 只有一条路走到底图可以分岔、可以回退、可以循环。一个典型的路由函数def route_after_analysis(state: AgentState) - str: if state.get(needs_more_data): return fetch_data if state.get(analysis_complete): return report_writer return human_review路由函数返回的是下一个节点的名字LangGraph 会根据返回值走向对应的边。循环在这里特别重要——ReAct 模式模型决定调工具、工具返回结果、模型再看结果继续决策的本质就是一个循环。在 LangGraph 里这个循环可以直接用图结构表达不需要你在代码里写 while 来包着一次次 LLM 调用。我建议团队每个成员先忘掉 LangChain 里 Chain 的思维惯性把整个业务当成一张状态图来画。每个节点问自己我消费什么状态、产出什么状态每条边问自己什么条件下走这条路。图设计清楚了代码只是翻译。反过来如果你发现图画完是个大直线那说明你根本不需要图一个 Chain 就够。3. 三种可落地的多智能体协作模式3.1 Supervisor 模式一个大脑多个专家最稳妥、也是我推荐大多数团队先试的模式一个 supervisor 节点负责理解用户请求、分派任务多个 worker 节点各自负责一个领域最后由 supervisor 或单独的聚合节点汇总产出。在 LangGraph 里supervisor 本质上也是一个 LLM 节点它通过路由函数或工具调用来决定下一个执行哪个 worker。关键设计原则是supervisor 不亲自干活只做决策——该问谁就问谁、该收手就收手。worker 之间不直接对话也不看彼此的完整上下文只拿到自己任务需要的那部分状态。这个模式的好处很实在每个 worker 的 prompt 和工具集可以做得很专指令遵循压力小输出更稳定supervisor 可以控制预算比如限制最多调几个 worker、最多几轮防止智能体陷入无限折腾新增一个领域只需要加一个 worker 节点和一条路由规则不改其他节点扩展成本低。我那个电网评估项目用的就是这个模式supervisor 先解析问题然后路由给拓扑分析、负荷预测、故障研判三个 worker最后聚合节点统一生成评估报告。每个 worker 只绑自己的数据查询工具效果比单智能体好很多。3.2 Handoff 模式谁接手谁负责Handoff 模式更适合接力棒式的任务典型场景是客服流程接待 → 意图识别 → 转接给售后、技术或退款等专门智能体。和 Supervisor 模式最大的区别是控制权完全转移前一个节点不再参与后续决策。LangGraph 里可以用Command实现这种转移from langgraph.types import Command def handoff_to_billing(state: AgentState) - Command: return Command( update{ current_owner: billing_agent, handoff_reason: 用户咨询退款政策 }, gotobilling_agent )Command同时更新了状态并指定下一个节点比先 return 再走条件边更直接语义也更清晰。我在实践里的一个体会是Handoff 模式必须显式记录当前谁在负责和为什么转交这两个字段否则追上下文时会非常痛苦。你看到一个用户会话中途换了智能体但不知道为什么要换、换了之后前一个智能体还记不记得这回事调试就像盲人摸象。3.3 工具即接口用工具调用打通智能体边界第三种模式经常被忽略把另一个智能体封装成一个工具当前智能体通过 tool call 去调用它。这是按需召唤专家的典型做法。from langchain_core.tools import tool tool def expert_advisor(query: str) - str: 当用户问题涉及专业领域时调用领域专家智能体。 result expert_graph.invoke( {messages: [{role: user, content: query}]}, config{thread_id: fsub-{uuid4()}} ) return result[messages][-1].content这个模式特别适合主智能体保持轻量的场景平时它只用少量通用工具处理常规问题遇到专业问题才动态拉起一个子图。注意子图必须用独立的 thread_id避免 checkpointer 状态和主图串在一起——这是我在工具即接口模式里踩过最大的坑子图状态混进主图之后整个会话的get_state_history都乱了。实际项目里我通常是三种模式混合用外层一个 supervisor 做全局调度中间用 handoff 做明显的角色转交某些节点内部再嵌套工具形式的子智能体。模式本身不冲突关键是每层之间的状态接口要干净——谁写什么字段、谁读什么字段必须白纸黑字定下来。4. 生产环境落地FastAPI 集成与状态持久化4.1 从 Notebook 到服务的改造要点demo 里graph.invoke()一把梭很爽生产环境面对的却是另一套问题并发请求、状态隔离、超时控制、失败重试、流式输出。这些东西在 Notebook 里统统不会出现一上服务全都来了。我的标准做法是用 FastAPI 包一层把 graph 编译成单例接口只暴露必要的入参from fastapi import FastAPI from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver app FastAPI() async def init_graph(): async with AsyncSqliteSaver.from_conn_string(checkpoints.db) as saver: graph builder.compile(checkpointersaver) return graph GRAPH await init_graph() app.post(/chat) async def chat(request: ChatRequest): config {configurable: {thread_id: request.session_id}} result await GRAPH.ainvoke( {messages: [{role: user, content: request.query}]}, configconfig ) return {answer: result[messages][-1].content}几个容易踩的坑逐个说不要把 graph 放在每个请求里新建。编译一次、复用实例状态隔离交给 checkpointer 做否则每个请求都在重复编译开销并发一上来直接崩。checkpointer 必须选持久化的。默认的 MemorySaver 重启即丢生产至少上 SQLiteQPS 再高就换 Postgres。异步 SQLite 写入在高并发下有瓶颈如果你的服务要扛住几十路并发建议直接上AsyncPostgresSaver连接池配好别在小文件数据库上死磕。FastAPI 本身是异步的LangGraph 也支持ainvoke和astream所以整个链路天然是异步的。但要注意 LLM 调用本身是 IO 密集操作如果某个第三方 SDK 是同步阻塞的最好放进线程池或者用asyncio.to_thread包一层别卡住事件循环。4.2 Checkpointer 与人工介入LangGraph 的 checkpointer 不只是为断点续传设计的它给多智能体项目带来了两个特别重要的能力人机协同和故障重放。人机协同用interrupt()实现from langgraph.types import interrupt def review_node(state: AgentState) - dict: user_decision interrupt( {draft: state[draft], question: 这份方案是否可以发布} ) if user_decision approve: return {status: approved} return {status: rejected, reason: user_decision}这个函数执行到interrupt时会暂停把当前状态持久化到 checkpointer然后返回给调用方一个待恢复的句柄。后续通过Command(resume...)或者再次 invoke 带回用户的决定图会从暂停的位置继续执行。整个过程中其他节点不会空跑非常适合智能体出草稿、人类做审批这类半自动流程。我在内部流程自动化那个项目里审批环节就这么做的效果非常好——机器做不了决定的时候停下来问人而不是硬着头皮编。故障重放就更实用。生产环境出了错拿着thread_id用同一个图实例重新 invoke能把每一次节点执行的状态都翻出来精确到哪个节点改了哪个字段。这个能力让多智能体系统的排错成本下降了一个量级我是真的依赖它。4.3 流式输出与并发控制多智能体系统延迟高用户体验全靠流式输出兜底。LangGraph 支持多种流式模式最常用的是astreamasync for chunk in GRAPH.astream( {messages: [{role: user, content: query}]}, configconfig, stream_modemessages ): if chunk[0].content: yield fdata: {chunk[0].content}\n\nstream_modemessages会逐 token 地输出 LLM 生成内容前端用 SSE 接住就行。但我的经验是多智能体场景下只吐最终答案不够最好把当前在哪个环节也一起吐出来。比如用stream_modeupdates捕获每个节点完成时的状态变化前端显示正在调用数据分析工具…正在生成评估报告…这类进度提示。用户知道系统还在干活耐心会显著提升——这一点在延迟动辄十秒以上的多智能体系统里几乎决定产品口碑。并发控制方面多个用户共享同一个 graph 实例加持久化 checkpointer 是安全的因为状态隔离在 thread_id 上互不干扰。真正的瓶颈在 LLM API 的 rate limit 和数据库连接池。建议给所有 LLM 调用加统一的限流中间件给 checkpointer 配连接池并给每个请求设置总超时上限比如 60 秒超时就返回一个友好的降级提示而不是让用户无限等待。5. 可观测性与调试多智能体系统排错的手段5.1 状态快照排查法多智能体系统排错第一步永远是拿到这个请求完整跑过的状态快照。LangGraph 的 checkpointer 天然支持这一点不用额外做埋点config {configurable: {thread_id: demo-001}} states list(GRAPH.get_state_history(config)) for state in states: print(state.values.get(messages)[-1].content[:100]) print(--- next:, state.next)get_state_history返回从最新到最早的所有状态版本每个版本包含当时的完整状态值和下一步要执行的节点。这几乎等于给每个请求装了行车记录仪任何一个中间结果出了问题都能精确地从历史里找到那一帧。我处理线上问题的标准流程是先复制 thread_id拉状态历史看最后一条消息是谁写的、当时图停在哪个节点然后决定是重放、修正状态、还是直接恢复。这个流程我大概用了两个月就形成了肌肉记忆效率提升非常明显。团队里其他人也慢慢养成了这个习惯——遇到问题先拉状态而不是瞎猜模型。5.2 结构化追踪与日志光有状态快照还不够多智能体系统的日志必须结构化。我维护一份统一格式的运行日志每条记录包含这些字段字段说明run_id整次请求的追踪 ID前端透传进来node_name当前节点名agent_role该节点所属的智能体角色model_name用到的模型input_tokens / output_tokenstoken 消耗latency_ms节点耗时status成功、失败或超时用 Python 的 logging 打全这些字段或者直接发到 OpenTelemetry 都行。关键是每次 LLM 调用前后都把日志打全这样后续做成本分析、延迟分析、错误归因都有数据可查。LangSmith 如果预算允许也可以接它能把图中的每次调用串成 trace可视化程度很高。但我的经验是别只依赖它。第三方平台可能因为网络问题丢 trace自己维护的结构化日志才是排错的第一手资料。5.3 两个典型的线上故障说两个我实际遇到过的故障给大家提个醒。第一个是死循环。某个 agent 节点在条件边里判断是否还有未处理的问题LLM 连续判定还有于是反复调用自己直到触发超时。排查时拉状态历史发现同一个节点执行了 40 多次。解决方法是给循环加计数器字段——每次进入循环加一超过阈值就直接走出口边不让 LLM 无限自我延伸。这是多智能体系统的必修课任何一个循环都必须有硬上限。第二个是结果污染。两个 worker 同时更新 messages 字段因为 reducer 写错成覆盖模式后完成的 worker 把先完成的覆盖掉了。排查时靠状态快照一眼就看出 messages 少了内容修复 reducer 就解决了。这类问题在单智能体里根本不存在多智能体里却很容易犯——并发更新共享状态reducer 语义不正确数据就悄悄丢了。6. 性能与成本优化低延迟不是玄学6.1 串行转并行多智能体延迟高的最大来源是串行执行。很多流程里两个子任务其实互不依赖但默认写法是一个接一个跑白白浪费时间。LangGraph 支持并行节点——只要两个节点之间没有依赖边它们会在同一个 super-step 里并发执行from langgraph.graph import StateGraph, START builder StateGraph(AgentState) builder.add_node(agent_a, agent_a) builder.add_node(agent_b, agent_b) builder.add_node(merge, merge_results) builder.add_edge(START, agent_a) builder.add_edge(START, agent_b) builder.add_edge(agent_a, merge) builder.add_edge(agent_b, merge)agent_a和agent_b并行执行merge等两个都完成后再跑。对于可以拆分的任务这一步能把总延迟从两个节点耗时相加变成取最大值。我在数据分析项目里把一轮请求从 12 秒压到 7 秒靠的就是这个改动一行 prompt 没动。改之前总觉得慢是模型的问题改完之后才意识到是架构的问题。当然并行不是银弹。两个节点写同一个字段时就别并行要么用 reducer 合并要么串行。并行之前先确认状态依赖关系别为了快把正确性搭进去。6.2 模型分工大模型做决策小模型干杂活另一个降低延迟和成本的关键是别让所有节点都用同一个旗舰模型。我的原则路由、分类、信息抽取这类简单任务用快的小模型比如 7B 级别的开源模型或者 API 的低档位型号复杂推理、生成最终方案用最强的大模型工具调用相关的模型至少用中档以上因为工具调用对指令遵循要求高。LangGraph 的节点函数里可以自由选择模型所以在图设计阶段就要把每个节点的模型选定。我算过一笔账一个五节点的流程把其中三个节点的模型从旗舰换成中档成本降了六成以上整体质量几乎不受影响。因为那些节点本来就是干脏活累活的大模型反而容易想太多输出不稳定。6.3 token 节省的几个具体手段最后说几个 token 节省的具体手段都是我在项目里验证过有效的状态瘦身。别把所有历史消息都塞进共享状态。定义 state 时区分短期工作记忆和长期记忆工作记忆在节点完成后可以裁剪掉只保留必要的结果摘要。压缩摘要。如果确实需要保留长对话历史定期用小模型把前面的消息压成摘要只保留最近 N 轮原文。缓存工具结果。查询型工具查数据库、查文档在同一个 session 内按参数哈希做缓存避免同一份数据被多个节点重复拉取。这个优化在工具调用密集的智能体上效果特别明显。精简工具描述。每多一个工具描述模型每轮调用都要多读一遍token 成本是累计的。只暴露当前节点真正需要的工具别把所有工具全部绑给每个智能体。7. 踩坑实录与我的几点建议7.1 五个常见的坑整理一下我见过、踩过最多的五个坑新团队照着避雷就行把 LangChain 的 Chain 思维带进 LangGraph。想到串联就连一条线最后画出一个巨大的线性链分支和循环能力完全没用上图的优势荡然无存。共享状态里塞太多东西。什么都往 state 里堆模型每次请求都要读一整套无关上下文token 飙升、响应变慢。状态设计要克制只放真正跨节点流转的数据。忽略 checkpointer 配置。demo 用 MemorySaver 没问题上了生产忘了换持久化重启丢状态所有用户会话全部断掉。这个问题我在生产环境见过不止一次。不给循环设上限。agent 卡在我还能再做一步的幻觉里直到超时才停下来。所有循环必须有最大轮次这是铁律。所有节点用同一个模型。又贵又慢简单任务用大模型反而容易想太多、输出格式不稳定。模型分工要趁早做。7.2 让多智能体系统可回归多智能体系统最让人头疼的是不确定性同样的输入可能跑出不同的结果。这对测试是个大挑战我的做法是给测试加确定性单元层测试节点函数。节点是普通函数直接 mock LLM 返回值验证状态更新逻辑和路由逻辑不依赖真实模型。集成层录制回放。用 VCR 风格的录制库把 LLM 和工具的响应录下来测试时回放保证流程可复现。这个库很多语言都有Python 里有现成的。回归基线。准备一组标准 query每次改动后跑一遍对比关键指标答案是否包含核心要点、是否调用了预期工具、有没有死循环、延迟有没有劣化。线上灰度。任何图的修改先切 10% 流量对比 token 消耗、延迟、用户反馈确认没问题再全量放开。这几件事做完多智能体系统才敢说可维护。否则每次改 prompt 都像重新开一个项目——这不是夸张是真实体感。7.3 最后几句真心话LangGraph 这套框架的价值不在于它让我们用上了多智能体这个词而在于它逼着你把业务流程画成一张状态图逼着你考虑状态、边界、恢复这些工程问题。这些素养对任何复杂系统都是通用的不亏。如果你正准备开始我的建议是先拿一个月只做一个小而真的场景——一个 supervisor 加两个 worker生产部署、流式输出、结构化日志一样不落地跑起来比在 Notebook 里搭十个花架子 demo 有用得多。等这个最小闭环稳了再往里面加模式、加节点、加优化。真实流量的反馈永远比精心设计的 demo 案例更能暴露问题。
