Lab 03 — Human-in-the-Loop: interrupt(), Command(resume=...) & Breakpoints

Phase 18 · Lab 03 · « Phase 18 README · « Lab 02

The problem

Some steps an agent must not take alone: spend money, send an email, delete a row, run a tool call it invented. The fix is not a smarter model — it is a pause. The graph runs up to the risky point, stops, shows a human exactly what it wants to do, and waits. The human approves, edits, or rejects; the graph resumes with that answer folded in.

That pause has to be durable. The human might answer in 3 seconds or 3 days, possibly after the process has restarted. So human-in-the-loop (HITL) is built directly on the Lab 02 checkpointer: interrupt(payload) stops the run, checkpoints the state, and returns the payload to the caller. The caller resumes with Command(resume=answer); the interrupted node re-executes, and this time interrupt(...) returns answer so the node continues past it.

What you build

You extend the Lab 02 engine with dynamic and static interrupts:

File map

Run it

pytest test_lab.py -v                       # your lab.py (red until you implement)
LAB_MODULE=solution pytest test_lab.py -v   # the reference (green)
python solution.py                          # worked example

Success criteria

• A node calling interrupt(payload) pauses the run: invoke returns a dict containing "__interrupt__": [Interrupt(value=payload)], and a checkpoint records the interrupted node as next.
• invoke(Command(resume=answer), config) continues to completion, and the resumed node saw answer (i.e. interrupt(...) returned it, not the payload).
• A different resume value routes differently (approve -> tool runs; reject -> it does not).
• interrupt_before=[n] pauses before n runs; interrupt_after=[n] pauses after; both resume with invoke(None, config) or Command(...).
• interrupt() without a checkpointer raises — the pause cannot be made durable.
• The whole pause/resume flow is deterministic (resume values keyed by thread, checkpoint ids from a counter). All 20 tests pass under both lab and solution.

How this maps to the real stack

This is the real LangGraph HITL API, faithfully.

Dynamic interrupt() — pause inside a node at a data-dependent point:

from langgraph.types import interrupt, Command

def approval(state):
    decision = interrupt({"action": "approve_tool_call", "tool": state["pending"]})
    return {"approved": decision == "approve"}

# 1) run until the interrupt:
result = graph.invoke({"messages": [...]}, config)
result["__interrupt__"]        # (Interrupt(value={'action': ...}),) — show this to the human

# 2) resume with the human's answer; the node re-runs and interrupt() returns it:
graph.invoke(Command(resume="approve"), config)

Command does more than resume — it steers:

Command(resume=value)          # feed the paused interrupt() a value
Command(update={"x": 1})       # edit state (through reducers) before continuing
Command(goto="other_node")     # override where the graph goes next

Static breakpoints — approval gates declared at compile time, no node edits:

graph = builder.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["tools"],   # pause before the tools node
    interrupt_after=["planner"],  # and/or after another
)
graph.invoke(inputs, config)      # stops at the breakpoint
graph.get_state(config).next      # ('tools',) — what's pending
graph.invoke(None, config)        # approve -> continue

Why HITL needs a checkpointer. The pause must survive time (and crashes) — the human isn't sitting in the same function call. The interrupt() value and the resume answer are stored in the checkpoint, so a resume is just "load the paused checkpoint, inject the answer, re-run the node." This is the same durable-execution machinery as Phase 08 (a workflow that pauses on an external signal) and the interaction patterns of Phase 10 (human-in-the-loop). The three canonical patterns all fall out of it:

• Approve / reject — interrupt() returns a decision that a conditional edge routes on.
• Edit — interrupt() returns corrected content (or use Command(update=...) / update_state).
• Review / verify — pause to show a draft, resume once a human signs off.

Limits of the miniature. A node re-runs from the top on resume — so any side effect before the interrupt() happens twice. Real LangGraph has the same caveat: keep pre-interrupt work idempotent, or put side effects after the interrupt. Our resume-value counter matches interrupts by call order within a single node; LangGraph tracks them per task across the checkpoint more robustly, and also supports interrupts inside parallel/subgraph tasks (via contextvars), which this single-threaded model does not. The pause/persist/resume mechanism is identical.

Extensions (your own machine)

• Add an edit flow: pause showing a draft tool call, let the human return a modified call via Command(resume=edited), and run the edited version.
• Wire this to Lab 02's SqliteSaver extension and resume a paused thread from a fresh process — proving the pause is truly durable.
• Add a timeout / auto-reject: if get_state(config).next still shows the gate after N ticks of an injected clock, resume with Command(resume="reject") automatically.

Interview / resume signal

"Implemented LangGraph human-in-the-loop from scratch: a dynamic interrupt() that pauses a running graph mid-node, persists the pause to the checkpointer, and surfaces the payload; a Command(resume=...) path that re-executes the node so interrupt() returns the human's answer; plus static interrupt_before/interrupt_after breakpoints — the approve/edit/review gates that make an autonomous agent safe to ship, all durable and deterministic."