Skip to main content
LangGraph Checkpointer 用于保存 Agent 执行状态,支持 HITL(Human-in-the-Loop)中断恢复和会话持久化。

1. 概述

1.1 Checkpointer 的作用

1.2 与其他组件的区别


2. 方案对比

2.1 存储方案选择

2.2 推荐:PostgresSaver

理由
  1. 已有 PostgreSQL 数据库(Supabase/Drizzle),无需新增依赖
  2. LangGraph 官方支持,稳定可靠
  3. 延迟可接受(5-20ms)
  4. 可通过 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_idcreated_at 上创建索引。

4. HITL 工作流程


5. 迁移步骤

5.1 从 DrizzleCheckpointSaver 迁移到 PostgresSaver

  1. 安装依赖 - 安装 langgraph-checkpoint-postgres>=1.0.0
  2. 修改 base_service.py - 将 DrizzleCheckpointSaver 引用替换为 AsyncPostgresSaver
  3. 更新 get_checkpointer 方法 - 使用 AsyncPostgresSaver.from_conn_string() 初始化
  4. 运行数据库迁移 - PostgresSaver 会自动创建所需表结构
  5. 删除旧代码 - 移除 DrizzleCheckpointSaver 和相关 Next.js API

6. 监控与调试

可以通过 SQL 查询 checkpoint 历史记录,包括查看某个会话的所有 checkpoint、查找中断状态的 checkpoint、以及清理过期的 checkpoint 数据。建议添加日志来追踪 checkpoint 的加载、保存和中断操作。

7. 总结