langchain-human-in-the-loop

name: langchain-human-in-the-loop description: Add human oversight to LangChain agents using HITL middleware - includes interrupts, approval workflows, edit/reject decisions, and checkpoints language: python

langchain-human-in-the-loop (Python)

Overview

Human-in-the-Loop (HITL) lets you add human oversight to agent tool calls. When agents propose sensitive actions (like database writes or sending emails), execution pauses for human approval, editing, or rejection.

Key Concepts:

human_in_the_loop_middleware: Pauses execution for human decisions
Interrupts: Checkpoint where agent waits for human input
Decisions: approve, edit, or reject tool calls
Checkpointer: Required for persistence across interruptions

Code Examples

Basic HITL Setup

from langchain.agents import create_agent, human_in_the_loop_middleware
from langgraph.checkpoint.memory import MemorySaver
from langchain.tools import tool

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email."""
    # Send email logic
    return f"Email sent to {to}"

agent = create_agent(
    model="gpt-4.1",
    tools=[send_email],
    checkpointer=MemorySaver(),  # Required for HITL
    middleware=[
        human_in_the_loop_middleware(
            interrupt_on={
                "send_email": {
                    "allowed_decisions": ["approve", "edit", "reject"],
                },
            }
        )
    ],
)

Running with Interrupts

from langgraph.types import Command

config = {"configurable": {"thread_id": "session-1"}}

# Step 1: Agent runs until it needs to call tool
result1 = agent.invoke({
    "messages": [{"role": "user", "content": "Send email to john@example.com saying hello"}]
}, config=config)

# Check for interrupt
if "__interrupt__" in result1:
    interrupt = result1["__interrupt__"][0]
    print(f"Waiting for approval: {interrupt.value}")

# Step 2: Human approves
result2 = agent.invoke(
    Command(resume={"decisions": [{"type": "approve"}]}),
    config=config
)

# Tool now executes and agent completes
print(result2["messages"][-1].content)

Editing Tool Arguments

# Human edits the arguments
result2 = agent.invoke(
    Command(resume={
        "decisions": [{
            "type": "edit",
            "args": {
                "to": "alice@company.com",  # Fixed email
                "subject": "Project Meeting - Updated",
                "body": "...",
            },
        }]
    }),
    config=config
)

Rejecting with Feedback

# Human rejects
result2 = agent.invoke(
    Command(resume={
        "decisions": [{
            "type": "reject",
            "feedback": "Cannot delete customer data without manager approval",
        }]
    }),
    config=config
)

Multiple Tools with Different Policies

agent = create_agent(
    model="gpt-4.1",
    tools=[send_email, read_email, delete_email],
    checkpointer=MemorySaver(),
    middleware=[
        human_in_the_loop_middleware(
            interrupt_on={
                "send_email": {
                    "allowed_decisions": ["approve", "edit", "reject"],
                },
                "delete_email": {
                    "allowed_decisions": ["approve", "reject"],  # No edit
                },
                "read_email": False,  # No HITL for reading
            }
        )
    ],
)

Streaming with HITL

# Stream until interrupt
for mode, chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Send report to team"}]},
    config=config,
    stream_mode=["updates", "messages"],
):
    if mode == "messages":
        token, metadata = chunk
        if token.content:
            print(token.content, end="", flush=True)
    elif mode == "updates":
        if "__interrupt__" in chunk:
            print("\nWaiting for approval...")
            break

# Resume after approval
for mode, chunk in agent.stream(
    Command(resume={"decisions": [{"type": "approve"}]}),
    config=config,
    stream_mode=["messages"],
):
    # Continue streaming
    pass

Gotchas

1. Missing Checkpointer

# ❌ Problem: No checkpointer
agent = create_agent(
    model="gpt-4.1",
    tools=[send_email],
    middleware=[human_in_the_loop_middleware({...})],  # Error!
)

# ✅ Solution: Always add checkpointer
from langgraph.checkpoint.memory import MemorySaver

agent = create_agent(
    model="gpt-4.1",
    tools=[send_email],
    checkpointer=MemorySaver(),  # Required
    middleware=[human_in_the_loop_middleware({...})],
)

2. No thread_id

# ❌ Problem: Missing thread_id
agent.invoke(input)  # No config!

# ✅ Solution: Always provide thread_id
agent.invoke(input, config={"configurable": {"thread_id": "user-123"}})

3. Wrong Resume Syntax

# ❌ Problem: Wrong resume format
agent.invoke({"resume": {"decisions": [...]}})  # Wrong!

# ✅ Solution: Use Command
from langgraph.types import Command

agent.invoke(
    Command(resume={"decisions": [{"type": "approve"}]}),
    config=config
)