LangGraph Module 6: Human-in-the-Loop with Interrupts

Step 1: Know the requirements

Interrupts need a checkpointer, because the graph's state must be saved while it waits, and a thread_id so you can resume the right run. Compile with a checkpointer and always pass the same thread_id when resuming.

Step 2: Pause with interrupt

Call interrupt inside a node with any JSON-serializable value, such as a question and the details of the pending action. Execution stops there. When resumed, the value you resume with becomes the return value of interrupt.

from typing import Literal
from typing_extensions import TypedDict
from langgraph.types import interrupt, Command

class State(TypedDict):
    action: str
    status: str

def propose(state: State) -> dict:
    return {"action": "Delete 3 old log files"}

def approval(state: State) -> Command[Literal["execute", "cancel"]]:
    decision = interrupt(
        {"question": "Approve this action?", "action": state["action"]}
    )
    return Command(goto="execute" if decision else "cancel")

def execute(state: State) -> dict:
    return {"status": "executed"}

def cancel(state: State) -> dict:
    return {"status": "cancelled"}

Step 3: Build and run until it pauses

Wire the graph and invoke it. When an interrupt fires, the result contains an __interrupt__ entry with the value you passed, which your application shows to the user.

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver

builder = StateGraph(State)
builder.add_node("propose", propose)
builder.add_node("approval", approval)
builder.add_node("execute", execute)
builder.add_node("cancel", cancel)
builder.add_edge(START, "propose")
builder.add_edge("propose", "approval")
builder.add_edge("execute", END)
builder.add_edge("cancel", END)

graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "approval-1"}}

result = graph.invoke({"action": "", "status": ""}, config)
print(result["__interrupt__"][0].value)

Step 4: Resume with the decision

Send a Command with resume set to the person's answer, using the same thread_id. The graph continues from the interrupt, and the answer becomes the interrupt call's return value. Command(resume=...) is the only Command form meant to be used as input to invoke.

result = graph.invoke(Command(resume=True), config)
print(result["status"])   # executed

Step 5: Follow the rules for safe interrupts

When a graph resumes, the node that called interrupt starts again from its beginning, so code before the interrupt runs twice. Make side effects before an interrupt idempotent or put them after it. Do not wrap interrupt in a bare try and except, because it works by raising an exception, and do not conditionally skip or reorder interrupt calls inside a node. Prefer a conditional edge over a while loop to ask again.