LangGraph Checkpointer 用于保存 Agent 执行状态,支持 HITL(Human-in-the-Loop)中断恢复和会话持久化。
1. 概述
1.1 Checkpointer 的作用
1.2 与其他组件的区别
2. 方案对比
2.1 存储方案选择
2.2 推荐:PostgresSaver
理由:- 已有 PostgreSQL 数据库(Supabase/Drizzle),无需新增依赖
- LangGraph 官方支持,稳定可靠
- 延迟可接受(5-20ms)
- 可通过 SQL 查询历史 checkpoint,便于调试
3. 实现方案
3.1 替换 DrizzleCheckpointSaver 为 PostgresSaver
使用 LangGraph 官方提供的AsyncPostgresSaver,通过 DATABASE_URL 环境变量连接数据库。初始化时自动调用 setup() 创建所需的数据库表。
3.2 数据库表结构
PostgresSaver 会自动创建checkpoints 表,包含以下主要字段:
thread_id- 会话标识checkpoint_id- 检查点标识parent_checkpoint_id- 父检查点标识checkpoint- 检查点数据 (JSONB)metadata- 元数据 (JSONB)created_at- 创建时间
(thread_id, checkpoint_id) 的复合键,并在 thread_id 和 created_at 上创建索引。
4. HITL 工作流程
5. 迁移步骤
5.1 从 DrizzleCheckpointSaver 迁移到 PostgresSaver
- 安装依赖 - 安装
langgraph-checkpoint-postgres>=1.0.0 - 修改 base_service.py - 将 DrizzleCheckpointSaver 引用替换为 AsyncPostgresSaver
- 更新 get_checkpointer 方法 - 使用
AsyncPostgresSaver.from_conn_string()初始化 - 运行数据库迁移 - PostgresSaver 会自动创建所需表结构
- 删除旧代码 - 移除 DrizzleCheckpointSaver 和相关 Next.js API