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

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

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

{
  "dependencies": ["."],
  "graphs": {
    "agent": "./src/agent/graph.py:graph"
  },
  "env": ".env",
  "checkpointer": {
    "path": "./src/agent/checkpointer.py:generate_checkpointer"
  }
}

langgraph dev --no-browser