# Langchain Human In The Loop

> Add human oversight to LangChain agents using HITL middleware - includes interrupts, approval workflows, edit/reject decisions, and checkpoints

- Skill: `christian-bromann/langchain-human-in-the-loop-2` (Agent Skill)
- Install (CLI): `npx skillmds@latest add christian-bromann/langchain-human-in-the-loop-2`
- Raw SKILL.md: https://api.skillmd.com/api/skills/christian-bromann/langchain-human-in-the-loop-2/raw
- Safety review: pending (external: skill-scanner PASS, skillspector PASS)
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- Author: christian-bromann (https://skillmd.com/u/christian-bromann)
- Updated: 2026-09-22
- Page: https://skillmd.com/skills/christian-bromann/langchain-human-in-the-loop-2

---


# 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

```python
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

```python
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

```python
# 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

```python
# 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

```python
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

```python
# 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

```python
# ❌ 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

```python
# ❌ 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

```python
# ❌ 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
)
```

## Links to Documentation

- [Human-in-the-Loop Guide](https://docs.langchain.com/oss/python/langchain/human-in-the-loop)
- [LangGraph Interrupts](https://docs.langchain.com/oss/python/langgraph/interrupts)