17. 状态持久化与检查点
在第 15 章中,我们将代理循环重构为 StateGraph;在第 16 章中,我们使用预构建组件和多分支路由构建了更加复杂的东西。到目前为止,我们编写的每个图都有一个共同的局限:图不会保留它的状态。
图只在单次 invoke() 调用期间持有并管理状态,一刻也不会更久。当你调用它时,LangGraph 会创建一个全新的状态,运行各个节点,根据规约器(reducer)规则将每个返回值合并到状态中,然后将最终状态交还给调用方。一旦交出该状态,图便不再记得它。下一次 invoke() 会从一个全新的状态开始,与上一次调用毫无关联。
由此产生了两个问题。第一,messages 也是状态的一部分,所以代理无法记住你之前说过的任何内容。第二,如果某次运行中途失败,它在此之前完成的一切都会消失。假设第三个节点抛出异常:前两个节点产生的结果也随之消失,你不得不从头开始。回到 15.1 节,我们曾把"从中断中恢复"列为选择 LangGraph 的理由之一——这正是我们当时想到的问题。
LangGraph 在框架层面处理了这个问题。为图附加一个检查点器(checkpointer),LangGraph 就会在执行的每一步自动保存状态的快照。这些保存的快照比 invoke() 调用存活得更久,所以下一次调用可以从上一次停下的地方继续。这种特性——状态在单次运行之外依然留存——被称为持久化(persistence)。
其实你之前已经用过检查点器了。在第 11 章中,当我们为对话式 RAG 代理赋予多轮记忆时,我们传入了 create_agent(..., checkpointer=InMemorySaver()) 和一个 thread_id。那时候你只需要知道检查点器会按 thread_id 保存对话历史;我们从未解释它是如何工作的。而在第 16 章中,当我们引入 checkpointer 参数时,我们说过"我们会在第 17 章讲解它的工作原理"。本章就是那一章。
它分为三个部分。在 17.1 节,我们为图附加检查点器,并借助 thread_id 进行多轮对话。在 17.2 节,我们打开保存的检查点,查看代理在任意给定时刻所知道的内容——检查点是你追查代理为何行为异常的主要工具。在 17.3 节,我们把一个中途失败的图从停下的地方恢复,而不是从头开始。
17.1) 跨调用携带状态
17.1.1) 会遗忘的图
本章开篇我们说过图不会保留它的状态。让我们用代码来验证这一点。
下面这个图的形状与 15.2 节中的 say_hello 图相同。唯一的区别是节点返回的是 LLM 响应而不是固定字符串。
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-5-mini")
def llm_call(state: MessagesState):
response = model.invoke(state["messages"])
return {"messages": [response]}
builder = StateGraph(MessagesState)
builder.add_node(llm_call)
builder.add_edge(START, "llm_call")
builder.add_edge("llm_call", END)
graph = builder.compile() # 无检查点器现在让我们进行一段两轮对话。第一次调用时告诉模型我们的名字,然后在第二次调用时问它我们的名字是什么。
# 第一次调用——我们说出自己的名字
graph.invoke({"messages": [{"role": "user", "content": "Hi, my name is Bob."}]})
# 第二次调用——我们反过来问它
result = graph.invoke({"messages": [{"role": "user", "content": "What's my name?"}]})
print(result["messages"][-1].content)输出:
I'm sorry, but I don't have your name. Could you tell me what it is?第二次调用时模型收到的唯一消息是 "What's my name?"。第一次调用的状态已不在图中,所以更早的消息——那些带着名字的消息——从未到达模型。
我们当然可以自己修复这个问题。保留第一次调用返回的 messages,并在第二次调用时一并传入。这正是我们在第 8 章中管理对话历史的方式。但那样一来,我们就得为每段对话和每个用户编写自己的代码来存储和检索历史。这正是检查点器为你分担的工作。
17.1.2) 检查点与检查点器
检查点器(checkpointer)是一个专门负责保存状态的对象。你创建一个实例——例如 InMemorySaver()——并将它传给 builder.compile(checkpointer=...),从而把它附加到你的图上。
一旦附加了检查点器,图就会在执行过程中复制整个状态并将其保存下来。每一份这样保存的副本被称为检查点(checkpoint)。可以把它想象成一张照片:那一瞬间的整个状态,原封不动地保存下来。
电子游戏中的自动存档正是恰当的心理图景。每当你通过一个有意义的节点,游戏就会悄悄记录你的进度,这样你就可以退出以后再回来,或者死掉时不必从头开始。检查点器对图做的事情正是如此。
那么什么算是"有意义的节点"?LangGraph 把图的执行划分为若干阶段,每个阶段被称为一个超级步(super-step)。每当一个超级步完成时,就会保存一个检查点。
之所以称之为超级步而不只是步,是因为单个阶段可以同时运行多个节点。在像 17.1.1 节那样节点排成一条直线的图中,运行一个节点就是一个超级步。但在多个节点并行运行的图中,所有这些节点合在一起构成单个超级步。
所以即使单次调用 invoke() 也会留下好几个检查点。我们会在 17.2 节把它们取出来,看看每一个究竟包含什么。
我们一直用作示例的 InMemorySaver 是最简单的检查点器。顾名思义,它把检查点存储在进程的内存(RAM)中。无需安装,也无需配置,这使它非常适合学习和本地开发。代价是每个保存的检查点都会在进程重启时消失。我们会在 17.1.5 节介绍生产环境的替代方案。
17.1.3) 添加检查点器
附加检查点器只需要两样东西。
- 创建一个检查点器实例并将其传给
compile()。 - 每次调用
invoke()时传入一个包含thread_id的config。
我们稍后会讲到为什么需要第二样东西。现在,你只需知道它告诉检查点器你想继续它保存的哪一段对话。
让我们把这两样都应用到 17.1.1 节的图上。
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-5-mini")
def llm_call(state: MessagesState):
response = model.invoke(state["messages"])
return {"messages": [response]}
builder = StateGraph(MessagesState)
builder.add_node(llm_call)
builder.add_edge(START, "llm_call")
builder.add_edge("llm_call", END)
# 1. 创建一个检查点器并将其传给 compile()
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
# 2. 向 invoke() 传入携带 thread_id 的 config
config = {"configurable": {"thread_id": "1"}}
graph.invoke(
{"messages": [{"role": "user", "content": "Hi, my name is Bob."}]},
config,
)
result = graph.invoke(
{"messages": [{"role": "user", "content": "What's my name?"}]},
config,
)
print(result["messages"][-1].content)输出:
Your name is Bob.和 17.1.1 节相同的两次调用,却是不同的结果。这一次名字记住了。
原因如下。附加了检查点器的图会在运行 llm_call 节点之前加载保存的状态。该状态已经包含了第一轮的交流。然后我们传入的新消息会被合并进去。正如 15.2.2 节所见,messages 字段带有 add_messages 规约器,所以新消息会被追加到已有列表末尾。模型最终收到了三条消息:问候语、它自己的第一次回复,以及新的问题。
我们只发送了新消息,而 LangGraph 从上一个检查点加载了更早的对话。 我们在第 8 章手动管理的对话历史,现在由 LangGraph 管理了。
17.1.4) thread_id:区分对话的标识符
thread_id 是一个用来区分不同对话的标识符。取值由你决定。我们在 17.1.3 节用的是 "1",但任何字符串都可以。用相同的 thread_id 调用图,你就是在继续那段对话;用不同的 thread_id 调用,你就开启了一段独立的对话。
让我们来验证一下。我们将让 Alice 和 Bob 通过同一个图进行不同的对话。
def send(thread_id: str, text: str) -> str:
config = {"configurable": {"thread_id": thread_id}}
result = graph.invoke(
{"messages": [{"role": "user", "content": text}]},
config,
)
return result["messages"][-1].content
# Alice 的对话
send("alice", "My favorite color is teal.")
# Bob 的对话——不同的 thread_id
send("bob", "My favorite color is orange.")
# 分别再问一次
print("Alice:", send("alice", "What's my favorite color?"))
print("Bob: ", send("bob", "What's my favorite color?"))输出:
Alice: Your favorite color is teal.
Bob: Your favorite color is orange.两段对话都经过了同一个 graph 对象和同一个检查点器,却从未混淆。thread_id 是检查点器用来存储和查找状态的主键(primary key)。不同的键,完全独立的存储。
那么如果你把这个值省略掉会发生什么?
graph.invoke({"messages": [{"role": "user", "content": "Hello"}]})输出:
ValueError: Checkpointer requires one or more of the following 'configurable' keys: thread_id, checkpoint_ns, checkpoint_id图根本不会运行。一旦附加了检查点器,thread_id 就不是可选的——它是必需的。
鉴于我们刚才所见,这是合乎情理的。在运行节点之前,检查点器必须加载保存的状态——而 thread_id 正是告诉它加载哪段对话的状态的东西。
这就是一个聊天机器人服务的基本形态:一个图、一个检查点器,以及每个用户或每个聊天室对应一个 thread_id。
17.1.5) InMemorySaver 的局限与生产环境替代方案
我们之前说过 InMemorySaver 把检查点保存在内存中。这个选择带来了两个局限。
重启进程,一切就都没了。 重新部署服务或重启服务器,迄今为止积累的每段对话都会消失。
不同进程之间无法共享它。 真实的服务会把进来的请求分散到多个进程上。每个进程都有自己的内存,所以进程 A 保存的对话对进程 B 是不可见的。用户可以每次都发送相同的 thread_id,却仍然眼睁睁看着对话崩溃,这取决于恰好是哪个进程接收了请求。
这就是为什么生产环境使用将检查点存储在数据库中的检查点器。
SqliteSaver/AsyncSqliteSaver(langgraph-checkpoint-sqlite)——把所有内容存储在单个文件中。适合运行在单台服务器上的小型服务,或本地原型。PostgresSaver/AsyncPostgresSaver(langgraph-checkpoint-postgres)——把检查点存储在数据库服务器中。增加更多服务器后,每个进程仍能看到相同的检查点。这是 LangGraph 文档推荐用于生产环境的检查点器。
它们都实现了与 InMemorySaver 相同的接口。你的图代码、你的节点,以及你使用 thread_id 的方式都保持不变。唯一改变的是你如何创建检查点器。
你在第 16 章认识的
create_agent以同样的方式使用检查点器。向它的checkpointer参数传入一个检查点器,并向invoke()传入携带thread_id的config。这正是第 11 章中对话式 RAG 代理能记住更早轮次的原因。
本章接下来我们会继续使用 InMemorySaver。无论检查点恰好存放在哪里,你与它们打交道的方式都是一样的。
17.2) 检查状态与调试
在 17.1.2 节我们说过检查点器会在每个超级步保存状态。让我们把这些检查点取出来看一看。
打开保存的检查点并非只是出于好奇。代理会行为异常。它们会陷入永不停止的循环反复调用工具,会丢失更早轮次的线索,会走上一条你从未预料的分支。要弄清楚原因,你需要知道那一刻状态是什么样子——而检查点里就有答案。
LangGraph 给了你两个方法。
graph.get_state(config)——返回那段对话的最近一个检查点。graph.get_state_history(config)——返回那段对话的每一个检查点,最新的在前。
两者都需要 config 里有 thread_id,因为它们必须知道你要的是谁的检查点。
这两个方法返回的检查点以 StateSnapshot 对象来表示。
17.2.1) StateSnapshot:检查点内部有什么
让我们看看检查点实际包含什么。这个例子里我们不使用 LLM。LLM 每次返回的东西都不一样,当我们想逐个字段仔细过一遍时,这毫无帮助。因此我们改为构建一个返回固定值的小图。
状态将有两种字段。foo 没有规约器,所以它会被覆盖;bar 有一个规约器,所以它会累积。这与我们在 15.2.2 节给 messages 附加的 add_messages 规约器是同样的安排——这里我们用 Python 的 operator.add 作为规约器来拼接列表。
from operator import add
from typing_extensions import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
class State(TypedDict):
foo: str # 无规约器 → 覆盖
bar: Annotated[list[str], add] # add 规约器 → 累积
def node_a(state: State):
return {"foo": "a", "bar": ["a"]}
def node_b(state: State):
return {"foo": "b", "bar": ["b"]}
builder = StateGraph(State)
builder.add_node(node_a)
builder.add_node(node_b)
builder.add_edge(START, "node_a")
builder.add_edge("node_a", "node_b")
builder.add_edge("node_b", END)
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "1"}}
graph.invoke({"foo": "", "bar": []}, config)
snapshot = graph.get_state(config)
print(snapshot)输出:
StateSnapshot(
values={'foo': 'b', 'bar': ['a', 'b']},
next=(),
config={'configurable': {'thread_id': '1', 'checkpoint_ns': '',
'checkpoint_id': '1f17da03-9654-65d8-8002-9e59231bb481'}},
metadata={'source': 'loop', 'step': 2, 'parents': {}},
created_at='2026-07-12T03:17:33.637368+00:00',
parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '',
'checkpoint_id': '1f17da03-9653-6ca0-8001-7c24c8ec66f2'}},
tasks=(),
interrupts=()
)八个字段。让我们一个一个来看。
values——这个检查点处的状态。bar是['a', 'b'],因为add规约器累积了两个节点各自返回的内容;foo是'b',因为它没有规约器,所以最后一次写入胜出。这个字段回答的是"那一刻状态是什么样子?"next——一个元组,列出这个检查点之后要运行的节点名称。空的()意味着没有任何内容要运行了,也就是说图已经完成。('node_b',)意味着node_b仍在前面等着。config——这个检查点的地址。thread_id标识对话,而checkpoint_id标识其中的哪一刻。LangGraph 每次保存检查点时都会自动分配checkpoint_id。metadata——关于这次运行的记录信息。source告诉你检查点来自哪里:"input"意味着它是从你交给invoke()的输入构建而来,而"loop"意味着它是在图运行时产生的。step是超级步的编号。created_at——检查点保存的时间。当你需要把它和日志对齐时很有用。parent_config——紧邻这个检查点之前那个检查点的config。顺着它你就可以沿着这次运行向后回溯。对于最开头的那个检查点,它是None。tasks——next中所列节点的执行记录。在检查点保存的那一刻,那些节点还没有运行;一旦它们运行,其结果就会附加到这个检查点上。成功的节点会把它的返回值留在result中,失败的节点会把它的异常留在error中。interrupts——图暂停以将控制权交还给人的地方。LangGraph 可以在运行中途停下来,等待某人批准某一步骤或提供某个值,这个字段记录了那些暂停。
你以属性的方式读取这些字段。metadata 是一个字典,所以你用键从中取值。
snapshot = graph.get_state(config)
print(snapshot.values) # {'foo': 'b', 'bar': ['a', 'b']}
print(snapshot.next) # ()
print(snapshot.metadata["step"]) # 2在这些字段中,调试时你最常用到的是 next。如果 next 不为空,说明图没有走到终点——它在中途某处停了下来。而当我们在 17.3 节恢复一个失败的图时,这个字段正是我们的起点。
17.2.2) 遍历检查点历史
get_state() 只显示最近的一个检查点。但调试往往意味着要问"我们是怎么走到这一步的?"——为此你需要这次运行的整个轨迹。get_state_history() 就能给你。
for snap in graph.get_state_history(config):
print(f"step={snap.metadata['step']:>2} "
f"next={str(snap.next):<16} values={snap.values}")输出:
step= 2 next=() values={'foo': 'b', 'bar': ['a', 'b']}
step= 1 next=('node_b',) values={'foo': 'a', 'bar': ['a']}
step= 0 next=('node_a',) values={'foo': '', 'bar': []}
step=-1 next=('__start__',) values={'bar': []}最新的检查点在最前面,所以要从下往上看,才能按顺序追随这次运行。
- step -1——就在
invoke()接收到输入之后。注意我们传入的{"foo": "", "bar": []}并没有出现在values中。把输入写进状态本身就是一个阶段,而那个阶段还没有运行。next中的__start__就是执行这件事的内部节点。bar显示为[],但那并不是我们传入的值。带有规约器的字段会以一个空值开始,以便让后续写入累积进去。foo没有规约器,所以它根本没有起始值——这就是为什么它没有出现在这里。 - step 0——
__start__已经运行,输入现在在状态中了。foo=''和bar=[]就是我们传入的值。node_a是接下来要运行的。 - step 1——运行
node_a的结果。foo现在是'a',bar是['a'],接下来是node_b。 - step 2——运行
node_b的结果。next为空,所以图完成了。
step 0 或 step 1 处非空的 next 并不意味着图在那里停了下来。在图仍在运行时拍下的检查点,自然会有一个节点排在下一个位置。当 17.2.1 节说"非空的 next 意味着图停了下来"时,它指的是出问题的那个点上的检查点。在历史的中间部分,next 只是向你展示图走了哪条路径。
17.3) 从失败处恢复
因为状态在每个超级步结束时都会被保存,所以中途失败并不会把已完成的工作一起带走——它仍然留在检查点里。没有理由从头开始。你从停下的地方接着往下。
17.3.1) 用 invoke(None, config) 恢复
恢复很简单:在输入的位置传入 None。
graph.invoke(None, config)它的意思是"没有新输入;从保存的状态继续"。当然,config 仍然需要一个 thread_id,因为 LangGraph 必须知道要继续哪段对话。
让我们设置一次失败,然后从中恢复。我们将构建一个双节点的图,它的第二个节点只在第一次运行时失败。它必须在重试时成功,否则我们就永远看不到恢复真的起作用了。
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
class State(TypedDict):
step_1_done: bool
step_2_done: bool
first_try = True # 用于让第一次运行失败的标志
def step_1(state: State):
print("step_1 running (expensive work)")
return {"step_1_done": True}
def step_2(state: State):
global first_try
if first_try:
first_try = False
print("step_2 failed (API timeout)")
raise RuntimeError("External API timed out")
print("step_2 running")
return {"step_2_done": True}
builder = StateGraph(State)
builder.add_node(step_1)
builder.add_node(step_2)
builder.add_edge(START, "step_1")
builder.add_edge("step_1", "step_2")
builder.add_edge("step_2", END)
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "job-42"}}
try:
graph.invoke({"step_1_done": False, "step_2_done": False}, config)
except RuntimeError as e:
print("Failed:", e)输出:
step_1 running (expensive work)
step_2 failed (API timeout)
Failed: External API timed outstep_1 成功了,step_2 抛出了异常。让我们用 17.2 节学到的 get_state() 找出图停在了哪里。
snapshot = graph.get_state(config)
print("next =", snapshot.next)
print("values =", snapshot.values)输出:
next = ('step_2',)
values = {'step_1_done': True, 'step_2_done': False}next 是 ('step_2',),这告诉我们图在运行 step_2 的中途停了下来。而在 values 中,step_1_done 是 True——step_1 的结果仍然留在检查点里。
现在我们用 None 恢复。
result = graph.invoke(None, config)
print("Final =", result)输出:
step_2 running
Final = {'step_1_done': True, 'step_2_done': True}step_1 running (expensive work) 从未打印出来。 step_1 没有第二次运行。LangGraph 加载了保存的状态,并从 step_2 接着往下。我们没有为那个代价高昂的第一步付两次费用。
17.3.2) 恢复时需要注意什么
当你恢复时,失败的那个节点会再次运行。如果那个节点调用了 LLM 或访问了外部 API,那些调用也会再次发生——而且它们返回的结果可能会有所不同。
陷阱就在这里。如果 step_2 发送了一封邮件然后才失败,恢复会发送第二封邮件。LangGraph 只保证它不会重新运行那些成功的节点。
所以任何可能再次运行的节点都需要是幂等的(idempotent):同一件事做两次应该让你处在同样的境地。在发送邮件之前检查邮件是否已经发出去了;在数据库表上加一个唯一键,让重复插入不可能发生。