Checkpoint
Checkpoint 是 Agent Run 执行过程中的状态快照，用于中断恢复、Human-in-the-loop 和调试回溯。它是 Agent 可靠性的基础设施。
#type / concept
#status / evergreen
#tech / ai
#tech / architecture
[!info] related notes
- 所属 MOC: Agent Runtime MOC
- 上游概念: Run, Run State
- 应用: Human-in-the-loop, Cancellation
- 对比: Trace — Trace 用于回溯，Checkpoint 用于恢复
Checkpoint
一句话定义
Checkpoint 是 Agent Run 在执行过程中的状态快照，保存了当前的对话历史、工具调用结果、中间状态和执行位置。它的核心用途是让中断的 Run 能够恢复执行，而不需要从头开始。
它解决什么问题
Agent 的 Run 可能跨越多个 Step，执行时间从几秒到几分钟不等。在这个过程中：
- Human-in-the-loop: Agent 需要暂停等待人类审批，审批后恢复
- 服务重启: 部署新版本时，进行中的 Run 需要恢复
- 错误重试: 某步失败后，从上一个成功的 Checkpoint 重试
- 调试回溯: 查看 Run 在某个时间点的状态
没有 Checkpoint，这些场景都只能从头重来。
核心原理
Checkpoint 保存什么
@dataclass
class Checkpoint:
run_id: str # 所属 Run
checkpoint_id: str # 快照 ID
created_at: datetime # 创建时间
# 执行位置
step_index: int # 当前步骤索引
node_name: str # 当前节点名称（图工作流中）
# 状态
state: dict # 完整状态快照
messages: list[Message] # 对话历史
tool_calls: list[ToolCall] # 工具调用记录
tool_results: list[ToolResult] # 工具执行结果
# 元数据
token_usage: TokenUsage # token 消耗
metadata: dict # 其他元数据Checkpoint 时机
Run 开始
│
▼
Step 1: LLM Call
│ ← Checkpoint (每次 LLM 调用后)
▼
Step 2: Tool Execution
│ ← Checkpoint (每次工具执行后)
▼
Step 3: LLM Call
│ ← Checkpoint (遇到 interrupt)
▼
[等待人类审批]
│
▼
恢复执行
│
▼
Step 4: Tool Execution
│ ← Checkpoint
▼
Run 完成LangGraph 的 Checkpoint 机制
from langgraph.checkpoint import SqliteSaver
# 创建 checkpoint 存储
checkpointer = SqliteSaver.from_conn_string(":memory:")
# 编译图时启用 checkpoint
app = graph.compile(checkpointer=checkpointer)
# 运行时自动保存 checkpoint
config = {"configurable": {"thread_id": "user_123"}}
result = app.invoke(input_data, config=config)
# 恢复执行
state = app.get_state(config)
result = app.invoke(None, config=config) # 从 checkpoint 恢复Checkpoint 存储选择
| 存储 | 适用场景 | 优缺点 |
|---|---|---|
| 内存 | 开发测试 | 快但不持久 |
| SQLite | 单机部署 | 简单但不支持并发 |
| PostgreSQL | 生产环境 | 可靠、支持并发、可查询 |
| Redis | 高性能场景 | 快但需要额外管理持久化 |
典型工程实现
Checkpoint Manager
class CheckpointManager:
def __init__(self, store: CheckpointStore):
self.store = store
async def save(self, run: Run) -> str:
checkpoint = Checkpoint(
run_id=run.run_id,
checkpoint_id=generate_id(),
created_at=datetime.now(),
step_index=run.current_step,
state=run.state.copy(),
messages=run.messages.copy(),
tool_calls=run.tool_calls.copy(),
token_usage=run.token_usage,
)
await self.store.save(checkpoint)
return checkpoint.checkpoint_id
async def restore(self, run_id: str) -> Optional[Checkpoint]:
return await self.store.get_latest(run_id)
async def list_checkpoints(self, run_id: str) -> list[Checkpoint]:
return await self.store.list(run_id)从 Checkpoint 恢复
async def resume_run(checkpoint: Checkpoint, user_decision: dict):
# 1. 恢复状态
run = Run.from_checkpoint(checkpoint)
# 2. 注入用户决策（如果是 HITL 恢复）
if user_decision:
run.messages.append(tool_result_message(
tool_call_id=checkpoint.pending_tool_call_id,
result=user_decision,
))
# 3. 继续执行
async for event in engine.run(run.context, start_from=checkpoint.step_index):
yield event常见设计模式
1. 自动 Checkpoint
每个关键步骤（LLM 调用、工具执行）后自动保存。
2. 增量 Checkpoint
只保存与上一个 Checkpoint 的差异，减少存储开销。
3. Checkpoint 清理
定期清理过期的 Checkpoint，只保留最近 N 个。
常见坑
- 不保存 LLM 输入: 恢复时无法知道 Prompt 是什么
- Checkpoint 太频繁: 每个 token 都保存，存储开销大
- Checkpoint 太少: 中断后丢失太多进度
- 不做清理: Checkpoint 越积越多，存储爆满
- 不测试恢复: 保存了但恢复逻辑有 bug
和其他概念的关系
- vs Trace: Trace 用于事后回溯，Checkpoint 用于中断恢复
- vs Run State: Run State 是当前状态，Checkpoint 是状态的历史快照
- vs HITL: HITL 依赖 Checkpoint 实现中断-恢复
- vs Cancellation: 取消后可以从 Checkpoint 恢复
总结
Checkpoint 是 Agent 可靠性的基础设施。没有 Checkpoint，Agent 的每次中断都意味着从头再来。