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

LangGraph Checkpoint Conformance为自定义 Checkpointer 实现编写与运行一致性测试套件【免费下载链接】langgraphBuild resilient agents.项目地址: https://gitcode.com/GitHub_Trending/la/langgraphLangGraph 把图的状态持久化职责抽象为BaseCheckpointSaver这一存储契约而langgraph-checkpoint-conformance是官方提供的一套一致性Conformance测试套件用于自动验证任意BaseCheckpointSaver子类是否真正实现了这一契约。本文基于 checkpoint-conformance 的 README 与其完整源码讲解如何通过checkpointer_testvalidate注册并验证自己的 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-checkpoint2.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.pycheckpointer_test装饰器把一个返回 Checkpointer 实例的异步生成器注册为测试工厂validate核心运行器对注册的工厂执行全部能力测试并返回报告。写法一普通 asyncio 入口import asyncio from langgraph.checkpoint.conformance import checkpointer_test, validate checkpointer_test(nameMyCheckpointer) 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(nameMyCheckpointer) 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是aputput_writes是aput_writesget_tuple是aget_tuplelist是alistdelete_thread是adelete_threaddelete_for_runs否adelete_for_runscopy_thread否acopy_threadprune否aprunedelta_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的扩展方法只要子类没有真正覆盖它就会被判定为未实现对应测试自动标记为跳过detectedFalse, tests_skipped1。这也意味着只有put、put_writes、get_tuple、list、delete_thread五项是硬性门槛其余能力的缺失不会被算作失败。方法在基类中的位置若想快速确认各方法签名可对照 BaseCheckpointSaver 源码aget_tupleL429、alistL443、aputL468、aput_writesL491、adelete_threadL511、adelete_for_runsL522、acopy_threadL540、apruneL560、aget_delta_channel_historyL651。同时基类还公开了同步版本get_tuple/list/put/ …子类可二选一实现但 conformance 套件统一走 async 路径验证这提醒实现者注意别只实现同步分支导致 Graph 异步执行时回退到基类占位。使用选项Options1. 进度输出Progress outputProgressCallbacks提供三档预设见 report.pyfrom langgraph.checkpoint.conformance.report import ProgressCallbacks # Dot 风格每通过一个测试打印 ., 失败打印 F report await validate(my_checkpointer, progressProgressCallbacks.default()) # 详细风格逐条打印测试名✓/✗失败时附带堆栈 report await validate(my_checkpointer, progressProgressCallbacks.verbose())不传progress或使用ProgressCallbacks.quiet()则为静默模式。实现层面这三档只是对on_capability_start / on_test_result / on_capability_end三个回调的不同编排report.py因此你完全可以自定义回调把进度接入自己的日志系统。值得一提的细节在默认输出中未实现的能力会显示为⊘ put_writes (not implemented)之类的行而 dot 风格的./F字符是flushTrue边测边打印的。2. 跳过部分能力Skip capabilities当某个可选能力你明知未实现、不想让其干扰观察时可在装饰器上声明checkpointer_test(nameMyCheckpointer, skip_capabilities{prune}) async def my_checkpointer(): yield MyCheckpointer(...)从 validate 的执行分支 可见被跳过的能力会直接进入detectedFalse, passedNone, tests_skipped1的结果并continue不会触发能力探测与运行。3. 只运行指定能力Run specific capabilitiesreport await validate(my_checkpointer, capabilities{put, list})该参数传入后验证器只会把运行集合限制为{Capability.PUT, Capability.LIST}字符串会经Capability(c)转换校验适用于开发期对单个能力做快速迭代回归。4. Lifespan一次性初始化与清理对数据库建表这类昂贵的一次性 setup如果放在 factory 里执行会在每个能力套件间反复触发。此时应使用lifespanasync def db_lifespan(): await create_database() yield await drop_database() checkpointer_test(namePostgresSaver, lifespandb_lifespan) async def pg_checkpointer(): async with PostgresSaver.from_conn_string(CONN_STRING) as saver: yield saver生命周期语义见 initializer.py 的 enter_lifespanlifespan 在整个 validate 运行期间只进入一次yield前执行 setup运行结束后在finally中推进生成器执行 teardown而 factory 每个能力套件都会重新调用一次以获得干净实例。两者叠加恰好构成官方的推荐数据库测试模式db_lifespan管建库/删库factory 管每次创建新的PostgresSaver连接。validate 的运行原理源码级剖析validatevalidate.py的完整流程可概括为四步确定待测能力集合若传入了capabilities参数则取其子集否则默认覆盖全部Capability枚举顺序即声明顺序见 capabilities.py 的 Capability。进入 lifespanasync with registered.enter_lifespan()包裹整轮验证保证一次性 setup/teardown 在首尾执行。逐能力运行对每个在集合内且未被skip_capabilities排除的能力通过registered.create()创建全新 Checkpointer →DetectedCapabilities.from_instance(saver)探测能力 → 若未探测到则记录tests_skipped1并跳过 → 否则查找_RUNNERS映射表validate.py中对应的run_*_tests执行函数。汇总结论每个 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。报告对象读取验证结论CapabilityReportreport.py是理解验证结果的统一入口提供三个关键方法passed_all_base()全部五项基础能力测试均通过才返回True。这是 README 示例中断言使用的标准也是判断能否作为 LangGraph 官方支持的 Checkpointer的底线passed_all()所有被探测到的能力测试都通过才返回True未探测到的可选能力不计入失败conformance_level()返回可读等级字符串取值FULL全部通过、BASEPARTIAL基础全过、部分扩展缺失/失败、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_*_testsrunner。以被验证得最充分的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_removedrun_id 与扩展 metadata key的保留。其余八项能力的测试意图也可以从测试名直接读出test_put_writes.pypending writes 的写入、读取与排序test_get_tuple.py不存在返回None、无checkpoint_id时返回最新、指定 id 精确读取、parent/pending writes 与命名空间约束test_list.py按 thread/namespace 过滤、时间倒序排列、metadata 多 key 过滤、before与limit分页、空结果与结果中包含 pending writestest_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按策略裁剪历史 checkpointtest_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 时间戳等合理默认值构造完整Checkpointgenerate_config(thread_id, checkpoint_ns, checkpoint_id)L39-L52构造携带thread_id/checkpoint_ns/checkpoint_id的RunnableConfiggenerate_metadata(source, step, **extra)L55-L63构造含source、step、parents及自定义扩展 key 的元数据put_test_checkpoint / put_test_checkpointsL66-L137封装写入 parent 链接 channel 版本一致性的落库过程支持批量写入多线程/多命名空间/多版本链assert_checkpoint_equal / assert_tuple_equalL140-对 checkpoint 的五要素版本、id、channel_versions、versions_seen、channel_values与 tuple 的 config/metadata/parent/pending writes 做语义级比对例如允许版本号 int↔str 归一化而非字典浅比较。如何用它验证自己的 Checkpointer完整清单结合上文为自研 Checkpointer 跑一遍官方一致性验证只需四步安装langgraph-checkpoint-conformancePython ≥ 3.10需langgraph-checkpoint2.0.0在测试环境配好pytestpytest-asyncio实现BaseCheckpointSaver子类至少覆盖五项基础能力对应的方法按需实现acopy_thread/aprune/adelete_for_runs/aget_delta_channel_history获得扩展能力注册一个checkpointer_test(name...)装饰的异步生成器 factory返回全新实例、yield 后清理有建库类昂贵开销时另配lifespan运行与收口调用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 用例 回答真实实现如何做到——三者闭环即可为自己的持久化后端建立可回归、可审计的质量基线。【免费下载链接】langgraphBuild resilient agents.项目地址: https://gitcode.com/GitHub_Trending/la/langgraph创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考