Home > Wiki > Toolchain > Langgraph > Concepts > Interrupts

Interrupts

Interrupts enable human-in-the-loop workflows in LangGraph -- pausing execution, waiting for human input, and resuming with the provided value.

interrupt() Function

The interrupt() function pauses graph execution at the current point and waits for a resume value:

from langgraph.types import interrupt

def review_node(state):
    # Pause and wait for human input
    approved = interrupt(f"Review: {state['draft']}")
    return {"approved": approved}

When hit, the graph sends an interrupt event. The workflow is checkpointed, safe for restart.

Command(resume=...)

Resume a paused graph by providing a value via Command:

from langgraph.types import Command

# Resume with True for approval:
graph.invoke(Command(resume=True), config)

# Resume with a string:
graph.invoke(Command(resume="Looks good, proceed"), config)

# Resume with structured data:
graph.invoke(Command(resume={"action": "edit", "edits": [...]}), config)

Patterns

Approval

def generate_email(state):
    draft = llm.invoke(state["prompt"])
    result = interrupt(f"Approve draft?\n\n{draft}")
    return {"email": draft, "sent": bool(result)}

# User reviews, then:
graph.invoke(Command(resume=True), config)

Tool Review

def tool_executor(state):
    tools = state["planned_tools"]

    for tool_call in tools:
        confirm = interrupt(f"Execute {tool_call['name']} with args {tool_call['args']}?")
        if confirm:
            result = execute(tool_call)
            state["results"].append(result)

    return {"results": state["results"]}

Editing State

def draft_blog(state):
    draft = write_post(state["topic"])
    edits = interrupt(f"Draft ready. Edit?\n{draft}")
    if edits:
        return {"draft": apply_edits(draft, edits)}
    return {"draft": draft}

# Resume with edits:
graph.invoke(Command(resume="Fix typo in paragraph 2"), config)

Multiple Interrupts in Sequence

A single node can call interrupt() multiple times:

def multi_step(state):
    step1 = interrupt("Step 1 input:")
    state["step1"] = step1

    step2 = interrupt("Step 2 input:")
    state["step2"] = step2

    return state

# Resume step by step:
graph.invoke(Command(resume="value_1"), config)
graph.invoke(Command(resume="value_2"), config)

Each invoke with Command resumes past the next interrupt() call.

Dynamic Interrupts

Use conditional logic to decide whether to interrupt:

def conditional_review(state):
    if state.get("risk_score", 0) > 0.7:
        action = interrupt(f"High risk ({state['risk_score']}). Override?")
        if not action:
            return {"abort": True}

    return process(state)

Interrupts can be skipped entirely, triggered conditionally, or called with different prompts based on state.

Interrupts in Subgraphs

interrupt() works inside subgraphs. The interrupt propagates up to the parent:

# Subgraph:
subgraph_builder = StateGraph(SubState)
subgraph_builder.add_node("check", lambda s: {"result": interrupt("Confirm?")})

# Parent resumes the subgraph interrupt:
graph.invoke(Command(resume=True), config)

The parent manages the checkpoint and resume since the subgraph shares the parent's thread.

Combining Interrupts with Checkpointers

Interrupts rely on the checkpointer. The graph state is saved when interrupt() fires:

builder = StateGraph(State)
builder.add_node("review", review_node)
graph = builder.compile(checkpointer=SqliteSaver.from_conn_string("db.sqlite"))

config = {"configurable": {"thread_id": "workflow-1"}}
graph.invoke(inputs, config)
# Hits interrupt, checkpoint saved

# Resume from any client, any time:
graph.invoke(Command(resume=True), config)

Important: Code Before interrupt() Re-executes

On resume, execution restarts from the beginning of the node. Any code before the first interrupt() call re-executes:

def bad_pattern(state):
    # This line RE-EXECUTES on resume!
    state["audit_log"].append({"action": "review_started", "time": now()})

    approved = interrupt("Approve?")
    return {"approved": approved}

def good_pattern(state):
    approved = interrupt("Approve?")
    # This line only runs once (after resume):
    state["audit_log"].append({"action": "review_completed", "time": now()})
    return {"approved": approved}

Rule: Put interrupt() as early as possible in the node. Keep pre-interrupt code minimal and idempotent.

Related: Persistence, Durable Execution, Memory