第 14 章 · 综合实战三：容错、审批与恢复

把第 6、10 章的机制合成一条生产形态的工单流水线：历史检索节点带重试策略，起草节点产出草稿，升级节点在批准/拒绝前暂停，用检查点跨两次调用恢复。全程离线。

机制与设计选择

fetch 节点首次调用抛 TimeoutError，RetryPolicy(max_attempts=3, retry_on=TimeoutError) 自动重试到成功；escalate 用 interrupt() 暂停，两次调用必须同一 thread_id。审批只改变状态字段，不代替人工做决定；节点恢复时会从头重放，所以幂等责任仍在写操作的实现方。注意 LangGraph 1.2 的默认 retry_on 并不覆盖 TimeoutError，必须显式指定。

示例

python

from typing import TypedDict
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.types import RetryPolicy, interrupt, Command

class State(TypedDict):
    subject: str
    stage: str
    approved: bool

fetch_attempts = {"n": 0}

def fetch_customer_history(state: State):
    fetch_attempts["n"] += 1
    if fetch_attempts["n"] == 1:
        raise TimeoutError("simulated transient failure")
    return {"stage": "history_loaded"}

def draft_reply(state: State):
    return {"stage": "drafted"}

def escalate(state: State):
    decision = interrupt({"stage": "awaiting_approval", "reason": "high-severity ticket"})
    return {"approved": bool(decision), "stage": "approved" if decision else "rejected"}

builder = StateGraph(State)
builder.add_node("fetch", fetch_customer_history,
                 retry_policy=RetryPolicy(max_attempts=3, initial_interval=0.01, retry_on=TimeoutError))
builder.add_node("draft", draft_reply)
builder.add_node("escalate", escalate)
builder.add_edge(START, "fetch")
builder.add_edge("fetch", "draft")
builder.add_edge("draft", "escalate")
builder.add_edge("escalate", END)
graph = builder.compile(checkpointer=InMemorySaver())

for thread, decision in [("T-1", True), ("T-2", False)]:
    cfg = {"configurable": {"thread_id": thread}}
    paused = graph.invoke({"subject": "severe crash", "stage": "new", "approved": False}, cfg)
    assert paused["__interrupt__"], thread
    assert paused["__interrupt__"][0].value["stage"] == "awaiting_approval", thread
    result = graph.invoke(Command(resume=decision), config=cfg)
    assert result["approved"] is decision, thread
    print(thread, "->", result["stage"], "| fetch attempts:", fetch_attempts["n"])
assert fetch_attempts["n"] >= 2
print("retry + approval + resume ok")

运行与验收

直接运行：两次线程都应完成，T-1 为 approved、T-2 为 rejected，fetch attempts 至少 2，说明重试确实发生。断言同时检查暂停点 stage 与恢复后的 approved 值，不只看最终打印。

容易踩的坑

RetryPolicy 只能覆盖你声明过的瞬时故障：1.2 的默认 retry_on 不重试 TimeoutError，漏写 retry_on 等于不重试；把认证错误也放进重试只会放大失败。interrupt 恢复必须沿用原线程，新建 thread_id 会从头开始而不是从暂停点继续。内存检查点不适合跨进程，迁数据库时保留同一套 thread_id 语义。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. 首次 fetch 失败后发生什么？

2. 恢复用哪个线程？

3. interrupt 后节点从哪开始？

4. 内存检查点适合跨进程吗？

动手实践

为 T-3 线程演示同一线程的第二次工单处理。
把重试耗尽后的错误处理器接上，落一条失败记录。
把 MemorySaver 换成 PostgreSQL 检查点，说明连接与租户隔离。

官方参考

核验日期：2026-09-17，示例已在本机离线运行验证。

第 14 章 · 综合实战三：容错、审批与恢复 ​

机制与设计选择 ​

示例 ​

运行与验收 ​

容易踩的坑 ​