Interrupts: Pausing Execution

Capsule overview

Your research agent runs 5 nodes in sequence: it breaks down the query, searches sources, analyzes findings, generates the report, and publishes it. The whole process takes 3 minutes. At step 2, the agent decides to search a papers API that charges $0.10 per query. It runs 50 queries — $5.00 that nobody approved.

With interrupt(), the agent pauses before the expensive search: "I plan to search 50 papers on Google Scholar (estimated cost: $5.00). Proceed? [yes/no/reduce]". You answer "cut it to 10." The agent adjusts and continues — $1.00 instead of $5.00, with the same quality.

interrupt() is the function that pauses your graph mid-execution. The graph saves its state in a checkpoint, waits for human input indefinitely, and resumes when the human answers. It's the foundation of every HITL pattern you'll implement in this module.

In the previous capsule you understood why HITL matters and when to apply it. Now you learn the how: the mechanics of interrupt(), Command(resume=), and the full pause/resume flow.

The API: interrupt() and Command

Two imports. That's all you need for HITL:

from langgraph.types import interrupt, Command

• interrupt(value) — pauses the graph and sends value to the caller (the human). value can be any JSON-serializable data: string, dict, list, number, boolean.
• Command(resume=value) — resumes the paused graph. value becomes the return value of interrupt() inside the node.

How interrupt() works: step by step

1. The graph runs nodes normally
   [START] → [node_A] → [node_B] → ...

2. A node calls interrupt("message for the human")
   [node_B] runs → interrupt("Do you approve?") → PAUSE

3. The graph SUSPENDS
   State saved in a checkpoint (thanks to the checkpointer)
   The interrupt value shows up in result["__interrupt__"]

4. The human sees the message and makes a decision
   This can take seconds, minutes, hours, or days

5. The human sends Command(resume=value)
   graph.invoke(Command(resume="yes"), config)

6. The node RE-RUNS from the top
   interrupt() now returns "yes" (the resume value)
   The node continues with that value

7. The graph goes on to run the remaining nodes
   [node_B] completes → [node_C] → ... → [END]

Step 6 is critical and counter-intuitive: the entire node re-runs from the top, not from the line where interrupt() was. The interrupt() returns the resume value instead of pausing. Any code before interrupt() runs again.

CRITICAL: the checkpointer is mandatory

Without a checkpointer, interrupt() doesn't work. The graph can't pause if it has nowhere to save its state.

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

Absolute rule: if you use interrupt(), configure a checkpointer. If you compile without one and a node calls interrupt(), you'll get an error. It's a technical dependency, not a recommendation.

Basic example: a full UX simulation

This is the most important example in the capsule. It shows the whole flow: the agent plans, pauses, the human decides, and the agent continues.

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


class ResearchState(TypedDict):
    query: str
    search_plan: str
    human_decision: str
    results: str
    status: str


def plan_search(state: ResearchState) -> dict:
    plan = (
        "Sources to query:\n"
        "  1. Google Scholar ($0.10/query × 10 = $1.00)\n"
        "  2. Web search (free)\n"
        "  3. ArXiv (free)\n"
        f"Total estimated cost: $1.00\n"
        f"Query: '{state['query']}'"
    )
    return {"search_plan": plan, "status": "plan_ready"}


def approve_search(state: ResearchState) -> dict:
    decision = interrupt(
        f"Search plan:\n{state['search_plan']}\n\n"
        "Proceed? [yes / no / edit]"
    )
    return {"human_decision": decision, "status": "decision_received"}


def execute_search(state: ResearchState) -> dict:
    if state["human_decision"] == "no":
        return {"results": "Search cancelled.", "status": "cancelled"}

    if state["human_decision"].startswith("edit:"):
        edited = state["human_decision"].replace("edit:", "").strip()
        return {
            "results": f"Search run with the edited plan: {edited}",
            "status": "completed_edited"
        }

    return {
        "results": f"Search completed for: '{state['query']}'. 15 results found.",
        "status": "completed"
    }


builder = StateGraph(ResearchState)
builder.add_node("plan", plan_search)
builder.add_node("approve", approve_search)
builder.add_node("execute", execute_search)
builder.add_edge(START, "plan")
builder.add_edge("plan", "approve")
builder.add_edge("approve", "execute")
builder.add_edge("execute", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "research_001"}}

print("=== STEP 1: Run until the interrupt ===")
result = graph.invoke(
    {"query": "State of the art in RAG 2025", "search_plan": "",
     "human_decision": "", "results": "", "status": ""},
    config
)
print(f"Status: {result['status']}")
for item in result.get("__interrupt__", []):
    print(f"\n--- AGENT ASKS ---\n{item.value}\n")

print("=== STEP 2: Human answers 'yes' ===")
result = graph.invoke(Command(resume="yes"), config)
print(f"Status: {result['status']}")
print(f"Results: {result['results']}")
# Expected output:
# === STEP 1: Run until the interrupt ===
# Status: plan_ready
#
# --- AGENT ASKS ---
# Search plan:
# Sources to query:
#   1. Google Scholar ($0.10/query × 10 = $1.00)
#   2. Web search (free)
#   3. ArXiv (free)
# Total estimated cost: $1.00
# Query: 'State of the art in RAG 2025'
#
# Proceed? [yes / no / edit]
#
# === STEP 2: Human answers 'yes' ===
# Status: completed
# Results: Search completed for: 'State of the art in RAG 2025'. 15 results found.

The first invoke() runs plan and approve. When approve calls interrupt(), the graph pauses. The result includes __interrupt__ with the agent's message. The second invoke() with Command(resume="yes") resumes the graph — approve gets "yes" back from interrupt(), and then execute runs normally.

Three paths from the same interrupt

The example above supports 3 answers. Each thread_id creates an independent path:

init_state = {
    "query": "RAG 2025", "search_plan": "",
    "human_decision": "", "results": "", "status": ""
}

for decision, tid in [("yes", "demo_yes"), ("no", "demo_no"), ("edit: ArXiv only", "demo_edit")]:
    cfg = {"configurable": {"thread_id": tid}}
    graph.invoke(init_state, cfg)
    r = graph.invoke(Command(resume=decision), cfg)
    print(f"'{decision}' → {r['status']}: {r['results'][:60]}")
# Expected output:
# 'yes' → completed: Search completed for: 'RAG 2025'. 15 results found
# 'no' → cancelled: Search cancelled.
# 'edit: ArXiv only' → completed_edited: Search run with the edited plan: ArXiv only

HITL isn't binary. The human can approve, reject, or modify.

What happens during the pause

When interrupt() runs, the graph freezes:

• ✅ The checkpointer saved the graph's entire state
• ✅ The node with the interrupt is suspended — later nodes haven't run
• ✅ You can inspect it with graph.get_state(config)
• ✅ You can wait minutes, hours, or days
• ❌ With MemorySaver, if the process dies → state lost
• ✅ With PostgresSaver, it survives even restarts

You can check which node is waiting:

snapshot = graph.get_state(config)
print(f"State: {snapshot.values}")
print(f"Waiting node: {snapshot.next}")  # ('approval',) if paused, () if finished

snapshot.next tells you which node is waiting for the resume. If it's an empty tuple (), the graph already finished.

Node re-execution: the most important rule

When the graph resumes, the node containing interrupt() re-runs from the top. It doesn't pick up at the exact line — the whole node runs again, and interrupt() returns the resume value instead of pausing.

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

call_counter = 0


class State(TypedDict):
    result: str


def my_node(state: State) -> dict:
    global call_counter
    call_counter += 1
    print(f"  my_node ran (time #{call_counter})")

    decision = interrupt("Continue?")
    print(f"  interrupt returned: '{decision}'")
    return {"result": decision}


builder = StateGraph(State)
builder.add_node("my_node", my_node)
builder.add_edge(START, "my_node")
builder.add_edge("my_node", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "reexec_demo"}}

print("--- First invocation (it pauses) ---")
graph.invoke({"result": ""}, config)

print("\n--- Second invocation (resume) ---")
result = graph.invoke(Command(resume="approved"), config)
print(f"\nResult: {result['result']}")
# Expected output:
# --- First invocation (it pauses) ---
#   my_node ran (time #1)
#
# --- Second invocation (resume) ---
#   my_node ran (time #2)
#   interrupt returned: 'approved'
#
# Result: approved

The node ran twice. The print before interrupt() runs both times.

Implication: any code before interrupt() must be idempotent — safe to run multiple times without problematic side effects. Don't make API calls, don't insert records, don't send emails before interrupt(). Put that after.

Interrupt values: what you can send and receive

What you pass to interrupt() is what the human sees. What the human puts in Command(resume=) is what interrupt() returns. Both must be JSON-serializable:

# Send to the human — string or dict
decision = interrupt("Do you approve this action?")
decision = interrupt({"action": "search", "cost": 5.00, "question": "Proceed?"})

# Receive from the human — any JSON-serializable type
graph.invoke(Command(resume="yes"), config)                       # string
graph.invoke(Command(resume=True), config)                        # bool
graph.invoke(Command(resume={"approved": True, "max": 2.0}), config)  # dict

Inside the node, interrupt() returns exactly what the human sent in Command(resume=).

Multiple interrupts in the same graph

A graph can have interrupts in different nodes. Each Command(resume=) advances to the next one:

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


class MultiState(TypedDict):
    plan_approved: str
    budget_approved: str
    status: str


def approve_plan(state: MultiState) -> dict:
    return {"plan_approved": interrupt("Do you approve the plan? [yes/no]")}


def approve_budget(state: MultiState) -> dict:
    return {"budget_approved": interrupt("Do you approve the budget: $5.00? [yes/no]")}


def execute(state: MultiState) -> dict:
    both_yes = state["plan_approved"] == "yes" and state["budget_approved"] == "yes"
    return {"status": "executed" if both_yes else "cancelled"}


builder = StateGraph(MultiState)
builder.add_node("approve_plan", approve_plan)
builder.add_node("approve_budget", approve_budget)
builder.add_node("execute", execute)
builder.add_edge(START, "approve_plan")
builder.add_edge("approve_plan", "approve_budget")
builder.add_edge("approve_budget", "execute")
builder.add_edge("execute", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "multi_interrupt"}}

r1 = graph.invoke({"plan_approved": "", "budget_approved": "", "status": ""}, config)
print(f"Interrupt 1: {r1['__interrupt__'][0].value}")

r2 = graph.invoke(Command(resume="yes"), config)
print(f"Interrupt 2: {r2['__interrupt__'][0].value}")

r3 = graph.invoke(Command(resume="yes"), config)
print(f"Final status: {r3['status']}")
# Expected output:
# Interrupt 1: Do you approve the plan? [yes/no]
# Interrupt 2: Do you approve the budget: $5.00? [yes/no]
# Final status: executed

The graph pauses twice. Each Command(resume=) advances to the next interrupt.

Post-approval routing with Command(goto=)

Command doesn't just resume — it also redirects. If a node returns Command(goto="node_name"), the graph jumps to that node:

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


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


def approval_gate(state: ApprovalState) -> Command[Literal["execute", "cancel"]]:
    decision = interrupt({
        "question": f"Do you approve: '{state['action']}'?",
        "options": ["approve", "reject"]
    })
    if decision == "approve":
        return Command(goto="execute")
    return Command(goto="cancel")


def execute_node(state: ApprovalState) -> dict:
    return {"status": f"Executed: {state['action']}"}


def cancel_node(state: ApprovalState) -> dict:
    return {"status": f"Cancelled: {state['action']}"}


builder = StateGraph(ApprovalState)
builder.add_node("approval", approval_gate)
builder.add_node("execute", execute_node)
builder.add_node("cancel", cancel_node)
builder.add_edge(START, "approval")
builder.add_edge("execute", END)
builder.add_edge("cancel", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

init = {"action": "Send report", "status": ""}

cfg_yes = {"configurable": {"thread_id": "route_yes"}}
graph.invoke(init, cfg_yes)
r = graph.invoke(Command(resume="approve"), cfg_yes)
print(f"Approve: {r['status']}")

cfg_no = {"configurable": {"thread_id": "route_no"}}
graph.invoke(init, cfg_no)
r = graph.invoke(Command(resume="reject"), cfg_no)
print(f"Reject: {r['status']}")
# Expected output:
# Approve: Executed: Send report
# Reject: Cancelled: Send report

The type hint Command[Literal["execute", "cancel"]] declares the valid destinations. This is cleaner than conditional edges for approval gates.

Handling unexpected answers

A while True + interrupt() creates a validation loop that re-asks until it gets a valid answer:

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


class ValidatedState(TypedDict):
    result: str


def validated_interrupt(state: ValidatedState) -> dict:
    valid = ["yes", "no", "edit"]
    prompt = "Do you approve? [yes/no/edit]"

    while True:
        response = interrupt(prompt)
        if response in valid:
            return {"result": f"Valid answer: {response}"}
        prompt = f"'{response}' isn't valid. Options: {', '.join(valid)}"

Every invalid answer re-pauses the graph with a clearer error message. The loop only ends when the input is valid.

Troubleshooting

Problem 1: "interrupt() doesn't pause — it runs to the end"

Symptom: You invoke the graph expecting it to pause, but it runs every node without stopping. Cause: You didn't pass a checkpointer when compiling. Fix: Check that builder.compile(checkpointer=checkpointer) includes the checkpointer.

Problem 2: "The node with interrupt() runs twice"

Symptom: The print() calls or side effects inside the node happen twice. Cause: Expected behavior. On resume, the node re-runs from the top. interrupt() returns the resume value instead of pausing. Fix: Put side effects after interrupt(), not before. The pre-interrupt code must be idempotent.

Problem 3: "Command(resume=) throws a thread_id error"

Symptom: Error when calling graph.invoke(Command(resume="value"), config). Cause: The config doesn't have the same thread_id used in the original invocation. Fix: Use exactly the same config dict with the same thread_id.

Problem 4: "I don't see interrupt in the result"

Symptom: The first invoke() returns without the __interrupt__ field. Cause: The graph finished before reaching the node with interrupt() — a conditional route skipped it. Fix: Check with graph.get_state(config) that snapshot.next contains the node you expect.

Problem 5: "I wrap interrupt() in try/except and it doesn't work"

Symptom: The interrupt gets "swallowed" and the node continues without pausing. Cause: interrupt() works by raising a special exception. A try/except Exception catches it. Fix: Never wrap interrupt() in a generic try/except. Use specific exceptions (except ValueError) or put the try/except after the interrupt.

Exercises

Exercise 1: Three paths — approve, reject, edit (Easy)

Build a graph with 3 nodes (prepare → approve → execute). The prepare node sets recipient = "team@company.com". The approve node uses interrupt() to ask "Send email to {recipient}? [yes / no / edit:]". The execute node checks the decision: "yes" sends, "no" cancels, "edit:X" sends to X. Use 3 thread_ids to test all 3 paths.

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


class EmailState(TypedDict):
    recipient: str
    decision: str
    status: str


def prepare(state: EmailState) -> dict:
    return {"recipient": "team@company.com", "status": "prepared"}


def approve(state: EmailState) -> dict:
    decision = interrupt(
        f"Send email to {state['recipient']}. "
        "Proceed? [yes / no / edit:<recipient>]"
    )
    return {"decision": decision}


def execute(state: EmailState) -> dict:
    d = state["decision"]
    if d == "no":
        return {"status": "cancelled"}
    if d.startswith("edit:"):
        new = d.replace("edit:", "").strip()
        return {"recipient": new, "status": f"sent to {new}"}
    return {"status": f"sent to {state['recipient']}"}


builder = StateGraph(EmailState)
builder.add_node("prepare", prepare)
builder.add_node("approve", approve)
builder.add_node("execute", execute)
builder.add_edge(START, "prepare")
builder.add_edge("prepare", "approve")
builder.add_edge("approve", "execute")
builder.add_edge("execute", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

init = {"recipient": "", "decision": "", "status": ""}

for decision, tid in [("yes", "ex1_yes"), ("no", "ex1_no"), ("edit:boss@company.com", "ex1_edit")]:
    cfg = {"configurable": {"thread_id": tid}}
    graph.invoke(init, cfg)
    r = graph.invoke(Command(resume=decision), cfg)
    print(f"'{decision}' → {r['status']}")

# Expected output:
# 'yes' → sent to team@company.com
# 'no' → cancelled
# 'edit:boss@company.com' → sent to boss@company.com

Exercise 2: Multiple sequential interrupts (Medium)

Build a purchase graph with 3 nodes, each with its own interrupt: confirm_item ("Confirm: Laptop?"), confirm_payment ("Pay $999?"), confirm_shipping ("Ship to Mexico City?"). Invoke and resume 3 times. Check the final result.

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


class PurchaseState(TypedDict):
    item: str
    price: float
    city: str
    item_ok: bool
    payment_ok: bool
    shipping_ok: bool
    status: str


def confirm_item(state: PurchaseState) -> dict:
    r = interrupt(f"Confirm the item: {state['item']}? [yes/no]")
    return {"item_ok": r == "yes"}


def confirm_payment(state: PurchaseState) -> dict:
    if not state["item_ok"]:
        return {"status": "cancelled_at_item"}
    r = interrupt(f"Confirm the payment: ${state['price']:.2f}? [yes/no]")
    return {"payment_ok": r == "yes"}


def confirm_shipping(state: PurchaseState) -> dict:
    if not state["payment_ok"]:
        return {"status": "cancelled_at_payment"}
    r = interrupt(f"Ship to: {state['city']}? [yes/no]")
    ok = r == "yes"
    return {"shipping_ok": ok, "status": "completed" if ok else "cancelled_at_shipping"}


builder = StateGraph(PurchaseState)
builder.add_node("confirm_item", confirm_item)
builder.add_node("confirm_payment", confirm_payment)
builder.add_node("confirm_shipping", confirm_shipping)
builder.add_edge(START, "confirm_item")
builder.add_edge("confirm_item", "confirm_payment")
builder.add_edge("confirm_payment", "confirm_shipping")
builder.add_edge("confirm_shipping", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "ex2_purchase"}}
init = {
    "item": "Laptop Pro", "price": 999.00, "city": "Mexico City",
    "item_ok": False, "payment_ok": False, "shipping_ok": False, "status": ""
}

r1 = graph.invoke(init, config)
print(f"1: {r1['__interrupt__'][0].value}")

r2 = graph.invoke(Command(resume="yes"), config)
print(f"2: {r2['__interrupt__'][0].value}")

r3 = graph.invoke(Command(resume="yes"), config)
print(f"3: {r3['__interrupt__'][0].value}")

r4 = graph.invoke(Command(resume="yes"), config)
assert r4["status"] == "completed"
print(f"\n✅ Purchase completed: {r4['status']}")
# Expected output:
# 1: Confirm the item: Laptop Pro? [yes/no]
# 2: Confirm the payment: $999.00? [yes/no]
# 3: Ship to: Mexico City? [yes/no]
#
# ✅ Purchase completed: completed

Exercise 3: Validation with a retry loop (Medium)

Build a node that asks the human for a number between 1 and 10 using interrupt() in a loop. If the value is invalid, re-ask. Test with "abc" (not a number), 50 (out of range), and 7 (valid).

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


class NumberState(TypedDict):
    chosen_number: Optional[int]


def ask_number(state: NumberState) -> dict:
    prompt = "Pick a number between 1 and 10:"

    while True:
        answer = interrupt(prompt)
        try:
            num = int(answer)
        except (ValueError, TypeError):
            prompt = f"'{answer}' isn't a number. Pick between 1 and 10:"
            continue
        if 1 <= num <= 10:
            return {"chosen_number": num}
        prompt = f"{num} is out of range. Pick between 1 and 10:"


builder = StateGraph(NumberState)
builder.add_node("ask", ask_number)
builder.add_edge(START, "ask")
builder.add_edge("ask", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "ex3_number"}}

r = graph.invoke({"chosen_number": None}, config)
print(f"Question: {r['__interrupt__'][0].value}")

r = graph.invoke(Command(resume="abc"), config)
print(f"Re-ask: {r['__interrupt__'][0].value}")

r = graph.invoke(Command(resume="50"), config)
print(f"Re-ask: {r['__interrupt__'][0].value}")

r = graph.invoke(Command(resume="7"), config)
assert r["chosen_number"] == 7
print(f"✅ Number chosen: {r['chosen_number']}")
# Expected output:
# Question: Pick a number between 1 and 10:
# Re-ask: 'abc' isn't a number. Pick between 1 and 10:
# Re-ask: 50 is out of range. Pick between 1 and 10:
# ✅ Number chosen: 7

Exercise 4: Routing with Command(goto=) — three destinations (Advanced)

Build a graph with a gate node that uses interrupt() and returns Command(goto=...). If the human answers "approve", jump to fast_track. If "reject", to review. If "escalate", to escalate. Test all 3 paths with different thread_ids.

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


class RequestState(TypedDict):
    request: str
    status: str


def gate(state: RequestState) -> Command[Literal["fast_track", "review", "escalate"]]:
    decision = interrupt({
        "request": state["request"],
        "options": ["approve", "reject", "escalate"]
    })
    routes = {"approve": "fast_track", "reject": "review", "escalate": "escalate"}
    return Command(goto=routes.get(decision, "review"))


def fast_track(state: RequestState) -> dict:
    return {"status": "approved_fast"}


def review(state: RequestState) -> dict:
    return {"status": "sent_to_review"}


def escalate(state: RequestState) -> dict:
    return {"status": "escalated"}


builder = StateGraph(RequestState)
builder.add_node("gate", gate)
builder.add_node("fast_track", fast_track)
builder.add_node("review", review)
builder.add_node("escalate", escalate)
builder.add_edge(START, "gate")
builder.add_edge("fast_track", END)
builder.add_edge("review", END)
builder.add_edge("escalate", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

init = {"request": "Production access", "status": ""}

for decision, tid, expected in [
    ("approve", "ex4_a", "approved_fast"),
    ("reject", "ex4_r", "sent_to_review"),
    ("escalate", "ex4_e", "escalated"),
]:
    cfg = {"configurable": {"thread_id": tid}}
    graph.invoke(init, cfg)
    r = graph.invoke(Command(resume=decision), cfg)
    assert r["status"] == expected
    print(f"'{decision}' → {r['status']}")

print("\n✅ Routing with 3 destinations verified")
# Expected output:
# 'approve' → approved_fast
# 'reject' → sent_to_review
# 'escalate' → escalated
#
# ✅ Routing with 3 destinations verified

Summary

• interrupt(value) pauses the graph and sends value to the human. The graph saves its state in a checkpoint and waits indefinitely. It's the foundation of all HITL in LangGraph
• Command(resume=value) resumes the graph. The value becomes the return value of interrupt() inside the node. It can be any JSON-serializable data
• The checkpointer is mandatory. Without one, interrupt() can't save state. If you use interrupt(), configure MemorySaver (dev) or PostgresSaver (prod)
• The node re-runs from the top on resume. The code before interrupt() runs again. Pre-interrupt code must be idempotent. Side effects go after the interrupt
• A graph can have multiple interrupts in different nodes. Each Command(resume=) advances to the next interrupt
• Command(goto=) enables post-approval routing. The node dynamically decides which node to jump to based on the human decision
• Validation with retry: a while True + interrupt() creates loops that re-ask until they get a valid answer

Next capsule: Approval Gates: Validate Before Acting — the most common HITL pattern. You'll implement gates that ask permission before expensive actions, with cost estimation, approval levels, and conditional routing.

Additional resources

1. LangGraph — Interrupts — Official documentation for interrupt() and Command(resume=): flow, rules, and anti-patterns
2. interrupt() API Reference — API reference with signature, parameters, and examples
3. How to add human-in-the-loop — Practical guide with approval, review, and feedback patterns
4. LangGraph — Persistence — Documentation on the checkpointing that makes interrupts possible
5. How to wait for user input — Specific pattern for waiting on human input during execution

Module 9 — LangChain & LangGraph: From Chains to Agents