LangGraph 持久化与记忆管理详解：检查点、可恢复执行、Time Travel 与长期记忆

LangGraph 持久化与记忆管理：检查点、可恢复执行、Time Travel 与长期记忆

前面学习 LangGraph 时，我们主要关注了两个问题：

图怎么搭？
节点怎么流转？

但是到了真实项目里，只会搭图还不够。因为 Agent 或工作流经常会遇到这些问题：

运行到一半失败了，能不能从失败位置继续？
用户多轮对话，能不能记住上下文？
我想查看图每一步的状态，能不能看到？
我想回到某个历史节点重新执行，能不能做到？
程序重启后，之前的状态还在不在？

这些问题对应的就是 LangGraph 里的 持久化机制、检查点、可恢复执行和记忆管理。

这篇文章按由浅到深整理：检查点是什么、如何启用、如何查看历史状态、如何恢复执行、Time Travel、短期记忆、长期记忆、运行时上下文，以及 PostgreSQL 在持久化中的作用。

一、什么是可恢复执行？

可恢复执行可以简单理解成：

图运行过程中，每执行到关键步骤，就保存一次状态。
如果后面失败了，可以从之前保存的状态继续执行。

这就像打游戏时的存档。

没有存档时，失败就要从头开始。

有了存档后，可以从最近一次存档继续。

在 LangGraph 里，这个“存档”就叫：

Checkpoint，检查点

二、LangGraph 的持久化机制是什么？

LangGraph 的持久化机制，就是把图运行过程中的状态保存下来。

保存的内容通常包括：

• 当前状态值
• 下一步要执行哪些节点
• 当前检查点 ID
• 父检查点信息
• 当前任务信息
• 中断信息

这些信息合起来，就可以描述图在某一刻的完整运行状态。

核心组件主要有两个：

可以这样理解：

Checkpointer = 存档器
thread_id = 存档属于哪个会话
checkpoint_id = 某一次具体存档的编号

三、启用检查点：InMemorySaver

最简单的检查点存储器是：

from langgraph.checkpoint.memory import InMemorySaver

示例：

checkpointer = InMemorySaver()

config = {
"configurable": {
"thread_id": "123"
}
}

graph = builder.compile(checkpointer=checkpointer)

res = graph.invoke(
{"topic": "猫咪"},
config=config
)

这里有三个关键点。

1. InMemorySaver

checkpointer = InMemorySaver()

表示检查点保存在内存中。

优点是简单，适合学习和测试。

缺点是程序一关，数据就没了。

所以它不适合生产环境长期保存。

2. thread_id

config = {
"configurable": {
"thread_id": "123"
}
}

thread_id 用来标识一次会话。

如果是聊天机器人，可以把一个用户的一次对话看成一个 thread_id。

比如：

用户 A 的会话：thread_id = user_a_session_001
用户 B 的会话：thread_id = user_b_session_001

这样不同用户、不同会话的检查点就不会混在一起。

3. compile(checkpointer=…)

graph = builder.compile(checkpointer=checkpointer)

编译图时传入 checkpointer，表示这张图启用检查点功能。

如果没有传检查点存储器，后面就不能正常查看历史状态，也无法恢复执行。

四、一个并行图的检查点示例

假设图结构是：

START
↓
node_poem 和 node_joke 并行执行
↓
node_output 汇总结果
↓
END

输入：

graph.invoke({"topic": "猫咪"}, config=config)

运行过程中大概会产生这些检查点：

Step -1 -> Step 0 -> Step 1 -> Step 2

真实执行顺序是从小到大：

Step -1 -> Step 0 -> Step 1 -> Step 2

但查看历史记录时，get_state_history() 默认是倒序返回：

Step 2 -> Step 1 -> Step 0 -> Step -1

也就是：

最新检查点 -> 最旧检查点

五、StateSnapshot 是什么？

查看历史检查点时：

history_checkpoints = list(graph.get_state_history(config=config))

返回的是一组 StateSnapshot 对象。

StateSnapshot 可以理解成：

某一个检查点时刻的状态快照。

常见字段有：

六、检查点里的 values 和 next

看检查点时，最重要的是两个字段：

values
next

values

values 表示当前状态。

比如：

values = {
"topic": "猫咪",
"poem": "…",
"joke": "…"
}

说明图已经生成了诗和笑话。

next

next 表示从当前检查点继续执行时，下一步要执行哪些节点。

比如：

next = ("node_output",)

表示下一步执行 node_output。

如果：

next = ()

说明图已经执行结束。

七、检查点顺序为什么是 Step 2 -> Step 1 -> Step 0 -> Step -1？

因为 get_state_history() 默认按倒序返回。

真实运行顺序是：

Step -1 -> Step 0 -> Step 1 -> Step 2

历史查询顺序是：

Step 2 -> Step 1 -> Step 0 -> Step -1

原因很简单：查询历史时，通常最关心最新状态，所以 LangGraph 会把最新检查点放在最前面。

可以记成一句话：

图执行：从旧到新
查历史：从新到旧

八、Step -1、Step 0、Step 1、Step 2 分别代表什么？

以这个并行图为例：

START -> node_poem
START -> node_joke
node_poem -> node_output
node_joke -> node_output
node_output -> END

Step -1：输入检查点

还没有真正进入图计算。

通常可以看到：

values = {}
next = ("__start__",)

意思是：下一步准备进入图的内部起点。

Step 0：输入写入状态

输入已经进入状态：

values = {
"topic": "猫咪"
}

下一步：

next = ("node_poem", "node_joke")

说明要并行执行 node_poem 和 node_joke。

Step 1：并行节点执行完成

node_poem 和 node_joke 已经执行完。

状态大概是：

values = {
"topic": "猫咪",
"poem": "…",
"joke": "…"
}

下一步：

next = ("node_output",)

Step 2：最终输出完成

node_output 已经执行完。

状态大概是：

values = {
"topic": "猫咪",
"poem": "…",
"joke": "…",
"final_output": "…"
}

下一步：

next = ()

说明图结束。

九、查看最新检查点

如果只想看最新状态，可以用：

snapshot = graph.get_state(config=config)

它返回当前 thread_id 最新的 StateSnapshot。

常用查看方式：

print(snapshot.values)
print(snapshot.next)
print(snapshot.config)

如果 next = ()，说明这个线程已经运行结束。

十、根据 checkpoint_id 查看指定检查点

每个检查点都有自己的 checkpoint_id。

一般结构在：

snapshot.config["configurable"]["checkpoint_id"]

如果想回到某个具体检查点，可以把对应的 checkpoint_id 放进 config 里。

大概形式是：

target_config = {
"configurable": {
"thread_id": "123",
"checkpoint_id": "某个检查点ID"
}
}

snapshot = graph.get_state(config=target_config)

这样就能查看指定检查点的状态。

十一、持久化数据库：PostgresSaver

InMemorySaver 只适合临时测试。

如果希望程序重启后检查点还在，就需要用持久化存储，比如 PostgreSQL。

常见用法是：

from langgraph.checkpoint.postgres import PostgresSaver

大致流程：

1. 启动 PostgreSQL
2. 准备数据库连接地址
3. 创建 PostgresSaver
4. 执行 setup 初始化表
5. compile 时传入 checkpointer

示意代码：

DB_URI = "postgresql://用户名:密码@localhost:5432/数据库名"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
checkpointer.setup()
graph = builder.compile(checkpointer=checkpointer)

PostgreSQL 的作用就是：

把检查点真正保存到数据库里。

这样就算 Python 程序退出，后面仍然可以根据 thread_id 找回历史状态。

十二、失败后恢复运行

有了检查点后，如果图运行到一半失败，就可以从最近一次检查点恢复。

比如：

Step 0 已保存
Step 1 执行某个节点时报错

下次可以从 Step 0 或某个指定检查点继续。

这对于长任务很有用，比如：

• 多步骤文档处理
• 多轮 Agent 工具调用
• 数据分析流水线
• 人工审核流程
• 长时间运行的工作流

否则失败一次就从头跑，成本太高。

十三、Time Travel：回放和分叉

LangGraph 的 Time Travel 可以理解成：

回到历史检查点，重新看或重新跑。

它有两个重要概念：

Replay：回放
Fork：分叉

Replay：回放

Replay 是从某个历史检查点重新执行。

如果之前某些节点已经成功执行，并且检查点里记录了结果，LangGraph 可以避免重复执行已经完成的部分。

这适合调试：

我想看看从某一步开始，后面到底怎么跑。

Fork：分叉

Fork 是从某个历史检查点创建一条新的执行路线。

比如原路线是：

A -> B -> C

你回到 B，然后换一种输入或状态继续：

A -> B -> D

这就像从历史存档开了一条新支线。

适合：

• 对比不同策略
• 调试不同分支
• 人工修改状态后继续执行
• Agent 推理路径实验

十四、短期记忆

短期记忆通常指当前会话内的记忆。

在 LangGraph 里，短期记忆常常通过检查点实现。

比如聊天机器人中：

同一个 thread_id 下保存多轮 messages
下一轮调用时继续带上历史消息

这就是短期记忆。

特点：

• 和当前会话绑定
• 适合多轮对话上下文
• 通常依赖 thread_id
• 可以用检查点保存

十五、长期记忆

长期记忆不是只记住当前会话，而是跨会话保存用户偏好、事实、资料等。

比如：

用户喜欢 Python
用户偏好中文回答
用户常用 DeepSeek
用户正在学习 LangGraph

这些信息不应该只存在一次对话里，而应该保存到长期存储中。

长期记忆通常需要独立存储器，比如数据库。

在 LangGraph 项目中，可以把长期记忆理解为：

跨 thread_id 的持久数据。

短期记忆和长期记忆区别：

十六、访问长期记忆的常见流程

一个常见设计是：

用户输入
↓
检查当前状态里有没有用户偏好
↓
如果没有，从长期记忆数据库查询
↓
把查到的偏好写入状态
↓
模型节点基于偏好回答

伪流程：

def router(state):
if "preference" not in state:
return "load_memory_node"
return "llm_node"

意思是：

如果状态里没有记忆，先去加载长期记忆。
如果已经有了，就直接进入模型节点。

十七、运行时上下文 Runtime Context

除了 State，LangGraph 还支持运行时上下文。

状态 State 是图运行中会被节点读写、合并、保存的业务状态。

运行时上下文 context 更像是本次调用附带的环境信息。

比如：

• 当前用户 ID
• 用户等级
• 请求来源
• 当前语言
• 是否 VIP
• 当前租户 ID

定义上下文类型：

from typing import TypedDict

class Context(TypedDict):
user_id: str
is_vip: bool

创建图时声明：

builder = StateGraph(
state_schema=OverAllState,
context_schema=Context
)

节点里可以通过运行时对象访问：

def node_a(state, runtime):
user_id = runtime.context["user_id"]

调用时传入：

graph.invoke(
input_data,
context={
"user_id": "u_001",
"is_vip": True
}
)

可以这样区分：

State：图内部业务状态，会被节点更新
Context：本次调用的环境信息，一般不作为状态更新

十八、节点完整形态

LangGraph 的节点本质上就是函数。

最简单形态：

def node_a(state):
return {}

带配置：

from langchain_core.runnables import RunnableConfig

def node_a(state, config: RunnableConfig):
thread_id = config["configurable"]["thread_id"]
return {}

带运行时上下文：

def node_a(state, runtime):
user_id = runtime.context["user_id"]
return {}

节点的核心职责始终是：

读取状态或上下文
执行业务逻辑
返回状态更新

十九、PostgreSQL 为什么重要？

PostgreSQL 常用于保存 LangGraph 的检查点和长期记忆。

原因很直接：

• 内存存储不能跨进程
• 真实项目需要程序重启后还能恢复
• 多用户会话需要统一存储
• 长期记忆需要持久化数据库

如果只是学习：

InMemorySaver()

就够了。

如果是项目：

PostgreSQL / Redis / 其他持久化后端

会更合适。

二十、学习时最容易混的点

1. thread_id 不是 checkpoint_id

thread_id 是会话 ID。

checkpoint_id 是某一个具体检查点的 ID。

一个 thread_id 下面可以有多个 checkpoint_id。

2. get_state_history 是倒序

执行顺序：

Step -1 -> Step 0 -> Step 1 -> Step 2

查询历史：

Step 2 -> Step 1 -> Step 0 -> Step -1

3. InMemorySaver 不等于长期保存

InMemorySaver 是内存版。

程序退出后，检查点就没了。

如果想真正持久化，需要数据库。

4. 短期记忆和长期记忆不是一回事

短期记忆通常是一个会话里的上下文。

长期记忆是跨会话保存的用户信息或知识。

5. State 和 Context 不一样

State 会参与图状态更新和检查点保存。

Context 更像本次调用时附带的环境参数。

总结

LangGraph 的持久化和记忆管理，本质上是在解决一个问题：

图运行到某一步时，它当时的状态能不能被保存、查看、恢复和复用？

核心知识可以这样记：

Checkpointer：负责保存检查点
thread_id：区分会话
checkpoint_id：定位具体检查点
StateSnapshot：某一步的状态快照
values：当前状态
next：下一步节点
get_state_history：查看历史检查点，默认从新到旧
get_state：查看最新或指定检查点
Replay：回放历史执行
Fork：从历史检查点分叉新路线
短期记忆：当前会话上下文
长期记忆：跨会话持久信息
Runtime Context：本次调用的环境信息

学完这一部分后，LangGraph 就不只是能跑一个流程，而是开始具备真实 Agent 工程里的核心能力：多轮记忆、状态恢复、失败续跑、历史回放和长期记忆管理。