LangGraph实战：用状态图构建有状态智能体，掌握条件路由与持久化

LangGraph 出来这么久了，很多人还把它当成普通的流程编排工具在用，但真正让它区别于其他框架的，是“有状态智能体”这几个字。它用 Python 定义一张状态图，让智能体在执行多步任务时能记住上下文、携带中间结果，甚至在任意节点暂停和恢复。这篇文章我会从零开始，用一个可落地的订单查询智能体例子，带你拆解 LangGraph 的核心概念、状态管理方式、条件路由和持久化配置，适合那些已经用过 LangChain 的 Chain 但感觉不够灵活，或者正准备把 Agent 搬到真实业务场景里的朋友。

1. 为什么是 LangGraph：状态机思维，而不是流程拼接

1.1 从 Chain 到 Graph：编排思路的转变

先聊一个很多人踩过的问题。你用 LangChain 写 Chain 的时候，感觉一条链路走得很顺：模型处理一下，工具调用一下，输出拼接一下。但等你真正要做一个多步骤的 Agent，就会碰到几个很头疼的事。

第一，Chain 是线性的，A 完了必须 B，B 完了必须 C。可真实业务不是线性的。用户可能中途问一句“我另一个订单呢”，可能要求你换个查询条件，可能同样的问题要重试几次。第二，Chain 无状态。每一步之间靠“手动传参”硬接，一旦中间某一步出错，整个链路就断了，你想重试还得从头再来。第三，Chain 的结构是写死在代码里的，你想加一个“用户说不用了就直接退出”的分支，要改动的地方非常多。

LangGraph 的思路是：把你的整个智能体流程定义成一张“状态图”。图里面有节点（Node）和边（Edge）。节点是具体干活的地方，比如“识别意图”“查订单”“生成回答”；边是流程流转的控制线。它和我们熟悉的有限状态机是一个套路。每个节点做一件事，做完之后把结果写进一个共享的 State（状态对象），然后根据 State 的当前内容，决定下一步去哪个节点。这就有意思了——你不再是写死一条链路，而是设计一张“图”，图上可以有分支，可以有循环，甚至可以在某个节点停下来等用户输入，这完全是状态机的玩法。

打个比方吧。普通 Chain 就像一条没有分支的传送带，工件从一头进去，另一头出来，中间环节固定死。LangGraph 更像是带质检工序的自动化产线：工件经过一个工位，如果质检合格就往下走，不合格就送回去返工。工位之间不是直接递东西，而是把加工结果写在共享的操作记录表上，下一个工位看表干活。这个“操作记录表”就是 State，工位就是 Node，流向规则就是 Edge。

1.2 “有状态”到底指什么

很多人听到“有状态智能体”的第一反应是：不就是多轮对话里把历史消息带上吗？其实这只是很小的一部分。

在 LangGraph 里，State 是一个数据结构，可以包含你想让智能体记住的任何东西。比如：

• 多轮对话的消息列表，用来维持对话上下文；
• 当前的用户信息（user_id、登录状态、偏好设置）；
• 业务过程产生的中间变量（意图识别结果、查询到的订单数据、生成到一半的回答）；
• 一个计数器，用来限制循环次数，防止死循环。

这些数据在整个图的生命周期里是共享的。前端节点写入的字段，后面的节点可以直接读取。你不需要像传统 Chain 那样把每个中间结果都塞进函数的参数里。只要在 State 里定义好字段，节点返回一个字典，LangGraph 会自动把字典里的内容合并到 State 中。这个“自动合并”的动作是整个框架最核心的机制，也是后面所有分支和循环能跑起来的基础。

无状态智能体每次调用都是白纸一张，有状态智能体则保持对“当前进行到哪里了”的完整认知。举个例子，用户查完订单后问一句“那退款呢？”有状态智能体知道刚才用户查的是哪个订单，因此可以接着那个订单聊退款。无状态智能体可能又得重新问一遍订单号。

1.3 核心概念速览：State、Node、Edge、Conditional Edge

用语言聊概念不如先看代码骨架。LangGraph 里你有四个东西需要先建立印象。

StateSchema 用 TypedDict 定义，字段代表图内可见的所有数据。Node 就是一个普通 Python 函数，接收当前 State，返回一个字典，返回的字典会被合并到 State 中。Edge 是节点与节点之间的固定连接，告诉 LangGraph “这个节点跑完，下一步无条件去那个节点”。Conditional Edge 则带条件，它接收 State，返回字符串，这个字符串指定下一步去哪个节点。

下面是 LangGraph 官方风格的最小化的伪代码，你大概感受一下结构。

from typing import TypedDict from langgraph.graph import StateGraph, START, END class MyState(TypedDict): counter: int def node_a(state: MyState) -> dict: return {"message": "A 处理完成"} def node_b(state: MyState) -> dict: return {"message": "B 处理完成"} builder = StateGraph(MyState) builder.add_node("a", node_a) builder.add_node("b", node_b) builder.add_edge(START, "a") builder.add_edge("a", "b") builder.add_edge("b", END) graph = builder.compile() result = graph.invoke({"counter": 0})

这个例子完全没用到 LLM，但它已经包含 LangGraph 的全部基本玩法：定义状态、添加节点、连接边、编译成可执行图、invoke 触发执行。你把下面要讲的 LLM 节点、路由函数、持久化往里塞，它还是一个同样结构的图。所以说，千万不要一上来就只盯着 LangChain 的 Agent 封装看，把这张图的结构弄明白，后面才能写出可以维护的智能体。

2. 环境准备与最小有状态工作流

2.1 安装与版本选择

老规矩，先把环境搭好。我建议你新建一个虚拟环境再装依赖，别直接往全局环境里塞，后面踩到依赖冲突的概率会小很多。

python -m venv langgraph-demo source langgraph-demo/bin/activate # Windows 用 langgraph-demo\Scripts\activate pip install langgraph langchain-openai

如果你想用可视化工具来调试图结构，可以顺手把langgraph-cli装上。

pip install langgraph-cli

安装的时候留意一下版本。LangGraph 的迭代很活跃，早期 0.1.x 和现在 0.2.x 的 API 有一些差异，比如START和END的导入路径、条件边的绑定方式都调整过。我文中的示例代码基于较新的版本，你实际安装时最好以官方文档为准。如果你是在生产项目里使用，建议锁版本，别让依赖在半夜自动升级把 API 给你换了。

另外一个提醒：运行 LangGraph 本身不强制需要 LLM API，但你要接真实的大模型，还得配置对应的模型访问凭证。下面代码里我用的是langchain-openai，你换成别的模型提供商也完全可以，LangGraph 本身和具体模型解耦。

2.2 第一步：定义状态和节点

现在写一个最小的有状态工作流：让两个节点共享一个计数器，节点 A 把数字加一，节点 B 记录下数字的当前值。

from typing import TypedDict from langgraph.graph import StateGraph, START, END class CounterState(TypedDict): counter: int message: str def node_a(state: CounterState) -> dict: current = state.get("counter", 0) return {"counter": current + 1, "message": "node_a 处理完毕"} def node_b(state: CounterState) -> dict: return {"message": f"拿到 counter 的值：{state['counter']}"} builder = StateGraph(CounterState) builder.add_node("node_a", node_a) builder.add_node("node_b", node_b) builder.add_edge(START, "node_a") builder.add_edge("node_a", "node_b") builder.add_edge("node_b", END) graph = builder.compile() result = graph.invoke({"counter": 0}) print(result)

运行之后，result长这样：

{"counter": 1, "message": "拿到 counter 的值：1"}

注意一个关键细节：node_a返回的counter: 1被自动合并进了 State，所以node_b读取state["counter"]时拿到的是 1，而不是初始的 0。如果按普通函数思维，你会觉得 node_b 拿到的应该是state的初始值。但在 LangGraph 里，上一个节点返回的更新已经写回了 State。这就是“状态流动”最直观的体现。

2.3 跑通最小工作流，理解状态的增量合并

上面这个例子虽然简单，但我建议你亲手敲一遍，然后在node_b里打印一下接收到的state，观察一下里面的内容。

你会发现，节点函数里返回的字段，LangGraph 并不会替换整个 State，而是做了一次“增量合并”。也就是说：

• 你返回了{"counter": 1}，那么 State 里的counter字段变成 1；
• 你返回了{"message": "..."}，那么 State 里的message字段被更新；
• 其他字段如果没有被返回，则原封不动保留。

这个“增量合并”的机制很像是给共享的表格填内容。每个节点负责更新自己关心的那几列，其他列不受影响。这样设计的好处是：节点之间是松耦合的。每个节点只声明它负责写什么字段，不需要关心整个 State 的全量结构。在复杂的智能体里，这个特性非常重要。比如一个节点只负责更新日志字段，它根本不需要知道用户 ID、订单信息长什么样，直接返回{"log": "...", "timestamp": "..."}就行，完全不影响其他节点写入的数据。

理解了这一层，再去看条件路由、循环和持久化，思路就顺了。它们本质上都是在 State 这个共享数据表的基础上做文章。

3. 实战拆解：带条件分支和循环的订单查询智能体

3.1 场景设计与状态字段规划

光看计数器肯定不够过瘾，现在做一个更实际的东西：一个订单查询智能体。用户输入之后，机器人先判断意图，然后查订单，最后生成一段回复。整个流程里需要有两个分支：如果用户没登录，先引导登录；如果查询条件不完整，再追问补全。你会看到条件路由和循环是怎么在真实场景里发挥作用的。

我们规划一下状态字段。这里不需要一次把所有字段都定义得特别复杂，够用就行。

from typing import TypedDict, Optional, Annotated from langgraph.graph import StateGraph, START, END class OrderState(TypedDict): query: str user_id: Optional[str] intent: str order_info: Optional[str] response: str turn_count: int

这几个字段各有分工：

• query存放用户输入，可能是“查一下 A001 订单”，也可能是“我的订单到哪了”；
• user_id表示当前是否登录，为空时说明未登录；
• intent存放意图识别结果，比如query_order、unknown；
• order_info存放查出来的订单信息；
• response存放最终回复；
• turn_count用来记录追问次数，防止循环永远跑不完。

接下来我把节点拆开写。

3.2 实现意图识别、查询、登录、回复节点

第一个节点是意图识别。真实项目里这里可以接入一个 LLM 调用，演示阶段我用简单的关键词判断。

def intent_node(state: OrderState) -> dict: query = state.get("query", "") if ("订单" in query) or ("物流" in query) or ("到哪" in query): return {"intent": "query_order"} return {"intent": "unknown"}

第二个节点是查询订单。这里做一个假数据库来模拟返回结果。注意，如果用户只说了“查订单”却没有提供订单号，节点就不应该直接返回查询结果，而是返回一个缺失订单号的标记，让图流程决定下一步该怎么走。

FAKE_DB = { "A001": "订单 A001 已发货，预计明天下午到达", "A002": "订单 A002 正在仓库打包", } def query_order_node(state: OrderState) -> dict: query = state.get("query", "") # 简单粗暴地认为 query 里的 Axxx 就是订单号 if "A" in query: for key in FAKE_DB: if key in query: return {"order_info": FAKE_DB[key]} return {"order_info": "没有找到该订单，请确认订单号是否正确"} return {"order_info": None, "intent": "need_order_id"}

第三个是未登录时的处理节点。真实系统里应该是跳转登录页或会话层做鉴权，这里我用一个占位节点表示“需要用户先登录”。

def login_node(state: OrderState) -> dict: return {"response": "请先登录后再查询订单。"}

第四个是回复节点，它把上面的结果拼成一段最终回复。如果order_info不为空，就回复订单信息，否则回复提示。

def respond_node(state: OrderState) -> dict: if state.get("order_info"): response = f"查询结果：{state['order_info']}" elif state.get("intent") == "unknown": response = "暂时没听懂你的意思，你可以试试说：查一下 A001 订单" else: response = "请提供订单号，例如：A001" return {"response": response}

3.3 用条件路由决定下一步去哪

现在把这些节点放进图里。关键点是：intent_node跑完之后不是接着固定跑下一个节点，而是根据intent字段的内容选择不同的分支。

builder = StateGraph(OrderState) builder.add_node("intent", intent_node) builder.add_node("query", query_order_node) builder.add_node("login", login_node) builder.add_node("respond", respond_node) builder.add_edge(START, "intent") # 条件路由：根据 intent 字段跳转 builder.add_conditional_edges( "intent", lambda state: state.get("intent"), { "query_order": "query", "need_login": "login", "unknown": "respond", }, ) # query 查完去 respond builder.add_edge("query", "respond") # login 处理完也去 respond builder.add_edge("login", "respond") builder.add_edge("respond", END) graph = builder.compile()

这里add_conditional_edges的第二个参数是一个路由函数，它接收 State，返回一个字符串。第三个参数是映射表，把返回的字符串映射到实际节点名。你不需要把每一对都列完，但如果路由函数返回了一个映射表里不存在的字符串，LangGraph 会抛异常，找不到对应的下一步节点。这也是很多新手最容易踩的坑。

你现在可以试一下几种输入：

print(graph.invoke({"query": "查一下 A001 订单"})) # intent 为 query_order，走 query -> respond print(graph.invoke({"query": "你好"})) # intent 为 unknown，走 respond print(graph.invoke({"query": "查订单"})) # query 节点返回 intent=need_order_id，但直接去了 respond

3.4 引入循环：让机器人学会“追问”

上面的流程还有一个不完美的地方。用户说“查订单”而没有给订单号时，query_order_node返回了intent=need_order_id，但目前图里没有针对这个字段做条件跳转，所以它还是直接进了respond_node，回一句“请提供订单号”。这已经是能用的了，但它没有循环——用户重新输入订单号之后，整个状态图要从头开始重新跑，用户之前输入的“查订单”这个意图白费了。

一个更有状态的智能体应该是这样的：第一轮，用户说“查订单”，机器人回复“请提供订单号”；第二轮，用户说“A001”，机器人能结合上一轮的上下文，知道这是在补充订单号，然后直接去查询。

要实现这个效果，我们可以加一个“追问”循环：当query_order_node发现订单号缺失时，不让流程走到respond_node，而是让流程回到一个ask_order_node，由这个节点生成追问的回复，然后图停在ask_order_node，等下一轮用户输入，再把turn_count加一，继续往下走。

为了不把代码搞得太复杂，我们简化一点：用一个need_order_id的条件分支，让query_order_node在信息不足时回到自身循环，同时用turn_count限制最多循环两轮。

def ask_order_node(state: OrderState) -> dict: turn = state.get("turn_count", 0) return {"response": "请提供订单号，例如 A001，否则我没办法查到订单。", "turn_count": turn}

然后改一下图的结构：

builder.add_node("ask", ask_order_node) # query 节点走完，根据是否拿到 order_info 来决定下一步 builder.add_conditional_edges( "query", lambda state: "ask" if state.get("turn_count", 0) < 2 and not state.get("order_info") else "respond", {"ask": "ask", "respond": "respond"}, ) builder.add_edge("ask", "respond") # 这里 ask 结束后简单进 respond

我把这个逻辑简化了，实际项目中一般会再结合 LLM 和更完善的节点状态管理。但你已经能看出循环的意义：它让智能体在信息不足时停留在某个节点，而不是从头重启。这种“停在中间等用户补全信息”的能力，正是它区别于普通 Chain 的关键。

4. 状态管理与持久化：从内存字典到可恢复会话

4.1 为什么需要持久化

把状态放在内存里跑通很容易，但真实场景里没人会把状态只放在内存里。假设你做了一个客服智能体，用户早上问了一句“我的订单到哪了”，下午又问“现在呢？”如果你不持久化状态，那个订单上下文早就丢了，智能体只能干巴巴地回复“请问您的订单号是多少”。

更严重的问题在于进程重启。线上服务发布新版本，进程一重启，所有内存里的状态全部消失，所有正在进行的对话全部断开。你跟用户说到一半的对话，因为一次发布全部重来。这对生产系统来说是不可接受的。

LangGraph 给出的方案叫 Checkpointer。它的作用是把 State 在每一步的执行结果都保存起来，并且按会话 ID（thread_id）隔离。一旦状态保存下来，你就能做到：

• 在任意一轮对话后恢复完整状态；
• 按会话 ID 回到之前的执行步骤；
• 多用户之间互相隔离，A 用户的状态不会污染 B 用户。

4.2 Checkpointer 和 thread_id

Checkpointer 可以理解为一个“状态仓库”。每次节点执行完，LangGraph 会把最新的 State 写入这个仓库。当你下一次用同一个 thread_id 调用同一个图时，它会把之前的 State 读出来，恢复现场，再接着执行。

LangGraph 内置了几种 Checkpointer，用法都很像，区别在于存储介质和适用场景：

4.3 实际配置一个可恢复的会话

配置 Checkpointer 的代码非常简单。以MemorySaver为例：

from langgraph.checkpoint.memory import MemorySaver checkpointer = MemorySaver() graph = builder.compile(checkpointer=checkpointer)

然后调用的时候，传入一个带thread_id的配置：

config = {"configurable": {"thread_id": "session-001"}} graph.invoke({"query": "查订单"}, config=config) graph.invoke({"query": "A001"}, config=config)

第一次调用后，State 里已经记录了用户想查订单、以及当前状态图走到了需要订单号的步骤。第二次调用时，你会惊讶地发现，State 里的intent还是上一轮的query_order，query从“查订单”变成了“A001”，并且前一轮的response还保留着。你可以用下面的代码验证一下：

state = graph.get_state(config) print(state.values)

打印结果会包含这一条会话里积累下来的所有字段：query、intent、order_info、response、turn_count。这就是持久化最直观的效果。

生产环境里我建议用SqliteSaver或PostgresSaver。拿SqliteSaver举例，你只需要多一步连接：

from langgraph.checkpoint.sqlite import SqliteSaver with SqliteSaver.from_conn_string("checkpoints.sqlite") as checkpointer: graph = builder.compile(checkpointer=checkpointer)

看起来差别不大，但它把状态落到了磁盘，进程重启后状态还在。多实例部署时，则用PostgresSaver，让所有实例共享同一份状态数据。

4.4 字段更新策略：Reducer 是状态设计的核心

刚才我们提到增量合并规则是“返回字段直接覆盖旧字段”。但这个规则不是绝对的。LangGraph 允许你为每个字段定义自己的更新策略，这就是 Reducer。

比如我们要记录turn_count。每次进入某个节点，不是把turn_count覆盖成固定值，而是希望在上一个值的基础上累加。如果是普通覆盖，A 节点返回 1，B 节点返回 1，那永远都是 1，根本起不到计数作用。给turn_count加一个 Reducer 就能解决：

def increment_counter(existing_value: int, new_value: int) -> int: return (existing_value or 0) + new_value

然后在 StateSchema 里这么用：

from typing import Annotated class OrderState(TypedDict): query: str turn_count: Annotated[int, increment_counter] # ...

注意，Reducer 接收的第一个参数是 State 里的当前值，第二个参数是节点返回的新值。LangGraph 会把两者按你定义的规则合并，然后再写回 State。同理，处理消息列表时，langgraph.graph.message里提供的add_messages就是一个 Reducer，它会把新消息追加到消息列表末尾，而不是覆盖整份历史。

from langgraph.graph.message import add_messages class OrderState(TypedDict): messages: Annotated[list, add_messages] # ...

Reducer 设计的核心思想是：字段的更新策略属于 Schema 的一部分，而不是节点的一部分。节点只负责“要更新什么”，Reducer 负责“怎么更新”。这进一步降低了节点之间的耦合度，也让状态流转的规则有了清晰的定义位置。

5. 常见问题与排查技巧实录

5.1 状态字段被覆盖，值“丢”了

这是新手最容易踩的坑。你明明在节点 A 里写了user_id，到节点 B 里一读却是None。原因是可能有两个节点都返回了同一个字段，后执行的节点用返回值覆盖了前面的值。比如intent_node返回了{"intent": "query_order"}，而后面某个节点又返回了{"intent": None}，这种字段就会被覆盖成None。

排查方法很简单：在两个节点里分别打印state，看字段是什么时候丢的。如果确认多个节点都会写同一个字段，就要考虑是不是应该合并而不是覆盖。这种情况通常用 Reducer 就能解决。

5.2 条件路由永远走不通，或者总是走默认分支

条件路由的报错千奇百怪，但最常见的只有这么几个原因：

• 路由函数返回的字符串拼错了，和映射表里的键不一致；
• 条件路由的映射表里面漏了一项，导致某个分支没地方去；
• 路由函数里读取的字段不存在，永远是None，结果永远走进了兜底分支。

我建议你给路由函数写一个配套的print，在调试阶段把 state 里的关键字段打出来，确认到底读取到了什么值。另外，add_conditional_edges的映射表里最好补一个"unknown": "some_node"之类的兜底项，别让意外值直接打挂流程。

5.3 多轮对话“不记忆”，每次都像第一次见面

这个问题十有八九是没有配 Checkpointer，或者配了 Checkpointer 但每次调用都传了不同的thread_id。thread_id相当于一条会话的身份 ID，你换了 ID，就等于开了一个新会话，之前的状态自然全部清零。

排查时可以先打印一下你的config，确认每次传的thread_id是否是同一个变量。如果是同一个，再检查是否真的把 checkpointer 传进了compile()。这两个环节都查过之后，问题基本就能定位。

5.4 状态里出现 None，下游节点一读就崩

State 字段不是所有节点都会返回，如果你在某个节点读了order_info，但前面并没有节点写入过它，拿到的就是None。应对策略有两种：一是在 StateSchema 里给字段定义默认值，二是在读取字段时用state.get("order_info")配合默认值兜底。

class OrderState(TypedDict): order_info: Optional[str] = None

这里要注意，TypedDict 本身没有字段顺序的概念，但你可以在声明时把 Optional 类型标注好。配合dict.get的默认值，绝大多数空值问题都能避免。

5.5 调试技巧：从图结构、节点输入和状态快照入手

LangGraph 的调试跟普通函数不一样，你不能只盯着某一个函数内部看。我个人的调试顺序是这样的：

第一步，打印图结构。builder.compile()之后，你可以用可视化工具看整张图长什么样。有条件的话装一下langgraph-cli和配套的本地调试界面，它会一帧一帧告诉你每一步哪个节点被执行了、State 变成了什么样，这对理解复杂分支帮助巨大。

第二步，直接单测单个节点函数。因为节点函数本身就是一个普通的 Python 函数，你完全可以构造一个假的 State 字典，直接调intent_node({"query": "查订单"})，看返回的字典对不对。这样把每个节点测好之后，再拼成图去集成测试，问题范围会小很多。

第三步，用graph.get_state(config)查看指定会话的完整状态快照。这个命令可以帮你看“当前停在哪、State 里存了什么”，多轮对话场景下特别实用。

这些排查技巧都是我在实际项目里反复用过的。起步阶段遇到问题别慌，先确认字段在哪个节点丢的，再确认路由函数读到的是什么值，最后确认会话 ID 是否稳定，多数问题都能在这三步里定位。

最后再分享一点体会：LangGraph 的难点从来不在写节点，而在怎么设计状态。你在规划一个新项目时，别急着往图里加节点，先把 State 字段梳理清楚——哪些字段是长期记忆，哪些字段只是临时中转，哪些字段需要累加或追加更新，哪些字段只允许特定节点修改。这个 Schema 设计好了，后面的分支、循环、持久化都是顺手的事。可以先拿一个订单查询这类小项目练手，跑通状态、条件路由、Checkpointer 三个关键点之后，再去挑战更复杂的多 Agent 协作场景，会轻松很多。