LangGraph checkpoint 与故障恢复：崩溃、重启、幂等与重放实测

"""A three-step order workflow with visible side effects, shared by the recovery labs.
 
Every side effect is appended to LEDGER so the labs can show what ran, and how often.
"""
from typing import TypedDict
 
from langgraph.graph import END, START, StateGraph
 
LEDGER: list = []  # one entry per executed side effect
FAIL = {"ship": 0}  # node name -> remaining number of simulated crashes
 
 
class State(TypedDict, total=False):
    order_id: str
    charged: bool
    shipped: bool
    notified: bool
 
 
def charge(state: State):
    LEDGER.append(f"charge {state['order_id']}")
    return {"charged": True}
 
 
def ship(state: State):
    if FAIL["ship"] > 0:
        FAIL["ship"] -= 1
        raise RuntimeError("carrier API timeout")
    LEDGER.append(f"ship {state['order_id']}")
    return {"shipped": True}
 
 
def notify(state: State):
    LEDGER.append(f"notify {state['order_id']}")
    return {"notified": True}
 
 
def build(checkpointer=None):
    builder = StateGraph(State)
    for name, fn in [("charge", charge), ("ship", ship), ("notify", notify)]:
        builder.add_node(name, fn)
    builder.add_edge(START, "charge")
    builder.add_edge("charge", "ship")
    builder.add_edge("ship", "notify")
    builder.add_edge("notify", END)
    return builder.compile(checkpointer=checkpointer)

"""Crash in the middle of the flow: without a checkpointer vs. with one."""
from langgraph.checkpoint.memory import InMemorySaver
 
import order_flow
from order_flow import FAIL, LEDGER, build
 
 
def attempt(graph, payload, config=None):
    try:
        graph.invoke(payload, config)
        return "ok"
    except RuntimeError as exc:
        return f"crashed: {exc}"
 
 
print("A) no checkpointer: the retry starts from scratch")
FAIL["ship"] = 1
graph = build()
print("  run 1:", attempt(graph, {"order_id": "A1"}))
print("  run 2:", attempt(graph, {"order_id": "A1"}))
print("  ledger:", LEDGER)
 
print("B) with a checkpointer: resume continues at the failed node")
LEDGER.clear()
FAIL["ship"] = 1
graph = build(InMemorySaver())
config = {"configurable": {"thread_id": "order-A1"}}
print("  run 1:", attempt(graph, {"order_id": "A1"}, config))
snapshot = graph.get_state(config)
print("  saved state:", snapshot.values, "| next:", snapshot.next)
print("  resume:", attempt(graph, None, config))
print("  ledger:", LEDGER)
print("  final state:", graph.get_state(config).values)

A) no checkpointer: the retry starts from scratch
  run 1: crashed: carrier API timeout
  run 2: ok
  ledger: ['charge A1', 'charge A1', 'ship A1', 'notify A1']
B) with a checkpointer: resume continues at the failed node
  run 1: crashed: carrier API timeout
  saved state: {'order_id': 'A1', 'charged': True} | next: ('ship',)
  resume: ok
  ledger: ['charge A1', 'ship A1', 'notify A1']
  final state: {'order_id': 'A1', 'charged': True, 'shipped': True, 'notified': True}

"""Recover after the whole process dies: a SQLite checkpointer survives restarts.
 
Run without arguments, the script is the orchestrator: it launches itself twice as
separate OS processes ("crash" then "resume") that share one SQLite file.
"""
import subprocess
import sys
import tempfile
from pathlib import Path
 
CONFIG = {"configurable": {"thread_id": "order-A1"}}
 
 
def worker(phase: str, db: str) -> None:
    from langgraph.checkpoint.sqlite import SqliteSaver
 
    from order_flow import FAIL, LEDGER, build
 
    FAIL["ship"] = 1 if phase == "crash" else 0
    with SqliteSaver.from_conn_string(db) as saver:
        graph = build(saver)
        try:
            graph.invoke({"order_id": "A1"} if phase == "crash" else None, CONFIG)
            print(f"[{phase}] finished, state={graph.get_state(CONFIG).values}")
        except RuntimeError as exc:
            print(f"[{phase}] crashed: {exc}; next={graph.get_state(CONFIG).next}")
    print(f"[{phase}] side effects in this process: {LEDGER}")
 
 
def main() -> None:
    with tempfile.TemporaryDirectory() as tmp:
        db = str(Path(tmp) / "checkpoints.sqlite")
        for phase in ("crash", "resume"):
            proc = subprocess.run(
                [sys.executable, __file__, phase, db], capture_output=True, text=True, check=True
            )
            print(proc.stdout, end="")
 
 
if __name__ == "__main__":
    if len(sys.argv) == 3:
        worker(sys.argv[1], sys.argv[2])
    else:
        main()

[crash] crashed: carrier API timeout; next=('ship',)
[crash] side effects in this process: ['charge A1']
[resume] finished, state={'order_id': 'A1', 'charged': True, 'shipped': True, 'notified': True}
[resume] side effects in this process: ['ship A1', 'notify A1']

"""Pending writes: when one parallel node fails, the siblings that finished are not re-run on resume."""
import operator
from typing import Annotated, TypedDict
 
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
 
CALLS = {"fast": 0, "flaky": 0}
 
 
class State(TypedDict, total=False):
    results: Annotated[list, operator.add]
 
 
def fast(state: State):
    CALLS["fast"] += 1
    return {"results": ["fast"]}
 
 
def flaky(state: State):
    CALLS["flaky"] += 1
    if CALLS["flaky"] == 1:
        raise RuntimeError("upstream 503")
    return {"results": ["flaky"]}
 
 
builder = StateGraph(State)
builder.add_node("fast", fast)
builder.add_node("flaky", flaky)
builder.add_node("join", lambda state: {})
builder.add_edge(START, "fast")
builder.add_edge(START, "flaky")
builder.add_edge("fast", "join")
builder.add_edge("flaky", "join")
builder.add_edge("join", END)
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "t1"}}
 
try:
    graph.invoke({"results": []}, config)
except RuntimeError as exc:
    print("run 1 crashed:", exc)
result = graph.invoke(None, config)
print("resume result:", sorted(result["results"]))
print("calls:", CALLS)

run 1 crashed: upstream 503
resume result: ['fast', 'flaky']
calls: {'fast': 1, 'flaky': 2}

"""A checkpoint saves finished nodes, not half-finished ones: guard side effects with an idempotency key."""
from typing import TypedDict
 
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
 
PAYMENT_PROVIDER: dict = {}  # idempotency_key -> amount; stands in for a real payment API
CALLS = {"count": 0}
 
 
class State(TypedDict, total=False):
    order_id: str
    receipt: str
 
 
def charge_naive(state: State):
    CALLS["count"] += 1
    PAYMENT_PROVIDER[f"naive-{CALLS['count']}"] = 99  # every call creates a new charge
    if CALLS["count"] == 1:
        raise TimeoutError("response lost after the charge went through")
    return {"receipt": "ok"}
 
 
def charge_idempotent(state: State):
    CALLS["count"] += 1
    PAYMENT_PROVIDER[f"order-{state['order_id']}"] = 99  # same key: the provider deduplicates
    if CALLS["count"] == 1:
        raise TimeoutError("response lost after the charge went through")
    return {"receipt": "ok"}
 
 
def run(node, label: str) -> None:
    PAYMENT_PROVIDER.clear()
    CALLS["count"] = 0
    builder = StateGraph(State)
    builder.add_node("charge", node)
    builder.add_edge(START, "charge")
    builder.add_edge("charge", END)
    graph = builder.compile(checkpointer=InMemorySaver())
    config = {"configurable": {"thread_id": label}}
    try:
        graph.invoke({"order_id": "A1"}, config)
    except TimeoutError:
        graph.invoke(None, config)  # resume re-runs the node that failed
    print(f"{label}: node ran {CALLS['count']} times, charges at provider = {len(PAYMENT_PROVIDER)}")
 
 
run(charge_naive, "naive")
run(charge_idempotent, "idempotent")

naive: node ran 2 times, charges at provider = 2
idempotent: node ran 2 times, charges at provider = 1

"""interrupt() pauses a node; on resume the node restarts from its first line."""
from typing import TypedDict
 
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt
 
LOG: list = []
 
 
class State(TypedDict, total=False):
    decision: str
 
 
def review(state: State):
    LOG.append("before interrupt")  # runs again on resume
    answer = interrupt({"question": "Approve refund of 99?"})
    LOG.append("after interrupt")
    return {"decision": answer}
 
 
builder = StateGraph(State)
builder.add_node("review", review)
builder.add_edge(START, "review")
builder.add_edge("review", END)
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "t1"}}
 
paused = graph.invoke({}, config)
print("paused with:", paused["__interrupt__"][0].value)
print("log after pause:", LOG)
final = graph.invoke(Command(resume="approved"), config)
print("final state:", final)
print("log after resume:", LOG)

paused with: {'question': 'Approve refund of 99?'}
log after pause: ['before interrupt']
final state: {'decision': 'approved'}
log after resume: ['before interrupt', 'before interrupt', 'after interrupt']

"""Inspect checkpoint history, then replay from an earlier checkpoint (side effects run again)."""
from langgraph.checkpoint.memory import InMemorySaver
 
from order_flow import LEDGER, build
 
graph = build(InMemorySaver())
config = {"configurable": {"thread_id": "order-A1"}}
graph.invoke({"order_id": "A1"}, config)
 
history = list(graph.get_state_history(config))  # newest first
print("checkpoints (oldest first):")
for snap in reversed(history):
    print(f"  step={snap.metadata['step']:>2} next={snap.next} values={snap.values}")
 
before_ship = next(s for s in history if s.next == ("ship",))
LEDGER.clear()
graph.invoke(None, before_ship.config)
print("replay from the checkpoint before 'ship' executed:", LEDGER)

checkpoints (oldest first):
  step=-1 next=('__start__',) values={}
  step= 0 next=('charge',) values={'order_id': 'A1'}
  step= 1 next=('ship',) values={'order_id': 'A1', 'charged': True}
  step= 2 next=('notify',) values={'order_id': 'A1', 'charged': True, 'shipped': True}
  step= 3 next=() values={'order_id': 'A1', 'charged': True, 'shipped': True, 'notified': True}
replay from the checkpoint before 'ship' executed: ['ship A1', 'notify A1']

"""RetryPolicy handles transient errors inside one run; the default retry_on skips some exception types, including TimeoutError."""
from typing import TypedDict
 
from langgraph.graph import END, START, StateGraph
from langgraph.types import RetryPolicy
 
 
class State(TypedDict, total=False):
    value: str
 
 
def make_flaky(exc_type, failures: int):
    calls = {"n": 0}
 
    def flaky(state: State):
        calls["n"] += 1
        if calls["n"] <= failures:
            raise exc_type(f"transient failure #{calls['n']}")
        return {"value": "ok"}
 
    return flaky, calls
 
 
def run(label: str, exc_type, policy: RetryPolicy) -> None:
    node, calls = make_flaky(exc_type, failures=2)
    builder = StateGraph(State)
    builder.add_node("call_api", node, retry_policy=policy)
    builder.add_edge(START, "call_api")
    builder.add_edge("call_api", END)
    graph = builder.compile()
    try:
        result = graph.invoke({})
        outcome = f"succeeded with {result}"
    except Exception as exc:
        outcome = f"failed with {type(exc).__name__}"
    print(f"{label:<38} attempts={calls['n']}, {outcome}")
 
 
fast = {"initial_interval": 0.01, "jitter": False}
run("ConnectionError, default retry_on", ConnectionError, RetryPolicy(max_attempts=3, **fast))
run("TimeoutError, default retry_on", TimeoutError, RetryPolicy(max_attempts=3, **fast))
run("RuntimeError, default retry_on", RuntimeError, RetryPolicy(max_attempts=3, **fast))
run("TimeoutError, retry_on=TimeoutError", TimeoutError, RetryPolicy(max_attempts=3, retry_on=TimeoutError, **fast))
run("TimeoutError, max_attempts=2", TimeoutError, RetryPolicy(max_attempts=2, retry_on=TimeoutError, **fast))

ConnectionError, default retry_on      attempts=3, succeeded with {'value': 'ok'}
TimeoutError, default retry_on         attempts=1, failed with TimeoutError
RuntimeError, default retry_on         attempts=1, failed with RuntimeError
TimeoutError, retry_on=TimeoutError    attempts=3, succeeded with {'value': 'ok'}
TimeoutError, max_attempts=2           attempts=2, failed with TimeoutError

"""Configuration mistakes that make recovery silently fail or fail loudly."""
from langgraph.checkpoint.memory import InMemorySaver
 
from order_flow import FAIL, LEDGER, build
 
 
def show(label: str, fn) -> None:
    try:
        print(f"{label}: {fn()}")
    except Exception as exc:
        print(f"{label}: {type(exc).__name__}: {str(exc).splitlines()[0]}")
 
 
print("1) graph compiled without a checkpointer")
plain = build()
show("  get_state", lambda: plain.get_state({"configurable": {"thread_id": "t"}}))
 
print("2) checkpointer present, but no thread_id in the config")
saved = build(InMemorySaver())
show("  invoke", lambda: saved.invoke({"order_id": "A1"}))
 
print("3) resume with a different thread_id starts a new run")
FAIL["ship"] = 1
saved = build(InMemorySaver())
show("  first run (thread order-A1)", lambda: saved.invoke({"order_id": "A1"}, {"configurable": {"thread_id": "order-A1"}}))
FAIL["ship"] = 0
show("  resume on thread order-A1-retry", lambda: saved.invoke(None, {"configurable": {"thread_id": "order-A1-retry"}}))
show("  re-send the input on thread order-A1-retry", lambda: saved.invoke({"order_id": "A1"}, {"configurable": {"thread_id": "order-A1-retry"}}))
print("  ledger:", LEDGER)

1) graph compiled without a checkpointer
  get_state: ValueError: No checkpointer set
2) checkpointer present, but no thread_id in the config
  invoke: ValueError: Checkpointer requires one or more of the following 'configurable' keys: thread_id, checkpoint_ns, checkpoint_id
3) resume with a different thread_id starts a new run
  first run (thread order-A1): RuntimeError: carrier API timeout
  resume on thread order-A1-retry: EmptyInputError: Received no input for __start__
  re-send the input on thread order-A1-retry: {'order_id': 'A1', 'charged': True, 'shipped': True, 'notified': True}
  ledger: ['charge A1', 'charge A1', 'ship A1', 'notify A1']