以编程方式使用文档

部署代理到 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 或您自托管的平台。

后续步骤