企业官网建设流程全解析

做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(model="gpt-4o", temperature=0) 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_count，Agent节点每次执行时判断次数，如果超过限制直接返回一条“我无法完成此操作”的文本消息，不再绑定工具。这样条件路由仍然用tools_condition，但“是否继续调用工具”由Agent自己根据state决定，逻辑更清晰。

如果你确实想用自定义条件边，记住了：条件边函数只读取state并返回节点名，里面不要做复杂计算、不要发起外部请求，否则每个节点流转时都会多一次不可控的IO。

2.3 状态与持久化：从MemorySaver到数据库Checkpoint

工具调用图跑通后，接着要解决会话记忆问题。默认compile出来的是内存态，重启后一切归零。为了能在对话中传递历史，你需要给LangGraph加Checkpointer。

最省事的是MemorySaver：

from langgraph.checkpoint.memory import MemorySaver graph = builder.compile(checkpointer=MemorySaver()) config = {"configurable": {"thread_id": "session-001"}} result = graph.invoke( {"messages": [("user", "DV-1001状态正常吗？")]}, config=config )

这个thread_id就是会话ID。同一个thread_id继续invoke，LangGraph会自动把历史消息加载进状态。生产环境一般不用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)]}, config=config )

所有会话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)]}, config=config, version="v2" ): if event["event"] == "on_chat_model_stream": chunk = event["data"]["chunk"] if chunk.content: yield f"data: {chunk.content}\n\n" return StreamingResponse(event_stream(), media_type="text/event-stream")

这里我用了version="v2"，这是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)]}, config=config ) 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本身就是一个Agent+Tool循环，内部有自己的状态管理。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落地最常见的几类问题。

这个表不是凭空总结的，每一条都有对应的线上事故。我印象最深的是“会话串联”那次：当时图里用了全局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支持传入debug=True，会打印每一步的事件和状态变化。我遇到玄学问题时会先开启这个，把完整调用链拉出来，基本能定位到是模型返回了空消息还是工具返回了意外结构。

5.3 上线后的稳定性维护

上线不等于完事。LangGraph多Agent服务的稳定性，需要一套例行动作来维持。

第一，画一张“Agent版本变更表”。每次修改任何一个子Agent的prompt、工具、模型参数，都要记录线上行为是否有变化。因为多Agent系统的效果是由所有节点共同决定的，某个节点微调可能带来下游行为漂移，必须有回放机制兜底。

第二，做长尾输入监控。我在图执行层加了一个记录器，凡是用户问题触发了异常分支或者最终结果为空，都会单独落库。每周末分析这批数据，看哪些问题是模型没理解、哪些是工具数据缺失、哪些是状态管理缺陷。这种持续循环比任何一次大重构都有效。

第三，给图执行层加熔断。当某个工具连续失败达到阈值时，我让服务自动把对应Agent切换为“只读模式”，不再发起真实写操作，只返回缓存结果或提示人工介入。这样至少不会在问题扩大时把下游系统一起拖垮。

多Agent不是银弹，它把单Agent的“模型不可控”变成了“流程可控”和“单元可控”。我现在的体会是：能用单Agent解决的，绝不强行上多Agent；必须上多Agent的场景，优先保证每个子Agent简单、独立、可回放；最终的稳定性不靠模型有多聪明，而靠你愿意做多少工程加固。以上这些实践，都是从一次次线上事故里换来的，希望你能少踩几次。