LangGraph 是 2026 年这个时间点搭建多智能体应用时绕不开的编排框架。它的核心价值不是帮你写提示词而是解决一个非常现实的问题当多个智能体要协作同一件事状态怎么共享、下一步走哪个分支、跑挂了怎么恢复、人工审核在哪一步插入。如果你只会调大模型 API用几层 if/else 把任务串起来小 demo 没问题任务一多、分支一变复杂代码会迅速失控。这篇内容适合已经接触过 LangChain 或大模型应用开发、准备做多智能体项目的开发者也适合后端团队评估 LangGraph 能不能上生产。最值得先看的是状态设计、条件路由和持久化这三个概念后面的代码只是为了验证它们。下面按我实际落地的顺序拆一遍先纠正认知再理解组件然后跑通最小 Demo最后处理批量、部署和排查。1. 多智能体为什么需要编排层LangGraph 和 LangChain 的分工1.1 LangChain 是模型工具集LangGraph 是流程执行引擎很多人刚接触时会把 LangChain 和 LangGraph 混在一起问其实它们解决的问题完全不同。LangChain 更像一个“模型工具集”。它封装了模型调用、Prompt 模板、文档加载、向量库、消息结构、Tool 定义。你用 LangChain 可以很快写出“调用一次模型”的代码也能用 Chain 把两三次调用串起来。问题出在“串起来”这件事不够灵活。早期 Chain 的写法适合直线流程一旦要循环、要分支、要根据模型输出改变下一步代码就变得很难读。你也很难在中间暂停、恢复、或者记住前几轮对话状态。这些恰恰是多智能体协作里最常见不过的需求。LangGraph 就是专门做这件事的它提供一张状态图节点是你要执行的函数边是流程走向允许条件跳转、循环、中断和持久化。你可以把 LangGraph 当成一个“有状态的流程执行引擎”LangChain 的组件在它里面作为节点能力存在。所以更准确的理解是两者不是互相替代而是分层配合。LangGraph 负责流程LangChain 负责和模型、工具、消息格式打交道。实际项目里你经常会看到from langchain_openai import ChatOpenAI和from langgraph.graph import StateGraph同时出现在一个文件里。1.2 多智能体应用缺的不是模型是状态和路由多智能体是什么不是简单地把两个for循环嵌套起来多调几次 API。真正的多智能体协作需要几个关键能力多个智能体共享一份上下文状态比如任务描述、中间结果、最终产出。流程能够根据模型输出、工具结果、人工程序判断走到哪个节点。同一个分支可能被并发执行多次比如批量处理多个文件。任务在某个节点停下来等人工确认确认完再从原地继续。长时间运行的任务需要持久化重启后还能从 checkpointer 恢复。上面任何一个需求用普通 Python 代码硬写都做得到但综合起来会非常痛苦。LangGraph 把这些问题收敛成了几个概念State、Node、Edge、Conditional Edge、Checkpointer、Interrupt、Send。1.3 版本差异很大学习前先锁定参考文档这一点比很多教程讲的任何代码都重要。LangGraph 更新速度很快API 变动也频繁。你 2025 年初看到的Send导入地址到了 2026 年可能已经变化Command这种新 API 也在持续演进。我看到不少朋友第一反应是找“中文文档”“官方手册”但二手资料经常落后一两个版本。我的建议是先确定自己要安装的版本再以官方文档对应版本为准。比如你搜索“langgraph 中文文档”看到的 import 路径和本机实际安装的包不匹配报错几乎是一定的而不是你写错了。注意如果你拿到的代码报 ImportError先查 import 路径在当前版本是否存在再检查是不是旧版 API 被迁移到别的模块。这是 LangGraph 新手最常见的问题。2. 核心组件先看懂这六个State、Node、Edge、Checkpoint、Interrupt、Send2.1 State 和 reducer所有节点通过状态交换数据LangGraph 里最核心的抽象是 State。它不是一个普通字典而是一个定义了“怎么合并更新”的对象。最简单的 State 是一个 TypedDictfrom typing import TypedDict, Annotated from langgraph.graph.message import add_messages class State(TypedDict): messages: Annotated[list, add_messages] task: str result: str这里的Annotated[list, add_messages]是关键。它的意思是每当节点返回新的messages不是直接覆盖而是通过add_messages这个 reducer 函数合并进去。这样多个节点可以往同一个消息列表里追加内容而不会互相覆盖。如果你不写 reducer默认行为就是“后写的覆盖先写的”。单个字段无所谓但像消息列表、任务列表这种聚合数据必须要 reducer否则并行节点的返回会互相吃掉。很多人一上来写多智能体就遇到诡异的数据丢失问题十有八九是 reducer 没设计对。2.2 Node、Edge 和条件边图怎么走Node 就是你写的普通 Python 函数输入 State返回一个字典字典里的 key 会按 reducer 规则更新 State。Edge 表示确定走向比如 A 执行完一定走到 B。Conditional Edge 则是根据当前 State 或外部结果决定下一跳这是多智能体路由的核心。def route_next(state): if state[need_research]: return research return write graph.add_conditional_edges( supervisor, route_next, {research: research, write: write, finish: __end__}, )这里 “supervisor” 节点执行完后会调用route_next根据返回值走不同分支。OpenAI 模型输出一个结果代码决定流程去向。2.3 Checkpoint、Interrupt、Send普通流程图没有的能力Checkpoint把每一步执行后的 State 快照持久化下来。任务中断、服务重启、人工介入之后还能用同一个 Thread ID 恢复。Interrupt让图在某个节点暂停等你人工确认或补充信息。适合审核、确认、填表这类场景。Send动态创建多个并行分支。比如列表里有 10 个文件要处理你用 Send 按文件分别生成子任务而不是在同一个节点里写 for 循环。这三个能力是把 LangGraph 和图数据库、工作流引擎区分开的根本。没有它们LangGraph 就只是一个更复杂的链式调用。2.4 三种常见多智能体架构实际项目里多智能体架构基本就三类。第一类是 Supervisor 模式。一个主管智能体负责理解任务、拆解任务、决定下一步交给哪个 Worker然后收集结果继续决策。这是最稳妥的入门架构也最容易理解和维护。第二类是层级模式。多个 Supervisor 形成上下层关系上层拆大任务下层拆小任务。适合复杂任务但调试成本高。第三类是对等协作模式。多个智能体互相调用没有统一主管。灵活但容易出现死循环、上下文混乱、任务重复。新手不建议直接上。我的建议第一个多智能体项目优先用 Supervisor 模式。它已经能解决 80% 的实际问题而且出了问题好排查。对等协作等真正需要了再研究。3. 环境准备和最小多智能体 DemoSupervisor 加两个 Worker3.1 环境准备Python 版本、依赖、API Key先准备一个干净的 Python 环境。常见实践是 Python 3.10 或 3.11 以上太低会在部分依赖上碰到兼容问题。python -m venv .venv source .venv/bin/activate # Windows 用 .venv\Scripts\activate pip install -U langgraph langchain-openai注意这里不装一堆用不到的包。LangGraph 依赖 LangChain 的消息结构和模型封装所以langchain-openai通常必须装。如果你用国内模型提供商的 OpenAI 兼容接口也走这个包只是要配置 base_url。API Key 放在环境变量里别写死在代码中export OPENAI_API_KEY你的key export OPENAI_BASE_URL你的兼容接口地址 # 如果不使用默认服务按实际地址填写代码里也可以用ChatOpenAI(model..., base_url..., api_key...)的方式临时覆盖。实际生产环境建议用环境变量或密钥管理服务。3.2 一个可以直接跑起来的 Supervisor Demo下面这个例子是标准的 Supervisor 加两个 Worker一个负责查资料一个负责写文档。主管节点用模型判断下一步动作。from typing import TypedDict, Annotated from langchain_core.messages import HumanMessage, SystemMessage from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages class State(TypedDict): messages: Annotated[list, add_messages] next: str llm ChatOpenAI(modelgpt-4o-mini, temperature0) def supervisor(state: State): prompt 你是任务主管根据当前对话判断下一步 - research用户问题需要先查资料 - write可以进入写作阶段 - finish任务已经完成 只输出英文动作词。 response llm.invoke( [SystemMessage(contentprompt), *state[messages]] ) action response.content.strip().lower() if action not in {research, write, finish}: action finish return {next: action} def research_worker(state: State): # 真实项目里这里会接检索工具demo 里直接模拟 return { messages: [HumanMessage(content检索完成已找到相关背景资料。)] } def write_worker(state: State): content llm.invoke( [*state[messages], HumanMessage(content请基于已有资料输出文章初稿。)] ) return {messages: [content]} def route_supervisor(state: State): return state[next] graph StateGraph(State) graph.add_node(supervisor, supervisor) graph.add_node(research, research_worker) graph.add_node(write, write_worker) graph.add_edge(START, supervisor) graph.add_conditional_edges( supervisor, route_supervisor, {research: research, write: write, finish: END}, ) graph.add_edge(research, supervisor) graph.add_edge(write, supervisor) app graph.compile() result app.invoke( {messages: [HumanMessage(content帮我写一篇关于智能巡检的短文)]}, config{configurable: {thread_id: demo-001}}, ) print(result[messages][-1].content)跑起来之后主管节点会先判断用户需求可能直接进入write也可能先research再回到主管最后finish结束。3.3 跑完后看什么输出、消息列表和执行轨迹第一遍能跑通不代表流程是符合预期的。要重点看三样东西第一是最终消息。result[messages][-1].content是否是完整内容。第二是消息列表。中间有没有重复内容、缺内容。如果add_messagesreducer 没配好这里最能暴露问题。第三是执行轨迹。用stream_modeupdates可以拿到每一步for step in app.stream( {messages: [HumanMessage(content帮我写一篇关于智能巡检的短文)]}, config{configurable: {thread_id: demo-002}}, stream_modeupdates, ): print(step)你会看到每一步是哪个节点在跑、返回了什么。排查多智能体问题我几乎总是先开这个输出而不是盯着最终结果猜。4. 动态分支、并行调度和工具接入把 Send 和 Skill 讲透4.1 Send 什么时候用动态批量任务和普通 for 循环的区别多智能体处理批量任务时很多人的第一反应是写 for 循环。但 for 循环的问题是它在同一个节点内部串行处理而且所有任务共享一个大 State不方便单独跟踪每个任务的进度和错误。LangGraph 的答案是 Send。Send 允许你根据当前 State 动态生成多个相同类型的分支每个分支有独立输入可以并发执行。from langgraph.types import Send def dispatch_tasks(state): return [ Send(task_worker, {index: i, task: t}) for i, t in enumerate(state[task_list]) ] graph.add_conditional_edges(dispatcher, dispatch_tasks, [task_worker]) graph.add_edge(task_worker, aggregator)执行逻辑是dispatcher算出 n 个任务每个Send生成一个task_worker分支各分支并发跑最后统一回到aggregator汇总。用 Send 的好处有三个并行度由运行时控制每个分支的错误更容易隔离可以单独看每个分支的中间结果。缺点是状态合并要设计好尤其是多个分支同时返回相同 key 时。4.2 recursion_limit、并发冲突和 reducer 的关系多智能体容易出现两个问题无限循环和状态冲突。无限循环几乎都是 Supervisor 路由没有收敛。比如research和write来回跳没有达到finish的条件。LangGraph 对执行步数有默认上限撞到上限会报RecursionLimit类错误。这不是框架 bug而是你的业务流程没有设计出口。调试时可以临时调高上限看看流程是否最终收敛app.invoke( {messages: [HumanMessage(content写一篇短文)]}, config{configurable: {thread_id: demo-003}, recursion_limit: 50}, )但生产环境不要靠调高上限解决问题要从路由条件和业务逻辑上保证有限步内结束。更稳妥的做法是在主管提示词里明确“只有任务满足完成条件才输出 finish”并且给路由函数加默认 fallback。状态冲突问题我前面提过。多个 Worker 并行返回同一个字段时如果这个字段没有 reducer结果就是“后来者覆盖”。这种情况通常要在 State 里设计为列表并用add_messages或operator.add合并。from operator import add class BatchState(TypedDict): results: Annotated[list, add]这样每个 Worker 的结果都会追加而不是互相覆盖。4.3 给 Agent 增加 skill 的三种方式热搜词里有人问“langgraph 怎么增加 skill”其实就是给 Agent 加工具能力。常见方式有三种。第一种是用tool定义普通工具然后绑定到模型from langchain_core.tools import tool tool def query_device_status(device_id: str) - str: 查询某个设备的当前状态 return f设备 {device_id} 当前运行正常 llm_with_tools llm.bind_tools([query_device_status])第二种是用create_react_agent直接创建一个自带 ReAct 循环的 Agent把工具列表传进去LangGraph 自动处理“模型请求工具 - 执行工具 - 回到模型”的循环。from langgraph.prebuilt import create_react_agent agent create_react_agent(llm, tools[query_device_status])第三种是走 MCP 这类工具协议。现在的趋势是把外部能力封装成 MCP server然后用统一客户端接进来。好处是工具定义和 Python 代码解耦多个 Agent 可以复用同一套工具。我个人建议团队如果工具数量超过 10 个就认真考虑 MCP 方式减少每个 Agent 单独绑工具的维护成本。5. 记忆、持久化和人工审核从一次调用变成可用应用5.1 为什么必须理解 Thread ID 和 Checkpointer没有持久化的多智能体是一次性脚本不是应用。LangGraph 的持久化机制靠 Checkpointer 和 Thread ID 配合实现。先在编译时挂一个 checkpointerfrom langgraph.checkpoint.memory import InMemorySaver checkpointer InMemorySaver() app graph.compile(checkpointercheckpointer)然后调用时带thread_idconfig {configurable: {thread_id: user-123}} app.invoke({messages: [HumanMessage(content第一轮问题)]}, configconfig) app.invoke({messages: [HumanMessage(content接着上面继续)]}, configconfig)第二次调用时LangGraph 会从 checkpointer 里读取该 Thread 的历史 State所以模型知道“上面”指的是什么。InMemorySaver只适合学习和单机测试进程重启数据就没了。本地部署想持久化可以用 SQLite 类的 saver多实例部署一般会用 PostgreSQL 类 saver。具体以你安装版本支持的 saver 为准。注意凡是涉及 Interrupt 恢复thread_id 必须一致。换一个 thread_idLangGraph 会认为是一个全新对话根本找不到暂停点。5.2 Interrupt 和 Breakpoint人工审核插在哪一步很多业务场景不允许 Agent 全自动走完。比如“批量删数据”“生成对外发布的文案”“支付审核”都需要人工确认。LangGraph 的做法是 Interrupt。在节点里调用interrupt图会在该位置暂停把暂停信息返回给调用方。人工确认后用原 thread_id 再调用一次图会从暂停位置继续。from langgraph.types import interrupt def approval_node(state): confirmed interrupt({question: 确认要发布吗}) if not confirmed: return {status: rejected} return {status: approved}编译时也可以设置 breakpoint让图在指定节点之前自动停。区别是 Interrupt 更像业务逻辑主动暂停breakpoint 更像调试或强制审批位。落地时要特别注意暂停后外部系统要保存足够信息给人工看比如任务内容、影响范围、审批链接。恢复调用要带同样的 thread_id并且要把人工反馈传回Command(resume...)才能继续。5.3 长期记忆怎么存Checkpoint 保存的是单个 Thread 的对话状态适合短期上下文。但跨 Thread、跨用户的长期偏好和知识沉淀需要单独的存储。LangGraph 的实践是引入 Store比如内存版或持久化版本。Store 按 namespace 存数据可以理解为一张“长期知识表”。比如用户喜欢什么语气、上次选择什么方案都能存进去。如果只是学习内存 Store 就够生产环境要接持久化存储否则重启后长期记忆仍然丢失。这块 API 在不同版本差异较大落地前一定对照当前版本官方文档确认 Store 的写入和查询方式。6. 部署和服务化langgraph dev、uvicorn 和自托管怎么选6.1 开发命令和服务进程是两回事现在网上最常见的问题之一是“deerflow 里uv run uvicorn app和uv run langgraph dev有什么区别”。其实这是两条完全不同的启动路径。langgraph dev是 LangGraph CLI 提供的本地开发命令。它会读取根目录的langgraph.json启动一个带调试界面、持久化后端和 API 的开发服务。适合本地开发因为它自带了很多方便调试的能力比如可视化图结构、查看线程状态。uvicorn app:app是把你自己写的 FastAPI 应用当作普通 Web 服务启动。LangGraph 在这里只是一个库你需要自己实现接口、持久化、鉴权、日志。你完全可以用 uvicorn 跑一个没有 LangGraph 特性的普通 Python 服务。所以选择很简单本地开发用langgraph dev这类 CLI 工具减少搭建成本生产自托管时通常是自己写一个 FastAPI 壳把 LangGraph 编译后的 app 封装成业务接口再用 uvicorn 或 Docker 跑。6.2 langgraph.json 和服务配置使用 CLI 开发时根目录一般需要一个langgraph.json告诉工具依赖哪些包、哪个函数是图的入口{ dependencies: [.], graphs: { agent: ./src/agent.py:app }, env: .env }这个配置的意思是把src/agent.py里的app暴露成一个叫agent的图。CLI 启动后你能在本地接口上访问到这个图。如果你走纯 uvicorn 路线就不需要这个文件但你要自己处理环境变量加载、图编译时机、线程状态存储等问题。一个常见坑是图在启动时编译一次如果你的代码里有if __name__ __main__之外的重型初始化冷启动会明显变慢。6.3 资源、并发和日志上生产前要确认的事很多团队跑通 Demo 就急着上生产结果在并发和日志上栽跟头。资源方面多智能体比单次模型调用更吃资源。一个任务可能触发多次模型调用、多次工具调用。低配置机器能跑通单任务不代表能承接并发。上生产前至少要确认单任务平均耗时、单任务峰值内存、并发 5 和并发 20 时模型 API 的限流情况。并发方面不要一上来就开最大。先从小并发压测观察模型 API 响应时间、数据库连接数和任务失败率。如果失败率升高优先看是不是限流而不是盲目加机器。日志方面每个线程必须有可追踪的 thread_id。多智能体最大的排查难点就是“这个任务到底走到哪了”。日志里至少要记录每次节点进入和退出、路由函数返回的分支、模型调用的 token 数、工具执行结果摘要、耗时。没有这些出问题只能靠猜。6.4 非 Python 团队比如 Java怎么接入Java 团队经常会问“java langgraph 能不能直接用来写业务”。LangGraph 官方生态以 Python 为主Java 也有社区移植或仿写实现但能力覆盖、文档完整度、版本同步普遍不如 Python 版本。我的建议很直接如果你的核心业务代码是 Java不要让 Java 团队去深入维护 LangGraph 的 Python 代码。更稳的方案是把 LangGraph 多智能体服务部署成一个独立的 Python 服务通过 HTTP API 暴露业务接口Java 后端调它。比如 Java 服务调用 Python 服务的/agent/run接口提交任务并拿到任务状态Python 服务负责跑多智能体流程内部调用模型和工具。Java 侧只需要关心请求、响应、任务状态轮询不需要关心 LangGraph 内部细节。这样团队边界清晰调试也简单看接口日志定位是 Java 调用问题还是 Python 流程问题。7. 企业落地能不能替代 Flowable工厂 MES 场景怎么切入7.1 LangGraph 和 Flowable 不是同一种东西经常看到有人问“langgraph 代替 flowable”。先别急着选边这两者解决的问题差异很大。Flowable 是 BPMN 工作流引擎核心是固定流程、审批链、任务分配、超时提醒。它适合“流程确定、步骤固定、必须审计”的业务比如请假审批、费用报销、合同审批。这种场景要求稳定、可追溯、权限清楚没有太多“让模型决定下一步”的空间。LangGraph 适合的是“流程可能由模型输出动态决定”的场景。比如根据用户问题决定调用哪个 Agent根据工具查询结果决定下一步内容生成任务等。它灵活但灵活意味着行为不完全确定不适合纯审批流程。所以更合理的用法不是替代而是分工涉及多智能体决策、内容生成的部分用 LangGraph涉及固定审批流、组织和权限的部分用 Flowable 或类似引擎。也可以做成混合LangGraph 产出决策结果和内容Flowable 负责把这些内容推送到下一步人工审批。7.2 工厂/MES 场景的务实路线热搜里有人问“langgraph 结合 mes 布置在工厂”。我的态度是能做但要从数据分析和辅助决策切入不要一步到位控制设备。工厂环境最看重稳定性和安全。直接用大模型 Agent 控制产线设备风险太高目前也不适合作为生产主路径。更务实的是把 LangGraph 用在几个低风险业务上设备点检报告自动生成Agent 读取巡检数据自动生成异常描述和维修建议。质量异常根因分析多个 Agent 分别看工艺参数、原料批次、设备日志最后汇总成分析结论。生产计划排程辅助Agent 读取订单和产能数据生成多个候选排程方案由人工确认后执行。跨系统信息汇总把 MES、ERP、质量系统里的信息拉出来用自然语言回答车间主管的问题。这些场景的共同特点是Agent 只做分析和内容生成不直接控制 PLC、不发设备指令。所有建议必须有日志、有人工确认。7.3 落地前的最小检查清单不管企业场景多复杂我建议先确认这几件事流程是否允许模型决策出错出错成本多大每一步是否有日志并能按 thread_id 完整还原执行轨迹人工审核点是否明确任意节点是否都能暂停和恢复外部系统是只读还是可写可写操作有没有二次确认模型 API 超时和限流是否有重试策略批量任务失败时是否能跳过继续而不是整个流程报废如果上面任何一条回答不清楚就先不要全线铺开先用一个小场景验证完再扩。8. 常见报错和排查链路遇到问题先看哪里8.1 启动阶段依赖、导入和配置启动阶段最常见的报错是ModuleNotFoundError。遇到时先查三件事当前 Python 环境是否处于正确的虚拟环境安装的 langgraph 和 langchain-* 版本是否匹配import 路径在当前版本是否还存在。第二个常见问题是端口冲突。CLI 开发工具会占用固定端口如果你本机已经有服务占用启动会失败。看命令行输出里的端口提示换掉冲突服务或指定新端口。第三个坑是环境变量没加载。API Key 写在.env文件里但代码没加载。CLI 开发工具读取langgraph.json里的env配置如果你直接跑 uvicorn 就不用这套逻辑必须自己加载。8.2 运行时报错按这个顺序查我处理运行时报错时习惯按这个顺序而不是直接怀疑模型第一步看 State schema。报错提示某个 key 不在 State 类型里通常是节点返回了没有在 TypedDict 里定义的字段或者字段名拼错。修复方式是检查节点返回字典的 key。第二步看 reducer 问题。如果并行节点返回同一字段后结果丢失、变空、或被后来的覆盖去检查 State 里这个字段有没有Annotatedreducer。第三步看路由函数。Conditional Edge返回的值如果不在映射表里会报路由错误。给路由函数加默认 fallback不要赌模型一定会输出正确动作词。第四步看循环步数。出现RecursionLimit相关报错时先确认图的执行路径是不是一直在某个环里打转而不是立刻调大recursion_limit。第五步看模型输出本身。如果模型没有按提示词输出指定动作词路由就可能走到兜底分支。此时要改提示词和 fallback而不是改图结构。8.3 我的通用排查顺序下面这个顺序几乎可以覆盖多智能体 90% 的问题现象优先排查启动失败、导入报错虚拟环境、依赖版本、import 路径接口超时、任务卡住模型 API 调用日志、工具调用是否 hang 住、输出目录是否不可写输出内容为空输入消息格式、模型提示词、路由是否走到空分支输出内容重复add_messages reducer、节点是否重复调用并发任务结果覆盖State 字段是否缺 reducer 聚合中断后恢复失败thread_id 是否一致、resume 参数是否传回批量任务一半失败是否有失败跳过逻辑、每个子任务日志是否可追踪我最开始踩的坑就是“报错先怀疑模型”。后来发现多智能体的大部分问题出在状态定义、路由函数和持久化配置上。模型输出不稳定是常态你要做的就是把这层不稳定隔离在路由逻辑里不要让它直接决定流程能不能跑。如果只能记住一条经验我会说先用小样例跑通单条任务开stream_modeupdates把每一步执行轨迹看清楚再谈批量、并发和部署。这个顺序能帮你避开绝大多数所谓“框架坑”。