部署代理到 LangSmith 时,服务器提供一个内置的 Postgres 支持的检查点保存器,用于在图运行之间处理状态持久化。您可以用自己的 BaseCheckpointSaver 实现替换它,以使用不同的存储后端。
您提供一个异步上下文管理器的路径,它生成一个 BaseCheckpointSaver 实例,服务器自动管理其生命周期。
定义检查点保存器
从一个 **现有的** LangSmith 应用开始,创建一个定义异步上下文管理器以生成自定义检查点保存器的文件。如果您正在开始一个新项目,可以使用 CLI 从模板创建应用。
langgraph new --template=new-langgraph-project-python my_new_project
异步上下文管理器模式允许服务器在应用生命周期的适当时间点打开和关闭数据库连接:
# ./src/agent/checkpointer.py
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
针对一致性套件进行测试
大多数开源检查点保存器实现尚未实现 Agent Server 所需的所有操作。在配置检查点保存器之前,请针对一致性测试套件进行验证以确保兼容性。
安装软件包:
pip install langgraph-checkpoint-conformance
注册您的检查点保存器并运行验证:
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 测试运行:
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()
要查看该套件验证的基线和扩展操作的完整列表,请参阅 功能 section.
配置 langgraph.json
将 checkpointer 键添加到您的 langgraph.json 配置文件中。 path 指向您之前定义的异步上下文管理器 [](#define-the-checkpointer).
{
"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 或您自托管的平台。