第 10 章 Checkpointer 机制与 MemorySaver

10.1 Checkpointer 的定位

Checkpointer（检查点存储器） 是 LangGraph 持久化机制的核心抽象，其本质是在 Pregel 计算模型的超步边界提供"自动快照"能力。Pregel 模型将一次图执行划分为若干超步（Superstep）：每个超步内，被 Channel 更新触发的节点并发执行（Execute），所有节点完成后在屏障处同步，将各节点返回的增量写入按 Channel reducer 合并（Update），形成新版本的 State，然后进入下一超步。Checkpointer 的核心职责是在每个超步的 Update 阶段结束、Channel 合并完成、新版本 State 生成之后，将完整状态快照写入存储后端。这一机制是以下四类核心能力的共同基础：

持久化执行（Durable Execution）：进程崩溃、容器重启、机器宕机后，使用相同 thread_id 再次 invoke 时，Checkpointer 加载最近一次完整快照，图从断点继续执行，不会重跑已完成的节点，也不会丢失中间状态；
断点续跑（Human-in-the-Loop / interrupt/resume）：interrupt() 调用会在当前超步边界挂起执行，快照落盘后释放执行线程；人类审批或补充信息后，调用方通过 Command(resume=value) 传入恢复值，Checkpointer 加载快照、注入 resume 值、继续执行；
多轮对话记忆：同一 thread_id 的多次 invoke 自动接续历史 messages 与 State 字段，应用层无需手动拼消息历史；
时间旅行（Time Travel）：沿 checkpoint 版本链回溯到任意历史快照，读取或修改状态后从该点 fork 出新的执行分支，用于调试、A/B 对比、错误回放。

Checkpointer 的设计哲学是"自动、原子、边界清晰"：快照在超步边界写入，是原子操作（要么完整写入新版本、要么保持旧版本），节点执行中产生的中间状态不落盘，避免"执行到一半的脏状态"被保留。

10.2 BaseCheckpointSaver 接口

所有 Checkpointer 实现都继承自 langgraph.checkpoint.base.BaseCheckpointSaver，定义了四个同步核心方法与对应的异步版本（方法名加 a 前缀）：

方法	签名（简化）	作用
put	`put(config, checkpoint, metadata, new_versions) -> RunnableConfig`	写入一个完整 checkpoint 快照；返回包含新 checkpoint_id 的 config
put_writes	`put_writes(config, writes, task_id, task_path=None) -> None`	写入中间 pending writes（节点已完成但本超步尚未合并的写入），包括 interrupt 事件与 resume 值
get_tuple	`get_tuple(config) -> CheckpointTuple \| None`	根据 config（含 thread_id + 可选 checkpoint_id + checkpoint_ns）读取完整快照；未找到返回 None
list	`list(config, *, filter=None, before=None, limit=None) -> Iterator[CheckpointTuple]`	列出某 thread_id 的历史 checkpoint 链表，支持按 metadata 过滤、按 checkpoint_id 之前游标、分页 limit

异步版本为 aput / aput_writes / aget_tuple / alist，供 ainvoke/astream 等异步执行路径使用。同步与异步接口必须严格匹配：同步 Pregel 运行时调用同步方法，异步 Pregel 调用异步方法，不可混用。在 async 图中传入同步 ConnectionPool 会导致事件循环阻塞，严重时死锁（第 11 章详述）。

get_tuple 返回的 CheckpointTuple 是一个数据类，核心字段：config（本快照的 RunnableConfig）、checkpoint（Checkpoint 对象本身）、metadata（CheckpointMetadata）、parent_config（父快照的 config，根快照为 None）、pending_writes（未合并写入列表，interrupt 恢复时使用）。

10.3 Checkpoint 数据结构

Checkpoint 是一个 TypedDict，源码位于 langgraph/checkpoint/base/__init__.py。主要字段：

字段	类型	说明
v	int	checkpoint 格式版本号，当前版本为 2；跨大版本升级时会做 schema migration
id	str	checkpoint 唯一 ID（UUIDv6/7，单调递增、时间有序，便于按 ID 排序）
ts	str	ISO8601 时间戳（UTC）
channel_values	dict[str, Any]	每个 Channel 的当前值快照，即 State 字段的完整值（messages、业务字段均包含）
channel_versions	dict[str, str]	每个 Channel 的当前版本号（单调递增字符串），用于判断哪些 Channel 发生变化以触发订阅节点
versions_seen	dict[str, dict[str, str]]	每个节点已看到的 Channel 版本号映射（"节点 N 上次执行时 Channel C 的版本是 V"），节点调度时据此判断是否需要重新执行（版本未变则跳过，幂等性保证）
pending_sends	list[Send]	待执行的 Send 对象列表，用于持久化动态 Fan-out（Map-Reduce）中途状态；恢复时重新派发未完成的 Send 任务

配套的 CheckpointMetadata 字段：

字段	说明
source	快照来源枚举：`"input"`（初始输入，step=-1）、`"loop"`（超步循环产生）、`"update"`（update_state 手动产生）、`"fork"`（从历史快照 fork 产生）
step	超步编号，input 快照为 -1，首个超步结束为 0，依此类推
parents	dict[str, str]，子图场景下各父命名空间的 checkpoint_id 映射
run_id / thread_id	运行 ID 与线程 ID
created_at / written_at	创建时间戳
user_id / tenant_id 等	业务元数据（通过 config.metadata 注入）

版本链机制：每个 checkpoint 通过 parent_config.configurable.checkpoint_id 字段指向上一个 checkpoint，形成一条单链表。链表头是 source="input" 的初始快照，链表尾是最新快照。从最新 checkpoint 沿 parent_config 回溯可遍历该 thread 的完整执行历史。get_state_history 方法就是按这条链表迭代输出。时间旅行与 fork 机制：调用 invoke/stream 时在 config 中指定 checkpoint_id=X，运行时会加载快照 X 作为起点，执行中产生的新 checkpoint 以 X 为父节点，形成一条独立分支，不会影响 X 之后原链表上的快照。

10.4 MemorySaver（InMemorySaver）

MemorySaver（在部分版本中保留 InMemorySaver 别名以兼容旧代码）是 Checkpointer 的内存实现，位于 langgraph.checkpoint.memory，底层使用三个嵌套字典：checkpoints[thread_id][checkpoint_ns][checkpoint_id] = CheckpointTuple、writes[thread_id][checkpoint_ns][checkpoint_id] 存储 pending_writes。MemorySaver 不需要外部依赖，构造即启动，进程退出数据全部丢失，适用于以下场景：

本地开发期快速调试；
单元测试（测试用例之间可创建新 MemorySaver 实例实现状态隔离）；
Notebook/REPL 演示；
单进程无状态 CLI 工具（进程退出不需要保留历史）。

python

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
graph = builder.compile(checkpointer=memory)

MemorySaver 支持同步与异步双接口（异步接口通过线程池包装同步方法实现），开发期可直接配合 ainvoke 使用。

10.5 thread_id、checkpoint_id、checkpoint_ns

thread_id：执行线程标识，通过 config["configurable"]["thread_id"] 在每次 invoke/stream 时传入。每个 thread_id 对应一条独立的状态链表，复用相同 thread_id 会从最新 checkpoint 继续；传入新 thread_id 则启动全新执行（source="input" 新快照）。在对话场景下，thread_id 通常对应一次会话（session/conversation ID）。thread_id 是 Checkpointer 隔离数据的第一维度。
checkpoint_id：某一次具体快照的 ID，由 put() 自动生成（UUIDv6/7）。用户在调用 get_state/update_state/invoke 时可在 config 中显式指定 checkpoint_id，以读取特定历史快照或以该快照为起点 fork 执行。不指定时默认读写该 thread 的最新快照。
checkpoint_ns（checkpoint namespace）：子图命名空间隔离字段，由运行时自动维护。子图在父图中作为节点被调用时，其 checkpoint 的 checkpoint_ns 由父图命名空间、父图节点名、子图名按 : 拼接组成（如 "parent:subgraph_node:child_graph"）。命名空间隔离让父子图的 checkpoint 不会相互覆盖。用户代码一般不需要直接操作 checkpoint_ns。

10.6 get_state、get_state_history、update_state 与时间旅行

CompiledStateGraph 提供四类状态操作方法（同步+异步）：

get_state(config)：读取指定 config（必须含 thread_id，可选 checkpoint_id）对应的状态快照，返回 StateSnapshot 对象（含 values、next、config、metadata、created_at、parent_config、tasks、interrupts）。若指定 checkpoint_id 则读取该快照，否则读最新快照。values 是当前 State 值字典；next 是下一轮将被调度的节点名元组（interrupt 挂起时非空）；tasks 包含待执行任务详情；interrupts 包含当前未处理的 interrupt 值。
get_state_history(config, *, limit=None, before=None, filter=None)：返回该 thread_id 的历史 StateSnapshot 迭代器，按 checkpoint 时间倒序（最新在前）。limit 指定返回条数；before 指定游标 checkpoint_id 用于分页；filter 按 metadata 字段过滤。
update_state(config, values, as_node=None)：不执行任何节点，直接将 values 作为某节点的写入合并到 State，生成一个 source="update" 的新 checkpoint。as_node 参数指定该写入"伪装"为哪个节点的输出（影响 versions_seen 与后续节点触发，默认为 None/__start__）。该方法用于人工纠错、状态重置、修复异常数据。
fork 执行（时间旅行重放）：invoke/stream 时在 config 中指定 checkpoint_id=X，运行时从快照 X 加载 State 作为起点开始执行，产生的新 checkpoint 以 X 为 parent 形成分支。原链表从 X 之后的快照不受影响，实现"时光倒流、分支并行"。

10.7 完整示例：MemorySaver 多轮对话 + interrupt + 回溯 + 纠错

python

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_openai import ChatOpenAI

class State(TypedDict):
    messages: Annotated[list, add_messages]

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def chatbot(state: State) -> dict:
    """调用 LLM 生成回复。"""
    return {"messages": [llm.invoke(state["messages"])]}

def review_node(state: State) -> Command:
    """人工审批节点：AI 回复发出前要求人类审核。"""
    last_ai = state["messages"][-1].content
    # interrupt 将当前超步挂起，值会返回给调用方；resume 值作为 interrupt() 返回值
    human_answer = interrupt(
        {"type": "review", "proposed_answer": last_ai}
    )
    # resume 后 human_answer 为人类输入（可以是修改后的答案或"通过"指令）
    if human_answer == "通过":
        return Command(goto=END)  # 原回复保留，直接结束
    return Command(
        update={"messages": [AIMessage(content=human_answer)]},
        goto=END,
    )

builder = StateGraph(State)
builder.add_node("chatbot", chatbot)
builder.add_node("review", review_node)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", "review")
# review 通过 Command(goto=END) 结束

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# === 第一轮：发送消息，会在 review 节点中断 ===
config = {"configurable": {"thread_id": "conv-1"}}
graph.invoke(
    {"messages": [SystemMessage(content="你是一个友好的助手。"),
                  HumanMessage(content="你好，简单介绍下自己")]},
    config,
)
state = graph.get_state(config)
print("挂起状态 next 节点：", state.next)             # ('review',)
print("待审批回复：", state.tasks[0].interrupts[0].value["proposed_answer"][:50])

# === 恢复执行：人类修改答案后 resume ===
graph.invoke(Command(resume="我是基于 LangGraph 构建的 AI 助手。"), config)

# === 第二轮：复用 thread_id，自动接续历史 ===
graph.invoke(
    {"messages": [HumanMessage(content="我叫小明，请记住我的名字")]},
    config,
)
# 在 review 中断点恢复（此处演示直接通过）
state = graph.get_state(config)
graph.invoke(Command(resume="通过"), config)

# === 查看完整消息历史 ===
final = graph.get_state(config)
print("=== 消息历史 ===")
for m in final.values["messages"]:
    prefix = {"system": "SYS", "human": "USR", "ai": "AI ", "tool": "TL "}.get(m.type, "?  ")
    print(f"{prefix} | {m.content[:70]}")

# === 时间旅行：列出所有历史 checkpoint ===
print("=== 历史快照（时间倒序）===")
for h in graph.get_state_history(config):
    print(f" step={h.metadata['step']:2d}  source={h.metadata['source']:8s}  "
          f"next={h.next}  id={h.config['configurable']['checkpoint_id'][:8]}...")

# === 人工纠错：直接修改最新状态（例如删除一句错误回复）===
graph.update_state(
    config,
    {"messages": [AIMessage(content="（已人工修正：该回复已被编辑）")]},
    as_node="chatbot",
)
corrected = graph.get_state(config)
print("修正后最后消息：", corrected.values["messages"][-1].content)

10.8 MemorySaver 的局限与生产替代

MemorySaver 有明确的适用边界，超出边界必须切换到持久化后端：

进程退出即丢失：内存字典无法跨重启保留状态，服务重启后所有对话历史与断点消失；
单进程隔离：多 worker 部署（Gunicorn/Uvicorn 多进程、K8s 多副本、Fargate 多实例）下，每个进程持有独立的 MemorySaver 实例，请求被负载均衡到不同 worker 时会出现"thread_id 找不到"或"读到旧状态"，造成数据丢失与对话断裂；
内存无限增长：所有 channel_values 常驻进程内存，无法自动卸载或归档，长时间运行后进程内存会持续膨胀（特别是大 messages 列表场景），最终导致 OOM；
无事务保证：写入过程中若进程崩溃可能产生不一致快照（pending_writes 部分写入）；
无跨进程共享：异步任务（如 Celery worker）无法与 API 进程共享状态。

因此 MemorySaver 只适用于开发期、单测、Notebook 演示，任何生产部署必须切换到持久化共享后端（PostgresSaver、SqliteSaver、RedisSaver 等，详见第 11 章）。

本章小结

Checkpointer 在每个超步末的 Update 阶段自动保存 Channel 快照，是持久化执行、interrupt/resume 断点续跑、多轮对话记忆、时间旅行四类能力的共同基础。BaseCheckpointSaver 定义 put/put_writes/get_tuple/list 四个同步核心方法与 aput/aput_writes/aget_tuple/alist 四个异步版本。Checkpoint 数据结构包含 channel_values、channel_versions、versions_seen、pending_sends 等字段，通过 parent_config 形成版本链（单链表）。CheckpointMetadata 含 source、step、parents、run_id 等元数据。MemorySaver 是基于内存字典的实现，进程退出即丢失，适用于开发测试。thread_id 标识执行线程（checkpoint 隔离的第一维度），checkpoint_id 标识具体快照版本，checkpoint_ns 用于子图命名空间隔离。get_state/get_state_history/update_state 提供状态读取、历史回溯、人工纠错能力；通过在 config 中指定 checkpoint_id 可实现 fork 执行（时间旅行）。MemorySaver 不适合生产，必须替换为 PostgresSaver 等持久化后端。

常见错误

未配置 checkpointer 就使用 interrupt() 报错。现象：调用 interrupt() 节点时抛 NotImplementedError: Checkpointer is required for interrupt。原因：compile 时 checkpointer 参数保持默认 None，运行时没有地方保存中断快照。修复：compile 时传入 MemorySaver（开发/单测）或 PostgresSaver/SqliteSaver（生产）。
多个请求复用同一 thread_id 导致状态串线。现象：用户 A 的对话中出现用户 B 的历史消息，或两个会话互相看到对方中间状态。原因：把 thread_id 写成固定常量（如 "default"）或全局变量，所有用户共享同一执行线程。修复：每次新会话生成唯一 thread_id（推荐业务主键组合如 f"{tenant_id}:{user_id}:{session_id}"），从请求上下文读取并传入 config。
直接操作 checkpoint 数据结构内部字段导致版本错乱。现象：反序列化失败、节点重复执行或不执行、状态合并异常。原因：业务代码直接读取并修改 checkpoint.channel_values 或 channel_versions、versions_seen 等内部字段，绕过 reducer 与版本号机制。修复：所有状态修改都通过节点返回值或 update_state 方法进行，不直接触碰 checkpoint 内部结构；读取状态只通过 get_state().values。
MemorySaver 在生产多进程环境中状态隔离失效。现象：偶发"会话找不到"、"消息丢失"、"回复对应到错误问题"。原因：多 worker 各自持有独立 MemorySaver 实例，状态不跨进程共享。修复：生产环境必须切换到 PostgresSaver/SqliteSaver 等基于共享存储的后端；若使用多进程服务器（Gunicorn）但坚持用内存后端，必须将 worker 数设为 1（不推荐）。
多轮调用未传 config 导致对话"失忆"。现象：第二次 invoke 时模型看不到第一轮对话历史。原因：第二次调用没有传包含相同 thread_id 的 config，Checkpointer 无法关联到已有状态链，被当作全新对话处理。修复：每次 invoke/stream 必须传 config，session 期间 thread_id 保持一致；在 Web 服务中将 thread_id 存入用户 session。

生产建议

开发与单元测试阶段统一使用 MemorySaver，测试覆盖以下核心场景：同一 thread_id 多轮消息接续、interrupt/resume 正常往返、update_state 能正确纠错、get_state_history 能遍历完整历史、fork 执行能从指定 checkpoint 分支；
所有 checkpointer 调用都通过 config 显式传 thread_id，禁止依赖任何默认值；CI 中加入 lint 规则禁止生产配置使用 MemorySaver；
thread_id 命名包含业务主键（tenant_id/user_id/session_id），避免纯 UUID 导致问题排查时无法从 thread_id 直接定位业务对象；
对关键业务流程设计"状态修复脚本"：基于 get_state_history 查看快照链，用 update_state 在生产环境紧急纠错（如修正模型错误填入的字段、撤回误操作）；
MemorySaver 不设为任何服务的默认 checkpointer；应用启动时根据环境变量（CHECKPOINTER_BACKEND=memory|postgres|sqlite）选择后端；
在单测中为每个测试函数创建全新 MemorySaver 实例（或使用唯一 thread_id），避免测试用例之间状态污染。

C 语言

语言点

深度

新标准

面试整理

第 10 章 Checkpointer 机制与 MemorySaver ​

10.1 Checkpointer 的定位 ​

10.2 BaseCheckpointSaver 接口 ​

10.3 Checkpoint 数据结构 ​

10.4 MemorySaver（InMemorySaver） ​

10.5 thread_id、checkpoint_id、checkpoint_ns ​

10.6 get_state、get_state_history、update_state 与时间旅行 ​

10.7 完整示例：MemorySaver 多轮对话 + interrupt + 回溯 + 纠错 ​

10.8 MemorySaver 的局限与生产替代 ​

本章小结 ​

常见错误 ​

生产建议 ​