How do you implement human-in-the-loop approvals using persistence and resumption?

What you need to know

1from typing import TypedDict2from langgraph.graph import StateGraph, START, END3from langgraph.checkpoint.memory import InMemorySaver4from langgraph.types import interrupt, Command56class Refund(TypedDict):7    order_id: str8    amount_inr: int9    status: str1011def approve(state: Refund) -> dict:12    if state["amount_inr"] <= 5000:13        return {"status": "auto_approved"}14    decision = interrupt({"order_id": state["order_id"],15                          "amount_inr": state["amount_inr"],16                          "question": "Approve this refund?"})17    return {"status": "approved" if decision["ok"] else "rejected"}1819def pay(state: Refund) -> dict:20    if state["status"] in ("approved", "auto_approved"):21        print(f"paying {state['order_id']} with key refund-{state['order_id']}")22    return {}2324b = StateGraph(Refund)25b.add_node("approve", approve)26b.add_node("pay", pay)27b.add_edge(START, "approve"); b.add_edge("approve", "pay"); b.add_edge("pay", END)28graph = b.compile(checkpointer=InMemorySaver())2930cfg = {"configurable": {"thread_id": "refund-OD-88123"}}31first = graph.invoke({"order_id": "OD-88123", "amount_inr": 8500, "status": "new"}, cfg)32print(first["__interrupt__"][0].value)           # show this to the reviewer33print(graph.invoke(Command(resume={"ok": True}), cfg))

This runs as written.

Patterns

• Approve or reject — as above.
• Edit — the reviewer returns corrected values, e.g. {"ok": True, "amount_inr": 6000}.
• Review a tool call — an interrupt before the tool node shows the tool name and arguments.
• Ask the user — the same mechanism collects missing information.

For create_agent, HumanInTheLoopMiddleware(interrupt_on={"issue_refund": True}) pauses before that tool, and you resume with Command(resume={"decisions": [{"type": "approve"}]}); other decision types are edit, reject and respond.

Rules

• Payloads must be JSON-serialisable.
• Do not wrap interrupt() in try/except Exception: it works by raising a special exception, and a broad except swallows the pause.
• Several interrupts in one node are matched by order; parallel interrupts can be resumed together with a dict of {interrupt_id: value}.
• interrupt_before=[...] at compile time is a static breakpoint, useful for debugging, not for approvals.

A real-life example

A marketplace's refund agent auto-approves refunds up to Rs 5,000 and pauses above that. The API stores the interrupt payload in an "approvals" queue in the ops dashboard. On a festival-sale weekend 1,240 refunds paused; reviewers cleared them over 30 hours, some on the Monday morning. Each click sent Command(resume={"ok": ...}) to the same thread, on whichever pod received the request, because checkpoints were in Postgres. The pay node, after the approval, sent the gateway an idempotency key refund-<order_id>, so a double-click never paid twice.

Follow-up questions to expect

• "What if the human never answers?" — The thread stays paused at no compute cost. Add an expiry job that resumes with a timeout decision, such as auto-reject with a message to the customer.
• "How is the payload shown to the reviewer?" — Your API reads __interrupt__ from the result, or get_state(config).interrupts, and renders it; with stream, the interrupt arrives as an event.
• "Why not just block and wait for input?" — A blocked process ties up memory and dies on restart. A checkpointed interrupt costs nothing while waiting and survives deploys.