AI Agents LangGraph

Interrupts

Intermediate

This post covers Interrupts in LangGraph , explaining how to pause and resume graph execution for human-in-the-loop workflows. We explore waiting for approvals, maintaining interrupt state with checkpointing, building manual decision systems, handling interrupts in multi-agent workflows, common mistakes, and best practices for creating reliable interactive AI systems.

What Are Interrupts?

Interrupts in LangGraph are a powerful mechanism that allows you to pause graph execution at specific points and wait for external input (usually from a human) before continuing. They enable Human-in-the-Loop (HITL) workflows — one of the most important patterns for building reliable, safe, and controllable AI systems. Instead of letting the agent run autonomously from start to finish, you can insert breakpoints where human review, approval, or correction is required.

Human-in-the-Loop Workflows

Interrupts are the foundation of Human-in-the-Loop systems. Common use cases include:

Content moderation
Sensitive action approval
Fact-checking
Final answer review
Debugging and correction loops

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

graph = StateGraph(AgentState)

graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
graph.add_node("human_review", human_review_node)

graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", route_after_agent)
graph.add_edge("tools", "agent")
graph.add_edge("human_review", "agent")

Pausing Graph Execution

You define where to pause during compilation:

checkpointer = MemorySaver()

app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["human_review"],     # Pause before this node
    # interrupt_after=["tools"]            # Or after this node
)

When the graph reaches an interrupt point, execution stops and returns a special state.

Waiting for Human Approval

# Run until interrupt
config = {"configurable": {"thread_id": "user_123"}}

result = app.invoke(inputs, config)

# Check if interrupted
if result.get("__interrupt__"):
    print("Waiting for human approval...")
    # Show current state to user
    print(result["messages"][-1].content)

Resuming Interrupted Workflows

After human input, you resume the graph:

# Human provides feedback
human_feedback = {
    "messages": [HumanMessage(content="Approved. Please proceed with the plan.")]
}

# Resume execution
final_result = app.invoke(human_feedback, config)

You can also update state manually before resuming:

app.update_state(config, {"approved": True, "human_notes": "Looks good"})

Interrupt State Persistence

Interrupts work seamlessly with checkpointing. The entire state is saved when interrupted.

# After interruption, you can inspect the full state
state_snapshot = app.get_state(config)

print("Current State:", state_snapshot.values)
print("Next Node:", state_snapshot.next)   # Which node is waiting

This allows you to build robust systems that can be paused for days and resumed later.

Interrupts with Checkpointing

from langgraph.checkpoint.sqlite import SqliteSaver

checkpointer = SqliteSaver.from_conn_string("checkpoints.db")

app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["human_review", "final_approval"]
)

Now every thread (


     thread_id

) maintains its own execution state, even across server restarts.

Manual Decision Systems

You can create sophisticated decision workflows:

def human_review_node(state: AgentState):
    # This node is never actually executed by the LLM
    # It acts as a breakpoint
    return {"messages": state["messages"]}   # Pass through

def route_after_review(state: AgentState):
    # Human would update state before resuming
    if state.get("approved", False):
        return "final_answer"
    else:
        return "agent"   # Send back for revision

Interrupts in Multi-Agent Systems

app = graph.compile(
    checkpointer=MemorySaver(),
    interrupt_before=[
        "research_approval", 
        "code_review", 
        "final_output"
    ]
)

This allows different human reviewers to approve different agent outputs in a multi-agent team.

Common Interrupt Mistakes

Forgetting to use a checkpointer (state is lost on interrupt)
Not handling the __interrupt__ signal properly
Placing interrupts in the wrong nodes
Not providing clear context to the human reviewer
Resuming without updating state correctly

Best Practices for Interrupts

Always use a persistent checkpointer (MemorySaver, SqliteSaver, etc.)
Place interrupts before sensitive or high-stakes nodes
Provide rich context to the human when showing the interrupt state
Design clear resumption logic
Use meaningful node names for interrupt points (human_approval, content_moderation)
Test both approval and rejection paths
Log interrupt events for auditing

Example Production Pattern:

app = graph.compile(
    checkpointer=SqliteSaver.from_conn_string("checkpoints.db"),
    interrupt_before=["human_approval"]
)

# Run
result = app.invoke(inputs, config)

if "__interrupt__" in str(result):
    # Notify human via email/Slack/UI
    notify_human(result)
    
    # Wait for human response...
    human_input = get_human_feedback()
    
    # Resume
    app.invoke(human_input, config)

Interrupts turn autonomous agents into collaborative systems, combining the speed of AI with the judgment of humans.

AI agent LangChain LangGraph Python

← All training