langgraph 图的 checkpoint 机制

LangChain Code LangChain 文档

LangGraph 的 Checkpoint（检查点）机制是其内置持久化层的核心。简单来说，它像一个为你的图（Graph）状态提供的“Git”版本控制系统，会在图执行的每个“超级步”（Super-step）后，自动保存一份完整的状态快照。

1. 核心概念

要理解Checkpoint机制，首先需要掌握以下几个核心概念：

  • Checkpoint (检查点)：它是图状态在某个特定时间点的完整快照。快照包含所有通道值、通道版本等信息。每个检查点都有一个全局唯一的、单调递增的ID。
  • Super-step (超级步)：这是图执行的一个“节拍”。在一个超级步中，所有计划在该步执行的节点会（可能并行地）执行。LangGraph会在每个超级步的边界处创建一个检查点。
  • Thread (线程)：这是将一系列检查点组织起来的逻辑单元，通过唯一的 thread_id 来标识。它代表了一次独立的对话或工作流会话。通过 thread_id，你可以隔离和管理不同会话的状态。

2. Checkpointer 的作用与带来的能力

Checkpointer（检查点器）是实现持久化的组件。当你用 checkpointer 参数编译图时，持久化层就被激活了。它能为你带来以下关键能力：

  • 记忆 (Memory)：在多轮对话中，Agent 能记住之前的交互内容。
  • 容错 (Fault-tolerance)：当执行因错误中断时，可以从最后一个成功的检查点恢复。
  • 人机协同 (Human-in-the-loop)：允许人类在检查点处审查、中断或修改状态，然后继续执行。
  • 时间旅行 (Time Travel)：你可以回溯到任何一个历史检查点，查看当时的状态，甚至从这个点“分叉”（Fork）出一条新的执行路径来探索其他可能性。

3. 各种情况与场景详解

Checkpoint 机制在不同场景下发挥着不同作用。

3.1. 正常执行与自动保存

这是最基础的情况。当你为图配置了 Checkpointer 并正常调用时，LangGraph 会自动在每一个 Super-step 结束后保存状态。

3.2. 故障恢复与“待处理写入”(Pending Writes)

当一个超级步中的部分节点成功、部分节点失败时。为了不重复执行已经成功的节点，LangGraph 会保存这些成功节点的写入操作作为“待处理写入”(Pending Writes)。当从该检查点恢复时，这些待处理的写入会被应用，而失败的节点会被重新执行。

示例：在一个有“节点A”和“节点B”的超级步中，“节点A”成功写入了数据，但“节点B”崩溃了。Checkpoint 会记录下“节点A”的写入作为 pendingWrites。恢复执行时，系统会应用“节点A”的写入，并重新运行“节点B”。

3.3. 时间旅行 (Time Travel)

你可以回到过去的任何一个检查点，查看当时的状态，或从此处分叉。

  • 查看历史状态：通过 graph.getState(config) 获取最新状态，或通过 checkpointer.list(config) 列出某个线程的所有检查点。
  • 从检查点恢复/分叉：获取到特定的 checkpoint_id 后，可以在下一次调用时指定从该检查点恢复。

示例：假设你的图执行了三个步骤，产生了 cp1, cp2, cp3 三个检查点。你发现 cp3 的结果不理想，想回到 cp2 时的状态重新尝试。你可以通过API获取 cp2 的ID，然后配置图从该检查点恢复执行。

4. Checkpoint Saver 实现

LangGraph 提供了多种 Checkpoint Saver，你可以根据场景选择：

Saver 实现包名异步支持推荐场景
InMemorySaverlanggraph-checkpoint是仅用于开发和测试
SqliteSaverlanggraph-checkpoint-sqlite否轻量级演示和小型项目
AsyncSqliteSaverlanggraph-checkpoint-sqlite是异步SQLite，不推荐用于生产
PostgresSaverlanggraph-checkpoint-postgres否生产环境，需完整历史
AsyncPostgresSaverlanggraph-checkpoint-postgres是异步生产环境

此外，社区也提供了支持其他存储后端（如 DynamoDB, S3, MongoDB, Couchbase等）的实现。

5. 代码示例

5.1. 基本使用 (InMemorySaver)

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph

# 1. 定义你的图
builder = StateGraph(YourState)
# ... 添加节点和边 ...

# 2. 创建 Checkpointer
checkpointer = MemorySaver()

# 3. 编译图时传入 Checkpointer
graph = builder.compile(checkpointer=checkpointer)

# 4. 执行图，必须指定 thread_id
config = {"configurable": {"thread_id": "user_session_123"}}
initial_state = {"messages": [("user", "你好！")]}
# 图执行后，状态会自动保存
result = graph.invoke(initial_state, config)

5.2. 生产环境 (PostgresSaver)

from langgraph.checkpoint.postgres import PostgresSaver

# 使用 PostgreSQL 作为持久化存储
with PostgresSaver.from_conn_string("postgresql://user:pass@localhost/db") as checkpointer:
    # 首次使用时需要执行 setup 创建表
    checkpointer.setup()
    graph = builder.compile(checkpointer=checkpointer)
    # ... 后续执行 ...

5.3. 获取和恢复状态

# 获取指定线程的最新状态
config = {"configurable": {"thread_id": "user_session_123"}}
snapshot = await graph.aget_state(config)
print(f"当前 checkpoint ID: {snapshot.checkpoint_id}")

# 列出该线程的所有检查点
checkpoints = []
async for checkpoint in checkpointer.alist(config):
    checkpoints.append(checkpoint)

# 从特定检查点恢复执行 (假设 checkpoint_id 为 'xyz')
fork_config = {
    "configurable": {
        "thread_id": "user_session_123",
        "checkpoint_id": "xyz"  # 指定要从哪个检查点恢复
    }
}
# 后续的 invoke 会从该检查点继续

6. 最佳实践与总结

  • 选择合适的 Checkpointer：开发测试用 MemorySaver，生产环境务必使用 PostgresSaver 等持久化方案。
  • 善用 thread_id：为每个独立的会话或工作流使用唯一的 thread_id 来隔离状态。
  • 数据序列化：Checkpoint 数据默认使用 JsonPlusSerializer 进行序列化。对于敏感数据，可以包装 EncryptedSerializer 进行加密。
  • 利用时间旅行调试：当 Agent 行为异常时，可以利用“时间旅行”功能回溯到历史检查点进行排查。
标签：
LangChain Code LangChain 文档