LangGraph 正在成为 LLM 应用开发里绕不开的一个名字。很多人先学会 LangChain 的 chain 调用接着发现真实业务里的 Agent 根本不是一条链走下去而是要根据模型输出决定下一步动作要循环、要分支、要更新状态、还要能中途停下来等人工确认。这个时候 LangGraph 的价值就体现出来了它把 Agent 编排从“链式调用”提升为“图状态机”把每个节点的执行、状态更新、条件跳转和循环都显式表达出来。这篇文章会从一个最简可运行的 LangGraph Agent 开始逐步加入条件路由、循环、子图和持久化最后给出常见报错排查路径和适合直接抄进生产项目的工程建议。读完你会理解 LangGraph 与 LangChain 的本质差别也能独立搭出一个具备工具调用、状态管理和人工介入能力的 Agent 骨架。1. 先理解 LangGraph 解决了什么问题1.1 为什么普通的 Chain 不够用LangChain 最基础的抽象是 Chain也就是把“提示词模板 模型 输出解析器”串联起来。对于固定流程比如“先把用户问题翻译成英文再让模型总结”Chain 完全够用。但它有一个隐含假设执行路径是预先确定的。每个步骤执行完下一个步骤是谁在编写代码时就已经写死了。真实 Agent 场景完全不同。以一个带搜索能力的问答助手为例用户问“帮我查一下最近 3 天的天气顺便写一首关于下雨的诗”。模型先判断需要调用天气查询工具。工具返回数据后模型要判断是否还要继续调用工具。如果数据不完整可能还要再查一次。数据齐全后模型才生成最终回复。这里的执行路径取决于模型每次的输出。第 3 步可能走“继续调用工具”分支也可能走“生成回复”分支。用 Chain 表达这个逻辑要么把所有分支写死在代码里要么只能依赖 Agent 内部的黑盒循环。前者难以扩展后者缺少可控性。LangGraph 的核心思路是把 Agent 执行过程建模成一张“图”。图中的节点是函数或模型调用边是执行顺序边上的条件决定要不要跳转。执行状态集中保存在一个 State 对象里任何节点都可以读取和更新。1.2 LangGraph 的图模型和状态机思想LangGraph 底层是一套基于图的状态机。关键概念有四个State应用全局状态所有节点共享。Node一个 Python 函数或可调用对象接收 State处理业务逻辑返回状态更新。Edge定义节点之间的连接。Conditional Edge根据 State 内容动态决定下一步走向哪个节点。对比 ChainLangGraph 最大的不同是“执行的路由逻辑由显式边表达而不是隐藏在模型输出中”。模型只负责产出意图具体跳转到哪个节点仍由开发者控制的规则决定。这样做的好处是每一条路径都可以被审查、测试和追踪。用一个通俗比喻Chain 是流水线传送带零件按固定顺序经过每个工位LangGraph 是带传感器的分拣中心包裹经过扫描后系统根据目的地自动把包裹送往不同出口。1.3 LangGraph 与 LangChain 的分工LangGraph 并不是要取代 LangChain。两者分工如下关注点LangChainLangGraph模型调用封装ChatModel、提示词模板、输出解析不负责复用 LangChain 模型封装工具调用工具定义、工具绑定只负责调度不关心工具本身实现流程编排Chain 固定串联图状态机支持分支、循环、并行状态管理每步独立状态传递零散全局 State 集中管理可控性路径固定或黑盒路径显式可干预、可回退实际项目里最常见的组合是LangChain 负责模型接入和工具封装LangGraph 负责 Agent 的编排逻辑。LangChain 里的模型对象、工具对象、记忆组件都能直接放进 LangGraph 节点里使用。2. 环境准备与依赖安装2.1 Python 版本与虚拟环境LangGraph 是 Python 库支持 Python 3.9 及以上版本。建议使用 3.10 或 3.11生态兼容性最好。不要直接装在系统 Python 里建议先创建独立虚拟环境避免依赖冲突。python -m venv .venv source .venv/bin/activateWindows 环境激活命令是.venv\Scripts\activate。激活后确认 Python 版本python --version2.2 安装 langgraph 和运行依赖核心依赖只需要 langgraph。为了跑通 Agent 示例还需要一个模型接入层。下面的命令安装 LangGraph 和 LangChain 的 OpenAI 接入包pip install langgraph langchain-openai如果网络环境不便访问 OpenAI 接口可以使用 langchain-ollama 接本地模型或者使用 langchain-anthropic。不同提供方的安装包不同但 LangGraph 侧的写法高度一致。安装完成后确认版本python -c import langgraph; print(langgraph.__version__)注意LangGraph 版本迭代比较快API 有小幅变动。落地方案前要锁定版本不要在生产环境随意升级尤其是从 0.2.x 升级到更高版本时要仔细阅读迁移说明。2.3 准备大模型 API Key下面示例使用 OpenAI 兼容接口。首先设置环境变量export OPENAI_API_KEYsk-你的key在 Windows PowerShell 中使用$env:OPENAI_API_KEYsk-你的key如果要接本地 Ollama 模型安装 ollama 后拉取模型然后通过 langchain-ollama 的 ChatOllama 接入。项目落地前要先确认模型的 function calling 能力是否稳定否则工具调用节点可能频繁出错。3. 最小可运行案例从一个带天气查询的 Agent 说起3.1 先定义工具用 LangChain 的 tool 装饰器定义一个最简单工具。这里以查询天气为例真实项目里工具可以换成数据库查询、HTTP 接口、文档检索等任意能力。from langchain_core.tools import tool tool def get_weather(city: str) - str: 查询指定城市的天气情况。 return f{city} 今天晴22 到 28 摄氏度空气质量良。关键点工具的 docstring 会作为模型理解工具用途的重要信息必须写清楚参数含义。工具只有两张卡——名字和说明写得越清楚模型调用就越准确。3.2 定义状态和节点定义全局状态。最基本的状态只要两个字段messages 保留完整对话历史sender 记录当前节点名称用于路由判断。from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] sender: str这里的Annotated[list, add_messages]很关键。它告诉 LangGraph每次更新 messages 时不是覆盖旧列表而是把新消息追加进去。如果不写 add_messages默认会把旧消息覆盖掉导致对话历史丢失。接着定义两个节点。第一个节点调用模型第二个节点调用天气工具from langchain_openai import ChatOpenAI model ChatOpenAI(modelgpt-4o-mini, temperature0) def call_model(state: AgentState): messages state[messages] response model.invoke(messages) return {messages: [response], sender: model} def call_tool(state: AgentState): last_message state[messages][-1] tool_calls last_message.tool_calls results [] for tool_call in tool_calls: tool_name tool_call[name] tool_args tool_call[args] tool_result get_weather.invoke(tool_args) results.append( { type: tool, name: tool_name, tool_call_id: tool_call[id], content: tool_result, } ) return {messages: results, sender: tool}把模型绑定到工具上模型才会识别出当前会话可以调用 get_weathermodel_with_tools model.bind_tools([get_weather])注意call_model 里要使用 model_with_tools否则模型不知道存在工具永远不会发出工具调用。3.3 构图循环才是 Agent 的关键现在构建图。流程是模型节点 - 判断是否有工具调用 - 有则进入工具节点 - 工具节点执行完回到模型节点 - 再次判断。from langgraph.graph import StateGraph, START, END from langgraph.graph.state import StateGraph def should_continue(state: AgentState): last_message state[messages][-1] if last_message.tool_calls: return call_tool return end builder StateGraph(AgentState) builder.add_node(model, call_model) builder.add_node(call_tool, call_tool) builder.add_edge(START, model) builder.add_conditional_edges( model, should_continue, {call_tool: call_tool, end: END}, ) builder.add_edge(call_tool, model)这里should_continue是条件路由函数。LangGraph 会根据它的返回值和映射字典决定下一步。如果模型输出带 tool_calls就去执行工具否则视为 Agent 已完成直接进入 END。3.4 编译并运行 Agent编译图并执行调用graph builder.compile() def run_agent(query: str): result graph.invoke( { messages: [{role: user, content: query}], sender: user, } ) return result[messages][-1].content print(run_agent(北京天气怎么样))注意invoke 的初始状态必须包含 messages 和 sendersender 不是必须字段但建议从开始就保留方便后续写路由规则。输出应该是一段天气描述说明完整循环已经跑通。3.5 可视化检查执行路径LangGraph 支持把图渲染成图片适合排查流程是否正确from IPython.display import Image, display display(Image(graph.get_graph().draw_mermaid_png()))如果不需要图片也可以打印节点连接关系for node in graph.get_graph().nodes: print(node)这一步是很好的验证手段。看图的形状就能发现模型节点和工具节点之间是否存在回边条件分支是否正确连到 END。4. 核心机制深入条件路由、循环、子图和并行分支4.1 条件路由的完整写法上面的 should_continue 是最简条件路由。真实业务中条件会更复杂例如根据关键词判断走哪个专业节点。根据检索结果置信度决定是生成回复还是让用户补充信息。根据会话状态决定是否进入人工客服。条件路由函数可以返回一个字符串也可以返回字符串列表。列表用于并行路由一次跳转同时触发多个节点。def route_by_intent(state: AgentState): last state[messages][-1].content if 查询 in last: return retrieve if 咨询 in last: return consult return fallback使用条件边时映射字典的 key 就是函数返回值value 是目标节点名。注意 key 必须与返回值完全一致否则 LangGraph 会报“Invalid update node”类错误。4.2 循环与递归限制循环是 Agent 与普通 Chain 最明显的差异。但循环必须有限制否则遇到模型反复调用同一个工具或者工具返回格式反复无法解析程序会无限循环。LangGraph 内置了递归限制result graph.invoke( {messages: [{role: user, content: query}]}, config{recursion_limit: 50}, )默认 recursion_limit 是 25。达到限制后会抛异常。这不是 bug是保护机制。遇到递归超限时优先排查是不是条件路由永远走向工具节点而不是盲目调大限制。from langgraph.errors import GraphRecursionError try: result graph.invoke(...) except GraphRecursionError: print(达到递归上限请检查路由逻辑)4.3 子图把复杂流程拆成可复用模块当 Agent 功能变多把全部节点画在一张图里会非常乱。子图可以解决模块化问题。把一段流程封装成子图再挂在主图的某个节点上。sub_builder StateGraph(AgentState) sub_builder.add_node(step1, step1_node) sub_builder.add_node(step2, step2_node) sub_builder.add_edge(START, step1) sub_builder.add_edge(step1, step2) sub_builder.add_edge(step2, END) sub_graph sub_builder.compile()主图中把子图实例作为节点加入main_builder.add_node(sub_process, sub_graph) main_builder.add_edge(model, sub_process)注意子图的输入输出必须与主图状态结构兼容。子图内部可以定义自己的私有状态但在主图视角它只是一个接收整张 State 并返回部分更新的黑盒节点。4.4 并行分支一个节点同时触发多个任务LangGraph 支持扇出路由。比如收到问题后要同时让多个专家模型分别回答再汇总。条件路由返回列表即可实现def route_to_parallel(state: AgentState): return [writer, reviewer, security_check]builder.add_conditional_edges( router, route_to_parallel, [writer, reviewer, security_check], )这三个目标节点会并行执行当所有节点完成后再汇聚到下一个节点。并行执行可以显著降低多步骤 Agent 的耗时但要注意并发带来的资源占用和模型限流配额问题。生产环境建议控制最大并发数避免一下子把模型 API 打满。5. 持久化让 Agent 记住上次会话5.1 持久化解决的问题默认情况下graph.invoke 每次调用都是全新状态。无论你上一轮问过什么新请求进来时 messages 都是空的。真实产品要求 Agent 记住用户历史上下文比如客服场景用户已经提供了订单号第二轮不需要重新问。LangGraph 的持久化通过 checkpointer 实现。checkpointer 负责把每一步的图状态保存到外部存储消息记录可以按 thread_id 恢复。5.2 基于 SQLite 的持久化示例安装依赖pip install langgraph-checkpoint-sqlite创建带 checkpointer 的图from langgraph.checkpoint.sqlite import SqliteSaver with SqliteSaver.from_conn_string(checkpoints.db) as checkpointer: graph builder.compile(checkpointercheckpointer) config {configurable: {thread_id: user-001}} result1 graph.invoke( {messages: [{role: user, content: 我叫张三}]}, configconfig, ) result2 graph.invoke( {messages: [{role: user, content: 我叫什么名字}]}, configconfig, )第二次 invoke 时因为使用了相同 thread_id图会先从 checkpointer 恢复 user-001 的历史状态模型能看到之前的对话。thread_id 可以理解为“会话 ID”不同用户、不同会话使用不同 ID。5.3 何时使用持久化生产环境中建议所有有状态 Agent 都加 checkpointer。但持久化也带来额外成本SQLite 适合单机开发和中小流量。生产环境建议使用 Postgres 等共享存储便于多实例横向扩展。存储内容包含完整对话历史和中间状态注意隐私保护和脱敏。6. 流式输出与中间状态订阅6.1 为什么需要流式输出大模型响应耗时通常 1 到 10 秒。如果让用户等完整响应返回后再显示体验会很差。流式输出可以把模型生成的 token 分块推送给前端用户能实时看到输出过程。LangGraph 支持两种粒度的流式读取stream_modevalues每次状态更新时返回整个状态。stream_modeupdates只返回发生变化的那一份更新。for chunk in graph.stream( {messages: [{role: user, content: 北京天气怎么样}]}, config{configurable: {thread_id: user-001}}, stream_modeupdates, ): for node_name, update in chunk.items(): print(node_name, update)调用后你会看到 model、call_tool 等节点依次输出。对前端来说更常用的是消息级流式使用 stream_modemessages 可以逐 token 拿到模型生成内容。6.2 订阅中间状态的价值调试时stream 输出能帮你精确看到每个节点的执行顺序和状态变化。一旦某一步出错你能立刻定位是模型节点还是工具节点。这个能力在生产环境监控中非常有用——可以用日志记录每个节点的耗时和状态形成完整的 Agent 执行链路追踪。7. 常见报错与排查路径7.1 “Invalid update node”类错误现象运行时提示图结构中找不到某个节点或者跳转目标节点不存在。可能原因条件路由映射字典里的目标节点名与 add_node 注册的名称不一致。条件路由函数返回值不在映射字典的 key 集合里。子图节点挂在主图时名称冲突。排查方式print(graph.get_graph().nodes)输出所有节点名逐项比对映射字典。命名最好统一使用小写加下划线避免拼写错误。7.2 递归超限recursion_limit exceeded现象程序执行到某一步突然抛 GraphRecursionError。可能原因模型反复发出同一个工具调用。工具节点执行完后又触发了同一个条件分支。条件路由函数判断维度不对导致永远走不回结束分支。模型与工具的 message 格式不符合模型 API 要求模型一直尝试重新生成。检查方式打印 stream 输出看最后几次循环落在哪些节点上。查看最近几条消息的 tool_call_id 和 role 是否正确。确认工具节点返回的消息是 roletool且 tool_call_id 与模型发起的调用 ID 一致。解决方向修正工具消息格式或者在条件路由里加入“同一工具调用重复超过 N 次就强制结束”的保护逻辑。7.3 模型返回空 tool_calls但业务要求必须调用工具现象模型直接生成文本回答没有调用工具导致后续步骤缺失。可能原因模型没有绑定工具使用了原始 model 而不是 model_with_tools。工具说明模糊模型识别不出当前问题需要调用工具。模型温度过高生成的 tool_call 格式不稳定。解决方式检查 bind_tools 是否生效优化工具名和 docstring把 temperature 调低到 0 或 0.1必要时用 few-shot 示例引导模型调用工具。7.4 状态被覆盖而非追加现象多轮对话后 messages 只剩最近一轮历史丢失。原因State 字段没有用 add_messages 注解。直接使用messages: list时后续节点返回新 messages 会覆盖旧值。解决方式确保状态定义如下。class AgentState(TypedDict): messages: Annotated[list, add_messages] sender: str7.5 工具执行报错导致整个 Agent 中断现象某个工具抛异常graph 调用直接失败。解决方式在工具节点内部加 try except把异常转换为可读文本返回给模型。让模型知道“工具执行失败原因是什么”由模型决定是换一种方式重试还是直接如实告知用户。def call_tool(state: AgentState): ... try: tool_result get_weather.invoke(tool_args) except Exception as e: tool_result f工具执行失败: {e} ...这样 Agent 不会因为单个工具异常而整体崩溃具备更好的鲁棒性。8. Agent 架构设计的最佳实践8.1 状态字段要精简避免把大对象塞进 StateState 会随 checkpointer 持久化字段越多存储开销越大。像文件内容、图片、长文档碎片不应该直接塞进消息列表建议只存引用或摘要内容放到外部存储节点里按需读取。8.2 工具节点要做幂等设计Agent 循环中同一个工具可能被调用多次。工具执行如果产生副作用例如扣款、发消息、创建订单重复执行会造成严重问题。工具实现要先检查是否已执行过或者利用幂等键去重。至少要做到同一参数重复调用不产生重复业务效果。8.3 所有外部调用都要设置超时模型 API、数据库查询、HTTP 接口都可能超时。在工具内部设置明确超时时间避免工具长时间挂起拖住整个图。import httpx with httpx.Client(timeout10) as client: resp client.get(https://api.example.com/data)8.4 条件路由逻辑要尽量薄条件路由函数最好只做“读状态、做判断、返回节点名”三件事。不要在条件路由里写复杂业务逻辑、发请求或者修改全局状态。路由是执行链路的控制点应该保持纯粹、简单、易测试。8.5 深入 Agent 执行链路要打印结构化日志每次节点执行记录以下内容节点名称输入消息摘要输出状态摘要耗时是否有异常日志统一 JSON 格式方便接入日志平台和链路追踪。这个实践在本地开发看似多余进入生产后是排查问题的救命稻草。8.6 学习环境与生产环境差异对照维度学习环境生产环境API Key个人 Key密钥管理服务不写入代码模型小模型、低并发按业务选型配限流和降级持久化SQLitePostgres 或 Redis日志print 控制台结构化日志 监控告警错误处理直接抛异常异常捕获、重试、兜底回复测试跑通即可单元测试 集成测试 回归测试安全不考虑提示注入防护、敏感信息脱敏9. 从 LangChain 迁移到 LangGraph 的典型改造9.1 改造前先画图把现有流程画成图。确定哪些步骤是固定顺序的边哪些步骤需要条件判断哪些步骤会循环。画图完成后再在 LangGraph 里建节点。很多人在改造时直接写代码结果边连得乱七八糟回头反复改。9.2 工具封装层保持不动LangChain 的 tool 封装可以直接复用到 LangGraph。工具层是最少改动的部分。只需把原来手动编排的工具调用逻辑搬到 LangGraph 的工具节点内。9.3 记忆组件替换为 checkpointerLangChain 中常用 ConversationBufferMemory 管理对话历史。LangGraph 里建议直接用状态字段 checkpointer 管理历史。messages 本身就包含了完整对话不需要额外记忆对象。10. LangGraph 项目练习路线建议如果刚学完本文按照以下顺序练习能更稳地掌握实现一个固定顺序两节点图走一遍 invoke。加入条件路由让 Agent 根据关键词走不同分支。加入工具调用和循环复现本文天气示例。加入 checkpointer验证多轮会话记忆。加入子图把一个多步骤流程拆到子图里。加入并行分支让两个节点同时执行。模拟工具抛异常验证 Agent 能否兜底回复。接入真实业务工具例如订单查询、商品搜索或知识库检索。每完成一步就用 stream 模式观察节点执行顺序。看到每一步的输入输出才算真正理解图执行过程。真实项目里最容易出问题的不是 LangGraph API 本身而是对 Agent 流程的抽象不够清晰。先想清楚状态里保存什么、路由条件是什么、哪些节点必须串行、哪些可以并行再动手写代码后面的调试成本会低很多。LangGraph 的价值不是让你写出更复杂的 Agent而是让你把复杂 Agent 的每一条执行路径都变成可看见、可控制、可恢复的工程组件。