Human-in-the-Loop With interrupt()

Pause, ask, resume

Some actions should not run without a person: issuing a refund, sending an external email, deleting data. Inside a node, calling interrupt(value) pauses the run, saves state in the checkpointer and surfaces value (the question or the proposed action) to the caller. Later you resume the same thread with Command(resume=answer): the node continues from the interrupt call with the answer as its return value. This needs a checkpointer and a thread_id. Because the pause survives restarts, approval can arrive minutes or days later through a web UI, Slack message or email. Show the human exactly what will happen, and record who approved it.

Pause, retry, stream

Interrupts pause for people, retry policies absorb glitches, and streaming shows progress.

Approve a refund, run

I ran this offline with langgraph 1.2.12 and langchain-core 1.6.6 in a Python virtual environment. No API key or model is needed because plain Python functions stand in for the model, so the output is repeatable. The first invoke stops at interrupt and returns the question in __interrupt__; get_state(...).next shows the approve node is waiting. A second invoke with Command(resume="yes") continues and the final status becomes refunded.

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

class State(TypedDict):
    amount: int
    status: str

def propose(state): return {"status": "waiting for approval"}

def approve(state):
    decision = interrupt({"question": f"Approve refund of {state['amount']}?"})   # pauses here
    return {"status": "refunded" if decision == "yes" else "rejected"}

g = StateGraph(State)
g.add_node("propose", propose); g.add_node("approve", approve)
g.add_edge(START, "propose"); g.add_edge("propose", "approve"); g.add_edge("approve", END)
app = g.compile(checkpointer=InMemorySaver())

cfg = {"configurable": {"thread_id": "refund-1"}}
first = app.invoke({"amount": 250, "status": "new"}, cfg)
print("paused with:", first["__interrupt__"][0].value)
print("next node:", app.get_state(cfg).next)
final = app.invoke(Command(resume="yes"), cfg)          # a human answers; the graph continues
print("final:", final)

Output:

paused with: {'question': 'Approve refund of 250?'}
next node: ('approve',)
final: {'amount': 250, 'status': 'refunded'}

Side effects go after the interrupt

When a node resumes, code before interrupt runs again. Put irreversible actions after the approval and keep the code before it free of side effects.

Quick check: What does interrupt() require to work?

• A checkpointer and a thread_id
• A GPU
• A second graph
• A recursion limit of one