> ## Documentation Index
> Fetch the complete documentation index at: https://cndoc-langchain.site/llms.txt
> Use this file to discover all available pages before exploring further.

# 如何使用自定义检查点器

> 在您的智能体部署中，使用自定义的 BaseCheckpointSaver 实现替换内置的 Postgres 检查点器。

当将智能体部署到 LangSmith 时，服务器提供了一个内置的、基于 Postgres 的检查点器，用于处理跨图运行的状态持久化。您可以将其替换为您自己的 [BaseCheckpointSaver](https://reference.langchain.com/python/langgraph/checkpoints/#langgraph.checkpoint.base.BaseCheckpointSaver) 实现，以使用不同的存储后端。

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

<Warning>
  自定义检查点器目前处于 **alpha** 阶段。此功能在次版本更新中可能会发生破坏性变更。
</Warning>

<Tip>
  要使用 MongoDB 而非 PostgreSQL 进行检查点存储，请参阅 [配置检查点器后端](/langsmith/configure-checkpointer)。本页面用于实现完全自定义的存储后端。
</Tip>

## 定义检查点器

从一个**现有的** LangSmith 应用程序开始，创建一个文件，定义一个异步上下文管理器，该管理器生成您的自定义检查点器。如果您是开始一个新项目，可以使用 CLI 从模板创建应用。

```bash theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
langgraph new --template=new-langgraph-project-python my_new_project
```

异步上下文管理器模式允许服务器在应用程序生命周期的适当时间点打开和关闭数据库连接：

```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
# ./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
```

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

大多数开源检查点器实现尚未实现智能体服务器所需的所有操作。在配置您的检查点器之前，请使用一致性测试套件对其进行验证，以确保兼容性。

安装包：

```bash theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
pip install langgraph-checkpoint-conformance
```

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

```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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 测试运行：

```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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()
```

要查看该套件验证的基础和扩展操作的完整列表，请参阅 [功能](#capabilities) 部分。

## 配置 `langgraph.json`

在您的 [`langgraph.json` 配置文件](/langsmith/application-structure#configuration-file-concepts) 中添加 `checkpointer` 键。`path` 指向您[之前定义](#define-the-checkpointer)的异步上下文管理器。

```json theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
{
  "dependencies": ["."],
  "graphs": {
    "agent": "./src/agent/graph.py:graph"
  },
  "env": ".env",
  "checkpointer": {
    "path": "./src/agent/checkpointer.py:generate_checkpointer"
  }
}
```

## 启动服务器

在本地测试服务器：

```bash theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
langgraph dev --no-browser
```

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

## 功能

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

### 基础功能（必需）

| 方法               | 描述       |
| ---------------- | -------- |
| `aput`           | 存储检查点    |
| `aput_writes`    | 存储待处理的写入 |
| `aget_tuple`     | 检索检查点    |
| `alist`          | 列出检查点    |
| `adelete_thread` | 删除线程     |

### 扩展功能（可选）

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

## 部署

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

## 后续步骤

* [使用自定义存储](/langsmith/custom-store) 替换内置的长期记忆存储。
* 了解 LangGraph 中的[持久化与记忆](/oss/python/langgraph/persistence)。

***

<div className="source-links">
  <Callout icon="terminal-2">
    [将这些文档连接](/use-these-docs)到 Claude、VSCode 等工具，通过 MCP 获取实时答案。
  </Callout>

  <Callout icon="edit">
    [在 GitHub 上编辑此页面](https://github.com/langchain-ai/docs/edit/main/src/langsmith/custom-checkpointer.mdx) 或 [提交问题](https://github.com/langchain-ai/docs/issues/new/choose)。
  </Callout>
</div>