human-in-loop-agents

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": "[email protected]",
        "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())