将智能体部署到 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())
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 | 修剪线程历史 | 线程历史修剪不可用 |
后续步骤
