这几年做 Agent 项目最常被问的问题往往不是“哪个模型更强”而是“你们的 Agent 到底怎么编排的”。如果只是写一个回答问题的小脚本直接用模型 SDK 就能搞定可一旦进入真实的业务系统你会立刻遇到多轮工具调用、分支决策、并行任务、会话恢复、异常重试这类工程问题。这时候没有一套能描述状态流转的框架代码很快就会失控。LangGraph 就是冲着这个场景来的。它不是又一个模型封装库而是一个以“图”为内核的 Agent 编排框架。它允许你把 Agent 的运行过程拆成节点、边、状态流转让 Agent 的开发从“靠 prompt 和运气”变成“可设计、可控制、可恢复”。这篇文章会沿着一条完整链路展开先说为什么企业级 Agent 需要图框架然后拆解 LangGraph 的核心组件再通过单 Agent 示例、持久化记忆、多 Agent 并行三个实战代码把 LangGraph 的常用套路讲清楚。最后补充企业落地时的工程问题和排查思路。无论你是刚接触 Agent 开发还是已经写过一些 Agent 脚本但总觉得不可控这篇文章都值得收藏。1. 为什么 2026 年还要重新理解 Agent 的底层架构很多开发者的 Agent 入门路径是这样的先学会调用大模型接口然后在 prompt 里塞一堆工具说明再写一个 while 循环去处理工具调用结果。你发现这个循环确实能让模型调用函数也能多轮执行但一旦加入条件判断、并发分支、用户中断恢复整个循环就变得又臭又长。原因很简单你把 Agent 的“业务逻辑”写死在了一段命令式代码里而 Agent 的本质是一个有状态的过程它可以被暂停、可以走分支、可以并行、可以因为某个工具失败而重试。用不透明的 while 循环来维护这种过程复杂度是平方级上升的。LangGraph 的解法是把 Agent 过程抽象成一张有向图。你脑子里可以装着一个类比普通流程引擎解决的是“固定审批流”节点之间怎么走是写死的。LangGraph 比流程引擎更灵活的地方在于图里的每一条边都可以是条件边条件由 LLM 的输出决定。换句话说图和模型是协作关系。如果再对比传统程序LangGraph 解决的最关键问题有三个第一显式状态。每个节点都能读取和修改一个全局 State这个 State 是整个图的“数据库”。第二可持续执行。Graph 可以通过 checkpoint 把执行快照持久化进程崩溃了、服务重启了、用户隔天再回来上下文都还在。第三可控编排。你可以精确控制哪些节点并发执行哪些节点需要等前面的结果哪一步必须停下来等人工确认。这一点是企业级 Agent 最看重的能力。所以我的判断很直接如果你未来的 Agent 项目要跑在生产环境LangGraph 目前的定位不是锦上添花而是基础设施层面的选择。2. LangGraph 和 LangChain 到底有什么区别几乎每个搜 LangGraph 的人都会问LangChain 是不是已经被取代了LangGraph 和 LangChain 是不是同一个东西先给结论LangChain 是一套面向 LLM 应用的开发工具集LangGraph 是专做 Agent 编排的图运行时框架。两者不是替代关系而是定位不同。LangChain 里大家最常用的是模型封装、提示词模板、向量存储、文档加载、输出解析这些能力让你能快速开发“LLM 应用”。LangChain 早期也有 AgentExecutor用来执行 ReAct 循环但随着 Agent 场景变得越来越复杂AgentExecutor 那种简单循环已经很难表达有状态、多分支、并行的过程。LangGraph 正是 LangChain 团队为了接住这些复杂场景而推出的下一代编排内核。它保留了 LangChain 的模型调用和工具抽象但重新定义了执行模型。你可以把 LangChain 看成工具箱把 LangGraph 看成装配线装配线上哪些工位做什么顺序是什么是否允许某个工位拆成多条支线同时运行都由图决定。做一个简单对比对比维度LangChain 传统 AgentLangGraph核心抽象Chain / AgentExecutorStateGraph 节点和边执行模型顺序执行或简单循环有向图状态机支持循环、分支、并行状态管理消息列表不够显式自定义 State任意字段均可持久化是否支持暂停恢复较弱通过 Checkpointer 原生支持适合场景原型、RAG 问答、简单工具调用复杂工具调用、多 Agent 协作、生产任务编排理解了这层区别你就明白了为什么很多项目从 LangChain 迁移到 LangGraph 时感觉不是换了一个 API而是换了一种设计思路。LangGraph 的源码核心也比 AgentExecutor 更清楚。它把“节点函数”和“状态变换”解耦节点的职责就是输入 State、输出 State 的部分更新。图引擎负责调度检查点负责保存条件边负责路由。整体设计思路非常接近状态机加 Actor 模型。3. 核心组件拆解State、Node、Edge、Checkpointer要真正把 LangGraph 用起来你必须建立一套自己的心法。下面按核心组件逐个拆。3.1 State 是全局共享的数据模型State 是整个图运行过程中共享的数据结构通常是一个 TypedDict。每个节点执行完后返回的字段会被合并到全局 State 里。from typing import TypedDict class AgentState(TypedDict): messages: list user_id: str order_id: str如果你希望某个字段在多个节点之间累积而不是覆盖就需要给这个字段定义 reducer。reducer 是指定“新值和旧值如何合并”的规则。消息列表的累积就是这样实现的只是 LangGraph 的基础写法会稍微绕一点用到了Annotated泛型。from typing import Annotated from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages]这个写法是所有 LangGraph 代码的基础。它表示每个节点返回的新消息会被追加到原有消息列表后面而不是整体替换。3.2 Node 是业务逻辑的单元Node 可以是一个普通函数也可以是可调用对象。一个节点接收当前 State返回一个字典字典里的 key 就是要更新的字段。节点的函数签名非常一致这比旧式 Chain 容易理解得多。def agent_node(state: AgentState): result llm_with_tools.invoke(state[messages]) return {messages: [result]}注意返回的 key 必须和 State 里定义的字段一致否则 LangGraph 会提示状态字段不存在。这是新手经常踩的一个坑。3.3 Edge 和 Conditional Edge 是流程控制普通边表示无条件跳转builder.add_edge(node_a, node_b)条件边表示根据某个函数返回值决定跳转到哪个节点。最常见的场景是判断最后一轮大模型输出里有没有 tool_calls有就进工具节点没有就结束。def should_continue(state): last_message state[messages][-1] return tools if last_message.tool_calls else end builder.add_conditional_edges(agent, should_continue, {tools: tools, end: END})这里顺便解释一个很多人困惑的点LangGraph 的循环不是靠代码 while 实现的而是靠图中边构成的环路。只要图里有agent - tools - agent这样的环它就会自动循环执行次数由条件边控制。3.4 Checkpointer 是持久化核心Checkpointer 是 LangGraph 和普通流程引擎拉开差距的关键组件。它记录的是图的完整执行快照包括当前状态、消息、已经执行到的节点位置。配合thread_id你可以实现多轮会话记忆和故障恢复。from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() graph builder.compile(checkpointercheckpointer)生产环境一般不会用 MemorySaver因为它在进程重启后就丢失了。生产可以使用 Postgres 等数据库实现的持久化 Checkpointer例如langgraph-checkpoint-postgres。3.5 Send API 是动态并行分发企业级 Agent 经常遇到一个场景用户上传了十个工单要求对每个工单做风险分析。如果使用循环节点逐个处理性能不够如果把十个分析任务写成固定节点又太死板。Send API 就是干这个的。它不是把任务写死在图上而是在某个节点运行时动态决定给同一个子节点发送多个独立任务每个任务有自己的单独状态副本。from langgraph.types import Send def dispatch_node(state): return [ Send(analyze_node, {case_id: case[id], content: case[content]}) for case in state[cases] ]需要说明的是发送出去的每个子任务都带着独立状态执行完后再把结果汇总到主状态。这种做法本质上就是把 Fan-out / Fan-in 模式搬进了 Agent 编排。4. 环境准备与基础安装LangGraph 的安装非常简单核心依赖是 Python 3.9 以上。这里以一个标准项目为例mkdir langgraph-demo cd langgraph-demo python -m venv .venv source .venv/bin/activate pip install langgraph langchain-openai python-dotenv如果你还需要用 Postgres 做持久化可以额外安装pip install langgraph-checkpoint-postgres安装完成后在项目根目录创建.env文件写入模型服务的 API Key。注意LangGraph 本身不绑定模型厂商按你实际使用的模型配置即可。OPENAI_API_KEYyour-key-here如果你用的是国内模型服务可以把模型切换成对应的 LangChain 兼容类核心 Graph 逻辑完全不用改。这其实也是 LangGraph 的一个优势编排层和模型层分离模型可以平滑替换。5. 代码实战从零构建一个可运行的 Agent这一节写的代码是一个电商售后 Agent 的最小实现。它具备两个工具查订单状态、计算退款金额。Agent 会先判断用户意图自主决定多次调用工具最后给出结论。建议把代码保存为app.py直接复制到你的项目里运行。# app.py import os from typing import Annotated, TypedDict from dotenv import load_dotenv from langchain_core.tools import tool from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langgraph.prebuilt import ToolNode load_dotenv() # ---------- 1. 定义全局状态 ---------- class AgentState(TypedDict): messages: Annotated[list, add_messages] # ---------- 2. 定义工具 ---------- tool def search_order(order_id: str) - str: 根据订单号查询订单状态。 # 生产环境请替换为真实订单服务调用 return f订单 {order_id} 当前状态已发货物流公司顺丰 tool def calculate_refund(order_id: str) - str: 根据订单号计算可退款金额。 # 生产环境请替换为真实结算服务调用 return f订单 {order_id} 可退款金额199 元 # ---------- 3. 初始化模型并绑定工具 ---------- llm ChatOpenAI(modelgpt-4o, temperature0) tools [search_order, calculate_refund] llm_with_tools llm.bind_tools(tools) # ---------- 4. 定义 Agent 节点 ---------- def agent_node(state: AgentState): result llm_with_tools.invoke(state[messages]) return {messages: [result]} # ---------- 5. 条件路由 ---------- def should_continue(state: AgentState): last_message state[messages][-1] if getattr(last_message, tool_calls, None): return tools return end # ---------- 6. 拼接图 ---------- 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, should_continue, { tools: tools, end: END, }) builder.add_edge(tools, agent) # ---------- 7. 编译 ---------- graph builder.compile()运行验证脚本建议保存为run.py# run.py from app import graph def main(): result graph.invoke({ messages: [ {role: user, content: 帮我查一下订单 10086 的状态并计算可退款金额。} ] }) print(result[messages][-1].content) if __name__ __main__: main()执行python run.py预期输出类似订单 10086 当前状态已发货物流公司顺丰 订单 10086 可退款金额199 元这段代码背后的执行链路是START - agent 节点 - 模型判断需要调用两个工具 - 进入 tools 节点 - 返回工具结果 - 回到 agent 节点 - 模型综合所有工具结果生成最终回答 - 进入 end。从工程角度看这个图已经把“循环”和“条件分支”都纳入可控范围。你不需要自己维护 while 循环也不需要自己拼装历史消息LangGraph 会按照消息累积规则自动维护 State。6. 状态持久化让 Agent 拥有真正的记忆能力上一节的示例是无状态运行。每次graph.invoke调用都是全新开始多轮对话时你只能手动把历史消息传进去。这在企业场景里完全不够用用户可能上午问了一个订单下午回来继续问系统要能够自动加载他的会话状态。LangGraph 的 Checkpointer 机制正好解决这个问题。# checkpoint_demo.py from langgraph.checkpoint.memory import MemorySaver from app import builder # 复用上面的图构建脚本 checkpointer MemorySaver() graph builder.compile(checkpointercheckpointer) config {configurable: {thread_id: customer-10086}} # 第一轮用户提问 graph.invoke( {messages: [{role: user, content: 帮我查一下订单 10086 的状态}]}, configconfig, ) # 第二轮用户继续追问注意没有传历史消息 result graph.invoke( {messages: [{role: user, content: 刚才那个订单能不能退款}]}, configconfig, ) print(result[messages][-1].content)这里的thread_id是会话隔离的关键。不同用户使用不同 thread_id会话互不干扰。图引擎会自动把每一轮执行后的快照保存到 Checkpointer 中下一轮执行时自动恢复。结合刚才的电商场景这个能力意味着第一用户的会话历史不需要业务系统手工管理。第二Agent 执行到一半如果节点报错可以从最近一次检查点恢复而不是把整个流程重跑一遍。第三横向扩容时同一个用户的请求会路由到同一个会话上下文。这在旧式 Agent 实现里往往是最难处理的问题之一。生产环境建议使用 Postgres 版本的 Checkpointerpip install langgraph-checkpoint-postgres连接配置可以参考官方文档这里给一个最常见的连接方式from langgraph.checkpoint.postgres import PostgresSaver DB_URI postgresql://user:passwordlocalhost:5432/langgraph with PostgresSaver.from_conn_string(DB_URI) as checkpointer: graph builder.compile(checkpointercheckpointer) # 图编译完成后业务代码可以直接调用 graph.invoke需要提醒的是MemorySaver 只适合本地开发和测试。生产环境一旦进程重启MemorySaver 里的所有会话都会丢失这在企业项目里是不被允许的。7. 多智能体架构Supervisor 模式与动态并行分发企业级 Agent 很少只有一个 Agent。常见架构里会出现多个子 Agent一个负责订单一个负责退款一个负责工单审核。这时候需要设计多智能体架构。多智能体的组织模式很多最常用的是 Supervisor 模式。在 LangGraph 里Supervisor 本身也是一个节点它的职责是决定下一步把任务交给哪一个子 Agent 节点。# multi_agent.py from typing import TypedDict from langgraph.graph import StateGraph, START, END class TeamState(TypedDict): messages: list task: str next: str def supervisor_node(state: TeamState): # 真实项目里可以用 LLM 判断下一步给谁 # 这里为了演示流程固定路由到 order_agent return {next: order_agent} def order_agent(state: TeamState): # 子 Agent 处理订单逻辑 return {messages: [{role: assistant, content: 订单 Agent 处理完成}]} def refund_agent(state: TeamState): return {messages: [{role: assistant, content: 退款 Agent 处理完成}]} builder StateGraph(TeamState) builder.add_node(supervisor, supervisor_node) builder.add_node(order_agent, order_agent) builder.add_node(refund_agent, refund_agent) builder.add_edge(START, supervisor) # 这里用了一个简化写法supervisor 根据 next 字段选择去处 builder.add_conditional_edges( supervisor, lambda state: state[next], { order_agent: order_agent, refund_agent: refund_agent, end: END, }, ) builder.add_edge(order_agent, supervisor) builder.add_edge(refund_agent, supervisor) graph builder.compile()在这个结构里子 Agent 完成工作后会把控制权交回 SupervisorSupervisor 再决定是交给另一个子 Agent、还是汇总结果并结束。这是最经典的多智能体闭环。除了 Supervisor企业场景还经常需要动态并行分发。前面提到的大批量工单分析就是一个典型例子这里给出 Send API 的完整用法# send_demo.py from typing import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.types import Send class CaseItem(TypedDict): case_id: str content: str class ReportState(TypedDict): cases: list[CaseItem] reports: list[str] def dispatch_node(state: ReportState): return [ Send( analyze_node, {case_id: case[case_id], content: case[content]} ) for case in state[cases] ] def analyze_node(state): # 实际项目里这里调用模型或业务算法执行分析 result f工单 {state[case_id]} 分析完成风险等级中 return {reports: [result]} def aggregate_node(state: ReportState): combined .join(state[reports]) return {reports: [f汇总报告{combined}]} builder StateGraph(ReportState) builder.add_node(dispatch, dispatch_node) builder.add_node(analyze_node, analyze_node) builder.add_node(aggregate, aggregate_node) builder.add_edge(START, dispatch) builder.add_conditional_edges(dispatch, lambda state: [analyze_node] * len(state[cases]), [analyze_node]) builder.add_edge(analyze_node, aggregate) builder.add_edge(aggregate, END) graph builder.compile()这里有一个很容易误解的知识点Send 和普通 Node 函数的 return 不同。Send 不是在当前节点里串行处理任务而是为每个任务生成一个独立的图执行子分支这些子分支可以并行运行。对于超级大的任务集还需要结合分布式队列做更底层的扩容但这个模式已经能解决 80% 的批量分析需求。很多人搜 LangGraph 时还会看到一个词Skill。从 LangGraph 的视角看增加“技能”本质上有两种实现路径如果你的技能是单一工具类能力就把它注册成 Tool 节点如果你的技能是一整套多步骤流程就把它设计成一个独立子图再挂到主管图上。这比传统 Agent 框架里的“技能复制粘贴”更工程化。8. 企业落地必看Agent 安全、可观测性与错误处理企业级 Agent 和玩具 Demo 的区别往往不在模型能力而在工程化程度。下面几条是我强烈建议你画进项目清单的。8.1 Agent 安全边界工具调用给了 Agent 操作外部系统的能力也就同时打开了权限边界。最佳实践包括工具函数内部必须做二次参数校验不能只依赖 LLM 生成的参数。LLM 生成工具参数本质上是一个概率行为它可能格式正确但内容超范围。涉及订单、支付、数据库变更的工具建议默认走人工确认节点。LangGraph 支持 Human-in-the-loop可以通过interrupt让图暂停等待人工输入这比在工具函数里弹确认框要干净得多。生产环境遵循最小权限原则。图服务使用的数据库账号、内部 API Token、缓存 Redis 密码都要和业务系统之间做好权限隔离。8.2 可观测性图框架的最大优势之一就是天然具备 trace 能力。LangGraph 官方推荐的 LangSmith 能记录每一节点的输入、输出、耗时和 Token 消耗。如果你不想依赖外部平台也可以自己在节点函数里打印结构化日志。import logging logging.basicConfig(levellogging.INFO) def agent_node(state): logging.info(enter agent_node, messages length: %s, len(state[messages])) result llm_with_tools.invoke(state[messages]) logging.info(agent_node done, tool_calls: %s, result.tool_calls) return {messages: [result]}生产环境的日志字段至少应该包含thread_id、当前节点名、会话用户标识、模型请求 ID、耗时、错误堆栈。这些字段是后续排查问题的第一手依据。8.3 错误重试与失败恢复LangGraph 在生产中常常出现 “agent execution terminated due to error” 这类错误。产生原因五花八门可能是模型超时、工具内部 500、反序列化失败也可能是不等式约束导致节点无法执行。比较稳妥的治理策略是在图外围包一层重试逻辑而不是让错误直接返回给用户。常见做法是给工具函数内部加 try-except同时用十秒级超时控制。记住图负责流程编排业务系统负责数据一致性两者要各司其职。9. 常见问题与排查思路问题现象可能原因排查方式解决方案Agent 调用工具后不再继续回答条件边没有识别到 tool_calls打印最后一轮消息的 tool_calls 字段检查 should_continue 返回值确认分支映射表包含“tools”状态字段报“未定义”错误节点返回的 key 不在 State 中检查节点返回字典的 key让返回 key 与 State 字段保持一致多轮对话上下文丢失没有配置 Checkpointer或 thread_id 不一致检查 compile 是否传入 checkpointer配置 Checkpointer确保同一用户使用固定 thread_id消息越积累越多Token 消耗过高没有做历史消息裁剪查看每次 invoke 输入的消息数量在节点里做消息摘要或滑动窗口裁剪工具执行超时外部 API 响应过慢查看日志中工具节点耗时给工具设置超时和重试必要时改为异步任务子任务全部串行执行非常慢使用了 for 循环而不是 Send API检查 dispatch 节点的返回类型改用 Send 生成独立子分支生产重启后会话记录全部丢失使用了 MemorySaver检查 Checkpointer 类型替换为 Postgres 等持久化实现Agent 出现循环不退出条件边一直返回工具节点查看消息消息列表里是否包含未处理的 tool_calls在 Agent 节点增加最大循环次数校验如果你在这些案例里看到了自己正在踩的坑说明你已经进入 LangGraph 的真实使用阶段了。框架本身的坑并不多大多数问题出在状态设计和对图执行模型的理解上。10. 最佳实践从 Demo 到生产的一套固定套路最后总结一套我认可的 LangGraph 项目实施节奏这套节奏适合大多数企业级 Agent 项目。第一不要一上来就画大而全的多智能体图。先拆一个最小闭环比如“用户提问 - 模型决定调用工具 - 工具返回 - 模型生成答案”把这个闭环跑通再逐步增加分支和子 Agent。第二State 设计要和业务数据结构对齐。State 里的字段就是整个 Agent 的共享内存字段命名要直观建议带上业务前缀例如order_id、user_id、approval_status。不要让 State 变成一个塞满各种临时变量的内存袋。第三条件边函数要做成纯函数。should_continue这类函数只依赖 State 输入返回固定的节点名字符串不要在函数内部混入复杂 IO。这样图逻辑才容易测试和排查。第四企业项目一定买持久化 Checkpointer 的账。哪怕初期数据量不大也建议直接用 Postgres不要用 MemorySaver 上线。第五为每个工具建立单元测试。工具的输入输出是 Agent 正确性的前提不要幻想模型一定能把参数生成正确工具内部必须做参数范围校验和异常处理。第六给图设置最大执行步数和超时。防止模型失控导致无限循环。LangGraph 编译后的图对象支持在invoke时传入recursion_limit这是一个有效的保护手段。第七上线前做回归测试集。把所有典型对话流程沉淀成测试用例每次修改图结构后都跑一遍防止新增分支影响老功能。第八如果你考虑 LangGraph 替代现有工作流引擎需要冷静评估。LangGraph 的强项是 Agent 编排而不是重流程审批。它的状态模型简单直接但缺少传统 BPM 引擎里的复杂事务、多版本流程定义和组织机构概念。不要盲目做替换。把 LangGraph 当作一个状态机加持久化运行时来用它能发挥的价值远大于当作一个远程调用工具来用。如果你正在规划自己的 Agent 项目建议先从最小闭环开始再按这篇文章的路径逐步加上检查点、并行分发和人工确认这套骨架足够支撑大多数企业场景。