如何使用自定义检查点保存器

在您的智能体部署中，将内置的 Postgres 检查点保存器替换为自定义的 BaseCheckpointSaver 实现。

将智能体部署到 LangSmith 时，服务器提供了一个内置的基于 Postgres 的检查点保存器，用于处理图运行之间的状态持久化。您可以用自己的 BaseCheckpointSaver 实现来替换它，以使用不同的存储后端。

您需要提供一个指向异步上下文管理器的路径，该管理器会生成一个 BaseCheckpointSaver 实例，服务器会自动管理其生命周期。

自定义检查点保存器目前处于 alpha 阶段。此功能可能在次要版本更新中出现破坏性变更。

若想使用 MongoDB 而非 PostgreSQL 进行检查点存储，请参阅 配置检查点保存器后端。本页面适用于实现完全自定义的存储后端。

定义检查点保存器

从一个 现有 的 LangSmith 应用程序开始，创建一个文件来定义生成自定义检查点保存器的异步上下文管理器。如果您要开始一个新项目，可以使用 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__()
        # 在此处初始化您的自定义检查点保存器
    ...

    @contextlib.asynccontextmanager
    async def aget(self, config: RunnableConfig):
        # 在此处编写创建连接池和初始化检查点保存器的自定义逻辑。
        yield


@contextlib.asynccontextmanager
async def generate_checkpointer():
    """生成一个 BaseCheckpointSaver，在服务器运行期间保持打开状态。"""
    async with AsyncSqliteSaver.from_conn_string("./checkpoints.db") as saver:
        await saver.setup()
        yield saver

针对一致性测试套件进行测试

大多数开源检查点保存器实现尚未支持 Agent Server 所需的所有操作。在配置您的检查点保存器之前，请先通过一致性测试套件验证其兼容性。

安装包：

pip install langgraph-checkpoint-conformance

注册您的检查点保存器并运行验证：

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())

该套件会自动检测您的检查点保存器实现了哪些扩展功能，并运行相应的测试。您也可以将其作为 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

服务器日志将确认您的自定义检查点保存器已激活。

功能

服务器在启动时会检查您的检查点保存器是否具备 基础（必需）和 扩展（可选）功能。如果缺少某个扩展功能，服务器将使用回退方案或禁用相应的特性。

基础功能（必需）

方法	描述
aput	存储检查点
aput_writes	存储待写入项
aget_tuple	检索检查点
alist	列出检查点
adelete_thread	删除线程

扩展功能（可选）

方法	描述	缺失时的回退方案
adelete_for_runs	删除特定运行的检查点	回滚多任务策略不可用
acopy_thread	复制线程	慢速回退（逐个重新插入检查点）
aprune	修剪线程历史	线程历史修剪不可用

部署

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

后续步骤

• 使用自定义存储 来替换内置的长期记忆存储。
• 了解 LangGraph 中的 持久化与记忆。

---

Edit this page on GitHub or file an issue.

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.