如何使用自定义 checkpointer

在你的 agent 部署中，用自定义的 BaseCheckpointSaver 实现替换内置的 Postgres checkpointer。

将 agent 部署到 LangSmith 时，服务器会提供一个内置的、由 Postgres 支持的 checkpointer，用于在图的多次运行之间处理状态持久化。你可以将其替换为你自己的 BaseCheckpointSaver 实现，以使用不同的存储后端。

你提供一个异步上下文管理器的路径，该管理器会 yield 一个 BaseCheckpointSaver 实例，服务器会自动管理其生命周期。

WARNING

自定义 checkpointer 处于 alpha 阶段。此功能可能会在次要版本更新中出现破坏性变更。

TIP

如需使用 MongoDB 而不是 PostgreSQL 进行 checkpoint 存储，请参阅配置 checkpointer 后端。本页介绍如何实现完全自定义的存储后端。

定义 checkpointer

从一个现有的 LangSmith 应用开始，创建一个文件，定义 yield 你的自定义 checkpointer 的异步上下文管理器。如果你要开始一个新项目，可以使用 CLI 从模板创建应用。

langgraph new --template=new-langgraph-project-python my_new_project

异步上下文管理器模式让服务器能够在应用生命周期的正确时点打开和关闭数据库连接：

# ./src/agent/checkpointer.py
import contextlib

class MyCheckpointer(BaseCheckpointSaver):
    def __init__(self):
        super().__init__()
        # Initialize your custom checkpointer here
    ...

    @contextlib.asynccontextmanager
    async def aget(self, config: RunnableConfig):
        # Your custom logic to create a connection pool and initialize your checkpointer here.
        yield

@contextlib.asynccontextmanager
async def generate_checkpointer():
    """Yield a BaseCheckpointSaver, open for the duration of the server."""
    async with AsyncSqliteSaver.from_conn_string("./checkpoints.db") as saver:
        await saver.setup()
        yield saver

使用一致性测试套件进行测试

大多数开源 checkpointer 实现尚未实现 Agent Server 所需的全部操作。在配置你的 checkpointer 之前，请用一致性测试套件验证它，以确保兼容性。

安装该包：

pip install langgraph-checkpoint-conformance

注册你的 checkpointer 并运行验证：

import asyncio

from langgraph.checkpoint.conformance import checkpointer_test, validate

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

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

asyncio.run(main())

该套件会自动检测你的 checkpointer 实现了哪些扩展能力，并运行相应的测试。你也可以将其作为 pytest 测试运行：

import pytest

from langgraph.checkpoint.conformance import checkpointer_test, validate

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

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

要查看该套件验证的基础操作和扩展操作的完整列表，请参阅能力部分。

配置 langgraph.json

将 checkpointer 键添加到你的 langgraph.json 配置文件。path 指向你先前定义的异步上下文管理器。

{
  "dependencies": ["."],
  "graphs": {
    "agent": "./src/agent/graph.py:graph"
  },
  "env": ".env",
  "checkpointer": {
    "path": "./src/agent/checkpointer.py:generate_checkpointer"
  }
}

启动服务器

在本地测试服务器：

langgraph dev --no-browser

服务器日志会确认你的自定义 checkpointer 已生效。

能力

服务器会在启动时检查你的 checkpointer 的基础（必需）和扩展（可选）能力。如果缺少某项扩展能力，服务器会使用回退方案或禁用相应功能。

基础能力（必需）

方法	说明
aput	存储 checkpoint
aput_writes	存储待处理的写入
aget_tuple	检索 checkpoint
alist	列出 checkpoint
adelete_thread	删除线程

扩展能力（可选）

方法	说明	缺失时的回退
adelete_for_runs	删除特定 run 的 checkpoint	回滚多任务策略不可用
acopy_thread	复制线程	较慢的回退（逐个重新插入 checkpoint）
aprune	清理线程历史	线程历史清理不可用

部署

你可以将此应用按原样部署到 LangSmith 或自托管平台。

后续步骤

• 构建自定义 checkpointer，包括 delta channel 支持。
• 使用自定义 store 替换内置的长期记忆存储。
• 了解 LangGraph 中的持久化与记忆。

---