How to Implement Human-in-the-Loop in LangGraph Using the interrupt() Pattern

Apr 16, 2026

Cowrie

Dev @ Bswen

Photo by Google DeepMind on Unsplash

The Problem: Unchecked AI Agent Actions

When I first deployed an AI agent that could execute SQL queries autonomously, I learned a painful lesson. The agent ran a DELETE statement on my development database within minutes of deployment. No approval step. No confirmation. Just execution.

This is the core problem with autonomous agents: they can cause real damage before you realize what happened:

• SQL queries that modify production databases
• API calls that trigger financial transactions
• Email sends that cannot be undone
• File operations that overwrite critical data

Most frameworks force you to duct-tape approval logic into your agent after realizing these risks. This retrofitting is painful because:

• No native pause/resume mechanisms exist
• Custom solutions require managing state across asynchronous boundaries
• Error handling becomes complex when human delays are involved
• The duct-taped solutions break under edge cases

A Reddit commenter captured this frustration: “LangGraph’s interrupt() is the part that doesn’t get enough credit. Once you need a human to approve a tool call before it fires, most other frameworks make you duct-tape it together. The interrupt/resume pattern actually models the state transition cleanly.”

The Direct Answer

Use LangGraph’s interrupt() function inside your node or tool to pause execution mid-workflow, then resume with Command(resume=...) to pass human decisions back.

This pattern cleanly models state transitions for human approval without duct-taping custom solutions. Other frameworks require complex workarounds for this capability; LangGraph handles it natively.

How interrupt/resume Works

The interrupt() function pauses graph execution at a specific node. The caller receives interrupt data via ["__interrupt__"] key. When ready, the caller invokes the graph again with Command(resume=...) to continue execution.

Pattern 1: Simple Approval Node

approval_node.py

from langgraph.types import interrupt, Command

def approval_node(state: State) -> Command[Literal["proceed", "cancel"]]:
    # Pause execution; payload shows up under result["__interrupt__"]
    is_approved = interrupt({
        "question": "Do you want to proceed with this action?",
        "details": state["action_details"]
    })

    # Route based on the response
    if is_approved:
        return Command(goto="proceed")
    else:
        return Command(goto="cancel")

Pattern 2: Tool Call Review Before Execution

This is the exact use case from the Reddit comment—approving tool calls before they fire:

tool_review.py

from typing import Union
from langgraph.types import interrupt

def review_tool_call(tool_call: ToolCall) -> Union[ToolCall, ToolMessage]:
    """Review a tool call before it executes."""
    human_review = interrupt({
        "question": "Is this correct?",
        "tool_call": tool_call,
    })

    review_action = human_review["action"]
    review_data = human_review.get("data")

    if review_action == "continue":
        return tool_call  # Execute as-is
    elif review_action == "update":
        # Human modified arguments
        return {**tool_call, **{"args": review_data}}
    elif review_action == "feedback":
        # Human rejected, provide feedback to LLM
        return ToolMessage(
            content=review_data,
            name=tool_call["name"],
            tool_call_id=tool_call["id"]
        )

Pattern 3: Wrapped Tool with Approval

wrapped_tool.py

from langchain.tools import tool
from langgraph.types import interrupt

@tool
def run_query_tool_with_interrupt(**tool_input):
    """SQL query tool requiring human approval before execution."""
    request = {
        "action": "run_query",
        "args": tool_input,
        "description": "Please review the SQL query"
    }

    response = interrupt([request])

    if response["type"] == "accept":
        # Approved - execute the tool
        tool_response = run_query_tool.invoke(tool_input)
    elif response["type"] == "edit":
        # Human modified the query
        tool_input = response["args"]["args"]
        tool_response = run_query_tool.invoke(tool_input)
    elif response["type"] == "response":
        # Rejected - return human feedback to LLM
        tool_response = response["args"]

    return tool_response

Why This Pattern Matters

The Reddit commenter’s insight about “clean state transition modeling” is accurate because:

1. No custom state management: LangGraph’s checkpointer handles persistence automatically
2. No duct-taping: The pattern is native to the framework, not retrofitted
3. No async complexity: Resume value returns directly to interrupt call
4. No polling loops: Graph execution genuinely pauses

Comparison with other approaches:

Approach	Human Approval Implementation
Raw Python	Custom pause logic, manual state persistence
LangChain (old)	No native support, requires wrapper code
LangGraph	interrupt() + Command(resume=...) built-in

Full Human-in-the-Loop Workflow

full_workflow.py

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

class ApprovalState(TypedDict):
    action_details: str
    status: Optional[Literal["pending", "approved", "rejected"]]

def approval_node(state: ApprovalState) -> Command[Literal["proceed", "cancel"]]:
    # Expose details for UI rendering
    decision = interrupt({
        "question": "Approve this action?",
        "details": state["action_details"],
    })

    # Route based on human decision
    return Command(goto="proceed" if decision else "cancel")

def proceed_node(state: ApprovalState):
    return {"status": "approved"}

def cancel_node(state: ApprovalState):
    return {"status": "rejected"}

# Build the graph
builder = StateGraph(ApprovalState)
builder.add_node("approval", approval_node)
builder.add_node("proceed", proceed_node)
builder.add_node("cancel", cancel_node)
builder.add_edge(START, "approval")
builder.add_edge("proceed", END)
builder.add_edge("cancel", END)

# Checkpointer required for resumable workflows
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# Usage: Two-phase invocation
config = {"configurable": {"thread_id": "approval-123"}}

# Phase 1: Invoke, pauses at interrupt
initial = graph.invoke(
    {"action_details": "Transfer $500", "status": "pending"},
    config=config,
)
print(initial["__interrupt__"])
# Shows the question to render in UI

# Phase 2: Resume with human decision
resumed = graph.invoke(Command(resume=True), config=config)
print(resumed["status"])  # -> "approved"

Tool Approval Pattern for High-Risk Operations

sensitive_api.py

from langchain_core.runnables import RunnableConfig
from langchain.tools import tool
from langgraph.types import interrupt

@tool
def sensitive_api_call(config: RunnableConfig, **tool_input):
    """API call requiring human approval before execution."""
    # Show tool call details for review
    request = {
        "action": "api_call",
        "endpoint": "/api/payment",
        "args": tool_input,
        "risk": "HIGH"
    }

    response = interrupt([request])

    # Handle three approval types
    if response["type"] == "accept":
        return actual_api_call(tool_input)
    elif response["type"] == "edit":
        modified_input = response["args"]
        return actual_api_call(modified_input)
    elif response["type"] == "response":
        return response["args"]  # User feedback to LLM
    else:
        raise ValueError(f"Unknown type: {response['type']}")

Common Mistakes to Avoid

Mistake 1: Forgetting the Checkpointer

Without a checkpointer, interrupted graphs cannot resume:

wrong_no_checkpointer.py

# WRONG: No persistence
graph = builder.compile()

# CORRECT: With checkpointer for resumable workflows
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

Mistake 2: Not Using thread_id

Each interrupted conversation needs a unique thread_id:

thread_id_example.py

config = {"configurable": {"thread_id": "approval-123"}}
initial = graph.invoke({"input": "data"}, config=config)
# Resume requires same thread_id
resumed = graph.invoke(Command(resume=True), config=config)

Mistake 3: Ignoring the interrupt Key

The interrupt payload is available via ["__interrupt__"]:

access_interrupt.py

initial = graph.invoke({"input": "data"}, config=config)
print(initial["__interrupt__"])
# -> [Interrupt(value={'question': ..., 'details': ...})]

Mistake 4: Not Handling All Resume Types

Tool review needs to handle three types: accept, edit, feedback:

handle_all_types.py

# Only accept case is common mistake
response = interrupt([request])
if response["type"] == "accept":
    return tool.invoke(tool_input)
# Missing edit and feedback handling!

Mistake 5: Interrupting After the Action

Place interrupt BEFORE the risky operation, not after:

wrong_interrupt_position.py

# WRONG: Interrupt after action
def execute_node(state):
    result = dangerous_action(state)
    interrupt("Was this OK?")  # Too late!
    return result

# CORRECT: Interrupt before action
def approval_node(state):
    approved = interrupt("Approve action?")
    if approved:
        return dangerous_action(state)
    return {"cancelled": True}

Decision Framework

Do you need human approval before tool execution?
├─ Yes -> Use interrupt() inside tool wrapper
│   └─ Pattern: interrupt() -> review -> accept/edit/feedback
└─ No -> Do you need human approval for workflow decisions?
    ├─ Yes -> Use interrupt() in approval node
    │   └─ Pattern: interrupt() -> Command(goto=...)
    └─ No -> Standard LangGraph workflow without interrupt

Simplified rules:

• Approving tool calls before execution -> interrupt() in tool wrapper
• Workflow branching decisions -> interrupt() in approval node
• No human intervention needed -> Standard graph without interrupt

Summary

LangGraph’s interrupt() pattern provides native human-in-the-loop capability that other frameworks require duct-taping to achieve. Use it inside nodes for workflow decisions, or wrap tools for approving risky operations before execution. The pattern cleanly models state transitions with built-in checkpointing, eliminating custom persistence code.

Next step: Add a checkpointer to your LangGraph graph and place interrupt() before any tool call that modifies production data.

Final Words + More Resources

My intention with this article was to help others share my knowledge and experience. If you want to contact me, you can contact by email: Email me

Here are also the most important links from this article along with some further resources that will help you in this scope:

• 👨‍💻 LangGraph interrupt() Documentation
• 👨‍💻 Tool Call Review Pattern
• 👨‍💻 Reddit Discussion: LangGraph vs LangChain for Custom Agents
• 👨‍💻 LangGraph Checkpoint Persistence

Oh, and if you found these resources useful, don’t forget to support me by starring the repo on GitHub!

Comments

Tags:

• ai-agents
• python
• automation

Pydantic AI vs LangChain: Which Framework is Better for Production AI Agents?

What is Docling Agent for Agentic Document Operations?