Human-in-the-Loop with LangGraph interrupt(): Designing Interrupts That Let Humans Intervene Before Agents Make Mistakes

An AI agent that handles everything autonomously may seem impressive, but the story changes fast when that agent is about to fire a DELETE query against your production database. When I first wired an agent into our team's service, I thought, "Surely it won't auto-execute all the way to this step?" — and broke into a cold sweat when it did. If you found this article by searching LangGraph HITL or langgraph human approval, you've probably had a similar experience, or you're trying to prevent one. The heart of this article is how to precisely control an AI agent approval gate at the code level.

That experience was when "Human Oversight" stopped feeling like a UX choice and started feeling like a matter of survival. It's no coincidence that EU AI Act Article 14 has started legally mandating human oversight for high-risk AI systems. Building a structure where a human can intervene just before AI makes a critical decision — that's the problem LangGraph's interrupt() solves. Let's look at how to control those intervention points precisely in code.

This article uses Python-based LangGraph examples. A JavaScript SDK exists separately, but all code examples are in Python. If you're new to LangGraph's core concepts, skimming the official documentation first will make this much easier to follow.

Core Concepts

What Is Human-in-the-Loop

HITL is a design pattern where an AI agent pauses its execution flow before performing a sensitive action and waits for a human's approval, modification, or feedback. You might think of it as "just popping up a confirmation dialog in the middle," but in LangGraph it's far more sophisticated than that.

LangGraph supports this natively via the interrupt() function. Calling interrupt(value) inside any node in the graph triggers four things in sequence:

1. Graph execution stops at that point
2. The value you passed (JSON) is saved to the Checkpointer
3. The caller receives the interruption information via the ["__interrupt__"] key in the response object
4. When the human makes a decision, execution resumes via Command(resume=<value>)

There's one important internal behavior to understand here. interrupt() uses an Exception mechanism internally to halt execution. If you don't know this and wrap the interrupt() call in a try/except, the interrupt itself won't work. It's one of the most common mistakes in practice, so it's worth flagging at the core concepts stage.

What is a Checkpointer? It's a layer that saves the entire graph execution state as a snapshot at a specific point in time. Thanks to this, even if the server restarts while a human is deliberating on an approval, execution can resume exactly from where it stopped.

Static Breakpoints vs. Dynamic interrupt()

Previously, a static breakpoint approach was used — always pausing before or after specific nodes. Since it stops at the same point unconditionally on every execution, people had to click through even when review wasn't actually needed.

interrupt() is dynamic and can be combined with conditional logic. Selective intervention like "stop only for DELETE queries, let SELECT pass through" becomes possible.

from langgraph.types import interrupt, Command
 
def sql_review_node(state):
    query = state["generated_query"]
    
    # Interrupt only for dangerous queries — conditional dynamic intervention
    if any(keyword in query.upper() for keyword in ["DELETE", "DROP", "UPDATE"]):
        decision = interrupt({
            "query": query,
            "risk_level": "HIGH",
            "message": "This is a dangerous query. Do you want to execute it?"
        })
        return {"approved": decision == "approved", "query": query}
    
    # Safe queries like SELECT are auto-approved
    return {"approved": True, "query": query}

Checkpointer Selection Guide

Choosing the right checkpointer for local vs. production is a point of frequent confusion in practice.

SQLite can be used immediately with :memory: without a DB file, making it convenient for prototyping, while PostgreSQL is essentially mandatory for real services that require server restarts and horizontal scaling.

Values You Can Pass to Command(resume=)

The value in Command(resume=<value>) can be any JSON-safe value: strings, dictionaries, lists, None, etc. You can pass not just simple strings like "approved" but also structured dictionaries like {"action": "approve", "comment": "LGTM"}. This value becomes the return value of the interrupt() call and is used in the node's internal logic.

Three Core HITL Patterns

Here are the patterns that come up most frequently in practice. Of these three, the first is by far the most commonly used.

Practical Application

Example 1: SQL Execution Agent — Stopping Only Dangerous Queries (Action Approval Pattern)

Honestly, I think this is the most realistic trigger for many teams adopting HITL. Cases where data was lost because an LLM-generated query was executed as-is are more common than you'd think. I personally once wasted 30 minutes because I was creating a new thread_id on every request and couldn't resume at all. The key point of this example is that you must use the same thread_id when resuming.

from langgraph.graph import StateGraph, END
from langgraph.types import interrupt, Command
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_anthropic import ChatAnthropic
from typing import TypedDict, Optional
 
class SQLAgentState(TypedDict):
    user_request: str
    generated_query: str
    approved: bool
    result: Optional[str]
 
llm = ChatAnthropic(model="claude-sonnet-4-6")
 
def generate_query_node(state: SQLAgentState):
    """Generate SQL query with LLM"""
    response = llm.invoke(
        f"Please generate a SQL query for the following request: {state['user_request']}"
    )
    return {"generated_query": response.content.strip()}
 
def review_node(state: SQLAgentState):
    """Request DBA review for dangerous queries"""
    query = state["generated_query"]
    dangerous_keywords = ["DELETE", "DROP", "UPDATE", "TRUNCATE"]
    
    if any(kw in query.upper() for kw in dangerous_keywords):
        # Never wrap interrupt() in try/except
        # It uses an exception internally to halt execution — catching it breaks the mechanism
        decision = interrupt({
            "query": query,
            "risk": "HIGH",
            "tip": "It's a good idea to first check the number of affected rows with a SELECT"
        })
        return {"approved": decision == "approved"}
    
    return {"approved": True}
 
def execute_node(state: SQLAgentState):
    """Execute only approved queries"""
    if not state["approved"]:
        return {"result": "Query was rejected. Please modify and try again."}
    return {"result": f"Execution complete: {state['generated_query']}"}
 
# Graph construction
builder = StateGraph(SQLAgentState)
builder.add_node("generate", generate_query_node)
builder.add_node("review", review_node)
builder.add_node("execute", execute_node)
builder.set_entry_point("generate")
builder.add_edge("generate", "review")
builder.add_edge("review", "execute")
builder.add_edge("execute", END)
 
# SQLite checkpointer (for local development — replace with AsyncPostgresSaver for production)
checkpointer = SqliteSaver.from_conn_string(":memory:")
graph = builder.compile(checkpointer=checkpointer)
 
# Recommended: generate thread_id as UUID and map 1:1 to user session
# You must use the same thread_id when resuming
import uuid
thread_id = str(uuid.uuid4())
config = {"configurable": {"thread_id": thread_id}}
 
result = graph.invoke({"user_request": "clean up old users"}, config=config)
 
# If the __interrupt__ key is present, human judgment is required
if "__interrupt__" in result:
    print("Awaiting approval:", result["__interrupt__"])
    
    # When the DBA approves, resume with the same config (same thread_id!)
    final = graph.invoke(Command(resume="approved"), config=config)
    print("Final result:", final["result"])

Example 2: Travel Booking Agent — Only Book After Plan Confirmation (State Editing Pattern)

This is a pattern where the agent researches flights and hotels, presents the plan to the user, and only calls the actual booking API once approved. The key is that no external calls happen before approval.

from langgraph.types import interrupt, Command
from typing import TypedDict, Optional
 
class TravelState(TypedDict):
    destination: str
    travel_plan: Optional[dict]
    booking_confirmed: bool
 
def research_node(state: TravelState):
    """Research flights and hotels (read-only — safe)"""
    plan = {
        "flight": "ICN → NRT, 2026-05-10, KRW 280,000",
        "hotel": "Shinjuku Grand Hotel, 3 nights KRW 450,000",
        "total": "KRW 730,000"
    }
    return {"travel_plan": plan}
 
def approval_node(state: TravelState):
    """Show plan to user and await approval or modification"""
    response = interrupt({
        "message": "Please review your travel plan. You can modify or approve it.",
        "plan": state["travel_plan"],
        "options": ["approve", "edit", "cancel"]
    })
    
    if response["action"] == "edit":
        return {
            "travel_plan": {**state["travel_plan"], **response["edits"]},
            "booking_confirmed": False
        }
    elif response["action"] == "approve":
        return {"booking_confirmed": True}
    else:
        return {"booking_confirmed": False}
 
def booking_node(state: TravelState):
    """Call actual booking API only for approved plans"""
    if not state["booking_confirmed"]:
        return {"result": "Booking was cancelled."}
    return {"result": f"Booking complete! {state['travel_plan']}"}

Unlike Example 1, response here is a dictionary structure. When resuming, pass Command like this:

# When the user chooses to approve
final = graph.invoke(
    Command(resume={"action": "approve"}),
    config=config
)
 
# When the user chooses to edit
final = graph.invoke(
    Command(resume={"action": "edit", "edits": {"hotel": "Shibuya Hotel, 3 nights KRW 380,000"}}),
    config=config
)

Key point of the State Editing pattern: You can receive user edits as the return value of interrupt() and apply them directly to the state. Beyond simple approve/reject, this naturally supports flows where humans edit data.

Example 3: Production Checkpointer Setup (PostgreSQL)

SQLite is sufficient for local development, but real services need persistent storage that survives server restarts.

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
import os
import asyncio
 
async def setup_production_graph():
    # Always manage credentials via environment variables. Never hardcode them.
    conn_string = os.environ.get("DATABASE_URL")
    
    async with AsyncPostgresSaver.from_conn_string(conn_string) as checkpointer:
        # Create tables once on first run
        await checkpointer.setup()
        
        graph = builder.compile(checkpointer=checkpointer)
        
        config = {"configurable": {"thread_id": "prod-task-42"}}
        result = await graph.ainvoke(
            {"user_request": "..."},
            config=config
        )
        return result

Pros and Cons Analysis

Advantages

Disadvantages and Caveats

The items in this table are things that are quite painful to encounter in production. Knowing them in advance will save you a lot of regret.

What is an Audit Log? It's historical data recording who made what decision and when, in chronological order. In regulated industries such as finance, healthcare, and legal, this record itself is a compliance requirement.

The Most Common Mistakes in Practice

1. Wrapping interrupt() in a try/except. interrupt() uses an exception mechanism internally to halt execution. Catching it with try/except breaks the interrupt entirely — execution passes through at the moment it should have stopped.

2. Generating a new thread_id on every request. To resume an interrupted graph, you must use the same thread_id. I didn't know this at first and kept creating a new ID on resume, which caused execution to start over from the beginning every time. Recommended practice is to generate it as a UUID and maintain a 1:1 mapping with the user session.

3. Passing non-JSON-safe values to interrupt(). Types like datetime, custom objects, and numpy arrays will fail to serialize. You need to convert them to dict/list/str/int/float/bool before passing.

Closing Thoughts

As AI agents become more powerful, the ability to design "where humans intervene" becomes a core competency for developers.

At first I thought, "Why not just let the agent handle everything?" — but once you experience a sensitive action executing automatically, the importance of HITL design becomes viscerally real. LangGraph's interrupt() is the tool that lets you control those intervention points precisely at the code level.

Three steps you can start right now:

1. Install with pip install langgraph langgraph-checkpoint-sqlite and run the SQL agent example from this article locally. You can test immediately without a separate DB using SqliteSaver.from_conn_string(":memory:").

2. Pick the one node in your existing agent code that made you most uneasy and attach interrupt() to it. You don't need to change everything.

3. If you're thinking about production, switching to AsyncPostgresSaver is essential. The FastAPI + Next.js approval dashboard template is also worth checking out for a much faster start.

Next article: LangGraph Multi-Agent Architecture — Orchestrating multiple agents with the Supervisor pattern and tracking responsibility

References

• LangGraph Official Docs — Interrupts
• LangChain Official Blog — Making it easier to build human-in-the-loop agents with interrupt
• LangChain Changelog — LangGraph v0.4: Working with Interrupts
• DEV Community — Interrupts and Commands in LangGraph: Building Human-in-the-Loop Workflows
• Towards Data Science — LangGraph 201: Adding Human Oversight to Your Deep Research Agent
• Towards Data Science — Building Human-In-The-Loop Agentic Workflows
• MarkTechPost — How to Build Human-in-the-Loop Plan-and-Execute AI Agents
• The Handover Blog — LangGraph Human in the Loop: A Complete Tutorial
• Elastic Search Labs — Human in the loop AI Agents with LangGraph & Elastic
• IBM Think — Oversee a prior art search AI agent with human-in-the-loop
• GitHub — langgraph-interrupt-workflow-template (FastAPI + Next.js)
• Towards AI — Human-in-the-Loop (HITL) with LangGraph: A Practical Guide
• Medium — LangGraph Breakpoints or Interrupt: How to Add Human-in-the-Loop Control
• SparkCo — Mastering LangGraph Checkpointing: Best Practices for 2025