做LangGraph多智能体落地这段时间踩过的坑比想象多。今天不聊概念直接聊工程实践怎么把LangGraph从demo变成能扛住生产的系统覆盖工具调用、FastAPI服务化、多智能体协同以及电网可靠运行这类真实行业场景。如果你已经会用LangChain写单轮Agent但不确定多智能体怎么编排、状态怎么管理、服务怎么部署、出问题了怎么排查那这篇可以当一份实战手册看。文里没有炫技都是我实际跑过的方案、调过的参数、踩过的坑。1. 为什么需要多智能体编排从单Agent到系统化协作1.1 单个Agent搞不定的三类问题先说实话不是所有业务都需要上多智能体。我见过不少团队一个ChatOpenAI就能解决的问题硬拆成三个Agent最后状态对齐、上下文传递、日志追踪全部乱套。真正需要多智能体协作的通常逃不过这三类问题。第一类是上下文过长。单个Agent一旦把工具返回、历史记录、外部知识全塞进一个prompt很快就把上下文窗口撑爆。比如电网设备巡检场景一个设备的历史告警、实时测点、检修记录加起来几万字如果只有一个Agent处理要么截断丢掉关键信息要么烧钱换大上下文模型。拆成“数据检索Agent”和“研判Agent”前者只负责捞数据并做精简摘要后者拿到的就是干净、有限长度的结果问题立刻变得可控。第二类是工具权限和职责边界无法收敛。一个Agent绑了十几个工具模型很容易在无关工具之间乱跳。我实测过工具超过8个后调用准确率明显下降尤其是两个工具的功能描述相近时模型经常选错。多智能体协作的意义不是把工具分给不同Agent就完了而是让每个Agent只维护一个很小、很明确的工具集比如“负荷预测Agent”只绑趋势查询和天气接口“报告生成Agent”只绑文档模板和格式化工具。这样每个Agent的决策空间被刻意压缩反而更稳定。第三类是流程需要人为干预和审批。很多生产环境不允许Agent自己一条路走到底中间需要校验、审核、人工确认。单Agent流程中这些分支逻辑都写在判断语句里代码越来越难看而且要单独维护一套状态机。LangGraph把这种流程变成了有向图节点之间谁先谁后、要不要走条件分支一目了然天然适合这类活儿。1.2 LangGraph能提供的核心能力LangGraph本质上是一套基于图的状态编排框架。你不需要重新发明状态机只需要定义State、节点和边。State是全局共享的数据结构节点是执行逻辑的单元边决定了节点之间的流转路径。相比自己手写while循环和if判断它最大的价值是把“Agent运行过程”变成了可描述、可控制、可恢复的图。我最常用的几个能力条件边根据上一步Agent的输出决定下一步走哪个节点比如“是否调用工具”“是否需要人工审核”。检查点Checkpoint每一步执行完后把状态持久化到存储中进程崩溃后可以从最近一个节点恢复。流式输出支持token级别的流式回调做对话型应用时体验至关重要。可观测性每一步的状态变化、节点耗时、事件流都能拉出来方便排查问题。还有一点容易被忽略LangGraph的多智能体和LangChain的AgentExecutor不是替代关系LangGraph是更底层的编排器也就是说你可以用LangGraph托起多个传统Agent。这意味着团队不需要推翻已有代码可以先在LangChain里保留工具调用逻辑再在外面套LangGraph的图结构把迁移成本降到最低。2. 工程实践一可观测的Agent工具调用链路2.1 一个最简工具调用图先给一个最基础的工具调用图。我建议所有团队都从这一步开始跑通别一上来就套Supervisor、分层、群聊那些高级模式。这个图做的事情很简单Agent判断要不要调用工具要就进工具节点工具返回后再回到Agent直到模型不再请求工具才结束。from typing import Annotated, TypedDict from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langgraph.prebuilt import ToolNode, tools_condition from langchain_openai import ChatOpenAI from langchain_core.tools import tool tool def get_device_status(device_id: str): 查询设备实时状态。device_id是设备编号例如DV-1001。 return {device_id: device_id, status: normal, load: 0.72} tools [get_device_status] class AgentState(TypedDict): messages: Annotated[list, add_messages] def agent_node(state: AgentState): model ChatOpenAI(modelgpt-4o, temperature0) model_with_tools model.bind_tools(tools) result model_with_tools.invoke(state[messages]) return {messages: [result]} builder StateGraph(AgentState) builder.add_node(agent, agent_node) builder.add_node(tools, ToolNode(tools)) builder.add_edge(START, agent) builder.add_conditional_edges( agent, tools_condition, {tools: tools, END: END} ) builder.add_edge(tools, agent) graph builder.compile()这里有几个关键点。State里的messages字段用了add_messages这个reducer它表示每次节点返回的消息不是覆盖旧消息而是追加到历史列表里。如果你自己定义状态千万别忘了给列表字段配reducer否则LangGraph默认会用新值覆盖旧值对话历史就丢了。tools_condition是LangGraph预置的条件路由函数模型返回的消息里有tool_calls就走tools没有就走END。我见过有人自己写判断逻辑其实没必要预置函数已经处理了边界情况直接用就行。2.2 条件边为什么用tools_condition而不是自己写刚开始我嫌预置条件太死板想自己控制“最多只能调两次工具”于是写了个自定义条件函数结果踩了深坑。LangGraph的条件边函数接收当前状态返回要去的节点名字。看起来很简单但在Agent Tool循环中你需要判断“最新一条AI消息里有没有tool_calls”而消息的存储结构、tool_calls的嵌套格式在LangChain不同版本里都有过调整。自己解析一是代码脆二是错误处理容易漏。后来我改成在Agent节点里加标记字段比如在state里维护tool_call_countAgent节点每次执行时判断次数如果超过限制直接返回一条“我无法完成此操作”的文本消息不再绑定工具。这样条件路由仍然用tools_condition但“是否继续调用工具”由Agent自己根据state决定逻辑更清晰。如果你确实想用自定义条件边记住了条件边函数只读取state并返回节点名里面不要做复杂计算、不要发起外部请求否则每个节点流转时都会多一次不可控的IO。2.3 状态与持久化从MemorySaver到数据库Checkpoint工具调用图跑通后接着要解决会话记忆问题。默认compile出来的是内存态重启后一切归零。为了能在对话中传递历史你需要给LangGraph加Checkpointer。最省事的是MemorySaverfrom langgraph.checkpoint.memory import MemorySaver graph builder.compile(checkpointerMemorySaver()) config {configurable: {thread_id: session-001}} result graph.invoke( {messages: [(user, DV-1001状态正常吗)]}, configconfig )这个thread_id就是会话ID。同一个thread_id继续invokeLangGraph会自动把历史消息加载进状态。生产环境一般不用MemorySaver因为它是纯内存多实例部署时各实例之间状态不同步。我更推荐用SqliteSaver或Postgres的checkpointer把状态持久化到共享存储。切换checkpointer时有两个小坑。第一状态里的数据必须是可序列化的像自定义对象、数据库连接这类东西不能直接放State里否则checkpoint写入会失败。第二不同checkpointer对并发会话的支持不一样SqliteSaver默认是线程锁高并发下要开WAL模式或用PostgresSaver。我上线前压测时就是这个锁导致同时访问同一个thread_id直接报错后来才知道同一thread_id本来就不该被并发调用前端必须对用户点击做防抖。3. 工程实践二FastAPI封装LangGraph服务3.1 服务化分层设计图在本地跑通后接下去就是把它变成一个可以被Web前端、后端服务调用的接口。我习惯把服务拆成三层API层、业务编排层、图执行层。API层负责鉴权、参数校验、SSE流式传输业务编排层负责组装会话ID、拼接系统提示词、处理业务异常图执行层只负责调用LangGraph的invoke或astream。别把LangGraph对象直接暴露在路由函数里。我见过一个项目FastAPI启动时生成graph实例然后在每个请求里直接调用graph.invoke结果几个线上问题全搅在一起没有统一的会话ID生成规则、没有超时控制、没有日志埋点出了问题只能靠猜。我现在的做法是写一个GraphRunner类class GraphRunner: def __init__(self, graph): self._graph graph async def run(self, session_id: str, user_message: str): config {configurable: {thread_id: session_id}} return await self._graph.ainvoke( {messages: [(user, user_message)]}, configconfig )所有会话ID生成、上下文清理、异常转换都在GraphRunner内部做。业务层只调这个类不碰LangGraph的API。这样后面换模型、改prompt、调整工具都不需要动FastAPI路由。3.2 SSE流式输出的实现对话型Agent最在意响应速度。用普通POST等完整回复用户等10秒才看到第一个字体验很差。FastAPI里做流式输出很方便配合LangGraph的astream_events可以做到边生成边推送给前端。from fastapi import FastAPI from fastapi.responses import StreamingResponse from pydantic import BaseModel app FastAPI() class ChatRequest(BaseModel): session_id: str message: str app.post(/chat) async def chat(req: ChatRequest): async def event_stream(): config {configurable: {thread_id: req.session_id}} async for event in graph.astream_events( {messages: [(user, req.message)]}, configconfig, versionv2 ): if event[event] on_chat_model_stream: chunk event[data][chunk] if chunk.content: yield fdata: {chunk.content}\n\n return StreamingResponse(event_stream(), media_typetext/event-stream)这里我用了versionv2这是LangChain新版本推荐的协议初版协议字段结构混乱很难解析。注意on_chat_model_stream事件拿到的chunk里不一定只有content有些模型还有tool_call片段如果你不判断chunk.content可能会往前端推一坨空行。SSE还有一个坑中间如果节点调用了工具会有几秒“静默期”因为工具执行期间没有token输出。前端如果判断超时断开连接就白等了。实践上我会在业务编排层加一个“开始调用工具”的提示事件通过SSE发一个注释行或者特殊标记让前端知道自己还在处理中。3.3 超时、并发和连接池的工程处理LangGraph执行一次复杂任务可能涉及多次大模型调用和工具访问单次可能超过30秒。如果FastAPI请求一直开着连接池会被占满。首要方案是给每次业务执行加总超时比如用asyncio.timeout控制。import asyncio class GraphRunner: async def run_with_timeout(self, session_id: str, user_message: str, timeout: int 30): config {configurable: {thread_id: session_id}} try: async with asyncio.timeout(timeout): return await self._graph.ainvoke( {messages: [(user, user_message)]}, configconfig ) except TimeoutError: # 记录日志并返回可读的降级消息 return {fallback: 处理超时请稍后重试}超时不能一刀切。如果会话里已经有大量历史消息模型推理时间会明显变长尤其是多Agent场景每个子任务都要调模型总耗时可能是单Agent的几倍。我一般会把超时设置为“预估步骤数 × 单步最大耗时”再乘1.5的余量。预估步骤数可以靠图结构的节点数量估但更准的方式是上线后按真实分位数动态调整。并发控制也需要单独做。LangGraph内部如果使用MemorySaver多请求并发时容易出现状态覆盖。我的做法是在服务层按thread_id加一个简单的异步锁保证同一个会话同时只允许一个任务在执行。不同会话之间不需要锁它们彼此独立。这个方案比全局限制并发量更实用。4. 工程实践三多智能体协同的电网可靠运行场景落地4.1 场景拆解电网可靠运行需要什么样的多Agent电网可靠运行是一个很有代表性的多智能体场景因为它天然分层底层有大量设备状态数据中层需要对故障进行研判上层需要生成调度预案和报告。如果只做一个大Agent让它既查数据、又做分析、还要写预案prompt会变成几十页而且一旦领域数据更新维护成本极高。我参与过的项目是这样拆解的四个专职Agent加一个协调Agent。数据采集Agent只绑设备台账、测点查询、告警记录三类工具负责把用户问题转成标准查询并把结果压缩成结构化摘要。故障研判Agent绑历史故障库和诊断规则工具输入是数据采集Agent给出的摘要输出是可能的故障原因和置信度。负荷预测Agent绑气象接口、历史负荷数据工具负责预测未来时段负荷给调度决策提供边界条件。报告生成Agent绑文档模板工具把前面所有Agent的结论转成标准格式的运维报告。协调Agent是唯一的入口它先判断用户问题属于数据查询、故障研判、负荷预测还是完整报告生成再决定哪些Agent按什么顺序执行。这个结构最大的好处是每个Agent的prompt都很短工具集很小模型调用准确率高。4.2 基于Supervisor的分层协同架构我用LangGraph实现的是典型的Supervisor模式协调Agent作为监督者其他Agent作为执行者。状态流转大致是用户请求进入协调Agent协调Agent维护一个任务清单逐个调用执行Agent最后汇总结果并输出。关键实现是执行Agent也被建模成子图。比如故障研判Agent本身就是一个AgentTool循环内部有自己的状态管理。LangGraph允许节点里直接调用另一个编译好的图这样每个子Agent可以独立测试也能复用。我当时用了一个invoke嵌套的方式# 子Agent图提前compile好 diagnosis_graph diagnosis_builder.compile() def diagnosis_node(state): # 重新组装子图需要的输入 result diagnosis_graph.invoke( {input: state[summary]}, config{configurable: {thread_id: state[thread_id] -diag}} ) return {diagnosis_result: result[output]}这里有一点必须提醒子图嵌套时主图和子图共用同一个State对象容易出现字段名冲突。我的习惯是给子图的state字段加上前缀比如summary、diagnosis_result不让字段跨层重名。否则主图状态里一旦出现messages字段子图也操作messages两条执行链路的消息就会混在一起排查起来极其痛苦。4.3 协同运行中的容错与降级设计多Agent链路越长失败概率越大。任何一个子Agent的模型调用超时、工具报错、输出格式异常都可能让整个流程中断。我在这个场景里做了三层容错。第一层是工具调用兜底。每个工具函数都要捕获异常并返回结构化错误信息不要让异常直接抛到图里。比如数据库查询失败工具返回{error: 查询超时}Agent看到这个结果后会决定重试还是换一种问法而不是整条链路崩溃。第二层是子Agent重试。对故障研判Agent这种核心节点我用LangGraph的重试机制最多重试两次。你可以给节点单独配置retry策略比如指数退避。这个效果很明显很多一次性超时重试后都能恢复正常。第三层是降级策略。协调Agent要能识别“某个子Agent结果为空”的情况选择用其他Agent的输出来补充或者直接向用户反馈“当前数据不足建议人工复核”。从业务角度看比返回一堆错误堆栈更负责任。我还做了一步很关键的验证用历史故障数据回放。把过去一年有代表性的故障记录、对应工具返回结果、最终处置报告整理成测试集每次改动图结构或prompt都跑一遍回归。这是多Agent系统上线前的底线没有回放测试你根本不知道改了一个Agent的prompt会不会影响另一个Agent的行为。5. 常见问题与排查技巧实录5.1 典型问题速查表这里整理一份我踩过的坑速查表基本覆盖LangGraph多Agent落地最常见的几类问题。问题现象解决方案状态字段被覆盖工具调用后历史消息丢失Agent“失忆”检查State里列表字段是否配置了add_messages等reducer并发会话互相干扰A会话的回答串到了B会话thread_id没有隔离或使用了单例checkpointer确保每个会话有独立thread_id工具返回内容过大单次网络请求超时整个Agent卡死在工具节点里限制返回条数及字段大小做分页或截断模型反复调用同一工具死循环token消耗翻倍在Agent节点维护工具调用次数达到阈值后强制停止子Agent输出主Agent看不懂下游节点收到非预期格式json解析失败统一子Agent的输出schema用Pydantic模型约束并在prompt里给示例SSE前端显示中断工具调用期间无输出前端判定超时在SSE中发送处理中事件或心跳包模型输出tool_calls但工具报错工具不存在或参数不匹配清空工具缓存检查工具函数签名和docstring避免同名工具这个表不是凭空总结的每一条都有对应的线上事故。我印象最深的是“会话串联”那次当时图里用了全局MemorySaver测试人员同时开了两个浏览器窗口线程ID生成规则没做好导致两个用户的对话串了。后来改成UUID且从请求头里严格提取用户标识才彻底解决。5.2 调试LangGraph图的方法LangGraph调试最大的痛点是“不知道现在走到哪个节点了”。幸运的是编译后的图对象自带可视化能力你可以输出图结构检查。# 打印图的文本结构 print(graph.get_graph().print_ascii()) # 或者保存为图片 png_data graph.get_graph().draw_mermaid_png() with open(graph.png, wb) as f: f.write(png_data)我每次改完图都会先生成一张图片肉眼确认边的连接和条件分支有没有接错。这比单纯看代码直观得多尤其是Supervisor模式节点多、边多光靠记忆很容易漏一条。运行时的调试我强烈建议用LangSmith或者至少给每个节点加日志。最简单的做法是在节点函数里打印节点名和当前状态摘要def agent_node(state): print([agent_node] start, messages count:, len(state[messages])) # ...多Agent场景下日志里要带上thread_id和节点名这一步不能省。否则线上排查的时候十几个并发一起打日志你根本分不清哪条日志属于哪个用户。还有一个技巧在线下复现问题时可以用graph.invoke的debug模式LangGraph支持传入debugTrue会打印每一步的事件和状态变化。我遇到玄学问题时会先开启这个把完整调用链拉出来基本能定位到是模型返回了空消息还是工具返回了意外结构。5.3 上线后的稳定性维护上线不等于完事。LangGraph多Agent服务的稳定性需要一套例行动作来维持。第一画一张“Agent版本变更表”。每次修改任何一个子Agent的prompt、工具、模型参数都要记录线上行为是否有变化。因为多Agent系统的效果是由所有节点共同决定的某个节点微调可能带来下游行为漂移必须有回放机制兜底。第二做长尾输入监控。我在图执行层加了一个记录器凡是用户问题触发了异常分支或者最终结果为空都会单独落库。每周末分析这批数据看哪些问题是模型没理解、哪些是工具数据缺失、哪些是状态管理缺陷。这种持续循环比任何一次大重构都有效。第三给图执行层加熔断。当某个工具连续失败达到阈值时我让服务自动把对应Agent切换为“只读模式”不再发起真实写操作只返回缓存结果或提示人工介入。这样至少不会在问题扩大时把下游系统一起拖垮。多Agent不是银弹它把单Agent的“模型不可控”变成了“流程可控”和“单元可控”。我现在的体会是能用单Agent解决的绝不强行上多Agent必须上多Agent的场景优先保证每个子Agent简单、独立、可回放最终的稳定性不靠模型有多聪明而靠你愿意做多少工程加固。以上这些实践都是从一次次线上事故里换来的希望你能少踩几次。