Chapter 14 · Capstone 3: fault tolerance, approval, and resume

Fuse the mechanisms from chapters 6 and 10 into a production-shaped ticket pipeline: the history node carries a retry policy, the draft node produces a draft, the escalation node pauses before approval, and a checkpointer resumes across two calls. Everything runs offline.

Mechanics and design choices

The fetch node raises TimeoutError on its first attempt; RetryPolicy(max_attempts=3, retry_on=TimeoutError) retries until success. escalate uses interrupt() to pause, and both calls must share one thread_id. Approval only writes a state field; it does not decide for the human. Because the interrupted node replays from the start, idempotency responsibility stays with whoever performs the external write. Note that the default retry_on in LangGraph 1.2 does not cover TimeoutError; it must be specified explicitly.

Example

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")

Run and verify

Run directly: both threads should complete, T-1 approved, T-2 rejected, with at least two fetch attempts proving the retry happened. Assertions check the paused stage and the post-resume approved value, not just the final print.

Pitfalls

RetryPolicy only covers the transient failures you declare: in 1.2 the default retry_on does not retry TimeoutError, so omitting retry_on means no retry; retrying authentication errors amplifies outages. Resuming an interrupt requires the original thread; a new thread_id restarts from the beginning. In-memory checkpoints do not cross processes; keep the same thread_id semantics when migrating to a database checkpointer.

🧪 随堂测验

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

1. What happens after the first fetch failure?

2. Which thread resumes?

3. Where does a node restart after interrupt?

4. Is the in-memory checkpointer cross-process?

Hands-on exercises

1. Show a second ticket handled on the same thread T-3.
2. Wire an error handler for exhausted retries and record the failure.
3. Swap MemorySaver for a PostgreSQL checkpointer, explaining connections and tenant isolation.

Official references

• fault-tolerance
• interrupts
• persistence

Verified on 2026-09-17; examples were run locally without network access.