Patrón Human-in-the-Loop

Descripción general

Los agentes LLM autónomos pueden tomar acciones con consecuencias reales: enviar emails, desplegar código, modificar bases de datos, hacer compras. El patrón Human-in-the-Loop inserta un checkpoint donde un humano revisa la acción propuesta del agente antes de que se ejecute. El agente se pausa, presenta su plan y espera aprobación, rechazo o modificación.

No cada paso merece una pausa. Dos señales deciden cuáles sí: un umbral de confianza y el nivel de riesgo de la acción. Leer un archivo o buscar en la web pasa automáticamente. Desplegar a producción o enviar comunicaciones externas siempre se detiene para revisión. Combina bien con el patrón LLM guardrails: los guardrails acotan lo que el agente puede hacer, este patrón decide quién lo autoriza.

Cuándo usarlo

• El agente puede tomar acciones irreversibles (borrar datos, enviar comunicaciones, desplegar)
• El costo de una acción equivocada excede el costo de esperar revisión humana
• Requisitos regulatorios o de compliance exigen supervisión humana
• El agente opera en un entorno desconocido donde la confianza es baja
• Ejemplos: agentes de despliegue, agentes de redacción de emails, agentes de migración de base de datos, agentes de transacciones financieras

Cuándo NO usarlo

• Acciones de solo lectura o totalmente reversibles; pausar por esas solo gasta atención del revisor
• Rutas críticas de latencia donde un humano en el loop es demasiado lento (autoscaling, routing en tiempo real)
• Cuando la necesidad real es enrutar entre herramientas; el patrón de selección de herramientas cubre esa decisión
• Prototipos tempranos que aún están definiendo su espacio de acciones; añade el checkpoint cuando las acciones se estabilicen

Solución

Python

from dataclasses import dataclass, field
from typing import Callable, Dict, List, Optional, Any
from enum import Enum

class RiskLevel(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"

class ApprovalStatus(Enum):
    APPROVED = "approved"
    REJECTED = "rejected"
    MODIFIED = "modified"

@dataclass
class AgentAction:
    name: str
    description: str
    risk: RiskLevel
    parameters: Dict[str, Any] = field(default_factory=dict)
    confidence: float = 1.0

@dataclass
class ApprovalRequest:
    action: AgentAction
    reason: str
    agent_state: Dict[str, Any] = field(default_factory=dict)

@dataclass
class ApprovalResponse:
    status: ApprovalStatus
    modified_parameters: Optional[Dict[str, Any]] = None
    feedback: str = ""

class HumanInTheLoop:
    def __init__(
        self,
        confidence_threshold: float = 0.7,
        auto_approve_low_risk: bool = True,
        reviewer: Optional[Callable[[ApprovalRequest], ApprovalResponse]] = None,
    ):
        self.confidence_threshold = confidence_threshold
        self.auto_approve_low_risk = auto_approve_low_risk
        self.reviewer = reviewer or self._default_reviewer

    def _default_reviewer(self, request: ApprovalRequest) -> ApprovalResponse:
        print(f"\n--- APPROVAL REQUIRED ---")
        print(f"Action: {request.action.name}")
        print(f"Description: {request.action.description}")
        print(f"Risk: {request.action.risk.value}")
        print(f"Confidence: {request.action.confidence:.0%}")
        print(f"Parameters: {request.action.parameters}")
        print(f"Reason: {request.reason}")
        print(f"-------------------------")

        choice = input("Approve? (y/n/m=modify): ").strip().lower()
        if choice == "y":
            return ApprovalResponse(status=ApprovalStatus.APPROVED)
        elif choice == "m":
            new_params = dict(request.action.parameters)
            key = input("Parameter to modify (or empty to skip): ").strip()
            if key and key in new_params:
                new_params[key] = input(f"New value for {key}: ").strip()
            return ApprovalResponse(
                status=ApprovalStatus.MODIFIED,
                modified_parameters=new_params,
            )
        return ApprovalResponse(status=ApprovalStatus.REJECTED, feedback="Rejected by reviewer")

    def needs_approval(self, action: AgentAction) -> bool:
        if action.risk == RiskLevel.HIGH:
            return True
        if action.risk == RiskLevel.MEDIUM and action.confidence < self.confidence_threshold:
            return True
        if action.risk == RiskLevel.LOW and not self.auto_approve_low_risk:
            return True
        return False

    def execute_with_approval(
        self,
        action: AgentAction,
        execute_fn: Callable[[AgentAction], str],
        reason: str = "",
    ) -> str:
        if not self.needs_approval(action):
            print(f"[AUTO] Executing: {action.name}")
            return execute_fn(action)

        request = ApprovalRequest(action=action, reason=reason)
        response = self.reviewer(request)

        if response.status == ApprovalStatus.REJECTED:
            return f"[REJECTED] {action.name}: {response.feedback}"

        if response.status == ApprovalStatus.MODIFIED and response.modified_parameters:
            action.parameters = response.modified_parameters

        print(f"[APPROVED] Executing: {action.name}")
        return execute_fn(action)

# Uso
def mock_execute(action: AgentAction) -> str:
    return f"Executed {action.name} with {action.parameters}"

hitl = HumanInTheLoop(confidence_threshold=0.8)

actions = [
    AgentAction("read_file", "Read config file", RiskLevel.LOW, {"path": "/etc/config.yml"}, 0.95),
    AgentAction("run_tests", "Execute test suite", RiskLevel.LOW, {"suite": "unit"}, 0.9),
    AgentAction("send_email", "Send notification to client", RiskLevel.HIGH, {"to": "client@example.com", "subject": "Update"}, 0.85),
    AgentAction("deploy", "Deploy to production", RiskLevel.HIGH, {"env": "prod", "version": "v2.1.0"}, 0.6),
    AgentAction("delete_records", "Delete stale records", RiskLevel.MEDIUM, {"table": "logs", "older_than": "30d"}, 0.5),
]

for action in actions:
    result = hitl.execute_with_approval(action, mock_execute, reason="Agent workflow step")
    print(f"  Result: {result}\n")

JavaScript

const RiskLevel = { LOW: "low", MEDIUM: "medium", HIGH: "high" };
const ApprovalStatus = { APPROVED: "approved", REJECTED: "rejected", MODIFIED: "modified" };

class AgentAction {
  constructor(name, description, risk, parameters = {}, confidence = 1.0) {
    this.name = name;
    this.description = description;
    this.risk = risk;
    this.parameters = parameters;
    this.confidence = confidence;
  }
}

class ApprovalResponse {
  constructor(status, modifiedParameters = null, feedback = "") {
    this.status = status;
    this.modifiedParameters = modifiedParameters;
    this.feedback = feedback;
  }
}

class HumanInTheLoop {
  constructor(options = {}) {
    this.confidenceThreshold = options.confidenceThreshold ?? 0.7;
    this.autoApproveLowRisk = options.autoApproveLowRisk ?? true;
    this.reviewer = options.reviewer || ((req) => {
      console.log(`\n--- APPROVAL REQUIRED ---`);
      console.log(`Action: ${req.action.name}, Risk: ${req.action.risk}, Confidence: ${(req.action.confidence * 100).toFixed(0)}%`);
      return new ApprovalResponse(ApprovalStatus.APPROVED);
    });
  }

  needsApproval(action) {
    if (action.risk === RiskLevel.HIGH) return true;
    if (action.risk === RiskLevel.MEDIUM && action.confidence < this.confidenceThreshold) return true;
    if (action.risk === RiskLevel.LOW && !this.autoApproveLowRisk) return true;
    return false;
  }

  executeWithApproval(action, executeFn, reason = "") {
    if (!this.needsApproval(action)) {
      console.log(`[AUTO] Executing: ${action.name}`);
      return executeFn(action);
    }

    const response = this.reviewer({ action, reason });

    if (response.status === ApprovalStatus.REJECTED) {
      return `[REJECTED] ${action.name}: ${response.feedback}`;
    }

    if (response.status === ApprovalStatus.MODIFIED && response.modifiedParameters) {
      action.parameters = response.modifiedParameters;
    }

    console.log(`[APPROVED] Executing: ${action.name}`);
    return executeFn(action);
  }
}

// Uso
function mockExecute(action) {
  return `Executed ${action.name} with ${JSON.stringify(action.parameters)}`;
}

const hitl = new HumanInTheLoop({ confidenceThreshold: 0.8 });

const actions = [
  new AgentAction("read_file", "Read config", RiskLevel.LOW, { path: "/etc/config.yml" }, 0.95),
  new AgentAction("send_email", "Notify client", RiskLevel.HIGH, { to: "client@example.com" }, 0.85),
  new AgentAction("deploy", "Deploy to prod", RiskLevel.HIGH, { env: "prod" }, 0.6),
  new AgentAction("delete_records", "Delete stale logs", RiskLevel.MEDIUM, { table: "logs" }, 0.5),
];

for (const action of actions) {
  const result = hitl.executeWithApproval(action, mockExecute, "Agent step");
  console.log(`  Result: ${result}\n`);
}

Java

import java.util.*;

public class HumanInTheLoop {

    enum RiskLevel { LOW, MEDIUM, HIGH }
    enum ApprovalStatus { APPROVED, REJECTED, MODIFIED }

    record AgentAction(String name, String description, RiskLevel risk,
                       Map<String, Object> parameters, double confidence) {}
    record ApprovalResponse(ApprovalStatus status, Map<String, Object> modifiedParams, String feedback) {}
    record ApprovalRequest(AgentAction action, String reason) {}

    private final double confidenceThreshold;
    private final boolean autoApproveLowRisk;
    private final java.util.function.Function<ApprovalRequest, ApprovalResponse> reviewer;

    public HumanInTheLoop(double confidenceThreshold, boolean autoApproveLowRisk,
                          java.util.function.Function<ApprovalRequest, ApprovalResponse> reviewer) {
        this.confidenceThreshold = confidenceThreshold;
        this.autoApproveLowRisk = autoApproveLowRisk;
        this.reviewer = reviewer != null ? reviewer : req -> {
            System.out.printf("%n--- APPROVAL REQUIRED ---%n");
            System.out.printf("Action: %s, Risk: %s, Confidence: %.0f%%%n",
                req.action().name(), req.action().risk(), req.action().confidence() * 100);
            return new ApprovalResponse(ApprovalStatus.APPROVED, null, "");
        };
    }

    boolean needsApproval(AgentAction action) {
        if (action.risk() == RiskLevel.HIGH) return true;
        if (action.risk() == RiskLevel.MEDIUM && action.confidence() < confidenceThreshold) return true;
        if (action.risk() == RiskLevel.LOW && !autoApproveLowRisk) return true;
        return false;
    }

    public String executeWithApproval(AgentAction action,
                                       java.util.function.Function<AgentAction, String> executeFn,
                                       String reason) {
        if (!needsApproval(action)) {
            System.out.println("[AUTO] Executing: " + action.name());
            return executeFn.apply(action);
        }

        var response = reviewer.apply(new ApprovalRequest(action, reason));

        if (response.status() == ApprovalStatus.REJECTED) {
            return "[REJECTED] " + action.name() + ": " + response.feedback();
        }

        System.out.println("[APPROVED] Executing: " + action.name());
        return executeFn.apply(action);
    }

    public static void main(String[] args) {
        var hitl = new HumanInTheLoop(0.8, true, null);

        var actions = List.of(
            new AgentAction("read_file", "Read config", RiskLevel.LOW, Map.of("path", "/etc/config.yml"), 0.95),
            new AgentAction("send_email", "Notify client", RiskLevel.HIGH, Map.of("to", "client@example.com"), 0.85),
            new AgentAction("deploy", "Deploy to prod", RiskLevel.HIGH, Map.of("env", "prod"), 0.6),
            new AgentAction("delete_records", "Delete stale logs", RiskLevel.MEDIUM, Map.of("table", "logs"), 0.5)
        );

        for (var action : actions) {
            String result = hitl.executeWithApproval(action,
                a -> "Executed " + a.name() + " with " + a.parameters(),
                "Agent step");
            System.out.println("  Result: " + result + "\n");
        }
    }
}

Explicación

El flujo de aprobación es un embudo:

El patrón controla la ejecución detrás de tres verificaciones:

1. Clasificación de riesgo: cada acción se etiqueta como riesgo bajo, medio o alto. Las de alto riesgo siempre requieren aprobación. Las de medio riesgo requieren aprobación solo cuando la confianza está bajo el umbral. Las de bajo riesgo se auto-ejecutan a menos que se configure lo contrario.
2. Flujo de aprobación: cuando se necesita aprobación, el agente se pausa y presenta los detalles de la acción (nombre, parámetros, confianza, razón) a un revisor. El revisor puede aprobar, rechazar o modificar los parámetros.
3. Ejecución o abort: si se aprueba (o se modifica y aprueba), la acción se ejecuta. Si se rechaza, el agente recibe feedback y puede ajustar su plan.

El revisor puede ser un humano vía CLI, una UI web, un botón de aprobación en Slack, o incluso un modelo secundario automático para decisiones menos críticas.

Variantes

Variante	Descripción	Caso de uso
Aprobación asíncrona	El agente pausa y resume cuando llega la aprobación vía webhook	Agentes en producción con UIs web
Aprobación en lote	Presentar múltiples acciones a la vez para una sola revisión	Reduce fatiga del revisor en workflows rutinarios
Cadena de escalado	Enrutar a revisor senior si el primero rechaza	Supervisión en niveles para operaciones sensibles
Auto-rechazo por timeout	Si no hay aprobación en N minutos, rechazar por defecto	Previene que agentes cuelguen indefinidamente

Buenas prácticas

• Clasifica el riesgo de forma conservativa: cuando dudes, marca como alto riesgo
• Incluye contexto en las solicitudes: muestra el razonamiento del agente, no solo la acción
• Define timeouts: previene que agentes esperen indefinidamente por input humano
• Registra todas las decisiones de aprobación: auditoría para compliance y mejora
• Permite a los revisores modificar parámetros: a veces la acción es correcta pero los detalles necesitan ajuste
• Reduce gradualmente los requisitos: conforme el agente demuestra confiabilidad, baja el umbral de confianza

Errores comunes

• Requerir aprobación para cada paso, causando fatiga del revisor y ejecución lenta
• No proporcionar suficiente contexto para que el revisor tome una decisión informada
• Sin timeout, dejando agentes atascados esperando revisores que no responden
• No registrar aprobaciones, haciendo imposible las auditorías
• Tratar rechazos como fallos en lugar de feedback para que el agente se ajuste