Human-in-the-Loop Agents

Build agents that know when to stop and ask for human judgment.

Why Human-in-the-Loop?

Critical for:

• High-stakes actions: Financial transactions, data deletion
• Compliance: Audit requirements, approval workflows
• Quality control: Review before publishing, sending
• Edge cases: When agent confidence is low
• Trust building: Users control what agents do

Core Patterns

Pattern 1: Interrupt Before Action

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt

class AgentState(TypedDict):
    messages: list
    pending_action: dict | None
    approved: bool

def plan_action(state: AgentState) -> dict:
    """Agent plans what to do."""
    # Determine action based on messages
    action = {
        "type": "send_email",
        "to": "user@example.com",
        "subject": "Important Update",
        "body": "..."
    }
    return {"pending_action": action, "approved": False}

def request_approval(state: AgentState) -> dict:
    """Interrupt and wait for human approval."""
    action = state["pending_action"]

    # This pauses execution and waits for human input
    approved = interrupt({
        "message": f"Approve this action?",
        "action": action,
        "options": ["approve", "reject", "modify"]
    })

    return {"approved": approved == "approve"}

def execute_action(state: AgentState) -> dict:
    """Execute the approved action."""
    if state["approved"]:
        result = execute(state["pending_action"])
        return {"messages": [{"role": "system", "content": f"Executed: {result}"}]}
    else:
        return {"messages": [{"role": "system", "content": "Action rejected"}]}

def should_execute(state: AgentState) -> str:
    return "execute" if state["approved"] else "end"

# Build graph
workflow = StateGraph(AgentState)
workflow.add_node("plan", plan_action)
workflow.add_node("approve", request_approval)
workflow.add_node("execute", execute_action)

workflow.set_entry_point("plan")
workflow.add_edge("plan", "approve")
workflow.add_conditional_edges("approve", should_execute, {
    "execute": "execute",
    "end": END
})
workflow.add_edge("execute", END)

# Compile with checkpointer (required for interrupts)
app = workflow.compile(checkpointer=MemorySaver())

Using the Interrupt

# Start the agent
config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke({"messages": [{"role": "user", "content": "Send the report"}]}, config)

# Agent pauses at interrupt - check state
state = app.get_state(config)
print(state.next)  # Shows we're waiting at 'approve' node
print(state.tasks)  # Shows the interrupt details

# Human reviews and approves
app.update_state(
    config,
    {"approved": True},
    as_node="approve"  # Resume from this node
)

# Continue execution
final_result = app.invoke(None, config)  # None = continue from checkpoint

Pattern 2: Edit Before Continue

def generate_draft(state: AgentState) -> dict:
    """Generate content for review."""
    draft = llm.invoke("Generate a response...")
    return {"draft": draft.content}

def human_edit(state: AgentState) -> dict:
    """Allow human to edit the draft."""
    current_draft = state["draft"]

    # Interrupt with editable content
    edited = interrupt({
        "type": "edit",
        "content": current_draft,
        "instructions": "Review and edit this draft before sending"
    })

    return {"draft": edited}

def send_response(state: AgentState) -> dict:
    """Send the (possibly edited) response."""
    send_message(state["draft"])
    return {"sent": True}

Pattern 3: Confidence-Based Interrupt

def agent_with_confidence(state: AgentState) -> dict:
    """Agent that knows when to ask for help."""
    response = llm.invoke(
        state["messages"],
        # Request confidence score
        response_format={"type": "json_object"}
    )

    result = json.loads(response.content)
    return {
        "answer": result["answer"],
        "confidence": result["confidence"]
    }

def check_confidence(state: AgentState) -> str:
    """Route based on confidence."""
    if state["confidence"] < 0.7:
        return "human_review"
    return "respond"

def human_review(state: AgentState) -> dict:
    """Low confidence - get human help."""
    review = interrupt({
        "type": "review",
        "agent_answer": state["answer"],
        "confidence": state["confidence"],
        "question": "Is this answer correct? Edit if needed."
    })

    return {"answer": review.get("edited_answer", state["answer"])}

Pattern 4: Batch Approval

class BatchApprovalState(TypedDict):
    items: list
    approved_items: list
    rejected_items: list

def collect_items(state: BatchApprovalState) -> dict:
    """Collect items that need approval."""
    items = get_pending_items()
    return {"items": items}

def batch_approve(state: BatchApprovalState) -> dict:
    """Present batch for approval."""
    decisions = interrupt({
        "type": "batch_approval",
        "items": state["items"],
        "instructions": "Review each item. Select items to approve."
    })

    # decisions = {"approved": [id1, id2], "rejected": [id3]}
    return {
        "approved_items": decisions["approved"],
        "rejected_items": decisions["rejected"]
    }

def process_approved(state: BatchApprovalState) -> dict:
    """Process only approved items."""
    for item_id in state["approved_items"]:
        process_item(item_id)
    return {}

Timeout and Escalation

import asyncio
from datetime import datetime, timedelta

class TimedApprovalState(TypedDict):
    action: dict
    requested_at: str
    approved: bool | None
    escalated: bool

async def request_with_timeout(state: TimedApprovalState) -> dict:
    """Request approval with timeout."""
    state["requested_at"] = datetime.now().isoformat()

    # Interrupt for approval
    try:
        result = interrupt({
            "type": "approval",
            "action": state["action"],
            "timeout_minutes": 30
        })
        return {"approved": result == "approve"}
    except TimeoutError:
        return {"approved": None, "escalated": True}

def check_approval_status(state: TimedApprovalState) -> str:
    if state["approved"] is True:
        return "execute"
    elif state["approved"] is False:
        return "reject"
    elif state["escalated"]:
        return "escalate"
    return "wait"

def escalate_to_manager(state: TimedApprovalState) -> dict:
    """Escalate after timeout."""
    notify_manager(state["action"])
    return {"messages": [{"content": "Escalated to manager for approval"}]}

UI Integration

Webhook-Based Approval

from fastapi import FastAPI, HTTPException
import uuid

app = FastAPI()
pending_approvals = {}

@app.post("/agent/start")
async def start_agent(request: dict):
    """Start agent, returns approval_id if interrupt."""
    thread_id = str(uuid.uuid4())
    config = {"configurable": {"thread_id": thread_id}}

    result = agent_app.invoke(request["input"], config)

    # Check if waiting for approval
    state = agent_app.get_state(config)
    if state.next:  # Interrupted
        approval_id = str(uuid.uuid4())
        pending_approvals[approval_id] = {
            "thread_id": thread_id,
            "interrupt_data": state.tasks[0].interrupts[0]
        }
        return {
            "status": "pending_approval",
            "approval_id": approval_id,
            "details": state.tasks[0].interrupts[0]
        }

    return {"status": "complete", "result": result}

@app.post("/agent/approve/{approval_id}")
async def approve_action(approval_id: str, decision: dict):
    """Handle approval decision."""
    if approval_id not in pending_approvals:
        raise HTTPException(404, "Approval not found")

    pending = pending_approvals.pop(approval_id)
    config = {"configurable": {"thread_id": pending["thread_id"]}}

    # Update state with decision
    agent_app.update_state(
        config,
        {"approved": decision["approved"]},
        as_node="approve"
    )

    # Continue execution
    result = agent_app.invoke(None, config)
    return {"status": "complete", "result": result}

Best Practices

1. Clear context: Tell humans exactly what they’re approving
2. Reasonable defaults: Pre-fill with agent’s best guess
3. Timeout handling: Don’t block forever
4. Audit logging: Record all approvals/rejections
5. Batch when possible: Don’t interrupt for every small thing
6. Graceful degradation: Have fallback if no response

When to Require Approval

Monitoring HITL Metrics

# Track approval patterns
metrics = {
    "total_interrupts": 0,
    "approved": 0,
    "rejected": 0,
    "edited": 0,
    "timed_out": 0,
    "avg_response_time_seconds": 0
}

# Analyze to improve agent
# High rejection rate = agent needs improvement
# High edit rate = agent close but not quite
# High timeout rate = approval process too slow