LangGraph实战：从零构建可控的Agent状态机编排

LangGraph 正在成为 LLM 应用开发里绕不开的一个名字。很多人先学会 LangChain 的 chain 调用，接着发现真实业务里的 Agent 根本不是一条链走下去，而是要根据模型输出决定下一步动作，要循环、要分支、要更新状态、还要能中途停下来等人工确认。这个时候 LangGraph 的价值就体现出来了：它把 Agent 编排从“链式调用”提升为“图状态机”，把每个节点的执行、状态更新、条件跳转和循环都显式表达出来。这篇文章会从一个最简可运行的 LangGraph Agent 开始，逐步加入条件路由、循环、子图和持久化，最后给出常见报错排查路径和适合直接抄进生产项目的工程建议。读完你会理解 LangGraph 与 LangChain 的本质差别，也能独立搭出一个具备工具调用、状态管理和人工介入能力的 Agent 骨架。

1. 先理解 LangGraph 解决了什么问题

1.1 为什么普通的 Chain 不够用

LangChain 最基础的抽象是 Chain，也就是把“提示词模板 + 模型 + 输出解析器”串联起来。对于固定流程，比如“先把用户问题翻译成英文，再让模型总结”，Chain 完全够用。但它有一个隐含假设：执行路径是预先确定的。每个步骤执行完，下一个步骤是谁，在编写代码时就已经写死了。

真实 Agent 场景完全不同。以一个带搜索能力的问答助手为例：

1. 用户问“帮我查一下最近 3 天的天气，顺便写一首关于下雨的诗”。
2. 模型先判断需要调用天气查询工具。
3. 工具返回数据后，模型要判断是否还要继续调用工具。
4. 如果数据不完整，可能还要再查一次。
5. 数据齐全后，模型才生成最终回复。

这里的执行路径取决于模型每次的输出。第 3 步可能走“继续调用工具”分支，也可能走“生成回复”分支。用 Chain 表达这个逻辑，要么把所有分支写死在代码里，要么只能依赖 Agent 内部的黑盒循环。前者难以扩展，后者缺少可控性。

LangGraph 的核心思路是把 Agent 执行过程建模成一张“图”。图中的节点是函数或模型调用，边是执行顺序，边上的条件决定要不要跳转。执行状态集中保存在一个 State 对象里，任何节点都可以读取和更新。

1.2 LangGraph 的图模型和状态机思想

LangGraph 底层是一套基于图的状态机。关键概念有四个：

• State：应用全局状态，所有节点共享。
• Node：一个 Python 函数或可调用对象，接收 State，处理业务逻辑，返回状态更新。
• Edge：定义节点之间的连接。
• Conditional Edge：根据 State 内容动态决定下一步走向哪个节点。

对比 Chain，LangGraph 最大的不同是“执行的路由逻辑由显式边表达，而不是隐藏在模型输出中”。模型只负责产出意图，具体跳转到哪个节点仍由开发者控制的规则决定。这样做的好处是每一条路径都可以被审查、测试和追踪。

用一个通俗比喻：Chain 是流水线传送带，零件按固定顺序经过每个工位；LangGraph 是带传感器的分拣中心，包裹经过扫描后，系统根据目的地自动把包裹送往不同出口。

1.3 LangGraph 与 LangChain 的分工

LangGraph 并不是要取代 LangChain。两者分工如下：

实际项目里最常见的组合是：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/activate

Windows 环境激活命令是.venv\Scripts\activate。激活后确认 Python 版本：

python --version

2.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_KEY="sk-你的key"

在 Windows PowerShell 中使用：

$env:OPENAI_API_KEY="sk-你的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(model="gpt-4o-mini", temperature=0) 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_weather：

model_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 和 sender，sender 不是必须字段，但建议从开始就保留，方便后续写路由规则。输出应该是一段天气描述，说明完整循环已经跑通。

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(checkpointer=checkpointer) config = {"configurable": {"thread_id": "user-001"}} result1 = graph.invoke( {"messages": [{"role": "user", "content": "我叫张三"}]}, config=config, ) result2 = graph.invoke( {"messages": [{"role": "user", "content": "我叫什么名字"}]}, config=config, )

第二次 invoke 时，因为使用了相同 thread_id，图会先从 checkpointer 恢复 user-001 的历史状态，模型能看到之前的对话。thread_id 可以理解为“会话 ID”，不同用户、不同会话使用不同 ID。

5.3 何时使用持久化

生产环境中建议所有有状态 Agent 都加 checkpointer。但持久化也带来额外成本：

• SQLite 适合单机开发和中小流量。
• 生产环境建议使用 Postgres 等共享存储，便于多实例横向扩展。
• 存储内容包含完整对话历史和中间状态，注意隐私保护和脱敏。

6. 流式输出与中间状态订阅

6.1 为什么需要流式输出

大模型响应耗时通常 1 到 10 秒。如果让用户等完整响应返回后再显示，体验会很差。流式输出可以把模型生成的 token 分块推送给前端，用户能实时看到输出过程。

LangGraph 支持两种粒度的流式读取：

• stream_mode="values"：每次状态更新时返回整个状态。
• stream_mode="updates"：只返回发生变化的那一份更新。

for chunk in graph.stream( {"messages": [{"role": "user", "content": "北京天气怎么样"}]}, config={"configurable": {"thread_id": "user-001"}}, stream_mode="updates", ): for node_name, update in chunk.items(): print(node_name, update)

调用后你会看到 model、call_tool 等节点依次输出。对前端来说，更常用的是消息级流式，使用 stream_mode="messages" 可以逐 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 是否正确。
• 确认工具节点返回的消息是 role=tool，且 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: str

7.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 状态字段要精简，避免把大对象塞进 State

State 会随 checkpointer 持久化，字段越多，存储开销越大。像文件内容、图片、长文档碎片，不应该直接塞进消息列表，建议只存引用或摘要，内容放到外部存储，节点里按需读取。

8.2 工具节点要做幂等设计

Agent 循环中，同一个工具可能被调用多次。工具执行如果产生副作用，例如扣款、发消息、创建订单，重复执行会造成严重问题。工具实现要先检查是否已执行过，或者利用幂等键去重。至少要做到：同一参数重复调用不产生重复业务效果。

8.3 所有外部调用都要设置超时

模型 API、数据库查询、HTTP 接口都可能超时。在工具内部设置明确超时时间，避免工具长时间挂起拖住整个图。

import httpx with httpx.Client(timeout=10) as client: resp = client.get("https://api.example.com/data")

8.4 条件路由逻辑要尽量薄

条件路由函数最好只做“读状态、做判断、返回节点名”三件事。不要在条件路由里写复杂业务逻辑、发请求或者修改全局状态。路由是执行链路的控制点，应该保持纯粹、简单、易测试。

8.5 深入 Agent 执行链路要打印结构化日志

每次节点执行，记录以下内容：

• 节点名称
• 输入消息摘要
• 输出状态摘要
• 耗时
• 是否有异常

日志统一 JSON 格式，方便接入日志平台和链路追踪。这个实践在本地开发看似多余，进入生产后是排查问题的救命稻草。

8.6 学习环境与生产环境差异对照

9. 从 LangChain 迁移到 LangGraph 的典型改造

9.1 改造前先画图

把现有流程画成图。确定哪些步骤是固定顺序的边，哪些步骤需要条件判断，哪些步骤会循环。画图完成后，再在 LangGraph 里建节点。很多人在改造时直接写代码，结果边连得乱七八糟，回头反复改。

9.2 工具封装层保持不动

LangChain 的 @tool 封装可以直接复用到 LangGraph。工具层是最少改动的部分。只需把原来手动编排的工具调用逻辑，搬到 LangGraph 的工具节点内。

9.3 记忆组件替换为 checkpointer

LangChain 中常用 ConversationBufferMemory 管理对话历史。LangGraph 里，建议直接用状态字段 + checkpointer 管理历史。messages 本身就包含了完整对话，不需要额外记忆对象。

10. LangGraph 项目练习路线建议

如果刚学完本文，按照以下顺序练习，能更稳地掌握：

1. 实现一个固定顺序两节点图，走一遍 invoke。
2. 加入条件路由，让 Agent 根据关键词走不同分支。
3. 加入工具调用和循环，复现本文天气示例。
4. 加入 checkpointer，验证多轮会话记忆。
5. 加入子图，把一个多步骤流程拆到子图里。
6. 加入并行分支，让两个节点同时执行。
7. 模拟工具抛异常，验证 Agent 能否兜底回复。
8. 接入真实业务工具，例如订单查询、商品搜索或知识库检索。

每完成一步，就用 stream 模式观察节点执行顺序。看到每一步的输入输出，才算真正理解图执行过程。

真实项目里最容易出问题的不是 LangGraph API 本身，而是对 Agent 流程的抽象不够清晰。先想清楚状态里保存什么、路由条件是什么、哪些节点必须串行、哪些可以并行，再动手写代码，后面的调试成本会低很多。LangGraph 的价值不是让你写出更复杂的 Agent，而是让你把复杂 Agent 的每一条执行路径都变成可看见、可控制、可恢复的工程组件。