LangGraph Checkpoint Conformance：为自定义 Checkpointer 实现编写与运行一致性测试套件

LangGraph 把图的状态持久化职责抽象为 BaseCheckpointSaver 这一存储契约，而 langgraph-checkpoint-conformance 是官方提供的一套一致性（Conformance）测试套件，用于自动验证任意 BaseCheckpointSaver 子类是否真正实现了这一契约。本文基于 checkpoint-conformance 的 README 与其完整源码，讲解如何通过 checkpointer_test + validate 注册并验证自己的 Checkpointer，剖析其能力检测与报告机制，帮助你在生产环境接入自研存储（或复用官方 SQLite / Postgres 后端）前获得可靠的"达标"凭证。

为什么需要 Conformance 测试

LangGraph 的 Checkpointer 负责在 Graph 执行过程中按 thread_id 持久化检查点，使 Agent 得以在中断（interrupt）、恢复与多轮对话中保留状态。在 BaseCheckpointSaver 的定义 中可以看到，它要求子类承担一系列职责：存储与读取 Checkpoint、维护 channel 版本（channel_versions / versions_seen）、记录 pending writes、隔离不同 checkpoint_ns 命名空间、支持按 checkpoint_id 做时间旅行等。

这些职责横跨多个方法，且很容易在实现时出现"单测全绿、接入 Graph 却行为异常"的偏差。Conformance 测试套件就是为此设计的：它把存储契约固化成一组与具体实现无关的测试，专门验证 blob 往返（round-trip）、元数据保留、命名空间隔离、增量 channel 更新等行为，从而回答一个问题——你的 Checkpointer 真的符合 LangGraph 的存储契约吗？

整个仓库本身就是这套工具的活体证明：

• capabilities.py 定义能力模型；
• spec/ 下每个文件对应一个能力的一组测试；
• checkpoint / sqlite / postgres 库的测试、postgres 版本、sqlite 版本 均用它验证官方实现；
• test_validate_memory.py 用它自检 InMemorySaver。

安装与运行前提

在仓库中通过 uv 直接安装即可：

uv add langgraph-checkpoint-conformance

该包是一个轻量测试库，其 pyproject.toml 声明：

• 运行环境要求 requires-python = ">=3.10"；
• 唯一硬依赖为 langgraph-checkpoint>=2.0.0（内含 BaseCheckpointSaver、Checkpoint、CheckpointTuple 等核心类型）；
• 测试侧依赖 pytest 与 pytest-asyncio，且在 [tool.pytest.ini_options] 中配置了 asyncio_mode = "auto" 与 addopts = "--strict-markers --strict-config"。

需要特别说明的是：由于 validate() 是异步函数，在使用 pytest 时必须让 asyncio 插件生效（官方示例使用 @pytest.mark.asyncio），否则测试无法执行。

快速开始：注册 Checkpointer 并运行校验

入口 API

包的公开接口只暴露两个符号（见 init.py）：

• checkpointer_test：装饰器，把一个返回 Checkpointer 实例的异步生成器注册为"测试工厂"；
• validate：核心运行器，对注册的工厂执行全部能力测试并返回报告。

写法一：普通 asyncio 入口

import asyncio
from langgraph.checkpoint.conformance import checkpointer_test, validate

@checkpointer_test(name="MyCheckpointer")
async def my_checkpointer():
    saver = MyCheckpointer(...)
    yield saver
    # yield 之后的代码会在测试完成后执行，可用作资源清理

async def main():
    report = await validate(my_checkpointer)
    report.print_report()
    assert report.passed_all_base()

asyncio.run(main())

写法二：pytest 集成

import pytest
from langgraph.checkpoint.conformance import checkpointer_test, validate

@checkpointer_test(name="MyCheckpointer")
async def my_checkpointer():
    yield MyCheckpointer(...)

@pytest.mark.asyncio
async def test_conformance():
    report = await validate(my_checkpointer)
    report.print_report()
    assert report.passed_all_base()

@checkpointer_test 装饰后的函数本质是一个 factory（而非单个实例）。从 initializer.py 的 RegisteredCheckpointer 实现看，验证器会在每个能力套件开始前通过 factory().__anext__() 重新生成一个全新 Checkpointer，并在 finally 中推进生成器触发 yield 之后的清理代码——这也是为什么 README 强调"cleanup runs after yield"。

仓库自带的冒烟自测 test_validate_memory.py 就是最精简的参照实现：它注册 InMemorySaver，运行 validate，断言 report.passed_all_base() 为真，失败时通过 report.to_dict() 输出诊断信息。

能力模型：必需项与可选扩展项

测试套件把 Checkpointer 的能力分为基础能力（Base，必选）与扩展能力（Extended，可选、自动探测）。README 给出的能力总表如下：

能力（Capability）	是否必需	对应异步方法
put	是	aput
put_writes	是	aput_writes
get_tuple	是	aget_tuple
list	是	alist
delete_thread	是	adelete_thread
delete_for_runs	否	adelete_for_runs
copy_thread	否	acopy_thread
prune	否	aprune
delta_channel_history	否	aget_delta_channel_history

该表与 capabilities.py 的定义 完全一一对应：BASE_CAPABILITIES 包含前五项，EXTENDED_CAPABILITIES 包含后四项，_CAPABILITY_METHOD_MAP 建立了"能力名 → BaseCheckpointSaver 方法名"的映射（测试套件始终调用各能力的 async 变体）。

扩展能力的自动探测机制

README 中说明：扩展能力通过"该方法是否被 BaseCheckpointSaver 覆盖"来判断，若未覆盖则跳过对应测试。源码实现比描述更严谨——注意 DetectedCapabilities.from_instance 与底层判定函数 _is_overridden（见 capabilities.py）对全部九项能力统一执行探测：

def _is_overridden(inner_type: type, method: str) -> bool:
    base = getattr(BaseCheckpointSaver, method, None)
    impl = getattr(inner_type, method, None)
    if base is None or impl is None:
        return impl is not None
    return impl is not base

也就是说，判定标准是"当前类型上的方法不等于基类默认实现"。因此即便是 BaseCheckpointSaver 中已给出默认行为（如抛 NotImplementedError）的扩展方法，只要子类没有真正覆盖它，就会被判定为"未实现"，对应测试自动标记为跳过（detected=False, tests_skipped=1）。这也意味着，只有 put、put_writes、get_tuple、list、delete_thread 五项是硬性门槛，其余能力的缺失不会被算作失败。

方法在基类中的位置

若想快速确认各方法签名，可对照 BaseCheckpointSaver 源码：aget_tuple（L429）、alist（L443）、aput（L468）、aput_writes（L491）、adelete_thread（L511）、adelete_for_runs（L522）、acopy_thread（L540）、aprune（L560）、aget_delta_channel_history（L651）。同时基类还公开了同步版本（get_tuple / list / put / …），子类可二选一实现，但 conformance 套件统一走 async 路径验证，这提醒实现者注意别只实现同步分支导致 Graph 异步执行时回退到基类占位。

使用选项（Options）

1. 进度输出（Progress output）

ProgressCallbacks 提供三档预设（见 report.py）：

from langgraph.checkpoint.conformance.report import ProgressCallbacks

# Dot 风格：每通过一个测试打印 ".", 失败打印 "F"
report = await validate(my_checkpointer, progress=ProgressCallbacks.default())

# 详细风格：逐条打印测试名（✓/✗），失败时附带堆栈
report = await validate(my_checkpointer, progress=ProgressCallbacks.verbose())

不传 progress 或使用 ProgressCallbacks.quiet() 则为静默模式。实现层面，这三档只是对 on_capability_start / on_test_result / on_capability_end 三个回调的不同编排（report.py），因此你完全可以自定义回调把进度接入自己的日志系统。值得一提的细节：在默认输出中，未实现的能力会显示为 ⊘ put_writes (not implemented) 之类的行，而 dot 风格的 ./F 字符是 flush=True 边测边打印的。

2. 跳过部分能力（Skip capabilities）

当某个可选能力你明知未实现、不想让其干扰观察时，可在装饰器上声明：

@checkpointer_test(name="MyCheckpointer", skip_capabilities={"prune"})
async def my_checkpointer():
    yield MyCheckpointer(...)

从 validate 的执行分支 可见，被跳过的能力会直接进入 detected=False, passed=None, tests_skipped=1 的结果并 continue，不会触发能力探测与运行。

3. 只运行指定能力（Run specific capabilities）

report = await validate(my_checkpointer, capabilities={"put", "list"})

该参数传入后，验证器只会把运行集合限制为 {Capability.PUT, Capability.LIST}（字符串会经 Capability(c) 转换校验），适用于开发期对单个能力做快速迭代回归。

4. Lifespan：一次性初始化与清理

对数据库建表这类昂贵的一次性 setup，如果放在 factory 里执行会在每个能力套件间反复触发。此时应使用 lifespan：

async def db_lifespan():
    await create_database()
    yield
    await drop_database()

@checkpointer_test(name="PostgresSaver", lifespan=db_lifespan)
async def pg_checkpointer():
    async with PostgresSaver.from_conn_string(CONN_STRING) as saver:
        yield saver

生命周期语义（见 initializer.py 的 enter_lifespan）：lifespan 在整个 validate 运行期间只进入一次，yield 前执行 setup，运行结束后在 finally 中推进生成器执行 teardown；而 factory 每个能力套件都会重新调用一次以获得"干净"实例。两者叠加恰好构成官方的推荐数据库测试模式：db_lifespan 管建库/删库，factory 管每次创建新的 PostgresSaver 连接。

validate 的运行原理（源码级剖析）

validate（validate.py）的完整流程可概括为四步：

1. 确定待测能力集合：若传入了 capabilities 参数则取其子集，否则默认覆盖全部 Capability（枚举顺序即声明顺序，见 capabilities.py 的 Capability）。
2. 进入 lifespan：async with registered.enter_lifespan() 包裹整轮验证，保证一次性 setup/teardown 在首尾执行。
3. 逐能力运行：对每个在集合内且未被 skip_capabilities 排除的能力，通过 registered.create() 创建全新 Checkpointer → DetectedCapabilities.from_instance(saver) 探测能力 → 若未探测到则记录 tests_skipped=1 并跳过 → 否则查找 _RUNNERS 映射表（validate.py）中对应的 run_*_tests 执行函数。
4. 汇总结论：每个 runner 返回 (passed, failed, failures) 三元组，包装为 CapabilityResult 存入 report.results[cap.value]。

每个 runner 的通用模式可见 spec/test_put.py 的 run_put_tests：按顺序遍历该能力的全部测试函数，逐个 await，成功则 passed += 1 并通过 on_test_result 回调通知进度，失败则捕获异常记录 test_name: exception 与完整 traceback。

报告对象：读取验证结论

CapabilityReport（report.py）是理解验证结果的统一入口，提供三个关键方法：

• passed_all_base()：全部五项基础能力测试均通过才返回 True。这是 README 示例中断言使用的标准，也是判断"能否作为 LangGraph 官方支持的 Checkpointer"的底线；
• passed_all()：所有被探测到的能力测试都通过才返回 True（未探测到的可选能力不计入失败）；
• conformance_level()：返回可读等级字符串，取值 FULL（全部通过）、BASE+PARTIAL（基础全过、部分扩展缺失/失败）、BASE（只有部分基础能力通过）或 NONE。

print_report()（report.py）输出分为 BASE CAPABILITIES 与 EXTENDED CAPABILITIES 两段，用 ✅（通过）、❌ (N failed)（失败）、⊘ (not implemented)（未实现）与 ⏭ (skipped) 逐行标注，末尾打印 Result: <level> (passed/total)。若需要把结果喂给 CI 或生成 JSON 报告，可用 to_dict()（report.py）得到含 conformance_level 与逐能力统计（passed / failed / skipped / failures）的可序列化字典。

spec 测试套件：九项能力覆盖哪些行为

仓库内 conformance/spec/ 目录按能力拆分测试文件，每份文件末尾都导出 ALL_*_TESTS 列表与 run_*_tests runner。以被验证得最充分的 put 为例，test_put.py 共包含 17 个测试，覆盖：

• 往返一致性：aput 后 aget_tuple 能还原相同的 checkpoint id、channel values、channel versions、versions_seen 与 metadata；
• 序列化正确性：str/int/list/dict 等多种 channel 值类型无损往返，版本号允许 int↔str 归一化比较；
• 命名空间：根命名空间（checkpoint_ns=''）、子命名空间（'child:abc'）与缺省行为均需正确落库；
• 多线程隔离：同一线程多 checkpoint 均可按 id 找回、不同 thread_id 互不干扰、父子 checkpoint 的 parent_config 链接正确；
• 增量 channel 更新：只有更新的 channel 才写入新 blob（对应 new_versions 语义），未变更的 channel 应从先前版本的 blob 重建（见 test_put_incremental_channel_update、test_put_new_channel_added、test_put_channel_removed）；
• run_id 与扩展 metadata key 的保留。

其余八项能力的测试意图也可以从测试名直接读出：

• test_put_writes.py：pending writes 的写入、读取与排序；
• test_get_tuple.py：不存在返回 None、无 checkpoint_id 时返回最新、指定 id 精确读取、parent/pending writes 与命名空间约束；
• test_list.py：按 thread/namespace 过滤、时间倒序排列、metadata 多 key 过滤、before 与 limit 分页、空结果与结果中包含 pending writes；
• test_delete_thread.py：删除 checkpoint 及其 writes、覆盖全部命名空间、不影响其它线程、对不存在线程为空操作；
• test_delete_for_runs.py：按 run 删除单个/多个 run 的数据并保留其它 run，空列表与不存在 run 为空操作，跨命名空间生效；
• test_copy_thread.py：复制整条线程的全部 checkpoint，同时保留 metadata、命名空间、pending writes 与顺序，源线程不被修改，源不存在时行为正确；
• test_prune.py：按策略裁剪历史 checkpoint；
• test_delta_channel_history.py（配合 _delta_fixtures.py）：验证 delta channel 历史写入按时间升序返回、以最近的快照为 seed、从根回溯无 seed 的 walk 行为、旧格式纯值到 delta 快照的迁移兼容等。

这种"单文件一能力 + 测试清单全量导出"的布局，让第三方开发者既可以整体运行，也可以临时只跑某个 test_fn 定位具体缺陷。

测试数据与断言辅助工具

为了让测试与实现彻底解耦，test_utils.py 提供了全套"造数据"与"比对"辅助函数，这些也是理解断言语义的钥匙：

• generate_checkpoint(...)（L19-L36）：用 UUIDv6 风格的 checkpoint_id、UTC 时间戳等合理默认值构造完整 Checkpoint；
• generate_config(thread_id, checkpoint_ns, checkpoint_id)（L39-L52）：构造携带 thread_id / checkpoint_ns / checkpoint_id 的 RunnableConfig；
• generate_metadata(source, step, **extra)（L55-L63）：构造含 source、step、parents 及自定义扩展 key 的元数据；
• put_test_checkpoint / put_test_checkpoints（L66-L137）：封装"写入 + parent 链接 + channel 版本一致性"的落库过程，支持批量写入多线程/多命名空间/多版本链；
• assert_checkpoint_equal / assert_tuple_equal（L140-）：对 checkpoint 的五要素（版本、id、channel_versions、versions_seen、channel_values）与 tuple 的 config/metadata/parent/pending writes 做语义级比对（例如允许版本号 int↔str 归一化），而非字典浅比较。

如何用它验证自己的 Checkpointer：完整清单

结合上文，为自研 Checkpointer 跑一遍官方一致性验证只需四步：

1. 安装 langgraph-checkpoint-conformance（Python ≥ 3.10，需 langgraph-checkpoint>=2.0.0），在测试环境配好 pytest + pytest-asyncio；
2. 实现 BaseCheckpointSaver 子类，至少覆盖五项基础能力对应的方法；按需实现 acopy_thread / aprune / adelete_for_runs / aget_delta_channel_history 获得扩展能力；
3. 注册一个 @checkpointer_test(name=...) 装饰的异步生成器 factory（返回全新实例、yield 后清理）；有建库类昂贵开销时另配 lifespan；
4. 运行与收口：调用 validate，用 report.print_report() 查看逐能力结果，并以 report.passed_all_base()（CI 硬门槛）或 report.passed_all() / report.conformance_level()（追求 FULL）作为断言；定位问题时配合 capabilities={"put"} 缩小范围、用 ProgressCallbacks.verbose() 拿到失败堆栈。

仓库内 InMemorySaver 的自检测试 与官方 SQLite / Postgres 后端的 conformance 用例 就是这套流程的现成范本，可以直接对照仿写。

版本策略与生态定位

该包当前版本为 0.0.2（见 pyproject.toml），采用 MIT 协议开源，代码风格与其它 LangGraph 库一致（ruff + hatchling + uv）。值得留意的是其依赖面刻意做得极小——运行态只依赖 langgraph-checkpoint，对 langgraph 本身的引用（如 delta snapshot 私有类型）仅在测试时按需导入，正如 pyproject 中针对 ty 规则的注释所言：扩展方法是否存在应交给运行时的能力探测来决定，而不是静态类型检查提前否决。这一设计哲学也解释了为什么 README 中反复强调"测试是自动探测并跳过"而非"报错即失败"。

对想要深入的人群，建议按三条线索继续阅读仓库：BaseCheckpointSaver 契约 回答"要实现什么"，conformance/spec/ 测试文件 回答"怎样算符合"，官方 checkpointer 的 conformance 用例 回答"真实实现如何做到"——三者闭环，即可为自己的持久化后端建立可回归、可审计的质量基线。