Human-in-the-Loop Agents with LangGraph Interrupts

What Are Human-in-the-Loop Agents?

Human-in-the-loop (HITL) agents are AI systems that can pause their execution to request and incorporate human judgment, approval, or additional input. This pattern is essential when building reliable and safe autonomous agents that operate in high-stakes environments. LangGraph, a stateful orchestration framework from the LangChain ecosystem, provides first-class support for HITL workflows through its interrupts mechanism. An interrupt allows a graph node to suspend execution, persist the agent’s state, and wait for external input before resuming.

Unlike traditional “fire-and-forget” agent loops, a LangGraph graph with interrupts can stop at predefined checkpoints, expose the current state to a human operator, and then continue from exactly where it left off. This makes it ideal for tasks like approval workflows, form filling, debugging, and any scenario where the agent must align with human preferences or policies.

Why Human-in-the-Loop Matters

• Safety and oversight: In domains like healthcare, finance, or legal advice, agents must not act unilaterally on sensitive decisions. Human approval ensures ethical boundaries and reduces risk.
• Compliance and auditing: Regulated industries require a verifiable record of human decisions. Interrupts create a clear audit trail of exactly when and why a human intervened.
• Accuracy and edge cases: Agents can struggle with ambiguous queries or rare scenarios. A human can provide missing context, correct misconceptions, or supply domain-specific knowledge that the model lacks.
• Interactive user experiences: Chatbots and assistants can ask clarifying questions mid-flow, turning a one-shot interaction into a collaborative conversation.

How LangGraph Implements Interrupts

LangGraph graphs are built from nodes (Python functions) connected by edges. A special function interrupt() can be called inside any node. When invoked, the graph halts, the current state is saved via a checkpointer, and control returns to the calling code. The developer can then inspect the state, collect human input, and resume execution by passing a Command object with the required data.

Key components:

• Checkpointer: Persists graph state (e.g., in memory, SQLite, or Postgres) so progress isn’t lost.
• interrupt(): A function that raises an internal signal to pause execution. It can take a message string to inform the human operator.
• Command: A resume instruction that carries the human-provided data and tells the graph where to continue.

Setting Up Your Environment

Install the required packages. You need langgraph, langchain-core, and optionally a model provider like langchain-openai if your agent uses an LLM.

pip install langgraph langchain-core langchain-openai

We’ll use an in-memory checkpointer for simplicity, but you can swap in SqliteSaver or PostgresSaver for production persistence.

Defining a State Graph with Interrupts

Let’s build a simple travel approval agent. The agent collects a travel request from the user, then pauses for a manager’s approval before booking the trip. The state holds the request details and an approval flag.

1. Define the State Schema

We use a TypedDict or Pydantic model to define the shape of the state that flows through nodes.

from typing import TypedDict, Optional
from langgraph.graph import StateGraph

class TravelState(TypedDict):
    destination: str
    budget: float
    approved: Optional[bool]  # None until human decides
    booking_status: str       # will be set after approval

2. Create Node Functions

We need nodes for: collecting user input, requesting approval (the interrupt point), and booking the trip.

def collect_request(state: TravelState, user_input: str = None):
    """Simulate receiving user request. In real app, this might parse a message."""
    # For demo, we set destination and budget directly
    return {"destination": "Paris", "budget": 2000}

def human_approval(state: TravelState):
    """Interrupt to wait for manager approval."""
    from langgraph.errors import interrupt
    # Show what needs approval
    approval_message = f"Approve travel to {state['destination']} with budget {state['budget']}?"
    # Pause execution here – control returns to the caller
    decision = interrupt(approval_message)
    # Once resumed, decision contains the human's input (True/False)
    return {"approved": decision}

def book_trip(state: TravelState):
    if state.get("approved"):
        # Simulate booking
        return {"booking_status": "Booked successfully"}
    else:
        return {"booking_status": "Booking cancelled"}

3. Assemble the Graph

Add nodes and edges. The human_approval node is where the interrupt occurs. After it resumes, we conditionally branch based on the approval decision.

builder = StateGraph(TravelState)

builder.add_node("collect_request", collect_request)
builder.add_node("human_approval", human_approval)
builder.add_node("book_trip", book_trip)

# Flow: collect → approval → book
builder.add_edge("collect_request", "human_approval")
builder.add_edge("human_approval", "book_trip")

# Set entry point
builder.set_entry_point("collect_request")

# Compile with a checkpointer (required for interrupts)
from langgraph.checkpoint.memory import MemorySaver
graph = builder.compile(checkpointer=MemorySaver())

Handling Interrupts in the Invocation Loop

When you invoke the compiled graph, execution will proceed until the interrupt() call. LangGraph then raises an Interrupt exception (or returns an interrupt event when streaming). Your outer loop must catch that exception, present the interrupt message to the human, collect their response, and resume with a Command.

Basic Invocation with Exception Handling

config = {"configurable": {"thread_id": "travel-001"}}  # thread_id ties to state persistence

# First invocation – runs until interrupt
try:
    result = graph.invoke({"destination": "", "budget": 0, "approved": None, "booking_status": ""}, config)
    # If no interrupt, result would be final state
    print("Final result:", result)
except Exception as e:
    # Interrupts are wrapped; check for Interrupt type
    from langgraph.errors import Interrupt
    if isinstance(e.__cause__, Interrupt):
        interrupt_event = e.__cause__
        print("Agent paused:", interrupt_event.args[0])  # The approval message
        # Now collect human input...

In practice, you’ll use the exception handling to display the message and then resume.

Resuming with Human Input Using Command

To resume, you create a Command object with the resume argument set to the value that interrupt() should return. You pass this command back to graph.invoke() using the same config (same thread_id).

from langgraph.types import Command

# After catching the interrupt, suppose we got human approval as True
human_decision = True  # This could come from a UI prompt

# Create resume command
resume_cmd = Command(resume=human_decision)

# Resume the graph from where it stopped
final_state = graph.invoke(resume_cmd, config)
print("Final state after resume:", final_state)
# Output will show approved=True and booking_status='Booked successfully'

Full Interactive Loop Example

This snippet demonstrates a complete pattern: invoke, catch interrupt, ask user via console, resume.

import sys
from langgraph.errors import Interrupt
from langgraph.types import Command

config = {"configurable": {"thread_id": "travel-002"}}

# Initial input (empty state)
initial_state = {"destination": "", "budget": 0, "approved": None, "booking_status": ""}

while True:
    try:
        result = graph.invoke(initial_state, config)
        print("Flow completed without interrupt:", result)
        break  # done
    except Exception as e:
        if isinstance(e.__cause__, Interrupt):
            interrupt_event = e.__cause__
            message = interrupt_event.args[0]
            print("⏸️  Agent paused:", message)
            # Get human input from terminal (simulate UI)
            user_resp = input("Your response (True/False): ").strip()
            if user_resp.lower() == "true":
                decision = True
            else:
                decision = False
            # Resume
            resume_cmd = Command(resume=decision)
            # Replace initial_state with the command for next invoke
            initial_state = resume_cmd
            # Loop continues, now graph.invoke(Command(...), config) resumes
        else:
            raise

The loop is necessary because after resume, the graph might hit another interrupt (if there are multiple). For a single interrupt, one catch-and-resume suffices.

Best Practices

• Use descriptive interrupt messages: Pass a clear, actionable string to interrupt() so the human knows exactly what is needed.
• Validate human input before resuming: Sanitize and cast the response (e.g., bool, string) to avoid unexpected types inside the graph.
• Always use a checkpointer: Without persistence, the state cannot be saved across the interrupt boundary, and resuming will fail.
• Leverage thread_id for session management: Use unique thread IDs per user/conversation to keep states isolated.
• Set timeouts for human response: In production, you might want to escalate or cancel if no input arrives within a timeframe. Implement this at the orchestration layer.
• Limit interrupt frequency: Don’t interrupt for trivial confirmations; batch decisions where possible to avoid frustrating the user.
• Test resume paths thoroughly: Ensure that after resume, the graph state updates correctly and downstream nodes behave as expected.
• Handle nested interrupts carefully: If your graph has multiple potential interrupt nodes, design your loop to handle them sequentially. LangGraph will pause at each one in order.

Conclusion

Human-in-the-loop agents built with LangGraph interrupts give you precise control over when and how humans intervene in an autonomous workflow. By combining interrupt(), persistent checkpoints, and the Command resume pattern, you can create agents that are both powerful and safe. Whether you’re building a financial approval dashboard, a medical triage assistant, or an interactive debugger, LangGraph’s interrupt mechanism turns a monolithic agent into a collaborative, auditable system. Start with the simple patterns shown here, adopt the best practices, and you’ll be well on your way to robust human-AI partnerships.