Human-Agent Interaction Patterns

Human-in-the-loop, human-on-the-loop, policy-based autonomy, and the autonomy spectrum

Open In Colab

Setup
Autonomy Spectrum Overview
Human-in-the-Loop (HITL)
Human-on-the-Loop (HOTL) Monitoring
Policy-Based Autonomy
LangGraph HITL with interrupt()

!pip install -q langchain-openai langgraph

import os
# os.environ["OPENAI_API_KEY"] = "your-key"

2. Autonomy Spectrum Overview

Level	Pattern	Human Role	Agent Autonomy
1	Human-in-the-loop	Approves every action	Minimal
2	Human-on-the-loop	Monitors, intervenes on anomaly	Moderate
3	Policy-governed	Sets rules, reviews logs	High
4	Fully autonomous	Post-hoc audit only	Full

3. Human-in-the-Loop (HITL)

The agent proposes actions and waits for human approval before executing.

from dataclasses import dataclass, field
from enum import Enum
from typing import Optional, Callable


class ActionType(Enum):
    SEARCH = "search"
    WRITE = "write"
    DELETE = "delete"
    EXECUTE = "execute"
    RESPOND = "respond"


@dataclass
class ProposedAction:
    action_type: ActionType
    description: str
    args: dict = field(default_factory=dict)
    risk_level: str = "low"  # low, medium, high


class HITLAgent:
    """Agent that requires human approval for actions."""

    def __init__(self, auto_approve_low_risk: bool = True):
        self.auto_approve_low_risk = auto_approve_low_risk
        self.history: list[dict] = []

    def propose(self, action: ProposedAction) -> bool:
        """Propose an action and get approval."""
        print(f"\n🤖 Proposed: [{action.risk_level.upper()}] {action.action_type.value}: {action.description}")

        if self.auto_approve_low_risk and action.risk_level == "low":
            print("  ✅ Auto-approved (low risk)")
            self.history.append({"action": action, "approved": True, "auto": True})
            return True

        # Simulate human decision (in production: UI prompt, Slack webhook, etc.)
        # For demo, auto-approve medium, reject high
        approved = action.risk_level != "high"
        status = "✅ Approved" if approved else "❌ Rejected"
        print(f"  👤 Human: {status}")
        self.history.append({"action": action, "approved": approved, "auto": False})
        return approved

    def execute_if_approved(self, action: ProposedAction) -> Optional[str]:
        if self.propose(action):
            return f"Executed: {action.description}"
        return None


# Demo
agent = HITLAgent(auto_approve_low_risk=True)

actions = [
    ProposedAction(ActionType.SEARCH, "Search knowledge base for 'RAG'", risk_level="low"),
    ProposedAction(ActionType.WRITE, "Save summary to report.md", risk_level="medium"),
    ProposedAction(ActionType.DELETE, "Delete outdated index", risk_level="high"),
    ProposedAction(ActionType.RESPOND, "Send answer to user", risk_level="low"),
]

for action in actions:
    result = agent.execute_if_approved(action)
    if result:
        print(f"  → {result}")

print(f"\n📊 History: {len(agent.history)} actions, {sum(1 for h in agent.history if h['approved'])} approved")

4. Human-on-the-Loop (HOTL) Monitoring

The agent runs autonomously but a monitor watches for anomalies and can kill the run.

import time
from dataclasses import dataclass, field
from typing import Optional


@dataclass
class MonitorState:
    """State tracked by the human-on-the-loop monitor."""
    steps_taken: int = 0
    max_steps: int = 10
    errors: list[str] = field(default_factory=list)
    warnings: list[str] = field(default_factory=list)
    cost_usd: float = 0.0
    max_cost_usd: float = 0.50
    kill_switch: bool = False

    def log_step(self, action: str, cost: float = 0.01):
        self.steps_taken += 1
        self.cost_usd += cost

    def check_health(self) -> dict:
        issues = []
        if self.steps_taken >= self.max_steps:
            issues.append(f"Max steps reached ({self.max_steps})")
        if self.cost_usd >= self.max_cost_usd:
            issues.append(f"Cost limit reached (${self.cost_usd:.2f})")
        if len(self.errors) > 3:
            issues.append(f"Too many errors ({len(self.errors)})")

        should_kill = len(issues) > 0 or self.kill_switch
        return {"healthy": not should_kill, "issues": issues}


def run_agent_with_monitoring(task: str, monitor: MonitorState) -> str:
    """Run agent steps with HOTL monitoring."""
    print(f"🚀 Starting: {task}")

    for i in range(15):
        # Check monitor
        health = monitor.check_health()
        if not health["healthy"]:
            print(f"\n🛑 MONITOR KILL: {health['issues']}")
            return f"Stopped at step {i}: {health['issues']}"

        # Simulate agent step
        monitor.log_step(f"step_{i}", cost=0.08)
        success = i != 7  # Simulate error at step 7

        if success:
            print(f"  Step {i+1}: ✅ (cost: ${monitor.cost_usd:.2f})")
        else:
            monitor.errors.append(f"Step {i+1} failed")
            print(f"  Step {i+1}: ⚠️ Error")

    return "Completed all steps"


monitor = MonitorState(max_steps=10, max_cost_usd=0.50)
result = run_agent_with_monitoring("Research RAG best practices", monitor)
print(f"\nResult: {result}")

5. Policy-Based Autonomy

Define policies that automatically allow, block, or escalate actions based on rules.

from dataclasses import dataclass, field
from typing import Callable, Optional


@dataclass
class PolicyRule:
    """A single policy rule for agent actions."""
    name: str
    condition: Callable[[dict], bool]  # Returns True if rule applies
    action: str  # "allow", "deny", "escalate"
    priority: int = 0  # Higher = checked first


class PolicyEnforcedAgent:
    """Agent governed by a set of policy rules."""

    def __init__(self, rules: list[PolicyRule]):
        self.rules = sorted(rules, key=lambda r: -r.priority)
        self.escalation_queue: list[dict] = []
        self.audit_log: list[dict] = []

    def evaluate(self, action: dict) -> str:
        """Evaluate an action against policies."""
        for rule in self.rules:
            if rule.condition(action):
                decision = rule.action
                self.audit_log.append({
                    "action": action, "rule": rule.name, "decision": decision,
                })

                if decision == "escalate":
                    self.escalation_queue.append(action)

                return decision

        # Default: deny
        self.audit_log.append({"action": action, "rule": "default", "decision": "deny"})
        return "deny"


# Define policies
rules = [
    PolicyRule(
        name="block_destructive",
        condition=lambda a: a.get("type") in ("delete", "drop", "truncate"),
        action="deny",
        priority=100,
    ),
    PolicyRule(
        name="escalate_writes",
        condition=lambda a: a.get("type") in ("write", "update", "create"),
        action="escalate",
        priority=50,
    ),
    PolicyRule(
        name="allow_reads",
        condition=lambda a: a.get("type") in ("search", "read", "list"),
        action="allow",
        priority=10,
    ),
    PolicyRule(
        name="cost_limit",
        condition=lambda a: a.get("estimated_cost", 0) > 0.10,
        action="escalate",
        priority=80,
    ),
]

agent = PolicyEnforcedAgent(rules)

test_actions = [
    {"type": "search", "query": "RAG papers", "estimated_cost": 0.01},
    {"type": "write", "target": "report.md", "estimated_cost": 0.02},
    {"type": "delete", "target": "old_index", "estimated_cost": 0.001},
    {"type": "search", "query": "expensive query", "estimated_cost": 0.50},
    {"type": "read", "target": "config.yaml", "estimated_cost": 0.001},
]

for action in test_actions:
    decision = agent.evaluate(action)
    icon = {"allow": "✅", "deny": "🚫", "escalate": "⚠️"}[decision]
    print(f"{icon} {decision:10s} | {action['type']:8s} | {action.get('target', action.get('query', ''))}")

print(f"\n📋 Escalation queue: {len(agent.escalation_queue)} items")
print(f"📝 Audit log: {len(agent.audit_log)} entries")

6. LangGraph HITL with interrupt()

Use LangGraph’s built-in interrupt() for human-in-the-loop workflows.

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)


class HITLState(TypedDict):
    messages: Annotated[list, add_messages]
    proposed_action: str
    approved: bool


def plan_node(state: HITLState) -> dict:
    """Agent plans the next action."""
    response = llm.invoke([
        {"role": "system", "content": "Plan the next action. Describe what you want to do."},
        *state["messages"],
    ])
    return {"proposed_action": response.content, "approved": False}


def human_approval_node(state: HITLState) -> dict:
    """Simulate human approval (in production: interrupt() + UI)."""
    action = state["proposed_action"]
    # Simulate: approve if action doesn't mention 'delete'
    approved = "delete" not in action.lower()
    status = "approved" if approved else "rejected"
    print(f"\n👤 Human {status} action: {action[:80]}...")
    return {"approved": approved}


def execute_node(state: HITLState) -> dict:
    if not state["approved"]:
        return {"messages": [{"role": "assistant", "content": "Action was rejected by human reviewer."}]}

    return {"messages": [{"role": "assistant", "content": f"Executed: {state['proposed_action'][:100]}"}]}


# Build graph
graph = StateGraph(HITLState)
graph.add_node("plan", plan_node)
graph.add_node("approve", human_approval_node)
graph.add_node("execute", execute_node)

graph.add_edge(START, "plan")
graph.add_edge("plan", "approve")
graph.add_edge("approve", "execute")
graph.add_edge("execute", END)

checkpointer = MemorySaver()
hitl_app = graph.compile(checkpointer=checkpointer)

# Run
result = hitl_app.invoke(
    {
        "messages": [{"role": "user", "content": "Search for recent papers on RAG evaluation."}],
        "proposed_action": "", "approved": False,
    },
    config={"configurable": {"thread_id": "demo-1"}},
)

print(f"\n Final: {result['messages'][-1].content}")

Table of Contents

2. Autonomy Spectrum Overview

3. Human-in-the-Loop (HITL)

4. Human-on-the-Loop (HOTL) Monitoring

5. Policy-Based Autonomy

6. LangGraph HITL with interrupt()