LangGraph 做多智能体最容易被忽略的其实是工程那一层。网上教程大多停在怎么画图、怎么把两个 agent 串起来可一放到生产环境状态管理、工具调用、超时恢复、并发隔离这些问题一个接一个冒出来。这篇文章不重复概念我直接整理几个从 demo 推到线上过程中沉淀下来的 LangGraph 多智能体工程实践围绕图怎么拆、状态怎么设计、工具调用怎么约束、和 FastAPI 怎么集成以及故障怎么排查来展开。里面每一条都是我在真实项目里跑过、踩过坑之后留下的做法适合已经入门 LangGraph、正打算把多智能体系统做扎实的同学。1. 先把“为什么选 LangGraph”这件事想清楚很多团队一上来就画了一张特别复杂的多智能体架构图但问他们为什么这么画、每个节点存在的意义是什么答不上来。工程实践的第一课不是写代码而是把“你的系统到底属于哪种多智能体模式”确认下来。1.1 多智能体协作的常见模式抛开花哨的概念生产环境里常见的多智能体协作模式其实就四种。第一种是主管/编排者模式Supervisor。一个主 agent 负责拆解任务、调度子 agent子 agent 做完把结果回吐给主管。这种模式最适合“任务可分级拆解、各环节专业性强”的场景比如先做意图识别、再做信息检索、再做内容生成、最后统一汇总。优点是流程可控缺点是主管容易成为瓶颈所有信息都要过它一手。第二种是流水线模式Pipeline。任务按固定顺序依次经过几个 agent每个 agent 只处理自己那一阶段类似工厂流水线。适合处理流程明确、步骤顺序固定的业务比如客服工单分类 agent 先打标签处理 agent 再生成回复审核 agent 最后检查。第三种是协商/讨论模式Debate。多个 agent 围绕同一个问题分别给出看法再经过几轮交换意见收敛出结论。适合方案评审、内容质量打分这类场景但要注意轮数一旦很多token 开销会非常吓人。第四种是网络模式Network节点之间可以任意跳转依赖条件边的判断动态决定下一步。这是前三种的泛化形态也是 LangGraph 最擅长表达的形态。我建议的做法是先用文字把业务逻辑画成“谁在什么条件下调用谁”再翻译成图。如果翻译过程要反复改方向、加分支说明业务本身还不清晰这时候上多智能体只会放大混乱。1.2 和 CrewAI、AutoGen 放在一起比LangGraph 赢在哪选型这件事我在项目里比较过 CrewAI、AutoGen 和 LangGraph简单说说体感差异。CrewAI 抽象度高角色、任务、流程都能用非常少的代码声明出来Demo 做得快是它的最大优势。但问题也很明显内部编排是黑盒出问题很难定位是哪个节点、哪条边导致的对控制流、状态合并、细粒度检查点的控制力偏弱遇到需要“中途停下来等人工确认”这种需求方案做出来总是很别扭。AutoGen 更偏研究向核心是对话式多智能体两个 agent 之间靠消息对话推进。这种模式在论文复现、学术探索里很好用但工程上要把 Conversation 模式塞进业务系统状态模型、持久化、可观测性都要自己搭成本不低。LangGraph 的思路是“一切皆图”。节点就是普通 Python 函数边就是显式的控制流状态是一个可以自定义的共享字典检查点机制能让你随时保存、恢复、分支回退。这种显式建模带来的直接好处是可调试、可插桩、可限制。生产系统最重要的就是可控性这一点 LangGraph 天然占优。另外LangGraph 和 LangChain 生态打通工具加载、输出解析、模型封装可以直接复用和 FastAPI 这类 Web 框架整合也顺手后面接入流式输出、异步任务都不需要额外造轮子。维度CrewAIAutoGenLangGraph上手速度快中等中等控制流自由度中中高状态与检查点弱弱强生产可观测性弱弱强与 FastAPI 集成一般一般自然1.3 先判断这个需求真的需要多智能体吗这是我必须泼冷水的一点至少一半的需求不需要多智能体。一个 agent 加一套设计良好的工具链就能解决。我在项目里用过这样的判断清单你可以直接抄任务是否可以被明确拆成不同专业角色拆不出来就别拆。是否存在需要并行执行的独立子任务串行任务用流水线或单 agent 更省心。是否真的需要多轮协商、递进分析三句话能说清的事不要搞讨论组。每个子任务是否有明确的输入输出契约连契约都不稳定拆分之后会更不稳定。如果以上大多数是“否”老老实实用单 agent。多智能体是架构手段不是产品卖点。等单 agent 确实撑不住、或者你明显看到“一个大脑管太多事导致上下文爆炸”时再考虑拆。2. 图结构设计状态、节点、路由与检查点LangGraph 的图画出来容易画得能被工程维护就是另一回事。这一章说的全是画图背后的设计决策。2.1 节点粒度agent 节点与确定性节点的区别我踩过的第一个坑是把所有逻辑都塞进一个“超级 agent 节点”。一个节点里既做大模型推理、又做数据清洗、又做格式转换表面看着简单实际调试时完全没法定位问题——你不确定是模型抽风还是代码有 bug。倒过来另一个极端是每个小函数都独立成节点。图是看着精致了但节点之间传来传去都是一个小字典运行开销和代码复杂度反而上去了。我的经验是分两类节点agent 节点和确定性节点。agent 节点的核心是“要调用大模型做决策”它值得单独成节点因为要单独控制模型参数、单独埋点统计 token确定性节点做的是解析、校验、格式化、查库这类不依赖模型的动作按业务语义聚合一个阶段一个函数就行不必拆到原子级别。举一个实际例子在我做过的智能工单系统里图里一共有六个节点意图识别agent、信息抽取确定性、方案生成agent、工具执行确定性、结果校验确定性、人工审核agent。每个节点的职责边界一句话能说清这就算粒度合格。2.2 State 设计能不放共享态就不放LangGraph 的核心机制是全局共享的 State所有节点读写同一个字典。这个机制很方便但也是多智能体互相污染的温床。我建议遵循一个原则最小共享原则。只在 State 里放需要跨节点流转的信息节点私有的中间结果不要往 State 里塞。否则图跑到第 N 层时State 里塞满了历史遗留字段你根本分不清哪个字段是哪个角色写的。具体设计上我会把 State 分成三个区域消息区messages 列表用来给大模型看的对话历史用Annotated[list, operator.add]做累加合并。工作区context 字典各 agent 把自己的结果写在这里但要约定好命名空间。比如检索 agent 写context[retrieval_result]分析 agent 写context[analysis_result]互相不覆盖。元信息区status、error、attempts 这类控制字段用于路由判断和终止条件。这里有一个容易踩的坑有人图省事把所有结果都 append 进 messages。看起来没问题但 messages 是直接喂给模型的塞进去的每一段都会变成 token 成本而且会让后续 agent 的注意力被历史噪音干扰。能用 context 承载的结构化结果就别进 messages。顺便一提 reducer 的用法。State 字段默认是覆盖写如果你希望一个字段是追加而不是覆盖就要用Annotated加一个 reducer 函数。最常见的operator.add能实现列表拼接但要注意如果你需要“去重后追加”或者“按 key 合并字典”得自己写 reducer不要指望内置函数能满足所有业务。2.3 条件路由与循环终止别让图成为脱缰野马条件边是 LangGraph 最灵活的地方也是最容易失控的地方。路由函数本质上是一个普通函数接收当前 State返回下一个节点的名字或者节点名列表。工程上有几个约束要提前定好。第一路由函数里不要做大模型调用。路由只应该基于已有的结构化字段做判断你的模型调用应该发生在 agent 节点里路由负责“看结果、选方向”。如果路由里再塞一层模型等于又多了一个隐性 agent调试复杂度直接翻倍。第二循环一定要有终止条件。很多业务场景需要 agent 先跑一步看结果不好再回来调整这种“不好就重试”的循环非常常见但必须设置轮数上限。LangGraph 提供了recursion_limit但你最好在业务状态里也维护一个attempts字段在节点内部先判断“这次要不要继续”别让它链式触发到底。第三路由返回的目标集合要和图中的节点名严格一致。这个错误我犯过不止一次路由返回了一个字符串但图里根本没这个节点运行时报错还不明显排查半天才发现是拼写问题。建议把所有节点名定义成模块级常量路由和建图都引用常量不要手写字符串。2.4 检查点、断点续跑与人工审批生产环境里图运行到一半可能因为 API 超时、内存溢出或者人工介入被打断这时候如果整个流程重新跑一遍成本无法接受。LangGraph 的检查点机制就是为解决这个问题的。有个认知要纠正不是编译时加了 checkpointer 就万事大吉。MemorySaver只适合开发调试服务一重启状态全没。上生产至少要换 SQLite 或 Postgres 的持久化 checkpointer这样才能做到“进程挂了从上一个检查点继续”。人工审批是多智能体系统里绕不开的需求。比如执行 agent 准备发通知、改数据库、调用支付接口这种动作不能直接让模型决定需要挂起等人工确认。LangGraph 官方提供的interrupt()就是干这个的在节点里调用interrupt图的执行会停住外部拿到暂停信息人工裁决后再用Command(resume...)恢复执行。我自己的使用习惯是所有“会产生不可逆影响”的工具调用前面都放一个独立的确认节点。不要在 agent 节点内部边调工具边 interrupt那样把业务逻辑和人工流程耦合在一起后续想调整审批策略非常痛苦。3. 工具调用多智能体最容易失控的地方多智能体系统里的每一个 agent 都会调用工具工具调用层就是风险最高的地方。模型可能选错工具、传错参数、反复调用同一工具、甚至把敏感动作提前执行。这一章讲怎么把工具层做到可控、可审计、可预算。3.1 工具描述与参数约束先在定义层把话说清楚模型调用工具的准确率很大程度取决于工具定义写得好不好。我见过很多团队在工具描述里就写一句话“查询用户信息”然后模型各种乱传参。工具描述应该包括这个工具什么时候该用、什么时候不该用、每个参数的含义和边界、返回值长什么样、失败时会发生什么。如果工具是 Python 函数建议用 Pydantic 定义参数模型。LangChain 的tool装饰器会自动从函数签名和 docstring 生成 JSON Schema所以函数签名要写得足够严谨参数名要语义化docstring 要写“触发条件”和“反例”。这比在 prompt 里反复叮嘱模型管用得多。返回值也要结构化。我给所有工具定了一个统一的返回格式(success: bool, result: dict | str, error: str | None)。这个格式有两个好处一是 agent 可以直接根据 success 字段决定下一步二是错误信息能作为一次正常的工具返回喂回给模型让模型自己尝试修正参数而不是直接抛异常打断整个图。3.2 上下文膨胀工具返回值和历史消息都要治理多智能体的 token 消耗通常比单 agent 高出数倍原因很简单每个 agent 都要携带一部分对话历史而工具返回的大块数据也会被反复传给模型。上下文膨胀的治理有几个实用手段。一是工具返回值裁剪。检索类工具最容易返回大段文本我会在工具内部先把结果压到摘要级别只保留 next 步骤真正需要的信息。与其把十万字文档全塞给模型不如让工具返回“命中的段落标题 每段首句 关键词”。二是每个 agent 只挂自己需要的工具集。主管 agent 不需要持有全部子 agent 的工具否则它的上下文里全是无关工具的 schema干扰判断还费 token。专业工具下放到专业 agent这本身就是在减少上下文噪音。三是定期压缩 messages。LangGraph 里没有内置的自动压缩但你可以写一个节点在 messages 长度超过阈值时把早期消息摘要成一条系统消息。这个节点放在路由之前每次循环检查一次成本控制效果显著。3.3 高危操作与人工确认给执行动作上一道闸多智能体系统的“自主性”是分层级的。低危操作比如查天气、算个数可以让 agent 自主执行中危操作比如发一封邮件可以先草拟再确认高危操作比如批量改数据、对外发通知、扣费必须人工确认。我的做法是把工具分为普通工具和白名单工具。白名单工具在高危节点里被调用之前节点先执行interrupt()等待人工确认工具不直接执行。这样你既保留了 agent 的自动化能力又把最终决定权握在手里。这里有个细节很多人忽视人工确认的触发条件写在工具定义里是不可靠的因为工具的调用者是模型模型不一定遵守你的“调用前需要确认”的要求。正确做法是在图的层面控制——把高危工具单独放在一个节点里只有图路由到这个节点时才可能被触发跟模型本身的行为解耦。3.4 可观测性日志、追踪与成本记账多智能体系统的调试比单 agent 难很多因为没有一条清晰的主链。生产环境里必须做到每一步都能还原。我一般会在每个节点入口和出口打印结构化的节点日志包含节点名、输入字段摘要、耗时、token 数。LangSmith 或者 Langfuse 这类追踪工具如果能接入最好但至少要在业务日志里把 thread_id、节点名、模型调用信息串起来出错时能按 thread_id 捞出一整条运行链路。成本记账这点容易被忽略。多智能体的 token 消耗和单 agent 不是一个量级我建议在每个 agent 节点统计本轮调用的输入 token、输出 token、估算成本写入 State 的元信息区整张图跑完再汇总。这样运营团队能看到一条工单处理到底烧了多少 token也方便你决定要不要削减历史窗口或者减少循环轮数。4. 从代码骨架到 FastAPI一次完整落地前面几章讲的都是设计原则这一章拿出一个可以照着抄的落地路径。我会用一个智能工单助理的样例把整个代码骨架串起来。4.1 样例一个智能工单助理的角色拆分假设要做一个内部工单系统用户提交问题系统先判断问题类型然后检索知识库生成回复方案如果方案涉及外部操作比如修改配置、发通知需要人工确认后执行。按前面说的模式我把它拆成四个角色主管 agent接收工单判断类型决定交给哪个下游。检索 agent查知识库返回结构化片段。方案 agent基于检索结果生成回复方案并列出需要执行的动作。执行 agent执行低危动作高危动作交给人工确认节点。这个拆法符合角色清晰、职责不重叠的原则。主管只做调度检索只做召回方案只做生成执行只做动作。任何一个环节出问题都可以单独重跑对应节点。4.2 图的核心代码骨架下面是一个精简但可运行的骨架from typing import Annotated, TypedDict import operator from langgraph.graph import StateGraph, START, END from langgraph.graph.state import CompiledStateGraph from langgraph.checkpoint.memory import MemorySaver NODE_SUPERVISOR supervisor NODE_RETRIEVAL retrieval NODE_PLAN plan NODE_EXECUTE execute NODE_HUMAN human_approval # 会中断等待人工的节点 class AgentState(TypedDict): messages: Annotated[list, operator.add] task: str context: dict status: str def supervisor(state: AgentState) - dict: # 这里内部可以调用大模型做意图分类也可以先用规则判断 # 关键是返回一个 decision 字段给路由用 decision classify(state[task]) # 简化示意 return {context: {decision: decision}} def retrieval(state: AgentState) - dict: docs search_knowledge_base(state[task]) return {context: {retrieval_result: summarize(docs)}} def plan(state: AgentState) - dict: reply generate_reply(state[task], state[context][retrieval_result]) return {context: {reply: reply, need_approval: check_risky(reply)}} def execute(state: AgentState) - dict: # 高危动作在节点内先 interrupt,由外部恢复 act state[context][reply].get(action) if act: confirm interrupt({action: act}) if confirm ! approved: return {status: rejected} run_low_risk_action(act) return {status: done} def route_after_supervisor(state: AgentState) - str: decision state[context].get(decision, ) if decision retrieval: return NODE_RETRIEVAL if decision plan_only: return NODE_PLAN return END def route_after_plan(state: AgentState) - str: if state[context].get(need_approval): return NODE_EXECUTE # execute 内部会 interrupt return END builder StateGraph(AgentState) builder.add_node(NODE_SUPERVISOR, supervisor) builder.add_node(NODE_RETRIEVAL, retrieval) builder.add_node(NODE_PLAN, plan) builder.add_node(NODE_EXECUTE, execute) builder.add_edge(START, NODE_SUPERVISOR) builder.add_conditional_edges(NODE_SUPERVISOR, route_after_supervisor) builder.add_edge(NODE_RETRIEVAL, NODE_PLAN) builder.add_conditional_edges(NODE_PLAN, route_after_plan) graph: CompiledStateGraph builder.compile(checkpointerMemorySaver())调用时用graph.invoke(input_state, config{configurable: {thread_id: ticket-1001}})。thread_id 很关键它决定了检查点按什么维度隔离。每个工单一个 thread_id恢复、续跑、人工审批都是基于这个维度。用interrupt()后外部通过graph.invoke(Command(resumeapproved), configsame_thread_config)恢复执行。注意恢复时要用同一个 thread_id否则 LangGraph 找不到挂起点。4.3 和 FastAPI 集成流式输出与长任务处理多智能体系统跑一条完整链路通常需要几十秒Web 接口不能做成同步请求等结果。我推荐的方案是短任务用 SSE 流式输出长任务用独立的任务队列加状态轮询。SSE 的方式和 LangGraph 的异步接口配合得很好。graph.astream_events可以实时产出每个节点的运行事件你把它转成 SSE 推给前端用户能看到“主管 agent 正在分析 → 检索 agent 正在查库 → 方案生成中”体验比干等一把梭好得多。from fastapi import FastAPI from fastapi.responses import StreamingResponse app FastAPI() app.post(/agent/run) async def run_agent(payload: dict): thread_id payload[ticket_id] config {configurable: {thread_id: thread_id}} async def event_stream(): async for event in graph.astream_events( {task: payload[question]}, configconfig, versionv2, ): if event[event] on_chain_end and name in event: yield fdata: {event[name]} done\n\n return StreamingResponse(event_stream(), media_typetext/event-stream)这是最简示意生产上还要处理客户端断开、超时、错误重试。但核心思路是图是异步的接口是流式的客户端和中间链路都按这个模型设计。4.4 并发、限流与线程安全LangGraph 本身是线程安全的但多智能体系统的并发瓶颈几乎都在大模型 API 上。你在 FastAPI 里开了一堆异步请求每个请求内部图又会并发调用多个模型如果不对总量做限制上游 API 很快就给你返回限流错误。我的做法是用一个全局信号量控制并发图运行数再给每个节点内部的大模型调用设置超时和重试。信号量可以按用户维度和全局维度分开全局限制打向 API 的总流量用户维度限制防止单用户把资源吃满。import asyncio _global_semaphore asyncio.Semaphore(10) async def run_with_limit(payload: dict): async with _global_semaphore: return await graph.ainvoke( {task: payload[question]}, config{configurable: {thread_id: payload[ticket_id]}}, )另一个工程细节不要把编译后的 graph 对象放在内存里就以为没事了。生产环境要评估 graph 编译成本、checkpointer 的存储容量、以及多进程部署时检查点数据能不能共享。如果用了内存 checkpointer多进程部署时每个 worker 各存各的thread_id 在另一个进程就不认识了。这时需要把 checkpointer 换成 Postgres 或 Redis 这类共享存储。5. 生产环境问题排查速查手册这一章直接给结论都是我实际踩过并解决的坑按症状、原因、解法的顺序写。5.1 图进了死循环症状是单个请求耗时异常长日志里反复出现同一个节点名。原因通常是条件路由里漏掉了返回END的分支或者循环节点的退出条件不满足。解法分三层一是路由函数里先写死默认返回END宁可多走一步短路也不要漏分支二是给compile传入recursion_limit作为兜底一般设 25-50三是在循环计数节点里维护attempts字段超过阈值强制走退货流程。5.2 Agent 反复调用同一个工具症状是日志显示同一个工具被连续调用五六次入参几乎一样。原因是模型没有拿到“这个工具已经调用过且结果没变”的信号它以为每次调用都是新的。我的解法是在工具层维护一个简单的调用缓存相同的入参在短时间内返回相同的结果并且结果里带上cached: true标记。同时在给模型的工具说明里加一句“如果该工具上一个结果已经满足需求不要再重复调用”。站在模型视角它需要明确的信息来停止重复动作。5.3 子 Agent 污染了共享状态症状是某个 agent 读到的 context 字段被另一个 agent 改写或者 messages 里混入其他角色的内部思考。解法回到 2.2 节说的最小共享原则。每个 agent 写 context 时严格使用自己的命名空间重要字段用 reducer 控制合并逻辑必要时在节点出口做一个 State 白名单清洗——只保留当前节点允许写入的字段其余全部丢弃。这个白名单清洗虽然多花一点代码但它保证了状态的确定性值得做。5.4 Token 消耗直线飙升症状是运营一看账单不对劲单条任务成本翻了几倍。原因通常是历史消息无限累积、工具返回大块文本、循环轮数过多。解法组合拳设置 messages 长度阈值并挂压缩节点工具返回统一摘要化降低循环轮数上限在高成本节点接入 token 统计超过预算时由路由自动降级为简单回复或者转人工。5.5 Agent 输出格式不稳定导致下游解析失败症状是解析 JSON 频繁报错原因是大模型偶尔会在 JSON 前后加说明文字或者字段名漂移。不要靠提示词解决所有格式问题。能用工具调用function calling约束的内容就优先用工具因为工具调用天然会返回结构化 schema 结果实在要用自由文本输出就在下游加一个确定性解析节点先尝试json.loads失败再用一次小模型修复再失败就进入人工兜底。顺序是结构化优先、确定性解析、修复兜底、人工兜底。5.6 断点恢复后的状态错乱症状是人工审批通过后图恢复执行但输出跟预期不一致。原因多半是恢复时传入的状态和挂起时不一致或者 checkpointer 版本不兼容。解法是约定恢复操作只传Command(resume...)不要重新传入完整的 input state另外升级 LangGraph 时跑一遍恢复流程的集成测试检查点数据结构一旦变化老数据要安排迁移脚本。下面是一张精简速查表可以直接贴在团队 Wiki 里症状最常见原因优先排查动作死循环路由漏掉 END / 退出条件失效检查条件边分支、recursion_limit重复调工具模型不知道结果未变工具层缓存 调用提示状态污染共享字段无命名空间白名单清洗 reducer成本飙升上下文膨胀、循环过多压缩节点 工具结果摘要解析失败自由文本格式漂移结构化输出 确定性解析恢复错乱恢复状态与挂起不一致只传 resume不重传 input6. 写在最后多智能体不是装饰品我个人在几个项目里反复验证过一条结论多智能体不是架构的装饰品你每多加一个 agent就要多付一份状态管理、工具契约、成本控制和故障排查的账。能用一个带工具的单 agent 解决就不要为了“看起来高级”硬拆当你确实需要拆的时候优先保证图的显式性、状态的可控性和工具层的可审计性这三条做到位多智能体系统才有资格上生产。最后再分享一个细节我习惯在图的每个节点打一条nodexxx, statusxxx, costxxx的结构化日志本地开发时看日志就能定位问题上线后通过日志采集系统做告警。这套东西看起来不起眼但在你半夜被叫起来排查一条卡住的工单时它就是救命稻草。LangGraph 给了你强大的图执行能力能不能驾驭它取决于你把工程基本功做得多扎实。