Lesson 8: Human-in-the-Loop Deep Dive

Human-in-the-Loop (HITL) is one of the most critical safety mechanisms for AI agents. When an agent proposes actions that could have real-world consequences, HITL ensures a human can review, modify, or reject those actions before they execute. This lesson explores the complete HITL implementation in Deep Agents, from basic configuration to advanced integration patterns.

1. What is Human-in-the-Loop?

Definition and Purpose

Human-in-the-Loop (HITL) is a middleware pattern that pauses agent execution when specific tool calls require human oversight. Instead of automatically executing every action the model proposes, HITL intercepts sensitive operations and waits for a human decision before proceeding.

The core workflow is:

1. Agent proposes a tool call (e.g., "delete file X")
2. HITL middleware checks if this tool requires approval
3. If yes, execution pauses and the human is prompted
4. Human reviews and decides: approve, edit, reject, or respond
5. Execution resumes based on the decision

Why HITL Matters for AI Safety

AI agents can take actions with irreversible consequences. Without HITL:

• Data Loss: An agent could delete critical files or database records
• Security Breaches: Sensitive information could be sent to unauthorized recipients
• Financial Impact: Unauthorized transactions or API calls with cost implications
• Compliance Violations: Actions that violate regulatory requirements
• Reputation Damage: Sending incorrect communications to customers or partners

HITL provides a safety net where humans maintain control over consequential decisions while still benefiting from AI automation for routine tasks.

Types of Operations That Need Approval

Deep Agents Code classifies the following as gated actions that benefit from HITL:

Read-only tools like ls, read_file, glob, and grep typically run without prompting.

2. Deep Agents Implementation

HumanInTheLoopMiddleware Class

The HumanInTheLoopMiddleware is the core component that intercepts tool calls and manages the approval flow. It integrates with LangGraph's interrupt mechanism to pause and resume execution.

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver


def read_email_tool(email_id: str) -> str:
    """Read an email by its ID."""
    return f"Email content for ID: {email_id}"


def send_email_tool(recipient: str, subject: str, body: str) -> str:
    """Send an email to a recipient."""
    return f"Email sent to {recipient} with subject '{subject}'"


def delete_file_tool(path: str) -> str:
    """Delete a file from the filesystem."""
    return f"Deleted {path}"


# Create agent with HITL middleware
agent = create_agent(
    model="anthropic:claude-sonnet-4-6",
    tools=[read_email_tool, send_email_tool, delete_file_tool],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                # All decisions allowed (approve, edit, reject, respond)
                "send_email_tool": True,

                # Only approve or reject, no editing
                "delete_file_tool": {"allowed_decisions": ["approve", "reject"]},

                # No approval needed for read operations
                "read_email_tool": False,
            },
            description_prefix="Tool execution pending approval",
        ),
    ],
    # REQUIRED: Checkpointer for state persistence across interrupts
    checkpointer=InMemorySaver(),
)

Configuration Options

The interrupt_on parameter accepts three value types:

InterruptOnConfig Options

interrupt_on = {
    "critical_tool": {
        # Which decisions the reviewer can make
        "allowed_decisions": ["approve", "edit", "reject"],

        # Custom description for the approval prompt
        "description": "CRITICAL: This operation requires DBA approval",

        # Optional predicate to conditionally interrupt
        "when": lambda request: should_interrupt(request),
    }
}

Using with Deep Agents

Deep Agents provides a streamlined interrupt_on parameter directly in create_deep_agent:

from langchain.tools import tool
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver


@tool
def remove_file(path: str) -> str:
    """Delete a file from the filesystem."""
    return f"Deleted {path}"


@tool
def fetch_file(path: str) -> str:
    """Read a file from the filesystem."""
    return f"Contents of {path}"


@tool
def notify_email(to: str, subject: str, body: str) -> str:
    """Send an email notification."""
    return f"Sent email to {to}"


# Checkpointer is REQUIRED for human-in-the-loop
checkpointer = MemorySaver()

agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    tools=[remove_file, fetch_file, notify_email],
    interrupt_on={
        "remove_file": True,  # All decisions allowed
        "fetch_file": False,  # No interrupts needed
        "notify_email": {"allowed_decisions": ["approve", "reject"]},  # No editing
    },
    checkpointer=checkpointer,  # Required!
)

Custom Approval Callbacks

For specialized workflows, you can implement custom HITL logic using the interrupt() primitive directly in your tools:

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


@tool(description="Request human approval before proceeding with an action.")
def request_approval(action_description: str) -> str:
    """Request human approval using the interrupt() primitive."""
    # interrupt() pauses execution and returns the value passed to Command(resume=...)
    approval = interrupt({
        "type": "approval_request",
        "action": action_description,
        "message": f"Please approve or reject: {action_description}",
    })

    if approval.get("approved"):
        return f"Action '{action_description}' was APPROVED. Proceeding..."
    else:
        reason = approval.get("reason", "No reason provided")
        return f"Action '{action_description}' was REJECTED. Reason: {reason}"

3. The Approval Flow

Sequence Diagram

The following diagram shows the complete HITL approval flow:

User Prompt Format

When an interrupt is triggered, the result contains structured information for the reviewer:

# Check for interrupts in the result
result = agent.invoke(
    {"messages": [{"role": "user", "content": "Delete temp.txt"}]},
    config=config,
    version="v2",
)

if result.interrupts:
    interrupt_value = result.interrupts[0].value

    # action_requests: List of pending tool calls
    action_requests = interrupt_value["action_requests"]
    # [
    #     {
    #         "name": "delete_file",
    #         "args": {"path": "temp.txt"},
    #         "description": "Tool execution pending approval..."
    #     }
    # ]

    # review_configs: Allowed decisions per tool
    review_configs = interrupt_value["review_configs"]
    # [
    #     {
    #         "action_name": "delete_file",
    #         "allowed_decisions": ["approve", "reject"]
    #     }
    # ]

Response Options

The four decision types and their effects:

Decision Examples

from langgraph.types import Command

# Approve: Execute as proposed
decisions = [{"type": "approve"}]

# Edit: Modify arguments before execution
decisions = [{
    "type": "edit",
    "edited_action": {
        "name": "send_email",
        "args": {
            "to": "team@company.com",  # Changed from everyone@company.com
            "subject": "Weekly Update",
            "body": "..."
        }
    }
}]

# Reject: Deny with feedback
decisions = [{
    "type": "reject",
    "message": "User rejected deleting this file. Do not retry deletion. Ask which file to archive instead."
}]

# Respond: For ask_user style tools
decisions = [{
    "type": "respond",
    "message": "The preferred color is blue."
}]

# Resume execution with decisions
result = agent.invoke(
    Command(resume={"decisions": decisions}),
    config=config,  # Same thread_id!
    version="v2",
)

Handling Multiple Tool Calls

When the agent proposes multiple tools requiring approval, all are batched together:

result = agent.invoke(
    {"messages": [{
        "role": "user",
        "content": "Delete temp.txt and send an email to admin@example.com"
    }]},
    config=config,
    version="v2",
)

if result.interrupts:
    action_requests = result.interrupts[0].value["action_requests"]

    # Two tools need approval - provide decisions in same order
    assert len(action_requests) == 2

    decisions = [
        {"type": "approve"},  # First: delete_file
        {"type": "reject", "message": "Do not send email without manager approval."}  # Second: send_email
    ]

    result = agent.invoke(
        Command(resume={"decisions": decisions}),
        config=config,
        version="v2",
    )

Timeout Handling

HITL leverages LangGraph's checkpointing for persistence, meaning execution can wait indefinitely for human input. For production systems, implement timeout logic in your application layer:

import asyncio
from datetime import datetime, timedelta

async def invoke_with_timeout(agent, input_data, config, timeout_seconds=300):
    """Invoke agent with timeout for human approval."""
    result = agent.invoke(input_data, config=config, version="v2")

    if result.interrupts:
        start_time = datetime.now()
        deadline = start_time + timedelta(seconds=timeout_seconds)

        # Poll for human decision (in production, use webhooks/callbacks)
        while datetime.now() < deadline:
            decision = await get_human_decision(result.interrupts)
            if decision:
                return agent.invoke(
                    Command(resume={"decisions": decision}),
                    config=config,
                    version="v2",
                )
            await asyncio.sleep(5)  # Check every 5 seconds

        # Timeout: auto-reject with explanation
        auto_reject = [
            {"type": "reject", "message": "Approval timeout. Action rejected automatically."}
            for _ in result.interrupts[0].value["action_requests"]
        ]
        return agent.invoke(
            Command(resume={"decisions": auto_reject}),
            config=config,
            version="v2",
        )

    return result

4. Conditional Interrupts and Pattern Matching

How Conditional Interrupts Work

Instead of interrupting every call to a tool, you can use a when predicate to interrupt only specific calls based on their arguments. The predicate receives a ToolCallRequest and returns True to interrupt or False to auto-approve.

from deepagents import create_deep_agent
from langchain.agents.middleware import ToolCallRequest
from langgraph.checkpoint.memory import MemorySaver


def writes_outside_workspace(request: ToolCallRequest) -> bool:
    """Pause writes to paths outside the workspace directory."""
    path = request.tool_call["args"].get("file_path", "")
    return not path.startswith("/workspace/")


def is_destructive_sql(request: ToolCallRequest) -> bool:
    """Pause SQL that modifies data."""
    query = request.tool_call["args"].get("query", "").upper()
    destructive_keywords = ["DELETE", "DROP", "TRUNCATE", "UPDATE", "INSERT"]
    return any(keyword in query for keyword in destructive_keywords)


agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    interrupt_on={
        "write_file": {
            "allowed_decisions": ["approve", "edit", "reject"],
            "when": writes_outside_workspace,  # Only interrupt writes outside /workspace/
        },
        "execute_sql": {
            "allowed_decisions": ["approve", "reject"],
            "when": is_destructive_sql,  # Only interrupt non-SELECT queries
        },
    },
    checkpointer=MemorySaver(),
)

Pattern Examples for Common Scenarios

File Path Patterns

import fnmatch

# Interrupt writes to sensitive paths
SENSITIVE_PATTERNS = [
    "**/.env*",
    "**/.git/**",
    "**/secrets/**",
    "**/config/*.json",
    "**/.github/workflows/*",
]

def is_sensitive_path(request: ToolCallRequest) -> bool:
    """Interrupt writes to sensitive file paths."""
    path = request.tool_call["args"].get("path", "")
    return any(fnmatch.fnmatch(path, pattern) for pattern in SENSITIVE_PATTERNS)

Email Domain Patterns

# Auto-approve internal emails, interrupt external
INTERNAL_DOMAINS = ["@company.com", "@company.internal"]

def is_external_email(request: ToolCallRequest) -> bool:
    """Interrupt emails to external recipients."""
    recipient = request.tool_call["args"].get("to", "")
    return not any(domain in recipient for domain in INTERNAL_DOMAINS)

Cost-Based Patterns

# Interrupt API calls above cost threshold
def exceeds_cost_threshold(request: ToolCallRequest) -> bool:
    """Interrupt expensive API operations."""
    args = request.tool_call["args"]

    # Example: interrupt large batch operations
    if "batch_size" in args and args["batch_size"] > 100:
        return True

    # Example: interrupt requests to paid APIs
    if args.get("api_tier") == "premium":
        return True

    return False

Approval Modes in Deep Agents Code

Deep Agents Code provides three approval modes for interactive sessions:

Toggle between Manual and Auto mid-session with Shift+Tab or Ctrl+T.

5. Integration with OpenShell and NemoClaw

How HITL Works with Sandbox

NemoClaw (the OpenShell sandbox) implements a deny-by-default approach that complements HITL:

• When the agent reaches an unlisted network endpoint, OpenShell blocks the request
• The operator can approve or deny requests in real-time through the gateway interface
• Approved endpoints are merged into the sandbox's policy as durable revisions
• Policy resets to baseline when the sandbox is recreated

Five Security Layers

NemoClaw enforces security across multiple protection layers:

Policy Engine + HITL Together

The Policy Engine provides binary-scoped endpoint rules that work alongside HITL:

# Example policy entry
- endpoint: "api.example.com"
  port: 443
  binaries:
    - path: "/usr/bin/curl"
      sha256: "abc123..."
  protocol: rest
  rules:
    - method: GET
      path: "/api/v1/read/*"
    - method: POST
      path: "/api/v1/safe-action"

Key Policy Engine features:

• Binary-Scoped Rules: Each network policy entry restricts which executables can reach endpoints using SHA256 hashes
• Path-Scoped HTTP Rules: Restrict HTTP methods and URL paths for fine-grained control
• L4 vs L7 Inspection: Configure inspection depth based on the protocol field

Administrative Scope Approval

NemoClaw auto-approves only limited scopes (operator.pairing, operator.read, operator.write) and never auto-approves operator.admin:

# List pending requests
openclaw devices list --json

# Verify the requestId, client, device, and scopes
# Then approve explicitly
openclaw devices approve <requestId>

Audit Logging

OpenShell provides comprehensive audit capabilities:

Memory Secret Scanner

A plugin blocks agents from writing secrets to persistent memory files, covering 14 high-confidence secret patterns including provider API keys.

CLI Secret Redaction

The CLI automatically redacts secret patterns (API keys, bearer tokens, provider credentials) from command output and error messages.

Security Audit

# View suppressed security findings with rationale
openclaw security audit --json | jq '.suppressedFindings'

6. Best Practices

What to Always Require Approval For

Safe Auto-Approve Patterns

interrupt_on = {
    # Safe to auto-approve
    "read_file": False,
    "list_directory": False,
    "search_code": False,
    "get_status": False,

    # Conditionally approve based on scope
    "write_file": {
        "allowed_decisions": ["approve", "edit", "reject"],
        "when": lambda r: not r.tool_call["args"].get("path", "").startswith("/workspace/")
    },

    # Approve internal, interrupt external
    "send_notification": {
        "allowed_decisions": ["approve", "reject"],
        "when": lambda r: r.tool_call["args"].get("channel") not in ["#general", "#team"]
    },
}

Rejection Message Guidelines

When rejecting an action, provide clear guidance:

# BAD: Vague rejection
{"type": "reject", "message": "No."}

# GOOD: Specific rejection with guidance
{"type": "reject", "message": "User rejected deleting production database. Do not retry deletion. Instead, ask which test database to clean up."}

# GOOD: Rejection with alternative
{"type": "reject", "message": "Cannot send email to all@company.com. Please send to team-leads@company.com instead."}

Checkpointer Selection

Production HITL Configuration

from deepagents import create_deep_agent, FilesystemPermission
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
import os


async def create_production_agent():
    # Use persistent checkpointer for production
    checkpointer = AsyncPostgresSaver.from_conn_string(
        os.environ["DATABASE_URL"]
    )
    await checkpointer.setup()

    agent = create_deep_agent(
        model="anthropic:claude-sonnet-4-6",
        tools=[...],
        interrupt_on={
            # Explicit allow-list approach
            "execute_shell": True,
            "write_file": {"when": is_sensitive_path},
            "send_email": True,
            "execute_sql": {"when": is_destructive_sql},
            "deploy": True,
        },
        # Filesystem permission interrupts
        permissions=[
            FilesystemPermission(
                operations=["write"],
                paths=["/secrets/**", "**/.env*"],
                mode="interrupt",  # Trigger HITL for these paths
            ),
        ],
        checkpointer=checkpointer,
    )

    return agent

Complete Working Example

"""
Complete Human-in-the-Loop Example
Demonstrates file operations with approval workflow
"""
from langchain.tools import tool
from deepagents import create_deep_agent
from langchain.agents.middleware import ToolCallRequest
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command
import uuid


# Define tools
@tool
def read_file(path: str) -> str:
    """Read contents of a file."""
    return f"[Contents of {path}]"


@tool
def write_file(path: str, content: str) -> str:
    """Write content to a file."""
    return f"Successfully wrote {len(content)} bytes to {path}"


@tool
def delete_file(path: str) -> str:
    """Delete a file from the filesystem."""
    return f"Deleted {path}"


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


# Conditional interrupt predicates
def is_protected_path(request: ToolCallRequest) -> bool:
    """Interrupt writes to protected directories."""
    path = request.tool_call["args"].get("path", "")
    protected = ["/etc/", "/system/", "/.env", "/secrets/"]
    return any(p in path for p in protected)


def is_external_recipient(request: ToolCallRequest) -> bool:
    """Interrupt emails to external addresses."""
    to = request.tool_call["args"].get("to", "")
    return "@mycompany.com" not in to


# Create agent
checkpointer = MemorySaver()

agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    tools=[read_file, write_file, delete_file, send_email],
    interrupt_on={
        "read_file": False,  # Always auto-approve reads
        "write_file": {
            "allowed_decisions": ["approve", "edit", "reject"],
            "when": is_protected_path,
        },
        "delete_file": {
            "allowed_decisions": ["approve", "reject"],
            # No 'when' - always require approval for deletes
        },
        "send_email": {
            "allowed_decisions": ["approve", "edit", "reject"],
            "when": is_external_recipient,
        },
    },
    checkpointer=checkpointer,
)


def run_with_approval():
    """Demonstrate the approval workflow."""
    thread_id = str(uuid.uuid4())
    config = {"configurable": {"thread_id": thread_id}}

    # User request that will trigger approval
    result = agent.invoke(
        {"messages": [{"role": "user", "content": "Delete the file /tmp/old_data.txt"}]},
        config=config,
        version="v2",
    )

    # Check for interrupts
    if result.interrupts:
        interrupt_value = result.interrupts[0].value
        action_requests = interrupt_value["action_requests"]
        review_configs = interrupt_value["review_configs"]

        print("=== APPROVAL REQUIRED ===")
        for i, action in enumerate(action_requests):
            config_for_action = review_configs[i]
            print(f"Tool: {action['name']}")
            print(f"Args: {action['args']}")
            print(f"Allowed decisions: {config_for_action['allowed_decisions']}")

        # Simulate user approval
        user_decision = input("Enter decision (approve/reject): ").strip().lower()

        if user_decision == "approve":
            decisions = [{"type": "approve"}]
        else:
            reason = input("Rejection reason: ")
            decisions = [{"type": "reject", "message": reason}]

        # Resume with decision
        result = agent.invoke(
            Command(resume={"decisions": decisions}),
            config=config,
            version="v2",
        )

    # Print final result
    print("\n=== FINAL RESULT ===")
    print(result.value["messages"][-1].content)


if __name__ == "__main__":
    run_with_approval()

Knowledge Check Quiz

Summary

• HITL pauses agent execution for human review of sensitive operations
• Four decision types: approve, edit, reject, respond - each with specific use cases
• Checkpointer required: State persistence enables pause/resume across interrupts
• Conditional interrupts: Use when predicates to interrupt only specific calls
• Sandbox integration: HITL complements network/filesystem policies for defense in depth
• Best practices: Always require approval for destructive, external, and financial operations

Next Steps

• Lesson 9: Memory and Context Management
• Lesson 10: Multi-Agent Orchestration
• Lesson 11: Production Deployment Patterns