LangGraph多智能体工程实践：状态设计与工具调用的关键要点

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 框架整合也顺手，后面接入流式输出、异步任务都不需要额外造轮子。

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(checkpointer=MemorySaver())

调用时用graph.invoke(input_state, config={"configurable": {"thread_id": "ticket-1001"}})。thread_id 很关键，它决定了检查点按什么维度隔离。每个工单一个 thread_id，恢复、续跑、人工审批都是基于这个维度。

用interrupt()后，外部通过graph.invoke(Command(resume="approved"), config=same_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"]}, config=config, version="v2", ): if event["event"] == "on_chain_end" and "name" in event: yield f"data: {event['name']} done\n\n" return StreamingResponse(event_stream(), media_type="text/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 里：

6. 写在最后：多智能体不是装饰品

我个人在几个项目里反复验证过一条结论：多智能体不是架构的装饰品，你每多加一个 agent，就要多付一份状态管理、工具契约、成本控制和故障排查的账。能用一个带工具的单 agent 解决，就不要为了“看起来高级”硬拆；当你确实需要拆的时候，优先保证图的显式性、状态的可控性和工具层的可审计性，这三条做到位，多智能体系统才有资格上生产。

最后再分享一个细节：我习惯在图的每个节点打一条node=xxx, status=xxx, cost=xxx的结构化日志，本地开发时看日志就能定位问题，上线后通过日志采集系统做告警。这套东西看起来不起眼，但在你半夜被叫起来排查一条卡住的工单时，它就是救命稻草。LangGraph 给了你强大的图执行能力，能不能驾驭它，取决于你把工程基本功做得多扎实。