LangGraph 设计与实现
第8章 Checkpoint 持久化
8.1 引言
在前面的章节中，我们深入剖析了 Pregel 执行循环如何驱动图的运行、Channel 如何承载状态的流转、以及任务调度器如何协调节点的并发执行。然而，所有这些精密的运行时机制都面临一个根本性问题：当进程终止、网络中断或用户需要暂停时，图的执行状态将随之消失。
Checkpoint（检查点）机制正是 LangGraph 对这一问题的系统性回答。它不仅仅是简单的"保存与恢复"，而是一套精心设计的持久化架构，涵盖了从数据结构定义、序列化协议、版本追踪到多后端存储适配的完整链路。Checkpoint 是 LangGraph 实现中断恢复、人机协作、时间旅行调试等高级特性的基石。
本章将从源码层面深入分析 LangGraph 1.1.6 的 Checkpoint 持久化体系。我们将逐一剖析 Checkpoint 数据结构的每个字段、BaseCheckpointSaver 基类的接口设计、JsonPlusSerializer 的序列化策略、以及 SQLite/Postgres 两种持久化实现的存储方案。理解 Checkpoint 体系不仅有助于正确使用 LangGraph 的持久化功能，更能帮助开发者在面对自定义存储后端、性能优化或调试状态恢复问题时做出正确的架构决策。
要真正理解 Checkpoint 为何如此设计，我们需要先思考状态持久化面临的核心挑战。首先，图的状态不是简单的键值对——它包含 Channel 的值、版本信息和节点的执行进度，这些信息必须作为一个原子单元被保存和恢复。其次，持久化的粒度需要精心选择——太粗会丢失中间状态，太细会拖慢执行速度。最后，序列化必须能够处理 Python 生态中丰富多样的类型系统，从简单的字典到复杂的 Pydantic 模型、从日期时间到 NumPy 数组，都需要可靠地序列化和反序列化。LangGraph 的 Checkpoint 体系对这些挑战给出了优雅而务实的回答。
本章要点
- Checkpoint 数据结构：理解
CheckpointTypedDict 的七个字段及其在状态恢复中的作用 - CheckpointSaver 基类：掌握
get_tuple/put/list/put_writes四大核心方法的契约 - CheckpointMetadata 与寻址：理解
thread_id/checkpoint_id/checkpoint_ns三维寻址体系 - 序列化体系：深入
JsonPlusSerializer基于 ormsgpack 的类型安全序列化协议 - 持久化模式：区分
sync/async/exit三种持久化策略的适用场景 - 存储实现：对比 InMemory/SQLite/Postgres 三种实现的架构差异
8.2 Checkpoint 数据结构
8.2.1 Checkpoint TypedDict
Checkpoint 是 LangGraph 持久化体系的核心数据结构。它定义在 langgraph.checkpoint.base 模块中，使用 Python 的 TypedDict 来提供类型安全的字典访问：
# langgraph/checkpoint/base/__init__.py
class Checkpoint(TypedDict):
"""State snapshot at a given point in time."""
v: int
"""The version of the checkpoint format. Currently 1."""
id: str
"""The ID of the checkpoint. Both unique and monotonically increasing."""
ts: str
"""The timestamp of the checkpoint in ISO 8601 format."""
channel_values: dict[str, Any]
"""The values of the channels at the time of the checkpoint."""
channel_versions: ChannelVersions
"""The versions of the channels at the time of the checkpoint."""
versions_seen: dict[str, ChannelVersions]
"""Map from node ID to map from channel name to version seen."""
updated_channels: list[str] | None
"""The channels that were updated in this checkpoint."""每个字段都承担着精确的职责：
v（格式版本）：Pregel 层实际写出的版本号是 4（libs/langgraph/langgraph/pregel/_checkpoint.py 的 LATEST_VERSION = 4）。libs/checkpoint/langgraph/checkpoint/base/__init__.py:579 里也有一个 LATEST_VERSION = 2，但它紧跟在 :577 那行分隔注释之后——那行写的是 "below are deprecated utilities used by past versions of LangGraph"——只服务旧版本，别把它当成当前格式版本。
这个版本号用于反序列化时的格式迁移——Pregel._migrate_checkpoint（pregel/main.py:1013-1020）就靠 checkpoint["v"] < 4 把老格式的 pending_sends 搬进 __pregel_tasks Channel。
id（检查点标识）：采用 UUID v6 生成，这是一个关键的设计选择。UUID v6 将时间戳编码到高位，使得 ID 既全局唯一又（在同一进程内）单调递增。这意味着可以直接通过字符串比较来确定检查点的先后顺序，无需额外的排序字段。
ts（时间戳）：ISO 8601 格式的 UTC 时间戳，主要用于调试和审计，不参与排序逻辑。
channel_values（通道值）：存储所有 Channel 在该时刻的快照值。这是状态恢复的关键数据。需要注意的是，并非所有 Channel 都会出现在这个字典中——只有在 channel_versions 中有记录（即至少被写入过一次）的 Channel 才会被保存。这种按需保存的策略避免了为未使用的 Channel 浪费存储空间。在 Postgres 实现中，None/str/int/float/bool 这类原始值直接内联在 checkpoints 表的 checkpoint JSONB 里，其余值才被挑进 blob_values（libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py:301-308），再通过 channel_versions 映射到独立的 checkpoint_blobs 表，实现跨检查点的去重。
channel_versions（通道版本）：记录每个 Channel 的当前版本号，用于增量更新判断。版本号的类型由 BaseCheckpointSaver 的泛型参数 V 决定，可以是 int、float 或 str。每当一个 Channel 被写入时，其版本号通过 get_next_version 方法递增。这个版本追踪机制是 LangGraph 实现增量持久化的基础——通过比较前后两个检查点的 channel_versions，可以精确知道哪些 Channel 发生了变化，从而只持久化变化的部分。
versions_seen（已见版本）：二维映射，记录每个节点上次执行时看到的各 Channel 版本。这是 Pregel 调度器判断"哪些节点需要重新执行"的核心依据。当 channel_versions[channel] > versions_seen[node][channel] 时，说明该 Channel 在节点上次执行之后被更新过，节点需要被重新调度。这个机制确保了图的执行遵循数据驱动的原则——只有当输入数据确实发生变化时，节点才会被触发。
updated_channels：记录在当前检查点中被更新的 Channel 列表。这个字段随 langgraph 0.6.4 加入（PR #5828 "perf: Save updated_channels to checkpoint"），用于优化 prepare_next_tasks 的性能——从检查点恢复时，可以借 trigger_to_nodes 由被更新的 Channel 直接找出候选节点，不必逐个遍历所有节点（prepare_next_tasks 里 if updated_channels and trigger_to_nodes 那个分支）。当值为 None 或为空时，回退到遍历全部节点。
classDiagram
class Checkpoint {
+int v
+str id
+str ts
+dict channel_values
+ChannelVersions channel_versions
+dict versions_seen
+list updated_channels
}
class CheckpointMetadata {
+Literal source
+int step
+dict parents
+str run_id
}
class CheckpointTuple {
+RunnableConfig config
+Checkpoint checkpoint
+CheckpointMetadata metadata
+RunnableConfig parent_config
+list pending_writes
}
CheckpointTuple --> Checkpoint
CheckpointTuple --> CheckpointMetadata
8.2.2 UUID v6：单调递增的标识符
LangGraph 没有使用标准库的 UUID v4，而是内联了一份 UUID v6 实现（base/id.py 的模块 docstring 写明改编自 oittaa/uuid6-python，内联进来是为了避免多一个安装依赖）。这个选择背后有工程考量：
# langgraph/checkpoint/base/id.py
def uuid6(node=None, clock_seq=None) -> UUID:
global _last_v6_timestamp
nanoseconds = time.time_ns()
timestamp = nanoseconds // 100 + 0x01B21DD213814000
if _last_v6_timestamp is not None and timestamp <= _last_v6_timestamp:
timestamp = _last_v6_timestamp + 1
_last_v6_timestamp = timestamp
if clock_seq is None:
clock_seq = random.getrandbits(14)
if node is None:
node = random.getrandbits(48)
time_high_and_time_mid = (timestamp >> 12) & 0xFFFFFFFFFFFF
time_low_and_version = timestamp & 0x0FFF
uuid_int = time_high_and_time_mid << 80
uuid_int |= time_low_and_version << 64
uuid_int |= (clock_seq & 0x3FFF) << 48
uuid_int |= node & 0xFFFFFFFFFFFF
return UUID(int=uuid_int, version=6)UUID v6 的核心特性是时间有序性：时间戳占据高位字节，使得按字符串排序就等同于按时间排序。这在数据库中带来两个关键优势：
- 索引友好：B-tree 索引在处理单调递增的主键时效率最高，避免了 UUID v4 造成的随机写入
- 天然排序：
max(checkpoints.keys())就能找到最新的检查点，无需额外查询
注意 clock_seq 参数的特殊用法：在 create_checkpoint 中，clock_seq=step 将步骤号编码进 UUID，而 empty_checkpoint 使用 clock_seq=-2 作为哨兵值。此外，模块级全局变量 _last_v6_timestamp 保证了即使在同一个 100 纳秒时间片内多次调用 uuid6()（或系统时钟回拨），同一进程内生成的 ID 也是严格递增的；跨进程则只能依赖各自的时钟，不做这层保证。这对于高频创建检查点的场景（如并行子图执行）至关重要。
从数据库的角度看，UUID v6 的时间有序性让新记录大致按顺序落入索引，不像 UUID v4 那样随机插入、引发频繁的 B-tree 页面分裂（id.py 里 uuid6 的 docstring 也写明 v6 是"reordered for improved DB locality"）。在默认的持久化模式下，检查点是高频创建的——每个执行步骤都会写一个检查点。与递增整数相比，UUID 不需要中央协调就能保证全局唯一，多进程部署时也无需共享计数器。
8.2.3 CheckpointTuple 与 CheckpointMetadata
Checkpoint 本身只存储图的运行时状态。围绕它，还有两个重要的伴生类型：
# libs/checkpoint/langgraph/checkpoint/base/__init__.py:35（CheckpointMetadata）
# 与 :112（CheckpointTuple）两段；中间 :65 是 Checkpoint 本身的定义
class CheckpointMetadata(TypedDict, total=False):
source: Literal["input", "loop", "update", "fork"]
step: int
parents: dict[str, str]
run_id: str
class CheckpointTuple(NamedTuple):
config: RunnableConfig
checkpoint: Checkpoint
metadata: CheckpointMetadata
parent_config: RunnableConfig | None = None
pending_writes: list[PendingWrite] | None = NoneCheckpointMetadata 的 source 字段记录了检查点的来源，这对于理解图的执行历史至关重要：
"input"：由图的输入触发创建（step = -1）"loop"：在 Pregel 循环的每次迭代中创建"update"：由手动状态更新（update_state）创建"fork"：从另一个检查点复制而来（以as_node="__copy__"更新状态时产生，对应pregel/main.py:1663-1687的__copy__分支；不带这个参数的普通状态更新产生的是"update"）
CheckpointTuple 则是检查点系统的"完整视图"，将检查点本体、元数据、配置和待处理写入打包在一起。使用 NamedTuple 而非 TypedDict 意味着元组本身的字段不能重新赋值；但里面的 checkpoint、metadata 仍是普通 dict，内容照样可以被改动，不可变性只到元组这一层。
pending_writes 尤为重要——它存储了尚未被合并到下一个检查点的中间写入，这是实现中断恢复的关键。每个待处理写入是一个 (task_id, channel, value) 三元组，记录了哪个任务向哪个 Channel 写入了什么值。当图从中断恢复时，这些写入会被重放，避免重新执行已完成的节点。parent_config 字段指向上一个检查点的配置，每个检查点只指向自己的父节点（分叉后多个检查点可以共享同一个父节点），允许沿着检查点历史回溯——这是时间旅行功能的数据基础。
graph TB
subgraph "Checkpoint 生命周期"
I[输入到达] -->|"source=input, step=-1"| CP0[Checkpoint 0]
CP0 -->|"source=loop, step=0"| CP1[Checkpoint 1]
CP1 -->|"source=loop, step=1"| CP2[Checkpoint 2]
CP2 -->|中断| INT[Interrupt]
INT -->|"恢复, source=loop, step=2"| CP3[Checkpoint 3]
end
subgraph "手动干预（从 CP2 分叉）"
CP2 -->|"update_state(), source=update, step=2"| FORK["Checkpoint 4（分叉）"]
FORK -->|"继续执行, source=loop, step=3"| CP5[Checkpoint 5]
end
style INT fill:#f9f,stroke:#333,stroke-width:2px
style FORK fill:#ff9,stroke:#333,stroke-width:2px
8.3 三维寻址体系
LangGraph 使用三个维度来唯一定位一个检查点：
8.3.1 thread_id：会话标识
thread_id 是最外层的分区键。它将不同的对话或工作流隔离在独立的命名空间中：
config = {"configurable": {"thread_id": "user-alice-conv-1"}}
graph.invoke(inputs, config)同一个 thread_id 下的检查点通常构成一条线性的执行历史链；从历史检查点分叉（见 §8.2.3 的图）之后，它们经 parent_config 串成一棵树。重用 thread_id 可以实现对话记忆的累积，而每次使用新的 thread_id 则开启独立的执行。在生产系统中，thread_id 的选择策略需要根据业务场景仔细考虑：对于客服对话，可以使用用户会话 ID；对于数据处理流水线，可以使用批次 ID 加任务 ID 的组合；对于定时任务，可以使用调度时间戳。需要注意的是，thread_id 是字符串类型，如果传入了非字符串值，PregelLoop 会自动将其转换为字符串。
8.3.2 checkpoint_ns：命名空间
checkpoint_ns（checkpoint namespace）用于支持子图。根图的命名空间为空字符串 ""，子图的命名空间由父图的命名空间和分隔符组成：
NS_SEP = "|" # 层级分隔符
NS_END = ":" # 命名空间与任务ID的分隔符
# 例如：根图节点 "agent" 的子图，其命名空间为：
# "agent:task-id-xxx|inner_agent:task-id-yyy"这种层级命名空间设计使得子图可以拥有独立的检查点历史，同时保持与父图的关联。
8.3.3 checkpoint_id：检查点标识
checkpoint_id 是 UUID v6 格式的标识符，在给定的 (thread_id, checkpoint_ns) 下唯一标识一个检查点。由于 UUID v6 的单调递增特性，省略 checkpoint_id 时默认返回最新的检查点。
graph LR
subgraph "三维寻址空间"
direction TB
TID["thread_id<br/>'user-alice'"]
NS["checkpoint_ns<br/>'agent:task1|tool:task2'"]
CID["checkpoint_id<br/>'1ef8a3b4-...'"]
TID --> NS --> CID
end
subgraph "检索逻辑"
Q1["get_tuple(thread_id)"] --> R1["返回最新 checkpoint"]
Q2["get_tuple(thread_id + checkpoint_id)"] --> R2["返回指定 checkpoint"]
Q3["list(thread_id, before=...)"] --> R3["返回历史 checkpoints"]
end
8.4 BaseCheckpointSaver 基类
8.4.1 核心接口设计
BaseCheckpointSaver 是所有检查点存储后端必须实现的抽象基类。它定义了四组核心操作（同步+异步）：
# libs/checkpoint/langgraph/checkpoint/base/__init__.py:122
class BaseCheckpointSaver(Generic[V]):
serde: SerializerProtocol = JsonPlusSerializer()
def __init__(self, *, serde: SerializerProtocol | None = None) -> None:
self.serde = maybe_add_typed_methods(serde or self.serde)
# 1. 读取单个检查点
def get_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
raise NotImplementedError
# 2. 列出检查点
def list(self, config, *, filter=None, before=None, limit=None
) -> Iterator[CheckpointTuple]:
raise NotImplementedError
# 3. 存储检查点
def put(self, config, checkpoint, metadata, new_versions
) -> RunnableConfig:
raise NotImplementedError
# 4. 存储中间写入
def put_writes(self, config, writes, task_id, task_path=""
) -> None:
raise NotImplementedError泛型参数 V 代表版本号的类型，可以是 int、float 或 str。不同的实现选择不同的版本策略：
# 基类默认：整数版本（字符串版本必须由子类自己实现）
# libs/checkpoint/langgraph/checkpoint/base/__init__.py:460（docstring 略），判定体在 :474-479
def get_next_version(self, current: V | None, channel: None) -> V:
if isinstance(current, str):
raise NotImplementedError
elif current is None:
return 1
else:
return current + 1
# InMemorySaver：带随机后缀的字符串版本
# libs/checkpoint/langgraph/checkpoint/memory/__init__.py:518-527
def get_next_version(self, current: str | None, channel: None) -> str:
if current is None:
current_v = 0
elif isinstance(current, int):
current_v = current
else:
current_v = int(current.split(".")[0])
next_v = current_v + 1
next_h = random.random()
return f"{next_v:032}.{next_h:016}"InMemorySaver 使用带随机后缀的字符串版本，其精妙之处在于：主版本号保证单调递增用于排序和比较；随机后缀则让从同一个历史检查点分叉出的两条分支不会撞版本——blobs 以 (thread_id, checkpoint_ns, channel, version) 为键，两条分支会算出相同的主版本号，没有后缀就会互相覆盖对方的 Channel 值。整数部分补零到 32 位，确保字符串排序等同于数值排序；后缀是 random.random() 的十进制表示（格式里的 016 只是最小宽度，实跑得到的形如 00000000000000000000000000000002.0.675530230124289）。这种设计在保持简单高效的同时，为 InMemorySaver 的 _load_blobs 方法提供了精确的版本匹配能力。
用 langgraph 1.1.6 实测过这一点：把 get_next_version 换成基类的整数版本后，用 update_state 从中间检查点分叉并继续执行，原分支最终检查点读出的值被分叉写入的值覆盖了；用默认的带后缀版本则不会。
8.4.2 put 方法的契约
put 方法是持久化的核心。它接收当前配置、新的检查点、元数据和新版本号，返回更新后的配置：
def put(
self,
config: RunnableConfig, # 当前的 thread_id + checkpoint_ns
checkpoint: Checkpoint, # 要保存的检查点
metadata: CheckpointMetadata, # 元数据（source, step, parents...）
new_versions: ChannelVersions, # 本次变更的 channel 版本
) -> RunnableConfig:
# 返回包含新 checkpoint_id 的配置
...new_versions 参数的设计体现了增量更新的思想：它只包含本次写入中实际发生变化的 Channel 版本，使得存储后端可以只更新变化的部分（如 Postgres 的 checkpoint_blobs 表）。在 PregelLoop._put_checkpoint 中，new_versions 通过 get_new_channel_versions 函数计算得出，它比较 checkpoint_previous_versions 和当前的 channel_versions，只返回版本号发生变化的 Channel。这种差量计算使得频繁创建检查点的开销大大降低——如果一个步骤只修改了少量 Channel，那么只有这些 Channel 的值需要被序列化和存储（InMemorySaver、PostgresSaver 如此；SqliteSaver 的 put 不看 new_versions，每次把整个检查点序列化后写入一行）。
8.4.3 put_writes 方法的契约
put_writes 是 LangGraph 区别于简单快照系统的关键方法。它存储的不是完整状态，而是中间的写入操作：
def put_writes(
self,
config: RunnableConfig, # 关联的检查点配置
writes: Sequence[tuple[str, Any]], # 写入列表：(channel, value)
task_id: str, # 产生写入的任务ID
task_path: str = "", # 任务路径（用于排序）
) -> None:
...特殊写入通过 WRITES_IDX_MAP 映射到负索引，避免与普通写入冲突：
WRITES_IDX_MAP = {
ERROR: -1, # 错误信息
SCHEDULED: -2, # 调度标记
INTERRUPT: -3, # 中断信息
RESUME: -4, # 恢复值
}这意味着一个任务的普通写入（索引 0, 1, 2, ...）和特殊写入（索引 -1, -2, -3, -4）可以共存而不冲突。这个映射关系在所有检查点实现中必须保持一致，因此它被定义在基类模块中作为共享契约。每种特殊写入类型只能有一个实例（由负索引的唯一性保证），而新的特殊写入会覆盖旧的——例如，一个任务的重试可能会更新 ERROR 写入的值。
8.4.4 辅助方法
基类还声明了线程管理和数据维护的辅助方法（各有 a 前缀的异步版本）。它们在基类里都只是 raise NotImplementedError；本专栏对照的源码里，InMemorySaver、SqliteSaver、PostgresSaver 只实现了 delete_thread，delete_for_runs/copy_thread/prune 目前只有 checkpoint-conformance 包里的一致性测试在调用：
# 删除整个会话的所有检查点
def delete_thread(self, thread_id: str) -> None: ...
# 删除指定运行的所有检查点
def delete_for_runs(self, run_ids: Sequence[str]) -> None: ...
# 复制会话
def copy_thread(self, source_thread_id, target_thread_id) -> None: ...
# 修剪检查点
def prune(self, thread_ids, *, strategy="keep_latest") -> None: ...classDiagram
class BaseCheckpointSaver~V~ {
+SerializerProtocol serde
+get_tuple(config) CheckpointTuple
+list(config, filter, before, limit) Iterator
+put(config, checkpoint, metadata, new_versions) RunnableConfig
+put_writes(config, writes, task_id, task_path) None
+delete_thread(thread_id) None
+get_next_version(current, channel) V
+with_allowlist(extra_allowlist) BaseCheckpointSaver
}
class InMemorySaver {
+defaultdict storage
+defaultdict writes
+dict blobs
}
class SqliteSaver {
+Connection conn
+bool is_setup
+setup() None
}
class BasePostgresSaver {
+str SELECT_SQL
+list MIGRATIONS
}
BaseCheckpointSaver <|-- InMemorySaver
BaseCheckpointSaver <|-- SqliteSaver
BaseCheckpointSaver <|-- BasePostgresSaver
8.5 序列化体系
8.5.1 SerializerProtocol
LangGraph 的序列化层通过 Protocol 类型实现松耦合：
@runtime_checkable
class SerializerProtocol(Protocol):
def dumps_typed(self, obj: Any) -> tuple[str, bytes]: ...
def loads_typed(self, data: tuple[str, bytes]) -> Any: ...注意返回值是 tuple[str, bytes]——类型标签和序列化后的字节分开存储。这个设计让存储后端可以根据类型标签选择不同的处理策略（例如数据库中区分 JSON 和二进制 BLOB）。
为了兼容旧版仅提供 dumps/loads 的序列化器，SerializerCompat 适配器自动包装：
# libs/checkpoint/langgraph/checkpoint/serde/base.py:29
class SerializerCompat(SerializerProtocol):
def __init__(self, serde: UntypedSerializerProtocol) -> None:
self.serde = serde
def dumps_typed(self, obj: Any) -> tuple[str, bytes]:
return type(obj).__name__, self.serde.dumps(obj)
def loads_typed(self, data: tuple[str, bytes]) -> Any:
return self.serde.loads(data[1])8.5.2 JsonPlusSerializer 深度解析
JsonPlusSerializer 是 LangGraph 的默认序列化器，它的名字有一定的历史误导性——在当前版本中，它主要使用 ormsgpack（高性能 MessagePack 实现）而非 JSON 作为主要编码格式。
# libs/checkpoint/langgraph/checkpoint/serde/jsonplus.py:50，dumps_typed 在 :228
class JsonPlusSerializer(SerializerProtocol):
def dumps_typed(self, obj: Any) -> tuple[str, bytes]:
if obj is None:
return "null", EMPTY_BYTES
elif isinstance(obj, bytes):
return "bytes", obj
elif isinstance(obj, bytearray):
return "bytearray", obj
else:
try:
return "msgpack", _msgpack_enc(obj)
except ormsgpack.MsgpackEncodeError:
if self.pickle_fallback:
return "pickle", pickle.dumps(obj)
raise
def loads_typed(self, data: tuple[str, bytes]) -> Any:
type_, data_ = data
if type_ == "msgpack":
return ormsgpack.unpackb(
data_, ext_hook=self._unpack_ext_hook,
option=ormsgpack.OPT_NON_STR_KEYS
)
elif type_ == "json":
return json.loads(data_, object_hook=self._reviver)
# ... 其他类型处理序列化分为四条路径：null（None 值）、bytes/bytearray（原始字节直通）、msgpack（默认路径）和 pickle（可选回退）。这个分层策略的设计非常精巧：None 值不需要任何编码，用空字节表示即可；原始字节数据不需要转换，直接传递可以避免不必要的编解码开销；大多数 Python 对象走 msgpack 路径以获得最佳的性能和类型安全性；只有在 msgpack 无法处理的极端情况下（且显式启用了 pickle_fallback）才会降级到 pickle。反序列化时通过类型标签精确分发，避免了猜测数据格式的开销和错误风险。
8.5.3 ormsgpack 扩展类型系统
ormsgpack 原生能直接编码的类型有限，而 dataclass、datetime、Enum、UUID 这几类它本能直接编码的，LangGraph 也通过 OPT_PASSTHROUGH_* 选项（serde/jsonplus.py:793-800）让它们转交自定义的 _msgpack_default，好把类名一并记下、反序列化时还原成原类型。对于 Python 的丰富类型生态（Pydantic 模型、datetime、Enum、dataclass 等），LangGraph 通过 MessagePack 的 Extension Type 机制进行扩展：
# Extension Type 代码定义
EXT_CONSTRUCTOR_SINGLE_ARG = 0 # Type(arg)
EXT_CONSTRUCTOR_POS_ARGS = 1 # Type(*args)
EXT_CONSTRUCTOR_KW_ARGS = 2 # Type(**kwargs)
EXT_METHOD_SINGLE_ARG = 3 # Type.method(arg)
EXT_PYDANTIC_V1 = 4 # Pydantic v1 模型
EXT_PYDANTIC_V2 = 5 # Pydantic v2 模型
EXT_NUMPY_ARRAY = 6 # NumPy 数组扩展类型大多编码为 (module, name, args) 三元组（Pydantic v2 与 datetime 多带一个方法名，NumPy 数组另有 (dtype, shape, order, buffer) 格式），这样在反序列化时可以动态导入对应的类并重构对象：
# libs/checkpoint/langgraph/checkpoint/serde/jsonplus.py:274
def _msgpack_default(obj: Any) -> str | ormsgpack.Ext:
if hasattr(obj, "model_dump") and callable(obj.model_dump):
# Pydantic v2: 序列化为 (module, name, dict, method)
return ormsgpack.Ext(
EXT_PYDANTIC_V2,
_msgpack_enc((
obj.__class__.__module__,
obj.__class__.__name__,
obj.model_dump(),
"model_validate_json",
)),
)
elif isinstance(obj, Enum):
# Enum: 序列化为 (module, name, value)
return ormsgpack.Ext(
EXT_CONSTRUCTOR_SINGLE_ARG,
_msgpack_enc((
obj.__class__.__module__,
obj.__class__.__name__,
obj.value,
)),
)
elif dataclasses.is_dataclass(obj):
# Dataclass: 序列化为 (module, name, {field: value})
return ormsgpack.Ext(
EXT_CONSTRUCTOR_KW_ARGS,
_msgpack_enc((
obj.__class__.__module__,
obj.__class__.__name__,
{f.name: getattr(obj, f.name) for f in dataclasses.fields(obj)},
)),
)
# ... 更多类型处理flowchart TD
OBJ[Python 对象] --> CHECK{类型判断}
CHECK -->|None| NULL["'null', b''"]
CHECK -->|bytes/bytearray| RAW["'bytes', raw_data"]
CHECK -->|基本类型| MP["ormsgpack.packb()"]
CHECK -->|Pydantic v2| EXT5["Ext(5, module+name+dict)"]
CHECK -->|Pydantic v1| EXT4["Ext(4, module+name+dict)"]
CHECK -->|dataclass| EXT2["Ext(2, module+name+kwargs)"]
CHECK -->|Enum| EXT0["Ext(0, module+name+value)"]
CHECK -->|datetime| EXT3["Ext(3, module+name+iso+method)"]
CHECK -->|其他| ERR["TypeError / pickle fallback"]
MP --> RESULT["('msgpack', bytes)"]
EXT5 --> RESULT
EXT4 --> RESULT
EXT2 --> RESULT
EXT0 --> RESULT
EXT3 --> RESULT
NULL --> RESULT
RAW --> RESULT
8.5.4 安全与白名单机制
反序列化涉及动态导入和对象构造，这带来安全风险。LangGraph 通过白名单机制控制哪些类型可以被反序列化：
# libs/checkpoint/langgraph/checkpoint/serde/jsonplus.py:50，__init__ 在 :61
class JsonPlusSerializer(SerializerProtocol):
def __init__(
self,
*,
pickle_fallback: bool = False,
allowed_json_modules: Iterable[tuple[str, ...]] | Literal[True] | None = None,
allowed_msgpack_modules: AllowedMsgpackModules | Literal[True] | None = ...,
) -> None:
...这里有一个容易看错的地方：allowed_msgpack_modules 的默认值不是 None 而是一个哨兵，构造函数里再按环境变量决定落到哪一档（serde/jsonplus.py:71-75）——LANGGRAPH_STRICT_MSGPACK 开启时落到 None（严格），否则落到 True（宽松）。也就是说开箱即用的默认是宽松模式，严格模式要显式打开。
三种安全级别为不同的部署环境提供了合适的安全/便利性平衡：
None（严格模式，LANGGRAPH_STRICT_MSGPACK=1时成为默认）：只允许SAFE_MSGPACK_TYPES（serde/_msgpack.py:14）中预定义的内置安全类型。它能阻止恶意构造的检查点数据借反序列化实例化任意类；宽松模式对未注册类型打出的警告也写明，将来的版本会阻止这类反序列化。内置安全类型包括标准库中的常见类型如datetime、UUID、Decimal、集合类型等；EXT_METHOD_SINGLE_ARG走的构造方法另有一张SAFE_MSGPACK_METHODS白名单（serde/_msgpack.py:82）。- 明确列表：
allowed_msgpack_modules=[("my_module", "MyClass")]逐个精确授权。这是在安全性和功能性之间折中的做法，开发者只需要将自己的状态类型加入白名单即可。严格模式下，图编译时 LangGraph 还会自动分析 Schema 中使用的类型并添加到白名单中（见下文）。 True（宽松模式，未开启严格开关时的默认）：允许所有类型但对未注册的类型记录警告日志，适用于开发和调试环境。
with_msgpack_allowlist 方法允许在运行时扩展白名单。它在 compile() 里被间接调用——但只在严格模式下：StateGraph.compile 先判断 _serde.STRICT_MSGPACK_ENABLED，再由 apply_checkpointer_allowlist → BaseCheckpointSaver.with_allowlist 把图的各个 Schema 中用到的类型加进白名单（详见第 5 章 §5.3.2）：
# BaseCheckpointSaver.with_allowlist
def with_allowlist(self, extra_allowlist):
serde = _with_msgpack_allowlist(self.serde, extra_allowlist)
if serde is self.serde:
return self
clone = copy.copy(self)
clone.serde = maybe_add_typed_methods(serde)
return clone8.6 持久化模式
8.6.1 三种 Durability 策略
LangGraph 定义了三种持久化策略，通过 Durability 类型控制：
Durability = Literal["sync", "async", "exit"]sync（同步模式）：每个步骤完成后，立即将检查点写入存储，阻塞直到写入完成。这是最安全但性能最低的模式。
async（异步模式）：检查点写入提交到后台执行（同步图走线程池，异步图走 asyncio 任务），不阻塞下一个步骤的开始。这是默认模式，在安全性和性能之间取得平衡。
exit（退出模式）：只在图退出时写入检查点。这提供最高性能，但如果进程意外崩溃，中间步骤的状态将丢失。
sequenceDiagram
participant G as Graph Loop
participant S as CheckpointSaver
participant DB as Storage
Note over G,DB: sync 模式
G->>S: put(checkpoint)
S->>DB: write
DB-->>S: ack
S-->>G: done
G->>G: next step
Note over G,DB: async 模式
G->>S: submit(put, checkpoint)
G->>G: next step (parallel)
S->>DB: write (background)
DB-->>S: ack
Note over G,DB: exit 模式
G->>G: step 1
G->>G: step 2
G->>G: step N
G->>S: put(final checkpoint)
S->>DB: write
8.6.2 PregelLoop 中的持久化实现
在 PregelLoop 中，持久化逻辑通过 _put_checkpoint 方法实现：
# langgraph/pregel/_loop.py
def _put_checkpoint(self, metadata: CheckpointMetadata) -> None:
# 传入的就是当前 metadata 对象，说明是退出时的补写
exiting = metadata is self.checkpoint_metadata
# ...（已保存则返回、写入 step/parents 等，略）
# 判断是否需要实际写入
do_checkpoint = self._checkpointer_put_after_previous is not None and (
exiting or self.durability != "exit"
)
# 创建新检查点
self.checkpoint = create_checkpoint(
self.checkpoint,
self.channels if do_checkpoint else None,
self.step,
id=self.checkpoint["id"] if exiting else None,
updated_channels=self.updated_channels,
)
if do_checkpoint:
# 计算增量版本
new_versions = get_new_channel_versions(
self.checkpoint_previous_versions,
channel_versions,
)
# 异步提交，不阻塞
self._put_checkpoint_fut = self.submit(
self._checkpointer_put_after_previous,
getattr(self, "_put_checkpoint_fut", None), # 等待上一次完成
self.checkpoint_config,
copy_checkpoint(self.checkpoint),
self.checkpoint_metadata,
new_versions,
)关键设计点：_checkpointer_put_after_previous 接收上一个 checkpoint save 的 Future，确保检查点按顺序写入。这在异步模式下尤为重要——即使写入是异步的，顺序性仍然得到保证。这种"串联异步"的模式避免了简单异步可能导致的乱序问题：如果步骤 2 的检查点在步骤 1 之前到达存储层，恢复时可能会跳过步骤 1 的状态。通过 Future 链，每个检查点的写入必须等待前一个完成后才能开始，既保证了顺序性，又不阻塞图的执行线程。
_put_checkpoint 方法中另一个值得关注的细节是对 exiting 的处理（它由 metadata is self.checkpoint_metadata 判定，并不是一个参数）。exit 模式下，循环中的各步只在内存里推进检查点、不做 Channel 快照也不落盘；等图退出时（正常结束、出错或被中断都会经过 _suppress_interrupt，pregel/_loop.py:877-886），才以 exiting=True 再调一次：检查点 ID 保持不变（复用当前检查点的 ID），补上 Channel 值的快照后写入。这避免了在图退出时创建一个冗余的检查点——退出时的状态与最后一个循环步骤的状态是相同的，只需要确保最终的 Channel 值被正确保存即可。若这个检查点已经保存过（checkpoint_id_saved 相同），则直接返回。
put_writes 方法同样在每次任务写入时被调用，但对 exit 模式有特殊处理：
def put_writes(self, task_id: str, writes: WritesT) -> None:
# 保存写入到内存
self.checkpoint_pending_writes.extend((task_id, c, v) for c, v in writes)
# 只在非 exit 模式下立即持久化
if self.durability != "exit" and self.checkpointer_put_writes is not None:
self.submit(self.checkpointer_put_writes, config, writes_to_save, task_id)8.7 存储实现
8.7.1 InMemorySaver：开发利器
InMemorySaver 将所有数据存储在内存中的字典结构中，适用于开发和测试：
class InMemorySaver(BaseCheckpointSaver[str]):
# thread_id -> checkpoint_ns -> checkpoint_id -> (checkpoint, metadata, parent_id)
storage: defaultdict[str, dict[str, dict[str, tuple]]]
# (thread_id, checkpoint_ns, checkpoint_id) -> (task_id, write_idx) -> write
writes: defaultdict[tuple, dict[tuple, tuple]]
# (thread_id, checkpoint_ns, channel, version) -> (type, bytes)
blobs: dict[tuple, tuple[str, bytes]]三层嵌套字典的设计模拟了关系数据库的表结构：
graph TB
subgraph "InMemorySaver 存储结构"
direction TB
S["storage<br/>[thread_id]"] --> NS["[checkpoint_ns]"]
NS --> CK["[checkpoint_id]<br/>(checkpoint, metadata, parent_id)"]
W["writes<br/>[thread_id, ns, ck_id]"] --> WE["[task_id, idx]<br/>(task_id, channel, value, task_path)"]
B["blobs<br/>[thread_id, ns, channel, version]"] --> BV["(type, bytes)"]
end
blobs 字典将 Channel 值与 Checkpoint 主体分离存储，这是一个重要的优化：多个检查点可能引用同一个 Channel 版本（如果该 Channel 在这些步骤之间没有被更新），通过分离存储避免了重复序列化——同一个版本的值只存一次，多个检查点通过 channel_versions 指向它。注意这是按版本号寻址，而不是 Git 那种按内容哈希寻址：内容相同但版本号不同的两次写入会各存一份。
在 InMemorySaver 的 put 方法中，可以清楚地看到这种分离是如何工作的：首先从 checkpoint 中弹出 channel_values，然后遍历 new_versions（只包含本次变化的 Channel），将每个 Channel 的值序列化后存入 blobs 字典。键是 (thread_id, checkpoint_ns, channel, version) 四元组，确保同一个 Channel 的不同版本独立存储。值得注意的是，如果一个 Channel 在 new_versions 中但不在 channel_values 中（比如 Channel 被清空了），会存储一个 ("empty", b"") 的标记值。
8.7.2 SqliteSaver：轻量级持久化
SqliteSaver 使用 SQLite 作为存储后端，适合单机部署场景：
# libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py:38（类）
# + :122（setup，docstring 已省略）
class SqliteSaver(BaseCheckpointSaver[str]):
def setup(self) -> None:
if self.is_setup:
return
self.conn.executescript("""
PRAGMA journal_mode=WAL;
CREATE TABLE IF NOT EXISTS checkpoints (
thread_id TEXT NOT NULL,
checkpoint_ns TEXT NOT NULL DEFAULT '',
checkpoint_id TEXT NOT NULL,
parent_checkpoint_id TEXT,
type TEXT,
checkpoint BLOB,
metadata BLOB,
PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id)
);
CREATE TABLE IF NOT EXISTS writes (
thread_id TEXT NOT NULL,
checkpoint_ns TEXT NOT NULL DEFAULT '',
checkpoint_id TEXT NOT NULL,
task_id TEXT NOT NULL,
idx INTEGER NOT NULL,
channel TEXT NOT NULL,
type TEXT,
value BLOB,
PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx)
);
""")
self.is_setup = True注意 PRAGMA journal_mode=WAL 的设置——WAL（预写式日志）模式允许读写并发，这对于图执行期间从另一个连接或进程同时查询状态很重要（同一个 SqliteSaver 实例内部的操作仍会被下文的线程锁串行化）。在默认的日志模式下，写入操作会阻塞读取操作，这意味着在保存检查点时无法查询图的状态。启用预写式日志模式后，读取操作可以与写入操作并行执行，只要读取的数据不依赖于尚未提交的写入即可。
SqliteSaver 使用线程锁来保证操作原子性：
# SqliteSaver.__init__，libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py:78
# （签名在源码里是多行带类型）
def __init__(self, conn, *, serde=None):
super().__init__(serde=serde)
self.jsonplus_serde = JsonPlusSerializer()
self.conn = conn
self.is_setup = False
self.lock = threading.Lock()
@contextmanager
def cursor(self, transaction: bool = True):
with self.lock:
self.setup()
cur = self.conn.cursor()
try:
yield cur
finally:
if transaction:
self.conn.commit()
cur.close()8.7.3 PostgresSaver：生产级方案
Postgres 实现是最复杂也最强大的存储后端。它使用三张数据表（外加一张 checkpoint_migrations 迁移记录表），并通过数据库迁移确保版本兼容性：
# libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py:37 起
# （每条 SQL 在源码里是多行，这里压成一行；shallow.py:39 另有一套精简版）
MIGRATIONS = [
# v0: 迁移管理表
"""CREATE TABLE IF NOT EXISTS checkpoint_migrations (v INTEGER PRIMARY KEY);""",
# v1: 检查点主表
"""CREATE TABLE IF NOT EXISTS checkpoints (
thread_id TEXT NOT NULL,
checkpoint_ns TEXT NOT NULL DEFAULT '',
checkpoint_id TEXT NOT NULL,
parent_checkpoint_id TEXT,
type TEXT,
checkpoint JSONB NOT NULL,
metadata JSONB NOT NULL DEFAULT '{}',
PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id)
);""",
# v2: Channel 值单独存储（去重优化）
"""CREATE TABLE IF NOT EXISTS checkpoint_blobs (
thread_id TEXT NOT NULL,
checkpoint_ns TEXT NOT NULL DEFAULT '',
channel TEXT NOT NULL,
version TEXT NOT NULL,
type TEXT NOT NULL,
blob BYTEA,
PRIMARY KEY (thread_id, checkpoint_ns, channel, version)
);""",
# v3: 写入表
"""CREATE TABLE IF NOT EXISTS checkpoint_writes (
thread_id TEXT NOT NULL,
checkpoint_ns TEXT NOT NULL DEFAULT '',
checkpoint_id TEXT NOT NULL,
task_id TEXT NOT NULL,
idx INTEGER NOT NULL,
channel TEXT NOT NULL,
type TEXT,
blob BYTEA NOT NULL,
PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx)
);""",
# v4-v9: 索引优化和字段扩展
"ALTER TABLE checkpoint_blobs ALTER COLUMN blob DROP not null;",
"SELECT 1;", # 占位迁移
"CREATE INDEX CONCURRENTLY IF NOT EXISTS checkpoints_thread_id_idx ...",
"CREATE INDEX CONCURRENTLY IF NOT EXISTS checkpoint_blobs_thread_id_idx ...",
"CREATE INDEX CONCURRENTLY IF NOT EXISTS checkpoint_writes_thread_id_idx ...",
"ALTER TABLE checkpoint_writes ADD COLUMN IF NOT EXISTS task_path TEXT ...",
]Postgres 实现的查询使用了优雅的子查询来一次性加载检查点及其关联数据：
SELECT
thread_id, checkpoint, checkpoint_ns, checkpoint_id,
parent_checkpoint_id, metadata,
-- 子查询加载 channel 值
(SELECT array_agg(array[bl.channel::bytea, bl.type::bytea, bl.blob])
FROM jsonb_each_text(checkpoint -> 'channel_versions')
INNER JOIN checkpoint_blobs bl
ON bl.thread_id = checkpoints.thread_id
AND bl.checkpoint_ns = checkpoints.checkpoint_ns
AND bl.channel = jsonb_each_text.key
AND bl.version = jsonb_each_text.value
) AS channel_values,
-- 子查询加载 pending writes
(SELECT array_agg(array[cw.task_id::text::bytea, cw.channel::bytea,
cw.type::bytea, cw.blob]
ORDER BY cw.task_id, cw.idx)
FROM checkpoint_writes cw
WHERE cw.thread_id = checkpoints.thread_id
AND cw.checkpoint_ns = checkpoints.checkpoint_ns
AND cw.checkpoint_id = checkpoints.checkpoint_id
) AS pending_writes
FROM checkpointserDiagram
checkpoints {
TEXT thread_id PK
TEXT checkpoint_ns PK
TEXT checkpoint_id PK
TEXT parent_checkpoint_id
TEXT type
JSONB checkpoint
JSONB metadata
}
checkpoint_blobs {
TEXT thread_id PK
TEXT checkpoint_ns PK
TEXT channel PK
TEXT version PK
TEXT type
BYTEA blob
}
checkpoint_writes {
TEXT thread_id PK
TEXT checkpoint_ns PK
TEXT checkpoint_id PK
TEXT task_id PK
INTEGER idx PK
TEXT channel
TEXT type
BYTEA blob
TEXT task_path
}
checkpoints ||--o{ checkpoint_blobs : "channel_versions -> version"
checkpoints ||--o{ checkpoint_writes : "checkpoint_id"
Postgres 的 UPSERT 策略也值得注意：检查点使用 ON CONFLICT DO UPDATE（允许覆盖），而普通写入使用 ON CONFLICT DO NOTHING（幂等），特殊写入（错误、中断等）使用 ON CONFLICT DO UPDATE（最新值覆盖）：
# PostgresSaver.put_writes（postgres/__init__.py:351-356）
# 注意：分支是「整批」而不是逐行——一批写入要么全是特殊写入走 UPSERT，
# 要么走 INSERT ... ON CONFLICT DO NOTHING
query = (
self.UPSERT_CHECKPOINT_WRITES_SQL
if all(w[0] in WRITES_IDX_MAP for w in writes)
else self.INSERT_CHECKPOINT_WRITES_SQL
)
# _dump_writes（postgres/base.py:249-260）里 idx 的取值
# 特殊写入取 WRITES_IDX_MAP 的负数，普通写入取 enumerate 的序号
WRITES_IDX_MAP.get(channel, idx)PregelLoop.put_writes 在提交前会先做一次去重（libs/langgraph/langgraph/pregel/_loop.py:311-312）：如果这一批写入全部是特殊写入，就按 channel 去重、后写覆盖先写，正好和 SQL 侧的 UPSERT 语义对上。
8.8 设计决策分析
8.8.1 为什么 Channel 值分离存储？
在 Postgres 实现中，checkpoint_blobs 表独立于 checkpoints 表。这看似增加了复杂度，但带来了显著收益：
- 去重：如果某个 Channel 在连续多步中没有变化，只存储一份数据
- 增量写入：
new_versions参数使得每次只写入变化的 Channel - 查询效率：通过
channel_versions映射精确加载需要的 blob
8.8.2 为什么 pending_writes 独立于 Checkpoint？
将写入与检查点分离存储是实现中断恢复的关键。当一个超步里部分任务已经完成、另一些任务出错或被中断时，已完成任务的写入被保存在 checkpoint_writes 表中（InMemorySaver 是 writes 字典）。恢复时，这些写入可以被重放而不需要重新执行已完成的节点；被中断的那个节点本身则会从头重新执行。
8.8.3 为什么选择 ormsgpack 而非 JSON？
ormsgpack 相比 JSON 有两个关键优势：
- 性能：ormsgpack 是派生自 orjson 的 Rust 实现 MessagePack 编码器，产物是紧凑的二进制
- 类型丰富：MessagePack 原生支持 bytes，datetime 等类型则通过 Extension Type 机制编码，无需 JSON 中常见的字符串转换
但 JSON 路径仍然保留：loads_typed 仍能解码类型标签为 "json" 的旧数据。在 Postgres 实现中，checkpoints 表的 checkpoint 和 metadata 列使用 JSONB 类型而非 BYTEA，这意味着这些数据以 JSON 格式存储在数据库中，支持 SQL 级别的查询和索引。例如，jsonb_each_text(checkpoint -> 'channel_versions') 这样的查询可以在不反序列化整个检查点的情况下提取 Channel 版本信息。这种混合使用 JSON 和 msgpack 的策略体现了一个务实的工程选择：元数据、结构信息和原始类型的 Channel 值用 JSON 以获得数据库级别的可查询性，其余 Channel 值经 serde（默认 msgpack）写进 checkpoint_blobs，以获得更好的序列化性能和类型保真度。
8.8.4 为什么版本号采用字符串格式？
InMemorySaver 的版本号格式 "{version:032}.{random:016}" 看似简单，却蕴含精妙设计：
- 补零到 32 位的整数部分确保字符串排序等同于数值排序
- 随机后缀让同一检查点分叉出的不同分支不撞版本号（见 §8.4.1）
- InMemorySaver、SqliteSaver、PostgresSaver 用的是同一套格式（基类默认则是整数版本）
8.9 实际应用中的注意事项
8.9.1 检查点的存储开销
在生产系统中，检查点的存储开销需要认真考虑。默认持久化模式下每个执行步骤都会创建一个检查点，每个检查点逻辑上包含所有 Channel 的快照。对于包含大量消息历史的对话系统，检查点的大小可能随着对话长度线性增长。InMemorySaver 的 blobs 和 Postgres 的 checkpoint_blobs 按版本去重，减轻了这个问题——如果一个 Channel 在多个步骤中没有变化，只会存储一份副本（SqliteSaver 没有这层去重）。但是，使用 operator.add 作为 reducer 的 Channel（如消息列表）每被写入一次就产生一个新版本，而新版本存的是追加后的整份列表。
为了控制存储增长，基类声明了 prune 方法，keep_latest 策略只保留每个命名空间的最新检查点，适用于不需要时间旅行功能的场景。但如 §8.4.4 所述，本专栏对照源码里的这三个实现都没有实现它，能直接用的是按线程删除的 delete_thread；按时间清理旧检查点需要在数据库侧自行处理。
8.9.2 序列化的兼容性
当应用代码更新后（例如修改了 Pydantic 模型的字段），旧的检查点可能无法被新版本的代码正确反序列化。JsonPlusSerializer 的 ext_hook 对此有一定的容错能力，但因类型而异（serde/jsonplus.py:583-690）：Pydantic 模型先正常构造，失败就退到 model_construct/construct（跳过校验），连类都找不到时才返回原始字段字典；dataclass、namedtuple 等走构造函数的类型一旦构造失败，异常被吞掉、直接返回 None。用 langgraph 1.1.6 实测：给 dataclass 删掉一个字段后读旧检查点，得到的是 {'k': None}，除了一条"未注册类型"的警告日志外没有任何异常。所以这种容错是有限的，甚至会静默丢数据。对于生产系统，建议在进行不兼容的 Schema 变更时，先完成所有活跃线程的执行，或者使用新的 thread_id 开始新的对话。
8.9.3 并发访问的安全性
InMemorySaver 不是线程安全的（尽管它的异步方法是同步包装的）。SqliteSaver 通过线程锁实现了基本的线程安全。Postgres 实现通过数据库事务和 UPSERT 语义实现了最高级别的并发安全。在多进程或分布式部署中，只有 Postgres 实现可以安全使用——SQLite 的文件锁在网络文件系统上不可靠，InMemorySaver 本身就不跨进程。
8.10 小结
本章深入剖析了 LangGraph 的 Checkpoint 持久化体系。我们看到，这不是一个简单的"保存/加载"机制，而是一个精心设计的分层架构：
- 数据层：
CheckpointTypedDict 精确定义了状态快照的结构，UUID v6 提供单调递增的标识 - 协议层：
BaseCheckpointSaver定义了四大核心操作的契约，泛型版本号适配不同后端 - 序列化层：
JsonPlusSerializer通过 ormsgpack 的扩展类型机制实现类型安全的序列化，白名单机制保障安全 - 寻址层：
thread_id/checkpoint_ns/checkpoint_id三维寻址支持多会话、多层子图的检查点管理，UUID v6 的单调递增特性使得按 ID 排序即可确定时间顺序 - 策略层：
sync/async/exit三种持久化模式在一致性与性能间灵活取舍，异步模式通过 Future 链保证写入顺序 - 存储层：从内存到 SQLite 到 Postgres，各实现在保持接口统一的同时针对各自场景做了深度优化
Checkpoint 机制是 LangGraph 实现"可暂停计算"的基石。从计算理论的角度看，Checkpoint 赋予了图计算一种"延续"（continuation）的能力——它可以捕获计算的中间状态，暂停执行，然后在将来的任意时刻从该状态恢复。这种能力是构建可靠的 AI Agent 系统的前提条件：Agent 的执行可能跨越数分钟（等待 API 响应）、数小时（等待人类审批）甚至数天（等待外部事件），没有持久化的状态管理，这些场景都无法可靠地支撑。
在下一章中，我们将看到 Checkpoint 如何与中断机制配合，实现真正的人机协作工作流。