LangGraph 的 checkpointer 设计是"接口归框架,实现归用户"。这个决策带来了巨大灵活性,但也制造了一个测试上的根本矛盾:框架支持 Memory、SQLite(同步/异步/AES 加密)、Postgres(同步/pipeline/连接池/异步各变体)、Redis 等十余种后端配置,而这些后端必须在语义层面表现完全一致——用同一个 thread_id 存进去的状态,用任何后端都要能原样取出来。
最直接的做法是为每个后端写一套测试。但这条路很快就走不通了:十几个后端意味着十几倍的测试维护成本,一旦协议规范有变动,每套测试都要改一遍,而且改漏了也不会立刻报错。
另一种常见方案是用类继承——定义一个 BaseCheckpointerTest,把公共测试逻辑放在基类方法里,各后端的测试类继承并重写 make_checkpointer() 工厂方法。这在初期看起来整洁,但随着测试维度增加,问题就暴露出来了。假设你需要测试 checkpointer 和 store 的组合,或者 checkpointer 和 cache 的组合,类继承就会陷入多重继承的泥潭,组合出来的类层次树越来越难以追踪。类继承擅长表达"是什么",却拙于表达"和什么组合"。
LangGraph 采用了不同的解法:pytest fixture 参数化。
看一下 libs/langgraph/tests/conftest.py 中的核心 fixture:
NO_DOCKER = os.getenv("NO_DOCKER", "false") == "true"
@pytest.fixture(
scope="function",
params=[
"memory",
"sqlite",
"sqlite_aes",
]
if NO_DOCKER
else [
"memory",
"memory_migrate_sends",
"sqlite",
"sqlite_aes",
"postgres",
"postgres_pipe",
"postgres_pool",
],
)
def sync_checkpointer(
request: pytest.FixtureRequest,
) -> Iterator[BaseCheckpointSaver]:
checkpointer_name = request.param
if checkpointer_name == "memory":
with _checkpointer_memory() as checkpointer:
yield checkpointer
elif checkpointer_name == "sqlite":
with _checkpointer_sqlite() as checkpointer:
yield checkpointer
elif checkpointer_name == "postgres":
with _checkpointer_postgres() as checkpointer:
yield checkpointer
# ...
这个 fixture 有几个设计决策值得细看。
第一,params 列表在定义时根据 NO_DOCKER 环境变量静态确定。这意味着后端列表的选择发生在 pytest 收集阶段,而不是每个测试函数运行时。如果 Docker 未启动,直接用 NO_DOCKER=true 运行,Postgres 相关的参数根本不会进入 params 列表,pytest 不会尝试连接数据库,测试也不会因为连接失败而报错,而是压根不生成这些测试用例。本地开发完全不受影响。
第二,scope="function" 确保每个测试函数都获得一个独立的 checkpointer 实例。这是故意的:共享 checkpointer 会让不同测试之间产生状态污染,而状态污染是最难调试的问题之一,因为失败现象取决于测试执行顺序。
第三,测试函数本身对后端完全无感知:
def test_time_travel(sync_checkpointer: BaseCheckpointSaver) -> None:
# 这里的代码对 sync_checkpointer 是 Memory 还是 Postgres 一无所知
# 也不需要知道
graph = builder.compile(checkpointer=sync_checkpointer)
# ...
graph LR
subgraph fixture["sync_checkpointer fixture (params=[memory, sqlite, postgres, ...])"]
P1["memory"]
P2["sqlite"]
P3["postgres"]
P4["postgres_pool"]
end
fixture -->|"自动展开"| T1["test_time_travel[memory]"]
fixture -->|"自动展开"| T2["test_time_travel[sqlite]"]
fixture -->|"自动展开"| T3["test_time_travel[postgres]"]
fixture -->|"自动展开"| T4["test_time_travel[postgres_pool]"]
pytest 会自动将这个函数扩展为 test_time_travel[memory]、test_time_travel[sqlite]、test_time_travel[postgres] 等多个测试用例。新增一个后端只需要在 params 列表里加一行,并在 fixture 的 if-elif 里加一个分支,所有依赖 sync_checkpointer 的测试函数都自动覆盖了新后端。
更重要的是多维度组合。libs/langgraph/tests/conftest.py 里同时定义了 sync_checkpointer、async_checkpointer、sync_store、async_store、cache 这几个参数化 fixture。测试函数可以同时声明多个参数化 fixture:
async def test_graph_with_store_and_cache(
async_checkpointer: BaseCheckpointSaver,
async_store: BaseStore,
cache: BaseCache,
) -> None:
...
pytest 会取这三个 fixture 的笛卡尔积,生成所有合法的后端组合测试。类继承在这里完全束手无策——你没有办法用继承优雅地表达三个维度的全组合,但 fixture 参数化自然就支持。
libs/checkpoint-conformance 是一个独立的包,专门存放 checkpointer 协议的合规测试。它存在的理由非常清晰:LangGraph 框架的 checkpointer 接口是公开的,任何人都可以实现自己的后端——可能是 DynamoDB、MongoDB 或者自定义的分布式存储。这些第三方实现需要有一种机制来验证自己是否正确。