Agent 状态持久化与人审（LangGraph Checkpoint / HITL / Time-Travel）

课程目标（14 天）

读完本课，你要能在白板上完整回答：

1. Agent 为什么必须持久化？一次性内存跑有什么灾难场景？
2. LangGraph 的 Checkpointer / Thread / State / Snapshot 如何协作？
3. 短期记忆（线程内）、长期记忆（跨线程）分别怎么存、怎么召回？
4. 如何实现 interrupt → 人审 → resume？HITL 插在哪些节点？
5. Time-Travel Debugging 怎么做？如何回到某一步重放不同分支？
6. 并发写入状态时如何避免冲突？分布式多实例如何协同？
7. 崩溃恢复时如何保证精确一次副作用（外部工具已调用怎么办）？

为什么需要（5 个真实灾难）

• 长任务（15 分钟 research agent）跑到第 12 分钟崩溃，从头再来，用户愤怒。
• Agent 调用"转账 tool"，写库成功但写状态前断电，重启后认为没转，再转一次。
• 用户在对话到 80 轮时想撤回第 20 轮的决定，没有 Time-Travel 只能删全局。
• 客服 Agent 需要上级审批退款，没 HITL 只能硬编码 if-else。
• 多实例部署下两个节点同时修改同一 thread 状态，覆盖写丢数据。

核心概念图

Thread (会话/任务)
  ├── Checkpoint 序列 (step0, step1, step2, ...)  ← Checkpointer 存储
  │     └── 每个 Checkpoint = {state, next_nodes, metadata, ts}
  ├── 短期记忆：state.messages / state.scratchpad
  └── 长期记忆：Store（跨 thread，按 user/namespace 查询）
 
interrupt 节点 → 暂停并写 checkpoint → 外部审批 → resume(thread_id, decision) → 继续

技术选型对照

14 天带练路径

Day 1 — 最小可跑 Checkpoint

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import StateGraph
 
graph = StateGraph(MyState)
# ... 加节点 ...
app = graph.compile(checkpointer=SqliteSaver.from_conn_string("state.db"))
 
cfg = {"configurable": {"thread_id": "u1-task-42"}}
for chunk in app.stream({"input": "..."}, cfg):
    print(chunk)

• 验收：杀进程后用同 thread_id 继续跑，状态仍在。

Day 2 — 理解 State Reducer

• 状态合并语义：Annotated[list, add_messages]、operator.add、自定义 reducer。
• 任务：写一个自定义 reducer：工具调用去重。
• 验收：能解释为什么直接 return {"messages": [...]} 不会覆盖现有 messages。

Day 3 — Thread 与多会话

• 一个用户 N 个任务 = N 个 thread_id。
• get_state(config) 查当前；get_state_history(config) 拿全部 checkpoint。
• 产出：会话列表页 + 详情页（为后续 Time-Travel 做准备）。

Day 4 — 短期记忆：Message Trimming 与摘要

• 问题：长对话会爆 context。
• 策略：
  1. trim_messages(max_tokens=4000, strategy="last")。
  2. 超阈值时摘要前 N 轮，替换为 SystemMessage("[历史摘要] ...")。
  3. 保留结构化 slot（用户名、订单号）不被摘要吃掉。
• 验收：100 轮对话仍然能答出第 3 轮提过的订单号。

Day 5 — 长期记忆：Store + 语义检索

from langgraph.store.memory import InMemoryStore
store = InMemoryStore(index={"embed": embeddings, "dims": 1536})
 
# 写
store.put(("user_42", "prefs"), "diet", {"value": "vegetarian"})
# 读（按 namespace + 语义）
hits = store.search(("user_42", "prefs"), query="dietary restriction", limit=3)

• 什么该写长期：用户偏好、常用项目、历史决策；什么不该：一次性中间计算。
• 验收：新开一个 thread 后仍能召回。

Day 6 — interrupt 实现人审

from langgraph.types import interrupt, Command
 
def risky_node(state):
    decision = interrupt({
        "action": "refund",
        "amount": state["amount"],
        "reason": state["reason"],
    })
    if decision["approved"]:
        return {"status": "refunded"}
    return {"status": "rejected"}
 
# 运行到 interrupt 会暂停
app.invoke({"amount": 500}, cfg)
# 审批后 resume
app.invoke(Command(resume={"approved": True}), cfg)

• 在哪儿加 interrupt：不可逆动作、金额阈值、合规关键字。
• 与 ai-application-security 的 LLM06 对齐。

Day 7 — 审批 UI / IM 回调

• 架构：interrupt 触发 → webhook/IM 推送 → 审批人点击 → 回调 resume(thread_id, decision)。
• 产出：FastAPI /approvals/{thread_id} + 前端审批卡片。
• 带 trace_id 做全链路可回溯。

Day 8 — Time-Travel Debugging

history = list(app.get_state_history(cfg))
# 回到第 5 步，换一个决策重跑
past = history[5]
new_cfg = {"configurable": {
    "thread_id": cfg["configurable"]["thread_id"],
    "checkpoint_id": past.config["configurable"]["checkpoint_id"],
}}
app.update_state(new_cfg, {"user_choice": "alternative"})
app.invoke(None, new_cfg)

• 用途：回归调试、做 A/B、让用户"撤销到某一步"。

Day 9 — 崩溃恢复与精确一次

• 问题：工具已调（转账成功），但写 checkpoint 前挂了。
• 方案：
  1. 幂等键：工具入参带 idempotency_key = hash(thread, step, args)，服务端去重。
  2. 先预占后提交：tool 执行返回"进行中"，下一 checkpoint 确认。
  3. 补偿事务：Saga 模式，失败回滚。
• 验收：强杀+重启 100 次，外部账户金额只变一次。

Day 10 — 并发与锁

• 场景：用户在 web 点按钮 + 定时器同时触发同 thread。
• 方案：
  • Postgres 行锁 + SELECT FOR UPDATE。
  • 乐观锁 + 版本号。
  • 队列串行化（Redis list / Kafka partition by thread_id）。
• 验收：两个并发 invoke 不丢状态。

Day 11 — 可观测性接入

• 每个 checkpoint 写 trace_id, thread_id, step, node, latency, token, cost。
• 接 LangSmith / OpenTelemetry，可视化节点时序。
• 能回答"这个 bad case 在第几步偏离？"。

Day 12 — 流式 + 持久化并行

• 需求：前端要流式 token，后端要持久化；两者不互斥。
• 用 app.astream_events(..., version="v2") 同时产出 token chunk 与 state delta。
• 与 ai-frontend-streaming-ux 联动。

Day 13 — 生命周期管理

• Thread 怎么清理？TTL（例：7 天不活跃归档到冷存）。
• 长期记忆容量控制：LRU + 语义去重。
• 合规：用户删除请求要级联清 checkpoint + store。

Day 14 — 综合项目 + 面试演练

• 做一个“15 分钟研究助手”：支持中断、审批、回放，串联 14 天所学能力。
• 白板 8 分钟讲"Agent 持久化架构"。

面试高频追问（深度版）

1. Checkpointer 和普通数据库写入有什么区别？ Checkpointer 是状态机整体快照，保证节点执行点一致；普通写入只是业务表。
2. 状态爆炸怎么办？ Reducer 去重 + 摘要 + 分层（热/冷状态） + 只存 delta。
3. 如何做 Exactly-Once 副作用？ 幂等键 + outbox 表 + saga 补偿，语言级做不到，要靠协议和外部系统。
4. Time-Travel 如何和长期记忆协调？ 回到旧 checkpoint 时，长期记忆不自动回退（那是跨线程的事实），需要显式 snapshot。
5. 为什么不用纯 Redis 而用 Postgres？ 事务 + 行锁 + JSONB 查询 + TTL + 审计，Redis 适合热缓存不适合真源。
6. interrupt 和传统消息队列审批有何区别？ interrupt 把状态机原地冻结，resume 从准确点继续；MQ 方案要自己重建上下文。
7. 多实例部署怎么防状态串？ 按 thread_id 做 sticky 或队列分区；checkpoint 写入强一致。

常见陷阱

• 把 SqliteSaver 直接上生产，多实例写冲突。
• thread_id 用用户 ID 而不是"用户+任务"，导致任务互串。
• 状态里塞原始 PDF / 大 JSON，checkpoint 膨胀到 MB 级。
• interrupt 后忘了超时机制，审批人不处理永远卡死。
• 长期记忆没做 namespace，跨用户泄露。

可交付产物

• graphs/research_agent.py（含 HITL + checkpoint）
• api/approvals.py（审批回调）
• ui/approval-card.tsx
• docs/agent-recovery-playbook.md
• eval/replay-test.py（Time-Travel 回归）

最新资料（2026-05 校验）

与 agent-architecture、langchain-langgraph、ai-application-security、ai-frontend-streaming-ux 联动。