LangChain Human-in-the-Loop

In production environments, some operations require human approval — such as sending emails, performing deletions, and processing payments.

Human-in-the-Loop (HITL) allows the Agent to pause at critical moments, wait for human approval, and then continue.

interrupt() — Pausing Execution in Tools

interrupt()This function lets a tool pause midway through execution, wait for external input, and then continue:

Example

from langgraph.types import interrupt

# Pause in the tool using interrupt()
def send_email(to: str, subject: str, body: str) -> str:
    """Send email (requires human approval)"""
    # Pause execution and send an approval request to the outside
    approval = interrupt({
        "action": "send_email",
        "to": to,
        "subject": subject,
        "body": body,
        "message": "Please confirm whether to send this email?"
    })

    # Wait for external approval to continue
    if approval.get("approved"):
        return f"Email sent to {to}"
    else:
        return f"Email sending rejected: {approval.get('reason', 'User cancelled')}"

How interrupt() works:

  1. The tool calls interrupt() → the Agent pauses execution
  2. The external system gets the interrupt information and displays it to the user
  3. After the user makes a decision, execution resumes via Command(resume=...)
  4. interrupt() returns the value passed by the user, and the tool continues execution

Complete Example — Approval Process

Example

from dotenv import load_dotenv
load_dotenv()

from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage


@tool
def delete_course(course_name: str) -> str:
    """Delete course (requires approval).

    Args:
course_name: the name of the course to delete
    """
    # Pause and wait for approval
    approval = interrupt({
        "action": "delete_course",
        "course": course_name,
        "message": f"Confirm deleting course '{course_name}'? This action cannot be undone."
    })

    if approval.get("confirmed"):
        return f"Course '{course_name}' deleted"
    else:
        return f"Deletion cancelled"


checkpointer = InMemorySaver()
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    tools=[delete_course],
    checkpointer=checkpointer,
    system_prompt="You are the admin assistant of the EXAMPLE tutorial.",
)

config = {"configurable": {"thread_id": "admin-001"}}

# Step 1: Initiate the deletion request (will trigger an interrupt)
print("=== Start Execution ===")
result = agent.invoke(
    {"messages": [HumanMessage(content="Please delete the course 'Outdated Java Tutorial'.")]},
    config=config,
)

# Check whether the Agent has paused
state = agent.get_state(config)
print(f"Status: {state.next}")  # ('tools',) means paused at the tools node
print(f"Interrupt info: {state.tasks)

# Step 2: Human approval (simulate user clicking "Confirm")
print("\n=== Human Approval ===)
resume_value = {"confirmed": True, "operator": "Administrator Zhang San"}
result = agent.invoke(
    Command(resume=resume_value),
    config=config,
)
print(f"Final reply: {result['messages'][-1].content}")

Output:

=== 开始执行 ===
Status: ('tools',)
中断信息: (Interrupt(value={'action': 'delete_course', ...}),)

=== 人工审批 ===
Final reply: The course "Outdated Java Tutorial" has been deleted.

interrupt_before / interrupt_after Parameters

Besides using interrupt() in tools, you can also set global interrupt points in create_agent():

Example

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
agent = create_agent(
    model="deepseek:deepseek-v4-flash",
    tools=[some_tool],
    checkpointer=checkpointer,

    # Pause before tool nodes (approval required before every tool call)
    interrupt_before=["tools"],

    # Pause after model nodes (can inspect after each model reply)
    # interrupt_after=["model"],
)
ParameterPause TimingApplicable Scenario
interrupt_before=["tools"]Before each tool executionAll tool calls require approval
interrupt_before=["model"]Before each model callManually review messages before model processing
interrupt_after=["model"]After each model replyReview model output before deciding whether to continue
interrupt_after=["tools"]After each tool executionCheck tool results before deciding the next step

Typical HITL Architecture

In real Web applications, HITL is usually implemented like this:

Example

# Backend: receive user message, process until the interrupt point, return interrupt info
def handle_user_message(thread_id: str, message: str):
    config = {"configurable": {"thread_id": thread_id}}
    result = agent.invoke(
        {"messages": [HumanMessage(content=message)]},
        config=config,
    )
    state = agent.get_state(config)

    # Check whether waiting for approval
    if state.tasks and state.tasks[0].interrupts:
        return {
            "status": "pending_approval",
            "interrupt": state.tasks[0].interrupts[0].value,
            "thread_id": thread_id,
        }

    return {
        "status": "completed",
        "reply": result["messages"][-1].content,
    }


# Backend: process user approval
def handle_approval(thread_id: str, approved: bool, reason: str = ""):
    config = {"configurable": {"thread_id": thread_id}}
    result = agent.invoke(
        Command(resume={"confirmed": approved, "reason": reason}),
        config=config,
    )
    return {"status": "completed", "reply": result["messages"][-1].content}

HITL requires a Checkpointer. Because when the Agent pauses at interrupt(), its state must be persisted in order to correctly resume execution.

Other Extensions