不用理论模型就聊真实落地。最近半年我们团队在电力调度辅助决策系统里用 LangGraph 把一套多智能体协作流程推上了生产环境。这中间经历了从迷恋 AutoGen 的群聊机制、到被复杂对话轮次折磨再到回归 LangGraph 的显式图控制最终稳定支撑每日数千次查询的过程。这篇文章不聊概念只讲几个只要做多智能体项目就绕不开的工程问题拓扑怎么选、状态怎么管、工具调用怎么不跑飞、断点续跑和并发怎么处理以及凭什么敢让 AI 直接面对生产数据。1. 为什么最终选型 LangGraph被 AutoGen 的轮次机制坑过一次之后先说结论如果你的智能体之间不是平等的闲聊关系而是有明确的主从、上下游、审批链LangGraph 的显式图结构比 AutoGen 的自动对话轮次可控得多。我们最早的原型用的是 AutoGen看中的是它的 GroupChat 机制——多个 Agent 像会议室里的人一样自由发言听起来很美好。但真跑到第二轮问题就来了轮次控制极其痛苦。你要么设置max_round让它自己聊结果它可能在一个分支里反复横跳要么手动干预发言顺序那基本等于放弃了框架的全部优势。而且 AutoGen 的对话历史是线性累积的一旦某个分支错误整个上下文的可信度都受影响没法精准回退。LangGraph 不一样。它把整个协作过程建模成一个有向图每个节点是一个处理函数每条边是条件路由。你可以精确控制谁在什么条件下调用谁每一步的输入输出都显式定义。这个特性对电网调度这种场景至关重要——我们不允许两个模型自由讨论后自行决定倒闸操作顺序每一步必须可控、可审查、可回退。另外一个实际原因是状态管理。LangGraph 的StateGraph围绕一个全局状态对象做流转每个节点返回的 dict 会合并进状态。这意味着你可以把原始查询、中间检索结果、每个智能体的结论、最终响应全部保留在状态里随时取用。AutoGen 在这方面的抽象要弱得多你主要靠 conversation history 传递信息结构化数据反而难放进去。注意选型时别只看 Demo 效果。AutoGen 的群聊在演示场景里非常惊艳但惊艳恰恰来自它的自由度。生产系统要的是约束不是自由。2. 多智能体拓扑设计我们最终选定的是路由 执行 审查三层结构LangGraph 官方文档里画了 Supervisor、Hierarchical、Handoff 好几种拓扑OpenAI 的 Swarm 还带了个 Agent 间交接的概念。但落到真实业务里我建议你忘掉这些花哨名词先想清楚一个问题你的系统里到底谁对最终结果负责我们第一版抄了Supervisor 多个 Worker的官方示例让一个调度员 Agent 决定把任务分给谁。很快发现问题Supervisor 的路由决策本身就会出错而且它错得毫无规律——有时候把设备状态分析任务分给了气象 Agent有时候又把气象任务分给了拓扑分析 Agent完全看模型当时的情绪。后来改成三层结构稳定下来了路由层Router只做一件事把用户查询分类到明确意图槽位。上下文只有一条系统提示 当前的用户原始输入不携带其他任何 Agent 的中间结果避免干扰。执行层Workers每个 Worker 是单一职责的专家型 Agent只处理一个领域设备台账、气象预警、拓扑分析、调度建议。审查层Reviewer一个独立的质检 Agent检查执行结果的完整性和内部一致性。这个结构的核心思想是路由决策不依赖上下文执行过程不跨域审查结果不参与执行。每一层都是独立可测的出了问题你知道是哪个环节的锅。实际运行下来Router 的准确率稳定在 97% 以上在 300 条测试集上Workers 各自只处理自己的领域Prompt 可以写得很聚焦。2.1 为什么不做 Chat HandoffHandoff 模式在客服场景很好用Agent 之间可以互相转单。但我们做下来发现这是把上下文传递错误问题从单 Agent 内部扩散到了整个系统——A 转给 B 时A 的中间推理过程如果带了错误信息B 无法察觉因为它默认 A 是可信的。在电力这种容错率极低的行业任何一环不可信都不可接受。宁可做显式的重新检索 - 独立推理 - 交叉验证也不做隐式的信息交接。2.2 状态字段按需最小化而不是把所有数据塞进去LangGraph 的 State 是个类字典结构很灵活但也容易让人滥用。我们初版把所有 Agent 的输出都平铺进 state后来状态里堆了二十多个字段有 AssistantMessage、HumanMessage、Agent 专属记忆、工具返回的长文本。Prompt 组装时不仅要筛字段还得注意消息顺序维护成本直线上升。现在我们的 State 只保留三类字段query原始查询永不改变intermediate_results检索、工具调用的结构化返回按产生顺序追加agent_outputs每个 Agent 的结论以 Agent 名字为 key消息列表不直接存全局 state而是每个节点内部自行组装。这样状态对象体积可控传给 LLM 的 token 消耗也降了大约 40%。3. 工具调用Tool Calling在多智能体里的工程化从裸调 Prompt 到受控执行多智能体能落地靠的不是 Agent 聊天能力而是工具调用质量。LangGraph 本身不提供工具注册体系它只是图编排引擎。你仍然需要自己管理 Function Calling 的 Tool Schema、执行过程、错误恢复。这一块我们踩的坑最深也是回报率最高的优化点。3.1 Tool Schema 必须用独立字段描述参数约束别依赖模型理解焱联网上很多 LangGraph 教程喜欢用tool装饰器一把梭函数签名直接给模型看。浅层 Demo 可以生产环境不行——模型会自由发挥参数。我们的设备台账工具接受device_id初版 schema 只写了设备的唯一标识结果 model 传了设备名称、设备 IP、甚至经纬度五花八门。后来所有工具的描述改成三个强制字段required、additionalProperties: false、以及每个参数的description都以枚举形式列出取值范围。OpenAI 和新浪的 Function Calling 都原生支持 JSON Schema你只要认真写 schema模型就不会乱填参数。实测 device_id 传错率从 18% 降到 2% 以内。3.2 工具执行必须和 Agent 状态隔离LangGraph 节点里执行工具时有个隐患工具异常如果直接抛出会污染整个 state。比如查询气象数据库超时异常信息一旦被塞进 ChatHistory模型会以为工具调用了但没返回说明该区域无气象数据然后基于这个错误认知继续推理产出一个看似合理实则错误的结论。我们的做法是给工具执行加一层 wrapper超时、限流、数据结构校验异常全部转化为结构化错误码返回错误详情单独存到intermediate_results的error字段同时附带上一次成功调用的缓存结果如果有。模型看到的是工具执行失败原因 code_5002近一次成功数据是 xxx它可以决定是否重试或换工具但系统不会因为一次工具异常就丢失上下文。3.3 并发工具调用一次性并行 vs 逐步串行LangChain 提供的bind_toolstool_results支持一次返回多个工具调用请求LangGraph 里可以并行执行。真实场景里我们推荐的做法是同域类的工具并行跨域类的工具串行。比如查询同一个设备的台账、实时状态、历史检修记录这三个是独立的并行没问题省了 2/3 的延迟。但先查设备状态再根据状态生成检修指导意见这个链路本质上有依赖关系必须串行。你可以在 LangGraph 里用两个节点分步处理或者在单个节点内控制工具执行顺序。我们统一在节点内部处理这样状态流转的粒度不用拆得太碎同时保持 LangGraph 图结构的清晰。4. 状态持久化与断点续跑Human-in-the-Loop 的真正工程实现多智能体系统跑在你的服务器上但最终决策可能落在人手上。电网调度场景里Agent 生成的倒闸操作票、检修建议绝对不允许自动执行必须等值班调度员确认。这就是典型的 Human-in-the-LoopHITL需求。LangGraph 的checkpointer提供了最基础的断点机制图执行到任意节点时暂停状态序列化到持久化存储默认 SQLite生产可换 Postgres之后可以从暂停点继续执行。这个能力听起来简单但真正要做到人审的时候千万不能出岔子有几个问题必须单独处理。4.1 Checkpoint 存储的是完整状态序列而不是当前快照LangGraph 的默认MemorySaver只存在内存里进程重启就丢生产必换SqliteSaver或自己实现 Postgres 存储。而且你要理解它的机制每次图节点的执行结果都会新增一个 checkpoint这相当于状态序列不只是最新状态的快照。好处是你可以回退到任意历史时刻重新执行坏处是数据量增长很快。我们的经验是定期清理旧 checkpoint只保留最近 N 次执行的和等待人工确认的活跃会话。from langgraph.checkpoint.sqlite import SqliteSaver # 生产环境不要用 MemorySaver重启即丢失 checkpointer SqliteSaver.from_conn_string(postgresql://user:passhost/langgraph)4.2 人工确认节点要用 interrupt_before 而不是硬编码停住我们第一版实现 HITL 是把等待确认写成一个节点用一个 while 循环轮询数据库确认状态。这是灾难——图线程阻塞所有并发请求都卡住。正确做法是使用图配置里的interrupt_before参数让图在进入审查节点之前暂停并返回控制权然后你的服务通过graph.invoke(..., config{configurable: {thread_id: xxx}})恢复执行。恢复执行不是重新从头跑而是从暂停的 checkpoint 继续。这意味着你必须保证在等待人工审查期间上下游节点的 Prompt 和业务逻辑不能被修改否则继续执行时可能出现同一份状态、两个版本 Prompt的隐性 bug。4.3 人工确认的输入怎么进状态LangGraph 2.x 里官方的做法是用Command(resumevalue)把人工确认的结果注入状态。你可以在 next 节点里读取这个值把它追加为 HumanMessage或者直接覆盖某个中间变量。这里有个容易踩的坑如果不显式用 Command(resume...)恢复执行时节点收到的输入里只有旧的intermediate_results没有人工反馈状态就断裂了。4.4 超时和幂等生产环境的隐藏要求HITL 不光是暂停和恢复你还得处理人一直不确认和用户重复提交两种情况。超时checkpoint 里要存created_at和expires_at超过 24 小时未确认的会话自动标记为已过期前端显示不可恢复。幂等用户点了两次确认后端必须保证只 resume 一次。我们自己包了一层run_id的分布式锁确认接口里先查run_id是否已消费防止同一个人工输入被 LangGraph 执行两遍。我这边的体会HITL 不是 Agent 系统的附加功能而是生产系统的骨架。如果你打算让 Agent 直接面向业务操作从第一天就把人审设计进图里而不是跑通了再补。5. 流式输出与可观测性从黑盒调用到过程可回溯多智能体系统有个天然问题单 Agent 你可以直接打日志看 Prompt 和 Completion但多 Agent 之间的路由、工具调用、状态迁移日志是分散的、交错在多个执行路径里的。没有可观测性生产事故排查就是灾难。我们前期上线时最痛苦的就是用户说回答错了但不知道错在哪一层。5.1 LangGraph 的事件流 API 是你唯一需要关心的接口LangGraph 的graph.stream()支持按事件、按节点、按更新三种模式。实测下来最有用的是stream_modeupdates——每次图流转到新节点你就能拿到该节点返回的状态增量。我们把它接到一个事件总线前端通过 WebSocket 实时展示当前哪个 Agent 在做什么、检索了哪些数据、工具返回了什么用户不再觉得 AI 是个黑盒。config {configurable: {thread_id: batch-20240112-001}} async for event in graph.astream(input, configconfig, stream_modeupdates): for node_name, updated_state in event.items(): # 推送前端同时落一份审计日志 logger.info(f[{node_name}] {updated_state})5.2 钩子函数是追踪工具调用链的关键LangGraph 的节点本身是普通 Python 函数所以你可以在每个节点里手动埋点。但如果工具调用发生在节点内部的model.bind_tools(..., tool_choice...)里一个节点里可能发生多次 LLM 调用和工具执行只打节点级日志会丢失细节。我们统一在工具 wrapper 里埋了三个钩子on_tool_start、on_tool_end、on_tool_error。每个钩子记录工具名、入参、出参摘要、耗时、错误码。LangSmith 也支持 tracing但生产环境我会选择把关键链路打到自己的日志系统因为 LangSmith 的采样率和大并发下的稳定性不如自建的日志可靠。5.3 审计日志必须保留原始查询 - 路由结果 - 工具入参 - 最终回答的完整链路多智能体系统一旦接入生产你的日志就不是给开发者看的了是给安全审计和业务复核看的。我们每条查询都会生成一个trace_id贯穿 LangGraph 状态、日志、前端展示、数据库记录。审计界面提供按 trace_id 重放执行过程的功能——这比看日志好用的多因为执行过程是结构化的状态变化不是一行行文本。6. 让它下地干活FastAPI 集成与并发配置的几个硬经验LangGraph 本身不关心你用什么 web 框架暴露接口但一旦和 FastAPI 集成有几个 LangGraph 特有的并发问题就出来了。我们踩过一个大坑记录一下。6.1 图实例是重资源不能每个请求都重新构造LangGraph 的编译图包含 PromptFactory、工具注册、模型客户端、checkpointer 连接池。如果在 FastAPI 的 request handler 里每次都调用builder.compile()你的服务会在高并发时直接 OOM。正确做法是把编译好的 graph 作为模块级单例进程启动时构建一次之后所有请求共享。如果不同业务线需要用不同的图也建议用工厂函数 LRU 缓存而不是即时编译_graph_cache: dict[str, CompiledStateGraph] {} def get_graph(domain: str) - CompiledStateGraph: if domain not in _graph_cache: _graph_cache[domain] build_domain_graph(domain) return _graph_cache[domain]6.2 thread_id 是并发的天然隔离但它不是并发控制LangGraph 用thread_id隔离会话状态不同 thread_id 的图执行互不干扰这天然适合多用户并发。但要注意线程隔离不等于线程安全——同一个 thread_id 同时被两个请求 invokecheckpoint 会相互覆盖状态直接乱掉。我们强制所有 WebSocket 请求在网关层按 thread_id 做互斥一个 thread_id 同一时间只允许一个执行请求在途。这比在 LangGraph 应用层做锁更接近入口能覆盖所有入口来源同步 HTTP、异步回调、定时任务。6.3 FastAPI 的异步接口和 LangGraph 的 sync/async 要分清如果你的业务大量依赖 LLM 调用和外部工具async 是必要的——同步阻塞会让 worker 线程池迅速耗尽。LangGraph 提供ainvoke/astream异步接口模型调用本身的耗时可以让出事件循环。但注意工具 wrapper 里如果用的是 sync 的 httpx 请求或 requests 库会阻塞 event loop必须用asyncio.to_thread包装def run_sync_query(sql: str) - Result: ... res await asyncio.to_thread(run_sync_query, sql)6.4 模型上下文预算多智能体放大了 Token 消耗多智能体的 token 消耗不是单 Agent 的简单相加Supervisor 路由会累积所有 Agent 的消息列表Worker 之间传递结构化结果也会逐步膨胀。我们在 400 次压测里统计过单轮完成一次路由-检索-分析-审查全链路大约消耗 3000~5000 token取决于中间检索的结果长度。省钱又保质量的两个措施节点的 System Prompt 越短越好领域知识走工具检索不写死 Prompt。状态里只保留最新一轮的消息列表中间轮次已经落库的不再反复发送。LangGraph 允许你在节点内自行控制传给 LLM 的 messages 子集不必把整个 state 递给模型。7. 让系统在真实数据上站稳测试基线、重试策略与降级方案多智能体系统上线不是写完图就完了真正的工程实践在测试和运维阶段。我们的核心经验是给每一种失败提前写好预案。7.1 路由层的回归测试集是唯一的救命稻草LLM 的非确定性让测试变得困难但路由层一定是确定性最强的。我们建立了 300 条真实历史工单的回归集每次修改路由层 Prompt必须在回归集上跑一遍准确率低于基线97%直接拒绝合并。这个数字听起来普通但对我们而言是硬指标——路由错了后面所有 Agent 的努力都白费。执行层的测试更复杂我们用关键字段断言而非全文比对检查 Agent 输出里是否包含了正确的设备编号、是否存在幻觉结构、是否引用了检索结果中没有的数据。7.2 LLM 调用必须封装重试但重试要限流大模型的 API 稳定性不代表 100%生产环境必须考虑 429、超时、网络抖动。我们用tenacity做了一个统一的重试包装首次失败等待 1s第二次等待 3s第三次直接放弃并把错误结构化返回。重试次数上限 3防止雪崩。但重试也会引入幂等问题——LLM 调用一次返回后网络断了你以为失败重试服务端实际已生成结果。这个问题在 LLM 层面没法完美解决我们能做的只是业务侧保证最终写入操作幂等检索操作本身天然幂等LLM 生成结果只做展示不落库两次。7.3 降级策略Agent 挂了系统不能挂单 Agent 挂了影响小但多智能体是有依赖链的上游挂了下游全部白干。我们给每个执行链路配备了降级路径路由层失败 → 回退到规则分类基于关键词的意图匹配兜底某个 Worker 失败 → 跳过该环节由审查层标记信息缺失并发给人工审查层失败 → 允许执行结果以未审查状态进入人工队列降级的关键不是设计得多优雅而是要让最终用户明确看到这个结果是降级产出的。我们在前端对降级结果的标记是醒目的黄色警示条避免 AI 结果被误当作全自动可靠结果。8. 几个没写进标题但你早晚会撞上的细节最后补充几个零碎的、但贯穿所有模块的判断没有系统性的逻辑却都是真实项目中验证过的。8.1 LangGraph 版本升级要慎重LangGraph 发展很快0.x 到 1.x 的接口变化不小。如果你上了生产不要追求最新版本锁死一个版本升级前必须跑完整回归。我们被 0.2.x 到 0.3.x 的StateGraphAPI 变更坑过一次改起来不难但排查成本很高。8.2 企业级部署建议自己实现 checkpointer 存储LangGraph 自带 SQLite 存储方便开发但生产环境建议基于 PostgreSQL 实现。不要用自建的 JSON 文件快照方案看起来方便并发和一致性都会出问题。直接复用 SqliteSaver 源码思路把存储层换成自己的连接池十行代码的事回报率很高。8.3 Agent 写代码不如代码写 Agent你不需要一个写代码的 AgentCode Agent你需要的是把业务规则写成代码让 Agent 通过工具调用这些规则。这条判断我们在早期走了弯路——尝试让 LLM 自己生成业务判断代码结果产出的代码经常是错的而且比直接写死的规则还难维护。后来所有确定性逻辑全部下沉为 Python 函数LLM 只负责分类意图、抽取参数、选择函数、汇总结果。这也带出一个多智能体设计的终极原则不要让 AI 做它不擅长的事只让它做好动词分类和上下文归纳。剩下的一切交给工程。