Human-in-the-Loop

AgentVisor™ supports LangGraph's interrupt/resume pattern for human-in-the-loop workflows. The interrupt/resume functionality is a LangGraph feature—AgentVisor provides the durable execution environment that preserves state between interruptions.

Overview

Human-in-the-loop allows agents to:

1. Pause execution at decision points
2. Request human input or approval
3. Resume with the human's response
4. Continue processing

Framework Note

This guide describes human-in-the-loop patterns using LangGraph's interrupt() function. The concepts (pausing for human input, approval workflows) apply broadly, but the specific implementation shown here is LangGraph-specific.

Basic Pattern

from langgraph.types import interrupt
from langgraph.graph import StateGraph, END


def request_approval(state: AgentState) -> dict:
    """Pause and wait for human approval."""
    interrupt({
        "action": state["pending_action"],
        "reason": "This action requires approval",
        "options": ["approve", "reject", "modify"]
    })
    # Execution stops here until resumed
    return {}


def process_response(state: AgentState) -> dict:
    """Process the human's response after resume."""
    # When resumed, the input contains the human's choice
    user_input = state.get("messages", [])[-1]
    if "approve" in str(user_input).lower():
        return {"approved": True}
    return {"approved": False}

Complete Example

A shopping list agent that pauses after each command:

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.types import interrupt, Command
from langchain_core.messages import AIMessage
from agentvisor.langgraph import AgentVisorCheckpointer


class ShoppingListState(TypedDict):
    messages: Annotated[list, add_messages]
    shopping_items: list
    last_response: str
    done: bool


def process_command(state: ShoppingListState) -> dict:
    """Process user command."""
    messages = state.get("messages", [])
    items = state.get("shopping_items", [])

    # Get last user message
    user_msg = ""
    for msg in reversed(messages):
        if hasattr(msg, "type") and msg.type == "human":
            user_msg = msg.content
            break
        elif isinstance(msg, dict) and msg.get("role") == "user":
            user_msg = msg.get("content", "")
            break

    user_msg = user_msg.lower().strip()

    # Process commands
    if user_msg.startswith("add:"):
        item = user_msg[4:].strip()
        items.append(item)
        response = f"Added '{item}'. You have {len(items)} item(s)."
    elif user_msg.startswith("delete:"):
        item = user_msg[7:].strip()
        if item in items:
            items.remove(item)
            response = f"Removed '{item}'. You have {len(items)} item(s)."
        else:
            response = f"'{item}' not found."
    elif user_msg == "list":
        if items:
            response = f"Shopping list: {', '.join(items)}"
        else:
            response = "Shopping list is empty."
    elif user_msg == "end":
        return {
            "shopping_items": items,
            "last_response": f"Goodbye! Final list had {len(items)} item(s).",
            "done": True
        }
    else:
        response = "Commands: add: <item>, delete: <item>, list, end"

    return {
        "shopping_items": items,
        "last_response": response,
        "done": False
    }


def wait_for_input(state: ShoppingListState) -> dict:
    """Pause and wait for next user input."""
    interrupt({
        "response": state.get("last_response", ""),
        "item_count": len(state.get("shopping_items", [])),
        "__interrupt__": True
    })
    return {}


def should_continue(state: ShoppingListState) -> str:
    if state.get("done", False):
        return "end"
    return "wait"


# Build graph
builder = StateGraph(ShoppingListState)
builder.add_node("process", process_command)
builder.add_node("wait", wait_for_input)
builder.set_entry_point("process")

builder.add_conditional_edges(
    "process",
    should_continue,
    {"wait": "wait", "end": END}
)
builder.add_edge("wait", "process")

graph = builder.compile(checkpointer=AgentVisorCheckpointer())

API Flow

Run 1: Initial Request

# Create thread
THREAD=$(curl -sX POST http://localhost:8090/threads | jq -r '.thread_id')

# First run
curl -sX POST "http://localhost:8090/threads/$THREAD/runs?wait=30s" \
    -H "Content-Type: application/json" \
    -d '{"input": {"messages": [{"role": "user", "content": "add: milk"}]}}'

Response:

{
  "status": "interrupted",
  "output": {
    "__interrupt__": true,
    "response": "Added 'milk'. You have 1 item(s).",
    "item_count": 1
  }
}

Run 2: Resume

curl -sX POST "http://localhost:8090/threads/$THREAD/runs?wait=30s" \
    -H "Content-Type: application/json" \
    -d '{"input": {"messages": [{"role": "user", "content": "add: eggs"}]}}'

Response:

{
  "status": "interrupted",
  "output": {
    "__interrupt__": true,
    "response": "Added 'eggs'. You have 2 item(s).",
    "item_count": 2
  }
}

Run 3: End

curl -sX POST "http://localhost:8090/threads/$THREAD/runs?wait=30s" \
    -H "Content-Type: application/json" \
    -d '{"input": {"messages": [{"role": "user", "content": "end"}]}}'

Response:

{
  "status": "completed",
  "output": {
    "response": "Goodbye! Final list had 2 item(s)."
  }
}

Approval Workflows

For approval-style workflows:

def request_approval(state: AgentState) -> dict:
    """Request approval for a sensitive action."""
    action = state["pending_action"]

    interrupt({
        "type": "approval_request",
        "action": action,
        "description": f"The agent wants to: {action['description']}",
        "options": [
            {"value": "approve", "label": "Approve"},
            {"value": "reject", "label": "Reject"},
            {"value": "modify", "label": "Modify"}
        ]
    })
    return {}


def handle_approval(state: AgentState) -> dict:
    """Handle the approval response."""
    # The user's response is in the latest message
    response = get_last_user_message(state)

    if response == "approve":
        # Execute the action
        execute_action(state["pending_action"])
        return {"action_result": "completed"}
    elif response == "reject":
        return {"action_result": "rejected"}
    else:
        # User wants to modify - collect new parameters
        return {"needs_modification": True}

Multi-Step Approvals

For complex workflows with multiple approval points:

def route_to_approval(state: AgentState) -> str:
    """Decide which approval is needed."""
    action = state["pending_action"]

    if action["cost"] > 1000:
        return "manager_approval"
    elif action["risk"] == "high":
        return "security_approval"
    else:
        return "auto_approve"

builder.add_conditional_edges(
    "prepare_action",
    route_to_approval,
    {
        "manager_approval": "request_manager",
        "security_approval": "request_security",
        "auto_approve": "execute"
    }
)

Timeouts

For time-sensitive approvals, implement timeout logic:

def request_with_timeout(state: AgentState) -> dict:
    """Request approval with timeout information."""
    interrupt({
        "type": "approval_request",
        "action": state["pending_action"],
        "timeout": "24h",
        "default_on_timeout": "reject"
    })
    return {}

Handle timeout on resume:

def check_timeout(state: AgentState) -> str:
    request_time = state.get("approval_requested_at")
    if datetime.now() - request_time > timedelta(hours=24):
        return "timeout"
    return "process_response"

Best Practices

Clear Interrupt Payloads

Include all context needed for the human to decide:

interrupt({
    "type": "approval",
    "summary": "Agent wants to send an email",
    "details": {
        "to": "user@example.com",
        "subject": "Report",
        "preview": "First 100 characters..."
    },
    "options": ["send", "edit", "cancel"],
    "context": {
        "thread_id": state.get("thread_id"),
        "previous_approvals": state.get("approvals", [])
    }
})

Graceful Degradation

Handle missing responses:

def process_response(state: AgentState) -> dict:
    response = get_last_user_message(state)

    if not response:
        # No response received, use default
        return {"action": "skip", "reason": "No response received"}

    # Process response...

Logging Human Decisions

Use structured logging to capture human decisions for observability:

import logging
from datetime import datetime

logger = logging.getLogger(__name__)

def log_approval(state: AgentState) -> dict:
    logger.info("human_approval", extra={
        "action": state["pending_action"],
        "decision": state["approval_decision"],
        "decided_at": datetime.now().isoformat()
    })
    return {}

This emits a structured log entry that can be captured by your logging infrastructure. For policy-based authorization, AgentVisor's PolicyEngine automatically logs authorization decisions.