使用 Checkpointer 实现跨请求恢复

本章目标

理解开发、单机和生产环境的持久化选择，并正确读取历史快照。

三种环境不要混用

SQLite 示例：examples/durable_approval.py

connection = sqlite3.connect(path, check_same_thread=False)
checkpointer = SqliteSaver(connection)
graph = builder.compile(checkpointer=checkpointer)

thread_id 是同一条执行链的持久游标：

config = {"configurable": {"thread_id": "tenant-a:T-100"}}

本项目把租户和工单共同编码进 thread_id，防止不同租户复用同一线程。真实系统仍必须在服务端验证当前身份有权访问该租户和工单。

读取历史时，get_state_history() 返回可迭代对象，不可直接下标：

history = list(graph.get_state_history(config))
checkpoint = history[3]

PostgreSQL 生命周期

src/support_agent/api.py 在 FastAPI lifespan 中创建异步保存器：

async with AsyncPostgresSaver.from_conn_string(database_url) as saver:
    await saver.setup()
    app.state.graph = build_support_graph(checkpointer=saver)
    yield

教程在启动时调用 setup() 以便首次运行。正式平台应把 Checkpointer 建表与迁移纳入独立发布任务，避免多个应用实例同时承担迁移职责。

做一次真正的“重启后恢复”实验

仅在同一个 Python 对象上调用两次 invoke() 不能证明持久化有效。examples/postgres_restart_recovery.py 会执行两个阶段：第一阶段创建 Checkpointer 并在审批点暂停；退出连接上下文后，第二阶段重新创建连接、Saver 和整张图，再使用同一个 thread_id 恢复。

docker compose up -d postgres
export DATABASE_URL='postgresql://langgraph:langgraph@localhost:5432/langgraph'
python examples/postgres_restart_recovery.py

预期先看到 paused，随后看到 resumed: success。为了模拟真实进程重启，可以把脚本拆成两个命令分别运行；关键标准是第二个进程没有接收原始 State，只依赖数据库和稳定线程标识。

thread_id 不是授权

攻击者如果能猜到 tenant-a:T-100，不应因此读取对应状态。API 必须先从已认证主体推导租户，再验证工单归属，最后在服务端构造 thread_id。不要允许客户端提交任意 Checkpoint ID、命名空间或租户前缀。

同一 thread_id 的并发运行还需要串行化。Checkpointer 保存状态，但不自动把所有业务动作变成互斥事务。可以使用运行队列、Agent Server 或带租约的数据库锁，并为重复提交返回现有 run_id 或 409。

状态演进与保留策略

生产 Checkpoint 会长期存在，State 字段因此需要版本策略：

• 新增可选字段时提供默认解释。
• 删除或改名字段前编写迁移或兼容读取逻辑。
• 不把无法长期反序列化的类实例写入 State。
• 记录 schema_version，在恢复入口拒绝无法安全升级的快照。
• 根据业务合规设置 TTL、归档和删除任务。

删除 Checkpoint 前先确认业务事实已写入业务数据库。Checkpoint 是执行上下文，不应成为退款单和审批单的唯一事实来源。

持久化故障矩阵

本章验收

• 使用两个独立图实例完成暂停和恢复。
• 能解释 Checkpoint、业务数据库记录和审计日志的职责差异。
• 为线程并发、数据保留和 State 版本升级写出策略。
• PostgreSQL 故障时服务明确不可用，而不是悄悄切换内存模式。