持久化与检查点

持久化是 LangGraph 区别于其他 Agent 框架的核心特性之一。通过 Checkpointer 机制，LangGraph 可以在任意步骤保存和恢复图的状态，实现对话记忆、故障恢复、时间旅行调试等功能。

为什么需要持久化

没有持久化时，Agent 的状态只存在于内存中，一旦程序退出就丢失了。持久化解决了以下问题：

问题	没有持久化	有持久化
多轮对话	每次从头开始	记住历史
长时间任务	中断就丢失	可以恢复
人工审批	无法暂停	暂停后恢复
调试排查	只看最终结果	回溯任意步骤
生产故障	无法恢复	自动重试

Checkpointer 概念

Checkpointer 是一个接口，负责在图的每个步骤之后保存状态快照。LangGraph 提供了几种内置实现：

python

from langgraph.checkpoint.memory import MemorySaver     # 内存
from langgraph.checkpoint.sqlite import SqliteSaver      # SQLite
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver  # PostgreSQL

核心概念

Thread：一个对话/会话线程，由 thread_id 标识
Checkpoint：某个时间点的状态快照
State History：一个 Thread 的所有 Checkpoint 序列

Thread "user-123"
  ├── Checkpoint 0 (初始状态)
  ├── Checkpoint 1 (节点A执行后)
  ├── Checkpoint 2 (节点B执行后)
  └── Checkpoint 3 (节点C执行后) ← 当前状态

MemorySaver（开发/测试用）

最简单的 Checkpointer，将状态保存在内存中。适合开发和测试，不适合生产环境（程序重启后数据丢失）。

python

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END

# 创建 Checkpointer
checkpointer = MemorySaver()

# 构建图
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_edge(START, "agent")
graph.add_edge("agent", END)

# 编译时传入 checkpointer
app = graph.compile(checkpointer=checkpointer)

# 运行时指定 thread_id
config = {"configurable": {"thread_id": "conversation-1"}}
result = app.invoke({"messages": [("user", "你好")]}, config=config)

# 同一个 thread_id 的后续调用会继续之前的对话
result2 = app.invoke({"messages": [("user", "你还记得我吗")]}, config=config)

SQLite（单机持久化）

SQLite 将状态保存到本地文件，适合单机部署的场景：

python

from langgraph.checkpoint.sqlite import SqliteSaver

# 方式 1：使用上下文管理器
with SqliteSaver.from_conn_string("checkpoints.db") as checkpointer:
    app = graph.compile(checkpointer=checkpointer)
    result = app.invoke(input_data, config=config)

# 方式 2：手动管理连接
import sqlite3
conn = sqlite3.connect("checkpoints.db")
checkpointer = SqliteSaver(conn)
app = graph.compile(checkpointer=checkpointer)

SQLite 的优势

持久化：数据保存在文件中，程序重启后仍在
零配置：不需要安装额外的数据库
轻量：适合个人项目和小规模部署

PostgreSQL（生产环境）

生产环境推荐使用 PostgreSQL，支持并发访问和高可用：

python

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

# 异步用法
async def main():
    async with AsyncPostgresSaver.from_conn_string(
        "postgresql://user:pass@localhost:5432/db"
    ) as checkpointer:
        app = graph.compile(checkpointer=checkpointer)
        result = await app.ainvoke(input_data, config=config)

# 同步用法
from langgraph.checkpoint.postgres import PostgresSaver

with PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost:5432/db"
) as checkpointer:
    app = graph.compile(checkpointer=checkpointer)
    result = app.invoke(input_data, config=config)

获取状态历史

通过 Checkpointer 可以查看一个 Thread 的所有历史状态：

python

# 获取某个 thread 的所有 checkpoint
config = {"configurable": {"thread_id": "conversation-1"}}

# 方式 1：获取最新状态
state = app.get_state(config)
print(state.values)    # 状态值
print(state.next)      # 下一个要执行的节点
print(state.config)    # 当前配置

# 方式 2：获取所有历史状态
history = list(app.get_state_history(config))
for snapshot in history:
    print(f"Step: {snapshot.metadata['step']}")
    print(f"Values: {snapshot.values}")
    print(f"Next: {snapshot.next}")
    print("---")

StateSnapshot 结构

python

@dataclass
class StateSnapshot:
    values: dict           # 状态值
    next: tuple[str, ...]  # 下一个要执行的节点
    config: dict           # 该 checkpoint 的配置
    metadata: dict         # 元数据（step、source 等）
    parent_config: dict    # 父 checkpoint 的配置
    tasks: tuple           # 待执行的任务

时间旅行（Time Travel）

LangGraph 支持从任意历史 checkpoint 恢复执行，实现"时间旅行"：

python

config = {"configurable": {"thread_id": "conversation-1"}}

# 获取所有历史
history = list(app.get_state_history(config))

# 选择第 2 个 checkpoint
checkpoint_to_replay = history[1]

# 从该 checkpoint 重新开始
result = app.invoke(
    None,  # 不需要新的输入
    config=checkpoint_to_replay.config,  # 使用历史 checkpoint 的配置
)

分支对话

时间旅行的一个实际应用是分支对话——从某个历史点开始，走一条不同的路：

python

# 原始对话
config = {"configurable": {"thread_id": "original"}}
app.invoke({"messages": [("user", "推荐一本Python书")]}, config=config)
app.invoke({"messages": [("user", "这本书多少钱")]}, config=config)

# 获取历史
history = list(app.get_state_history(config))

# 从第一个回复之后分支
branch_point = history[-2]  # 回复推荐书之后的 checkpoint

# 新的 thread，从分支点开始
branch_config = {"configurable": {"thread_id": "branch-1"}}
app.invoke(
    {"messages": [("user", "推荐一本Java书")]},  # 不同的后续问题
    config=branch_config,
    # 注意：需要从 branch_point 的状态开始
)

更新状态

除了读取历史，还可以手动修改状态：

python

config = {"configurable": {"thread_id": "conversation-1"}}

# 获取当前状态
state = app.get_state(config)

# 更新状态（例如：修改消息）
app.update_state(config, {
    "messages": [("assistant", "我修改了回复")]
})

# 验证更新
new_state = app.get_state(config)
print(new_state.values)

更新状态的应用场景

人工编辑：人工修改 AI 的回复后再继续
错误修正：发现错误后回退并修正
注入信息：在特定步骤注入外部数据

Checkpointer 的配置

thread_id 与 checkpoint_id

python

# 基本配置：只指定 thread_id
config = {"configurable": {"thread_id": "user-1"}}

# 精确配置：指定 thread_id + checkpoint_id
config = {
    "configurable": {
        "thread_id": "user-1",
        "checkpoint_id": "specific-checkpoint-id",
    }
}

多用户场景

python

def get_config(user_id: str, session_id: str) -> dict:
    return {
        "configurable": {
            "thread_id": f"{user_id}:{session_id}",
        }
    }

# 不同用户的不同会话
config_alice = get_config("alice", "session-1")
config_bob = get_config("bob", "session-1")

app.invoke(input_1, config=config_alice)
app.invoke(input_2, config=config_bob)

与 create_react_agent 集成

python

from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()

agent = create_react_agent(
    model=llm,
    tools=tools,
    checkpointer=checkpointer,  # 传入 checkpointer
)

# 多轮对话
config = {"configurable": {"thread_id": "chat-1"}}
agent.invoke({"messages": [("user", "你好")]}, config=config)
agent.invoke({"messages": [("user", "你还记得我吗")]}, config=config)

生产环境建议

环境	Checkpointer	说明
开发/测试	`MemorySaver`	最简单，重启丢失
单机部署	`SqliteSaver`	文件持久化，零配置
生产部署	`PostgresSaver`	支持并发、高可用
分布式	`PostgresSaver` + 连接池	多实例共享状态

性能注意事项

python

# ❌ 不推荐：每次调用都创建新的 checkpointer
for i in range(100):
    checkpointer = MemorySaver()  # 每次都新建
    app = graph.compile(checkpointer=checkpointer)

# ✅ 推荐：复用同一个 checkpointer
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)
for i in range(100):
    app.invoke(input_data, config=config)

小结

概念	说明
Checkpointer	持久化状态的接口
Thread	一个会话线程，由 thread_id 标识
Checkpoint	某个时间点的状态快照
MemorySaver	内存存储，开发测试用
SqliteSaver	SQLite 持久化，单机部署
PostgresSaver	PostgreSQL，生产环境
get_state()	获取当前/历史状态
update_state()	手动修改状态
时间旅行	从任意 checkpoint 恢复

→ 人机交互（Human-in-the-Loop）：interrupt 机制、审批工作流、断点恢复。

数据结构

数据库

离散数学

数字电路

概率论

数学建模

软件工程

面向对象程序设计

Python 与数据分析

计算机组成

操作系统

计算机网络

编译原理

图形学

自然语言处理

Linux 实验

分布式系统

Internet原理与技术

BlogWeb

Stars Saying

黑马商城

苍穹外卖

黑马点评

黑马商城单体版

JavaWeb AI Tlias

RuoYi DKD

LinkQwen

InterX

LeetCode Agent Lab

00-预备知识

01-PR1108-逐代码段解析

02-PR1135-Cube化逐代码段解析

03-两个Bug深度剖析

04-开发心得与数值性能实测

01-PR1169-模型推理性能优化

02-PR1194-固定Shape及分档推理适配

NumPy

PyTorch

Python 八股文

LeetCUDA

LeetGPU

Triton

持久化与检查点 ​

为什么需要持久化 ​

Checkpointer 概念 ​

核心概念 ​

MemorySaver（开发/测试用） ​

SQLite（单机持久化） ​

SQLite 的优势 ​

PostgreSQL（生产环境） ​

获取状态历史 ​

StateSnapshot 结构 ​

时间旅行（Time Travel） ​

分支对话 ​

更新状态 ​

更新状态的应用场景 ​

Checkpointer 的配置 ​

thread_id 与 checkpoint_id ​

多用户场景 ​

与 create_react_agent 集成 ​

生产环境建议 ​

性能注意事项 ​

小结 ​

下一篇 ​

持久化与检查点

为什么需要持久化

Checkpointer 概念

核心概念

MemorySaver（开发/测试用）

SQLite（单机持久化）

SQLite 的优势

PostgreSQL（生产环境）

获取状态历史

StateSnapshot 结构

时间旅行（Time Travel）

分支对话

更新状态

更新状态的应用场景

Checkpointer 的配置

thread_id 与 checkpoint_id

多用户场景

与 create_react_agent 集成

生产环境建议

性能注意事项

小结

下一篇