111 lines
5.2 KiB
Markdown
111 lines
5.2 KiB
Markdown
---
|
||
id: ai.langgraph.persistence-hitl
|
||
title: 检查点、人机协作和异常恢复：checkpointer、interrupt、RetryPolicy、任务状态
|
||
type: lesson
|
||
domain: ai
|
||
stack: [langgraph]
|
||
tags: [checkpointer, thread_id, interrupt, Command, RetryPolicy, 崩溃恢复, 任务状态]
|
||
level: advanced
|
||
prerequisites: [ai.langgraph.state-graph]
|
||
est_hours: 4
|
||
related: [ai.langgraph.scenario.refund-approval, ai.agent-ops.eval-quality-gate, ai.langgraph.interview]
|
||
week: [12]
|
||
order: 2
|
||
---
|
||
|
||
# 检查点、人机协作和异常恢复：checkpointer、interrupt、RetryPolicy、任务状态
|
||
|
||
> 基于 langgraph 1.2 和 langgraph-checkpoint-sqlite 编写，以官方文档为准：https://docs.langchain.com/oss/python/langgraph/persistence
|
||
|
||
## 解释说明
|
||
|
||
### 检查点（checkpointer）
|
||
|
||
编译时传入 `checkpointer`，每执行完一步（super-step）就把状态保存下来，按 `thread_id` 区分不同的会话 / 任务：
|
||
|
||
| 实现 | 用途 |
|
||
| --- | --- |
|
||
| `InMemorySaver` | 测试、演示；进程退出就没了 |
|
||
| `SqliteSaver`（`langgraph-checkpoint-sqlite`） | 本地、单机 |
|
||
| `PostgresSaver`（`langgraph-checkpoint-postgres`） | 生产 |
|
||
|
||
有了检查点就能：多轮对话记住上下文、**进程崩溃后从最近的检查点继续**（`invoke(None, config)`）、查看历史状态（`get_state_history`）、人工审批后继续。
|
||
|
||
### 人机协作（interrupt）
|
||
|
||
节点里调用 `interrupt(要给人看的数据)`，图会暂停并保存状态；人处理完后调用 `invoke(Command(resume=人的决定), config)` 继续，`interrupt()` 的返回值就是人的决定。
|
||
|
||
注意：恢复时**这个节点会从头重新执行**，`interrupt()` 之前的代码会再跑一次，所以那里不要放发邮件、扣款这类副作用操作。
|
||
|
||
### 异常恢复
|
||
|
||
| 问题 | 做法 |
|
||
| --- | --- |
|
||
| 模型超时、限流（临时错误） | 节点级 `RetryPolicy(max_attempts=3, retry_on=...)` 自动重试，带退避 |
|
||
| 重试仍失败 / 进程崩溃 | 已完成节点的结果在检查点中；修复后 `invoke(None, config)` 从失败节点继续 |
|
||
| 外部副作用（发送、扣款）可能重复 | 用业务 ID 做**幂等键**，重复执行也只生效一次 |
|
||
| 结构化输出不合法 | pydantic 校验，失败时重试一次或进入“转人工”分支，不让整个流程崩掉 |
|
||
|
||
### 任务状态
|
||
|
||
长任务要有明确的状态字段，前端和运维都靠它：`new → classified → drafted / waiting_approval → approved / rejected → sent / failed`。状态写在图的 State 里，同时同步到业务表，方便列表页查询和告警（例如“等待审批超过 24 小时”）。
|
||
|
||
## 案例
|
||
|
||
退款工单：800 元的退款需要主管审批。流程在审批节点暂停，主管第二天在后台点“同意”，服务已经重启过，但因为检查点存在 SQLite / PostgreSQL 中，用同一个 `thread_id` 就能继续；发送时进程崩溃，恢复后不会重新起草、也不会重复发送。
|
||
|
||
## 代码示例
|
||
|
||
核心片段（完整可运行版本 [examples/ticket-flow/ticket_flow.py](examples/ticket-flow/ticket_flow.py)，带 pytest 测试）：
|
||
|
||
```python
|
||
from langgraph.checkpoint.sqlite import SqliteSaver
|
||
from langgraph.types import Command, RetryPolicy, interrupt
|
||
|
||
|
||
def approve(state):
|
||
# 恢复时本节点从头执行：interrupt 之前不要有副作用
|
||
decision = interrupt({"ticket_id": state["ticket_id"], "draft": state["draft"]})
|
||
return {"status": "approved" if decision["ok"] else "rejected"}
|
||
|
||
|
||
g.add_node("draft", draft, retry_policy=RetryPolicy(max_attempts=3, retry_on=TransientError))
|
||
g.add_node("approve", approve)
|
||
|
||
cfg = {"configurable": {"thread_id": "T-1001"}}
|
||
with SqliteSaver.from_conn_string("checkpoints.sqlite") as cp:
|
||
app = g.compile(checkpointer=cp)
|
||
app.invoke(ticket, cfg) # 停在 approve
|
||
print(app.get_state(cfg).next) # ('approve',)
|
||
|
||
# ……服务重启后……
|
||
with SqliteSaver.from_conn_string("checkpoints.sqlite") as cp:
|
||
app = g.compile(checkpointer=cp)
|
||
app.invoke(Command(resume={"ok": True}), cfg) # 继续执行审批之后的节点
|
||
app.invoke(None, cfg) # 如果中途崩溃，从最近的检查点继续
|
||
```
|
||
|
||
示例运行输出（节选）：
|
||
|
||
```text
|
||
等待节点： ('approve',) 状态： waiting_approval
|
||
发送时崩溃： 进程崩溃（模拟）
|
||
崩溃后停在： ('send',) 状态： approved
|
||
最终状态： sent 事件： ['classified:refund', 'drafted', 'approved', 'sent']
|
||
```
|
||
|
||
## 易错点
|
||
|
||
- 没有传 `thread_id`：有检查点的图调用时必须提供，否则报错或者所有请求共享状态。
|
||
- 用 `InMemorySaver` 上生产：重启后所有进行中的任务丢失。
|
||
- `interrupt()` 之前做了副作用，恢复时重复执行。
|
||
- 恢复时传了新的输入而不是 `Command(resume=...)` 或 `None`：会被当成新的一轮输入。
|
||
- 重试所有异常：参数错误、权限错误重试没有意义，只重试临时错误。
|
||
- 状态里放了大对象（整篇文档、二进制）：每一步都要序列化保存，检查点表迅速膨胀。放 ID，内容存在别处。
|
||
|
||
## 面试常问
|
||
|
||
- 检查点保存的是什么？什么时候保存？
|
||
- interrupt 恢复时为什么节点会重新执行？会带来什么问题？
|
||
- Agent 执行到一半服务重启了，怎么保证任务不丢、不重复？
|