← Back to DevBytes

Human-in-the-Loop Agents with LangGraph Interrupts

Human-in-the-Loop Agents with LangGraph Interrupts

Building autonomous agents that can pause and ask for human guidance is essential for safe, reliable AI applications. LangGraph, the graph-based orchestration framework from LangChain, provides first-class support for human-in-the-loop workflows through its interrupt primitive. This tutorial walks you through everything you need to know to design, implement, and resume agentic workflows that involve human judgment.

What Are Human-in-the-Loop Agents?

A human-in-the-loop (HITL) agent is an AI system that can autonomously reason and act, but deliberately stops at critical junctures to request human approval, input, or correction. Instead of blindly executing a full pipeline, the agent pauses, exposes its current state, and waits for a human operator to decide how (or whether) to continue. This pattern is vital for high-stakes operations like approving financial transactions, reviewing generated code before deployment, or validating sensitive communication.

LangGraph's Interrupt Mechanism

LangGraph models agent logic as a directed graph of nodes and edges. An interrupt is a special node action that pauses graph execution and raises a GraphInterrupt exception. Unlike a crash, this exception is caught by LangGraph’s runtime, which then saves the current state to the configured checkpointer and returns control to the caller. The human (or external process) can inspect the state, make a decision, and then resume the graph from the exact point where it was interrupted.

The core function is interrupt(), which you call inside any node. It takes an optional string argument—a message describing what input is needed. Under the hood, interrupt throws a controlled exception that the graph runner interprets as a pause signal.

Why Interrupts Matter for Agentic Workflows

Getting Started: A Basic Approval Workflow

Let's build a minimal agent that drafts an email and then asks for human approval before sending it. The graph has three nodes: draft_email, review (where the interrupt happens), and send_email.

First, install the required package:

pip install langgraph langchain-core

Now define the state and graph:

import operator
from typing import TypedDict, Annotated

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

class EmailState(TypedDict):
    # The email draft that the agent produces
    draft: str
    # Human approval: True / False, set after interrupt
    approved: Annotated[bool, operator.add]

# Node: draft the email
def draft_email(state: EmailState):
    # In a real app, you'd call an LLM here
    draft_text = "Hello, this is the proposed newsletter content..."
    return {"draft": draft_text}

# Node: human review - pauses execution
def review(state: EmailState):
    # Show the draft and ask for approval
    decision = interrupt(f"Please review this draft:\n\n{state['draft']}\n\nApprove? (yes/no)")
    # The graph will pause here and wait.
    # When resumed, 'decision' will contain the human's response.
    approved = decision.strip().lower() == "yes"
    return {"approved": approved}

# Node: send the email (only if approved)
def send_email(state: EmailState):
    if state.get("approved"):
        print("Sending email... (simulated)")
    else:
        print("Email cancelled by human.")
    return state

# Build the graph
builder = StateGraph(EmailState)
builder.add_node("draft_email", draft_email)
builder.add_node("review", review)
builder.add_node("send_email", send_email)

builder.set_entry_point("draft_email")
builder.add_edge("draft_email", "review")
builder.add_edge("review", "send_email")
builder.add_edge("send_email", END)

# Use MemorySaver to persist state across pauses
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

The key line is decision = interrupt(...) inside the review node. When execution reaches this point, the graph halts, saves the state, and returns an Interrupt object. The human sees the message, provides input, and we resume with that input.

Resuming After an Interrupt

There are two common ways to resume a paused graph: using a Command object or calling update_state. Both require the same thread_id that was used during the initial invocation.

Approach 1: Resume with Command (recommended for most cases)

# Initial invocation – will pause at the interrupt
config = {"configurable": {"thread_id": "email-session-1"}}
try:
    graph.invoke({"draft": "", "approved": False}, config)
except GraphInterrupt:
    # Normal: the graph paused and waits for human input
    pass

# Now the human provides input and we resume
resume_value = "yes"  # this will be returned by interrupt()
graph.invoke(Command(resume=resume_value), config)

Approach 2: Resume via update_state (useful when you want to modify the state before resuming)

# After the graph paused, inspect the current state
current_state = graph.get_state(config)
print("Pending interrupt:", current_state.values)

# Update the state and specify which node to resume from
graph.update_state(
    config,
    {"approved": True},
    as_node="review"
)
# Now the graph continues from the review node, with the interrupt
# call returning the value you set in the state (or using the resume argument).

In both cases, the interrupt() call inside review receives the human's input (or the value you set) and execution proceeds to send_email.

Dynamic Routing Based on Human Input

Often you don't just need a yes/no approval; you want the human to choose the next step. LangGraph interrupts let you return a string that can be used for routing.

from typing import Literal

class RouterState(TypedDict):
    query: str
    next_action: Literal["search", "summarize", "human_fallback"]

def ask_human(state: RouterState):
    choice = interrupt(
        f"Query: {state['query']}\n"
        "Choose next action: 'search', 'summarize', or 'human_fallback'"
    )
    return {"next_action": choice.strip()}

def route_after_human(state: RouterState):
    return state["next_action"]

# Build graph with conditional edge after human node
builder = StateGraph(RouterState)
builder.add_node("ask_human", ask_human)
builder.add_node("search", search_node)
builder.add_node("summarize", summarize_node)
builder.add_node("human_fallback", human_fallback_node)

builder.set_entry_point("ask_human")
builder.add_conditional_edges("ask_human", route_after_human, {
    "search": "search",
    "summarize": "summarize",
    "human_fallback": "human_fallback"
})
builder.add_edge("search", END)
builder.add_edge("summarize", END)
builder.add_edge("human_fallback", END)

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

When the graph pauses at ask_human, the human sees the prompt and responds with one of the three keywords. The route_after_human function reads the returned value from state and directs the graph to the appropriate node. This pattern is incredibly powerful for building agents that can gracefully fall back to human operators.

Handling Multiple Interrupts in a Single Run

A single graph can contain more than one interrupt. For example, you might ask for approval of a plan, then later ask for approval of the final output. Each interrupt pauses the graph independently. You resume by sending a Command(resume=...) or update_state targeting the paused node.

LangGraph keeps track of which node is currently waiting. If you call graph.get_state(config), you'll see a next tuple pointing to the node that contains the interrupt(). This makes it easy to build UIs that display exactly what information the agent is waiting for.

Best Practices

Conclusion

Human-in-the-loop agents powered by LangGraph interrupts give you precise control over when and how a human collaborates with an AI. By inserting interrupt() calls at critical decision points, you transform a linear, opaque agent into a transparent, steerable system that can earn trust and meet compliance requirements. The combination of a well-defined graph, persistent state, and the simple resume pattern makes it straightforward to build safe, interactive agentic workflows. Start with a single approval node, then expand to dynamic routing and multiple checkpoints as your confidence grows. The result is an agent that feels less like a black box and more like a reliable teammate.

— Ad —

Google AdSense will appear here after approval

← Back to all articles