Cuándo Automatizar vs Cuándo Pausar

Descripción de la cápsula

Esta es la decisión de diseño más importante para agentes en producción: ¿qué debería disparar una interrupción humana?

Demasiadas interrupciones → el agente es inútil. Si te pide permiso para cada búsqueda en Google, para cada cálculo, para cada paso intermedio — más rápido lo haces tú. Demasiadas pocas interrupciones → el agente es peligroso. Si ejecuta transacciones financieras, borra datos, o envía emails sin preguntarte — un error cuesta caro.

El sweet spot está en medio, y no es intuitivo. Esta cápsula te da un framework concreto con cuatro criterios que puedes aplicar a cualquier acción de cualquier agente. No es teoría — es una función de Python que evalúa el riesgo de una acción y decide automáticamente si interrumpir o no.

En las cápsulas anteriores aprendiste la mecánica: interrupt() para pausar, Command(resume=...) para reanudar, breakpoints, edición de estado. Ahora tienes todas las herramientas. La pregunta ya no es "¿cómo interrumpo?" sino "¿cuándo debo interrumpir?"

El problema: dos extremos que no funcionan

Extremo 1: Interrumpir todo

Usuario: "Investiga AI agents"

Agente: "Voy a descomponer el tema en sub-preguntas. ¿Procedo?"
Usuario: "Sí"
Agente: "Voy a buscar en Wikipedia. ¿Procedo?"
Usuario: "Sí..."
Agente: "Voy a buscar en arXiv. ¿Procedo?"
Usuario: "SÍ"
Agente: "Voy a fusionar resultados. ¿Procedo?"
Usuario: "HAZLO YA"
Agente: "Voy a generar el resumen. ¿Procedo?"
Usuario: *cierra la aplicación*

Resultado: 5 interrupciones para una tarea que debería ser autónoma. El usuario pierde 3 minutos aprobando cosas obvias. La experiencia es peor que no tener agente.

Extremo 2: No interrumpir nada

Usuario: "Investiga AI agents y envía el reporte al equipo"

Agente: [busca en 5 fuentes] ✅
Agente: [llama API premium — $2.50] ✅
Agente: [genera reporte con datos incorrectos] ✅
Agente: [envía email a 50 personas con datos incorrectos] ✅

Usuario: "... acabo de enviar datos falsos a mi equipo"

Resultado: el agente actuó con autonomía total. Un dato incorrecto se propagó a 50 personas. El costo de corregirlo es enorme.

El objetivo: interrumpir en el punto exacto

Usuario: "Investiga AI agents y envía el reporte al equipo"

Agente: [busca en fuentes gratuitas] ✅ (auto-approve)
Agente: "Encontré 3 fuentes gratuitas. También puedo buscar en
         arXiv Premium ($2.50). ¿Procedo?"
Usuario: "No, las gratuitas bastan"
Agente: [genera borrador del reporte]
Agente: "Aquí está el borrador. ¿Lo envío a 50 personas?"
Usuario: [revisa, corrige un dato] "Ahora sí, envía"
Agente: [envía email corregido] ✅

Resultado: 2 interrupciones — ambas justificadas. La primera evitó un gasto innecesario. La segunda evitó propagar un error.

El framework de decisión: 4 criterios

Cada acción que un agente puede ejecutar se evalúa en 4 dimensiones. La combinación de las 4 produce un nivel de riesgo que determina si el agente debe pausar o continuar.

Criterio 1: Costo

¿La acción cuesta dinero?

El costo incluye: llamadas a APIs de pago, consumo de tokens de modelos caros, servicios externos con billing, y cualquier acción que genere un cargo.

Criterio 2: Reversibilidad

¿Se puede deshacer la acción?

La pregunta clave: "Si esta acción sale mal, ¿puedo deshacerla sin consecuencias?"

Criterio 3: Impacto

¿A cuántas personas o sistemas afecta?

Una búsqueda en Google afecta solo al agente (auto-contenido). Un email a un cliente afecta la relación con ese cliente (externo limitado). Un post en redes sociales afecta la reputación de la empresa (externo amplio).

Criterio 4: Confianza

¿Qué tan seguro está el agente de que la acción es correcta?

La confianza no siempre es un número explícito. Puede inferirse: si el agente encontró múltiples fuentes que coinciden, la confianza es alta. Si los resultados son contradictorios o ambiguos, es baja.

La matriz de decisión

Cada acción recibe un puntaje en cada criterio. El puntaje combinado determina la política:

Puntaje de riesgo = max(costo_score, reversibilidad_score, impacto_score) + (1 - confianza_score)

Si riesgo < 0.3  →  AUTO-APPROVE (ejecutar sin preguntar)
Si riesgo 0.3-0.7 →  QUICK APPROVE (mostrar plan, aprobación rápida)
Si riesgo > 0.7  →  FULL REVIEW (mostrar detalle completo, esperar aprobación explícita)

El operador max() en los tres primeros criterios significa que un solo criterio en rojo es suficiente para escalar. Una acción gratuita, auto-contenida, pero irreversible aún requiere aprobación.

Acciones clasificadas con el framework

Apliquemos el framework a acciones comunes de un agente de investigación:

La tabla hace evidente por qué "interrumpir todo" es incorrecto: la mitad de las acciones son safe para auto-approve. Y por qué "no interrumpir nada" es peligroso: las últimas tres líneas necesitan supervisión humana siempre.

Implementación: función de evaluación de riesgo

Traslademos el framework a código ejecutable:

from dotenv import load_dotenv
load_dotenv()

from dataclasses import dataclass
from enum import Enum


class CostLevel(Enum):
    NEGLIGIBLE = 0.0
    LOW = 0.3
    MEDIUM = 0.6
    HIGH = 1.0


class Reversibility(Enum):
    REVERSIBLE = 0.0
    PARTIAL = 0.5
    IRREVERSIBLE = 1.0


class Impact(Enum):
    SELF_CONTAINED = 0.0
    EXTERNAL_LIMITED = 0.4
    EXTERNAL_BROAD = 1.0


class RiskDecision(Enum):
    AUTO_APPROVE = "auto_approve"
    QUICK_APPROVE = "quick_approve"
    FULL_REVIEW = "full_review"


@dataclass
class ActionRisk:
    action_name: str
    cost: CostLevel
    reversibility: Reversibility
    impact: Impact
    confidence: float

    @property
    def risk_score(self) -> float:
        base = max(self.cost.value, self.reversibility.value, self.impact.value)
        confidence_penalty = 1.0 - self.confidence
        return min(round(base + confidence_penalty * 0.5, 2), 1.0)

    @property
    def decision(self) -> RiskDecision:
        score = self.risk_score
        if score < 0.3:
            return RiskDecision.AUTO_APPROVE
        elif score <= 0.7:
            return RiskDecision.QUICK_APPROVE
        else:
            return RiskDecision.FULL_REVIEW


actions = [
    ActionRisk("web_search_free", CostLevel.NEGLIGIBLE, Reversibility.REVERSIBLE, Impact.SELF_CONTAINED, 0.95),
    ActionRisk("api_search_paid", CostLevel.LOW, Reversibility.REVERSIBLE, Impact.SELF_CONTAINED, 0.90),
    ActionRisk("send_email_one", CostLevel.NEGLIGIBLE, Reversibility.PARTIAL, Impact.EXTERNAL_LIMITED, 0.80),
    ActionRisk("send_email_team", CostLevel.NEGLIGIBLE, Reversibility.PARTIAL, Impact.EXTERNAL_BROAD, 0.75),
    ActionRisk("delete_records", CostLevel.NEGLIGIBLE, Reversibility.IRREVERSIBLE, Impact.EXTERNAL_BROAD, 0.95),
    ActionRisk("call_expensive_api", CostLevel.HIGH, Reversibility.REVERSIBLE, Impact.SELF_CONTAINED, 0.90),
    ActionRisk("generate_draft", CostLevel.NEGLIGIBLE, Reversibility.REVERSIBLE, Impact.SELF_CONTAINED, 0.70),
]

print("=== Evaluación de riesgo por acción ===\n")
print(f"{'Acción':<22} {'Riesgo':>7} {'Decisión':<16}")
print(f"{'-'*22} {'-'*7} {'-'*16}")

for a in actions:
    print(f"{a.action_name:<22} {a.risk_score:>6.2f}  {a.decision.value}")
# Output esperado:
# === Evaluación de riesgo por acción ===
#
# Acción                  Riesgo Decisión
# ---------------------- ------- ----------------
# web_search_free          0.02  auto_approve
# api_search_paid          0.35  quick_approve
# send_email_one           0.60  quick_approve
# send_email_team          1.00  full_review
# delete_records           1.00  full_review
# call_expensive_api       1.00  full_review
# generate_draft           0.15  auto_approve

La función risk_score usa max() para los tres primeros criterios y agrega una penalización por baja confianza. Esto captura la intuición: una acción irreversible con alta confianza sigue siendo riesgosa (el costo del error es alto aunque sea improbable).

Integración con interrupt(): decisión condicional

Ahora conectamos la evaluación de riesgo con el mecanismo de interrupt() dentro de un grafo:

from dotenv import load_dotenv
load_dotenv()

from dataclasses import dataclass
from enum import Enum
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command


class CostLevel(Enum):
    NEGLIGIBLE = 0.0
    LOW = 0.3
    MEDIUM = 0.6
    HIGH = 1.0

class Reversibility(Enum):
    REVERSIBLE = 0.0
    PARTIAL = 0.5
    IRREVERSIBLE = 1.0

class Impact(Enum):
    SELF_CONTAINED = 0.0
    EXTERNAL_LIMITED = 0.4
    EXTERNAL_BROAD = 1.0

class RiskDecision(Enum):
    AUTO_APPROVE = "auto_approve"
    QUICK_APPROVE = "quick_approve"
    FULL_REVIEW = "full_review"

@dataclass
class ActionRisk:
    action_name: str
    cost: CostLevel
    reversibility: Reversibility
    impact: Impact
    confidence: float

    @property
    def risk_score(self) -> float:
        base = max(self.cost.value, self.reversibility.value, self.impact.value)
        confidence_penalty = 1.0 - self.confidence
        return min(round(base + confidence_penalty * 0.5, 2), 1.0)

    @property
    def decision(self) -> RiskDecision:
        score = self.risk_score
        if score < 0.3:
            return RiskDecision.AUTO_APPROVE
        elif score <= 0.7:
            return RiskDecision.QUICK_APPROVE
        else:
            return RiskDecision.FULL_REVIEW


def assess_risk(action_name: str, cost: float, reversible: bool, external: bool, confidence: float) -> ActionRisk:
    if cost < 0.01:
        cost_level = CostLevel.NEGLIGIBLE
    elif cost < 1.0:
        cost_level = CostLevel.LOW
    elif cost < 10.0:
        cost_level = CostLevel.MEDIUM
    else:
        cost_level = CostLevel.HIGH

    rev = Reversibility.REVERSIBLE if reversible else Reversibility.IRREVERSIBLE
    imp = Impact.EXTERNAL_BROAD if external else Impact.SELF_CONTAINED

    return ActionRisk(action_name, cost_level, rev, imp, confidence)


class State(TypedDict):
    topic: str
    sources: Annotated[list[str], operator.add]
    report: str
    actions_log: Annotated[list[str], operator.add]


def plan_and_search(state: State) -> dict:
    topic = state["topic"]
    actions = []

    free_search = assess_risk("web_search", cost=0.0, reversible=True, external=False, confidence=0.95)
    if free_search.decision == RiskDecision.AUTO_APPROVE:
        actions.append(f"[AUTO] web_search (riesgo: {free_search.risk_score:.2f})")
        sources = [f"Web: resultados gratuitos sobre '{topic}'"]
    else:
        response = interrupt({
            "action": "web_search",
            "risk_score": free_search.risk_score,
            "message": f"¿Buscar en web sobre '{topic}'?",
        })
        sources = [f"Web: resultados sobre '{topic}'"] if response.get("approved") else []
        actions.append(f"[APPROVED] web_search" if response.get("approved") else "[REJECTED] web_search")

    paid_search = assess_risk("premium_api", cost=2.50, reversible=True, external=False, confidence=0.85)
    if paid_search.decision == RiskDecision.AUTO_APPROVE:
        sources.append(f"Premium: datos pagados sobre '{topic}'")
        actions.append(f"[AUTO] premium_api")
    else:
        response = interrupt({
            "action": "premium_api",
            "risk_score": paid_search.risk_score,
            "cost": 2.50,
            "message": f"Buscar en API premium ($2.50) sobre '{topic}'. ¿Procedo?",
        })
        if response.get("approved"):
            sources.append(f"Premium: datos pagados sobre '{topic}'")
            actions.append(f"[APPROVED] premium_api (${2.50})")
        else:
            actions.append(f"[REJECTED] premium_api (${2.50} ahorrado)")

    return {"sources": sources, "actions_log": actions}


def generate_report(state: State) -> dict:
    source_list = "\n".join(f"  - {s}" for s in state["sources"])
    report = f"Reporte sobre '{state['topic']}':\n{source_list}\nFuentes: {len(state['sources'])}"
    return {"report": report, "actions_log": [f"[AUTO] generate_report"]}


graph_builder = StateGraph(State)
graph_builder.add_node("search", plan_and_search)
graph_builder.add_node("report", generate_report)
graph_builder.add_edge(START, "search")
graph_builder.add_edge("search", "report")
graph_builder.add_edge("report", END)

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

config = {"configurable": {"thread_id": "risk-demo"}}

print("=== Invocación 1: la búsqueda gratuita se auto-aprueba ===")
result = graph.invoke(
    {"topic": "AI agents", "sources": [], "report": "", "actions_log": []},
    config,
)

print(f"\nEstado actual: {graph.get_state(config).next}")
if graph.get_state(config).next:
    print("Interrupt detectado — el agente pide aprobación para la API premium")
    print("Reanudando con aprobación...")

    result = graph.invoke(
        Command(resume={"approved": True}),
        config,
    )

print(f"\nReporte:\n{result['report']}")
print(f"\nLog de acciones:")
for action in result["actions_log"]:
    print(f"  {action}")
# Output esperado:
# === Invocación 1: la búsqueda gratuita se auto-aprueba ===
#
# Estado actual: ('search',)
# Interrupt detectado — el agente pide aprobación para la API premium
# Reanudando con aprobación...
#
# Reporte:
# Reporte sobre 'AI agents':
#   - Web: resultados gratuitos sobre 'AI agents'
#   - Premium: datos pagados sobre 'AI agents'
# Fuentes: 2
#
# Log de acciones:
#   [AUTO] web_search (riesgo: 0.02)
#   [APPROVED] premium_api ($2.5)
#   [AUTO] generate_report

Observa el patrón: la búsqueda gratuita pasó sin preguntar (risk_score: 0.02). La API premium ($2.50) disparó un interrupt. El reporte se generó automáticamente (es reversible, sin costo, auto-contenido). Dos de tres acciones fueron automáticas — solo una requirió aprobación humana.

Calibración: reducir falsos positivos y negativos

Un framework de riesgo no se configura una vez y se olvida. Se calibra con datos reales.

Falsos positivos: interrupciones innecesarias

El agente pide permiso, el humano siempre dice "sí." Esto indica que el threshold para esa acción es demasiado bajo.

from dotenv import load_dotenv
load_dotenv()


def calculate_auto_approve_rate(approval_history: list[dict]) -> dict:
    """Calcula la tasa de aprobación por tipo de acción."""
    stats = {}

    for entry in approval_history:
        action = entry["action"]
        if action not in stats:
            stats[action] = {"total": 0, "approved": 0}
        stats[action]["total"] += 1
        if entry["decision"] == "approved":
            stats[action]["approved"] += 1

    recommendations = {}
    for action, data in stats.items():
        rate = data["approved"] / data["total"] if data["total"] > 0 else 0
        if rate >= 0.95 and data["total"] >= 10:
            rec = "AUTO_APPROVE — el usuario siempre aprueba. Eliminar interrupt."
        elif rate >= 0.80:
            rec = "MANTENER — la mayoría se aprueba pero hay rechazos importantes."
        elif rate >= 0.50:
            rec = "REVISAR — el usuario rechaza con frecuencia. Quizá el agente no debería intentar esta acción."
        else:
            rec = "ELIMINAR ACCIÓN — el usuario casi siempre rechaza."
        recommendations[action] = {"rate": rate, "total": data["total"], "recommendation": rec}

    return recommendations


history = [
    {"action": "web_search_paid", "decision": "approved"},
    {"action": "web_search_paid", "decision": "approved"},
    {"action": "web_search_paid", "decision": "approved"},
    {"action": "web_search_paid", "decision": "approved"},
    {"action": "web_search_paid", "decision": "approved"},
    {"action": "web_search_paid", "decision": "approved"},
    {"action": "web_search_paid", "decision": "approved"},
    {"action": "web_search_paid", "decision": "approved"},
    {"action": "web_search_paid", "decision": "approved"},
    {"action": "web_search_paid", "decision": "approved"},
    {"action": "web_search_paid", "decision": "approved"},
    {"action": "web_search_paid", "decision": "approved"},
    {"action": "send_email_team", "decision": "approved"},
    {"action": "send_email_team", "decision": "rejected"},
    {"action": "send_email_team", "decision": "approved"},
    {"action": "send_email_team", "decision": "approved"},
    {"action": "send_email_team", "decision": "rejected"},
    {"action": "send_email_team", "decision": "approved"},
    {"action": "send_email_team", "decision": "approved"},
    {"action": "send_email_team", "decision": "approved"},
    {"action": "send_email_team", "decision": "approved"},
    {"action": "send_email_team", "decision": "approved"},
    {"action": "delete_records", "decision": "rejected"},
    {"action": "delete_records", "decision": "rejected"},
    {"action": "delete_records", "decision": "approved"},
    {"action": "delete_records", "decision": "rejected"},
    {"action": "delete_records", "decision": "rejected"},
]

results = calculate_auto_approve_rate(history)
print("=== Calibración de interrupts ===\n")
for action, data in results.items():
    print(f"{action}:")
    print(f"  Tasa de aprobación: {data['rate']:.0%} ({data['total']} solicitudes)")
    print(f"  → {data['recommendation']}")
    print()
# Output esperado:
# === Calibración de interrupts ===
#
# web_search_paid:
#   Tasa de aprobación: 100% (12 solicitudes)
#   → AUTO_APPROVE — el usuario siempre aprueba. Eliminar interrupt.
#
# send_email_team:
#   Tasa de aprobación: 80% (10 solicitudes)
#   → MANTENER — la mayoría se aprueba pero hay rechazos importantes.
#
# delete_records:
#   Tasa de aprobación: 20% (5 solicitudes)
#   → ELIMINAR ACCIÓN — el usuario casi siempre rechaza.

El dato clave: web_search_paid tiene 100% de aprobación en 12 solicitudes. El interrupt es ruido — elimínalo y haz auto-approve. delete_records tiene 20% de aprobación — el agente no debería intentar borrar registros a menos que el usuario lo pida explícitamente.

Falsos negativos: acciones peligrosas no interceptadas

Estos son más difíciles de detectar porque solo los descubres cuando algo sale mal. La estrategia:

Regla de oro para falsos negativos:
  - Después de cada incidente, pregunta: "¿Había un interrupt que lo hubiera prevenido?"
  - Si sí → agrega el interrupt
  - Si no → la acción debería haberse diseñado diferente

Log de incidentes:
  - 2026-01-15: Agente envió reporte con datos incorrectos
    → Faltaba: interrupt para review de reporte antes de enviar
    → Fix: agregar FULL_REVIEW a "send_report"

  - 2026-01-22: Agente llamó API premium 15 veces en una sesión ($37.50)
    → Faltaba: límite de gasto por sesión con interrupt
    → Fix: agregar interrupt cuando gasto acumulado > $5.00

Niveles de autonomía por usuario

No todos los usuarios necesitan el mismo nivel de supervisión. Un usuario que lleva 200 sesiones con el agente necesita menos interrupciones que uno en su primera sesión.

from dotenv import load_dotenv
load_dotenv()

from dataclasses import dataclass


@dataclass
class UserAutonomy:
    user_id: str
    sessions_completed: int
    approval_rate: float
    trust_score: float

    @property
    def autonomy_level(self) -> str:
        if self.sessions_completed < 5:
            return "supervised"
        elif self.sessions_completed < 50 and self.approval_rate >= 0.85:
            return "standard"
        elif self.sessions_completed >= 50 and self.approval_rate >= 0.90:
            return "autonomous"
        else:
            return "standard"

    @property
    def risk_threshold_adjustment(self) -> float:
        """Ajusta el threshold de riesgo según autonomía.
        Usuarios autónomos toleran más riesgo sin interrupt."""
        adjustments = {
            "supervised": 0.0,
            "standard": 0.15,
            "autonomous": 0.30,
        }
        return adjustments[self.autonomy_level]


def should_interrupt(risk_score: float, user: UserAutonomy) -> bool:
    adjusted_threshold = 0.3 + user.risk_threshold_adjustment
    return risk_score >= adjusted_threshold


users = [
    UserAutonomy("nuevo", sessions_completed=2, approval_rate=1.0, trust_score=0.5),
    UserAutonomy("regular", sessions_completed=30, approval_rate=0.92, trust_score=0.8),
    UserAutonomy("experto", sessions_completed=150, approval_rate=0.95, trust_score=0.95),
]

test_risk_scores = [0.25, 0.40, 0.55, 0.70]

print("=== Autonomía por usuario ===\n")
print(f"{'Usuario':<12} {'Sesiones':>8} {'Nivel':<14} {'Threshold':>9}")
print(f"{'-'*12} {'-'*8} {'-'*14} {'-'*9}")
for u in users:
    threshold = 0.3 + u.risk_threshold_adjustment
    print(f"{u.user_id:<12} {u.sessions_completed:>8} {u.autonomy_level:<14} {threshold:>8.2f}")

print(f"\n{'Riesgo':<10}", end="")
for u in users:
    print(f" {u.user_id:>12}", end="")
print()
print(f"{'-'*10}", end="")
for _ in users:
    print(f" {'-'*12}", end="")
print()

for score in test_risk_scores:
    print(f"{score:<10.2f}", end="")
    for u in users:
        decision = "INTERRUPT" if should_interrupt(score, u) else "auto"
        print(f" {decision:>12}", end="")
    print()
# Output esperado:
# === Autonomía por usuario ===
#
# Usuario      Sesiones Nivel          Threshold
# ------------ -------- -------------- ---------
# nuevo               2 supervised          0.30
# regular            30 standard             0.45
# experto           150 autonomous           0.60
#
# Riesgo           nuevo      regular      experto
# ---------- ------------ ------------ ------------
# 0.25               auto         auto         auto
# 0.40          INTERRUPT         auto         auto
# 0.55          INTERRUPT    INTERRUPT         auto
# 0.70          INTERRUPT    INTERRUPT    INTERRUPT

El usuario nuevo (supervised) recibe interrupt en cualquier acción con riesgo ≥ 0.30. El usuario experto (autonomous) solo se interrumpe con riesgo ≥ 0.60. Mismo agente, mismas acciones, experiencia adaptada al usuario.

Consideraciones de compliance

En industrias reguladas, el framework de riesgo no es suficiente. Hay acciones que siempre requieren aprobación humana, independientemente del riesgo calculado:

Industria financiera (SOX, PCI-DSS):
  - Cualquier transacción > $0 → FULL_REVIEW
  - Acceso a datos de tarjetas → FULL_REVIEW
  - Modificación de registros contables → FULL_REVIEW

Salud (HIPAA):
  - Acceso a datos de pacientes → FULL_REVIEW
  - Comunicación con pacientes → FULL_REVIEW
  - Modificación de historiales → FULL_REVIEW

GDPR:
  - Procesamiento de datos personales → FULL_REVIEW
  - Transferencia de datos cross-border → FULL_REVIEW
  - Eliminación de datos (right to be forgotten) → FULL_REVIEW

La implementación es directa: agrega un override al framework de riesgo.

from dotenv import load_dotenv
load_dotenv()


COMPLIANCE_OVERRIDES = {
    "financial": {
        "any_transaction": "full_review",
        "card_data_access": "full_review",
        "ledger_modification": "full_review",
    },
    "healthcare": {
        "patient_data_access": "full_review",
        "patient_communication": "full_review",
        "record_modification": "full_review",
    },
}


def get_final_decision(
    action: str,
    risk_decision: str,
    compliance_domain: str = None,
) -> str:
    if compliance_domain and compliance_domain in COMPLIANCE_OVERRIDES:
        overrides = COMPLIANCE_OVERRIDES[compliance_domain]
        if action in overrides:
            return overrides[action]

    return risk_decision


test_cases = [
    ("web_search", "auto_approve", None),
    ("web_search", "auto_approve", "financial"),
    ("any_transaction", "quick_approve", "financial"),
    ("patient_data_access", "auto_approve", "healthcare"),
    ("generate_report", "auto_approve", "healthcare"),
]

print("=== Compliance overrides ===\n")
print(f"{'Acción':<25} {'Riesgo':<16} {'Dominio':<14} {'Final':<16}")
print(f"{'-'*25} {'-'*16} {'-'*14} {'-'*16}")
for action, risk, domain in test_cases:
    final = get_final_decision(action, risk, domain)
    domain_str = domain or "ninguno"
    override = " ⚠️" if final != risk else ""
    print(f"{action:<25} {risk:<16} {domain_str:<14} {final:<16}{override}")
# Output esperado:
# === Compliance overrides ===
#
# Acción                    Riesgo           Dominio        Final
# ------------------------- ---------------- -------------- ----------------
# web_search                auto_approve     ninguno        auto_approve
# web_search                auto_approve     financial      auto_approve
# any_transaction           quick_approve    financial      full_review      ⚠️
# patient_data_access       auto_approve     healthcare     full_review      ⚠️
# generate_report           auto_approve     healthcare     auto_approve

Los overrides de compliance tienen prioridad sobre el risk score. patient_data_access calculó auto_approve por el framework de riesgo, pero el dominio healthcare fuerza full_review.

Troubleshooting

Problema 1: "El agente interrumpe demasiado y los usuarios se quejan"

Síntoma: Feedback de usuarios: "el agente me pide permiso para todo."

Causa: El threshold de riesgo es demasiado bajo, o acciones de bajo riesgo están clasificadas incorrectamente.

Solución: Revisa la tasa de aprobación por acción (sección de calibración). Si una acción tiene >95% de aprobación con al menos 10 muestras, cámbiala a auto-approve.

Problema 2: "El risk_score siempre da lo mismo para acciones diferentes"

Síntoma: Acciones con perfiles de riesgo distintos producen el mismo score.

Causa: El max() en la fórmula domina. Si todas las acciones tienen al menos un criterio alto, todas dan score alto.

Solución: Ajusta los pesos. En vez de max(), usa un promedio ponderado que refleje tu dominio:

def weighted_risk(cost_v, rev_v, impact_v, confidence, weights=(0.3, 0.3, 0.3, 0.1)):
    base = cost_v * weights[0] + rev_v * weights[1] + impact_v * weights[2]
    return min(round(base + (1 - confidence) * weights[3], 2), 1.0)

Problema 3: "No sé qué confianza asignar a cada acción"

Síntoma: El criterio de confianza parece subjetivo.

Causa: Es subjetivo si lo asignas manualmente. No debería serlo.

Solución: Infiere la confianza de señales objetivas:

def infer_confidence(sources_found: int, sources_agree: bool, llm_certainty: str) -> float:
    base = min(sources_found * 0.2, 0.6)
    if sources_agree:
        base += 0.2
    if llm_certainty == "high":
        base += 0.2
    elif llm_certainty == "medium":
        base += 0.1
    return min(base, 1.0)

Problema 4: "Los niveles de autonomía crean experiencias inconsistentes"

Síntoma: Usuarios experimentados se confunden cuando ocasionalmente reciben un interrupt que normalmente no ven.

Causa: Acciones cercanas al threshold del usuario oscilan entre auto-approve e interrupt según contexto.

Solución: Agrega histéresis — una vez que una acción se auto-aprueba para un usuario, solo vuelve a interrupt si el riesgo sube significativamente (por ejemplo, 0.15 por encima del threshold, no justo en el límite).

Ejercicios

Ejercicio 1: Clasificar acciones con el framework (Fácil)

Dado este agente de atención al cliente, clasifica cada acción usando los 4 criterios y determina la decisión de riesgo:

1. Buscar en la base de conocimientos interna
2. Generar una respuesta al cliente
3. Enviar la respuesta al cliente por email
4. Crear un ticket de escalación
5. Aplicar un reembolso de $50
6. Cerrar la cuenta del cliente

Ejercicio 2: Implementar risk assessment personalizado (Medio)

Crea una función assess_action() que reciba: nombre de la acción, costo estimado, si es reversible, número de personas afectadas, y un score de confianza del LLM. Retorna la decisión (auto/quick/full). Pruébala con al menos 5 acciones diferentes.

from dotenv import load_dotenv
load_dotenv()


def assess_action(name: str, cost: float, reversible: bool, people_affected: int, llm_confidence: float) -> dict:
    if cost < 0.01:
        cost_score = 0.0
    elif cost < 1.0:
        cost_score = 0.3
    elif cost < 10.0:
        cost_score = 0.6
    else:
        cost_score = 1.0

    rev_score = 0.0 if reversible else 1.0

    if people_affected == 0:
        impact_score = 0.0
    elif people_affected <= 5:
        impact_score = 0.4
    else:
        impact_score = 1.0

    risk = max(cost_score, rev_score, impact_score) + (1 - llm_confidence) * 0.5
    risk = min(risk, 1.0)

    if risk < 0.3:
        decision = "auto_approve"
    elif risk <= 0.7:
        decision = "quick_approve"
    else:
        decision = "full_review"

    return {"name": name, "risk": round(risk, 2), "decision": decision}


tests = [
    ("web_search", 0.0, True, 0, 0.95),
    ("api_call", 0.50, True, 0, 0.90),
    ("send_notification", 0.0, False, 1, 0.85),
    ("batch_email", 0.0, False, 100, 0.80),
    ("refund", 25.0, False, 1, 0.75),
]

for args in tests:
    result = assess_action(*args)
    print(f"{result['name']:<20} riesgo={result['risk']:.2f}  → {result['decision']}")
# Output esperado:
# web_search           riesgo=0.02  → auto_approve
# api_call             riesgo=0.35  → quick_approve
# send_notification    riesgo=1.00  → full_review
# batch_email          riesgo=1.00  → full_review
# refund               riesgo=1.00  → full_review

Ejercicio 3: Agregar interrupt condicional a un grafo (Medio)

Crea un grafo con 3 nodos: search → process → send. El nodo send debe evaluar el riesgo de enviar (basado en el número de destinatarios en el estado). Si hay 1 destinatario, auto-approve. Si hay más de 5, interrupt para pedir aprobación. Prueba con 1 destinatario (debe ejecutar completo) y con 10 destinatarios (debe pausar).

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command


class State(TypedDict):
    data: str
    recipients: int
    result: str


def search(state: State) -> dict:
    return {"data": f"Resultados encontrados para el informe"}


def process(state: State) -> dict:
    return {"data": f"Informe procesado: {state['data']}"}


def send(state: State) -> dict:
    if state["recipients"] > 5:
        response = interrupt({
            "message": f"Enviar a {state['recipients']} destinatarios. ¿Procedo?",
            "recipients": state["recipients"],
        })
        if not response.get("approved"):
            return {"result": "Envío cancelado por el usuario"}

    return {"result": f"Enviado a {state['recipients']} destinatarios: {state['data'][:50]}"}


graph_builder = StateGraph(State)
graph_builder.add_node("search", search)
graph_builder.add_node("process", process)
graph_builder.add_node("send", send)
graph_builder.add_edge(START, "search")
graph_builder.add_edge("search", "process")
graph_builder.add_edge("process", "send")
graph_builder.add_edge("send", END)

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

print("=== Test 1: 1 destinatario (auto-approve) ===")
config1 = {"configurable": {"thread_id": "send-1"}}
result1 = graph.invoke({"data": "", "recipients": 1, "result": ""}, config1)
print(f"  Resultado: {result1['result']}")

print("\n=== Test 2: 10 destinatarios (interrupt) ===")
config2 = {"configurable": {"thread_id": "send-10"}}
result2 = graph.invoke({"data": "", "recipients": 10, "result": ""}, config2)

state = graph.get_state(config2)
if state.next:
    print(f"  Interrupt activo. Reanudando con aprobación...")
    result2 = graph.invoke(Command(resume={"approved": True}), config2)
    print(f"  Resultado: {result2['result']}")
# Output esperado:
# === Test 1: 1 destinatario (auto-approve) ===
#   Resultado: Enviado a 1 destinatarios: Informe procesado: Resultados encontrados para e
#
# === Test 2: 10 destinatarios (interrupt) ===
#   Interrupt activo. Reanudando con aprobación...
#   Resultado: Enviado a 10 destinatarios: Informe procesado: Resultados encontrados para

Ejercicio 4: Sistema de calibración con historial (Medio)

Implementa una clase InterruptCalibrator que registre cada decisión de aprobación y, después de N decisiones por acción, recomiende si cambiar la política. Prueba con un historial simulado de 20 decisiones para 3 tipos de acciones.

from dotenv import load_dotenv
load_dotenv()

from collections import defaultdict


class InterruptCalibrator:
    def __init__(self, min_samples: int = 5, auto_approve_threshold: float = 0.95):
        self.history = defaultdict(list)
        self.min_samples = min_samples
        self.auto_approve_threshold = auto_approve_threshold

    def record(self, action: str, approved: bool):
        self.history[action].append(approved)

    def recommend(self, action: str) -> str:
        decisions = self.history.get(action, [])
        if len(decisions) < self.min_samples:
            return f"DATOS INSUFICIENTES ({len(decisions)}/{self.min_samples})"

        rate = sum(decisions) / len(decisions)
        if rate >= self.auto_approve_threshold:
            return f"→ AUTO_APPROVE (tasa: {rate:.0%}, {len(decisions)} muestras)"
        elif rate >= 0.70:
            return f"→ MANTENER INTERRUPT (tasa: {rate:.0%})"
        elif rate >= 0.40:
            return f"→ REVISAR ACCIÓN (tasa: {rate:.0%} — muchos rechazos)"
        else:
            return f"→ ELIMINAR ACCIÓN (tasa: {rate:.0%} — casi siempre rechazada)"

    def report(self):
        for action in sorted(self.history.keys()):
            print(f"  {action}: {self.recommend(action)}")


cal = InterruptCalibrator(min_samples=5)

for _ in range(10):
    cal.record("search_paid", True)
cal.record("search_paid", True)
cal.record("search_paid", True)

for approved in [True, True, False, True, True, False, True]:
    cal.record("send_email", approved)

for approved in [False, False, True, False, False, False]:
    cal.record("delete_data", approved)

cal.record("new_action", True)
cal.record("new_action", False)

print("=== Reporte de calibración ===\n")
cal.report()
# Output esperado:
# === Reporte de calibración ===
#
#   delete_data: → ELIMINAR ACCIÓN (tasa: 17% — casi siempre rechazada)
#   new_action: DATOS INSUFICIENTES (2/5)
#   search_paid: → AUTO_APPROVE (tasa: 100%, 12 muestras)
#   send_email: → MANTENER INTERRUPT (tasa: 71%)

Ejercicio 5: Framework de riesgo para dominio regulado (Avanzado)

Extiende el framework de riesgo para soportar "compliance domains." Crea una función que reciba el riesgo calculado, el dominio (finanzas, salud, o ninguno), y el tipo de acción. Si el dominio tiene un override para esa acción, el override gana. Prueba con 6 combinaciones diferentes.

from dotenv import load_dotenv
load_dotenv()


COMPLIANCE_RULES = {
    "finance": {
        "transaction": "full_review",
        "account_modify": "full_review",
        "report_view": "quick_approve",
    },
    "healthcare": {
        "patient_data": "full_review",
        "prescription": "full_review",
        "schedule_view": "quick_approve",
    },
}


def final_decision(action: str, risk_decision: str, domain: str = None) -> dict:
    override = None
    if domain and domain in COMPLIANCE_RULES:
        rules = COMPLIANCE_RULES[domain]
        if action in rules:
            override = rules[action]

    return {
        "action": action,
        "risk_decision": risk_decision,
        "domain": domain or "none",
        "final": override if override else risk_decision,
        "overridden": override is not None and override != risk_decision,
    }


cases = [
    ("web_search", "auto_approve", None),
    ("transaction", "quick_approve", "finance"),
    ("transaction", "quick_approve", None),
    ("patient_data", "auto_approve", "healthcare"),
    ("schedule_view", "auto_approve", "healthcare"),
    ("report_view", "auto_approve", "finance"),
]

print("=== Decisiones con compliance ===\n")
print(f"{'Acción':<16} {'Riesgo':<16} {'Dominio':<12} {'Final':<16} {'Override?'}")
print(f"{'-'*16} {'-'*16} {'-'*12} {'-'*16} {'-'*9}")
for action, risk, domain in cases:
    r = final_decision(action, risk, domain)
    ov = "SÍ ⚠️" if r["overridden"] else "no"
    print(f"{r['action']:<16} {r['risk_decision']:<16} {r['domain']:<12} {r['final']:<16} {ov}")
# Output esperado:
# === Decisiones con compliance ===
#
# Acción           Riesgo           Dominio      Final            Override?
# ---------------- ---------------- ------------ ---------------- ---------
# web_search       auto_approve     none         auto_approve     no
# transaction      quick_approve    finance      full_review      SÍ ⚠️
# transaction      quick_approve    none         quick_approve    no
# patient_data     auto_approve     healthcare   full_review      SÍ ⚠️
# schedule_view    auto_approve     healthcare   quick_approve    SÍ ⚠️
# report_view      auto_approve     finance      quick_approve    SÍ ⚠️

Ejercicio 6: Simulación de agente con decisión dinámica (Avanzado)

Crea un grafo que simula un agente de compras: recibe una lista de items para comprar, evalúa el riesgo de cada compra (basado en precio), y decide si auto-aprobar o interrumpir. Items < $10 son auto-approved, items $10-$50 requieren quick approve, items > $50 requieren full review. El grafo debe procesar la lista completa, acumulando los items aprobados, y generar una orden de compra final.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command


class State(TypedDict):
    items: list[dict]
    approved_items: Annotated[list[dict], operator.add]
    current_index: int
    total_cost: float
    order_summary: str


def evaluate_item(state: State) -> dict:
    idx = state["current_index"]
    if idx >= len(state["items"]):
        return {"order_summary": "NO_MORE_ITEMS"}

    item = state["items"][idx]
    approved = []

    if item["price"] < 10.0:
        approved.append({**item, "decision": "auto_approved"})
    elif item["price"] <= 50.0:
        response = interrupt({
            "type": "quick_approve",
            "message": f"Comprar '{item['name']}' por ${item['price']:.2f}?",
            "item": item,
        })
        if response.get("approved"):
            approved.append({**item, "decision": "human_approved"})
    else:
        response = interrupt({
            "type": "full_review",
            "message": f"⚠️ Compra mayor: '{item['name']}' por ${item['price']:.2f}. Revisar y aprobar.",
            "item": item,
        })
        if response.get("approved"):
            approved.append({**item, "decision": "human_approved"})

    new_cost = state["total_cost"] + sum(i["price"] for i in approved)
    return {
        "approved_items": approved,
        "current_index": idx + 1,
        "total_cost": new_cost,
    }


def check_more(state: State) -> str:
    if state["current_index"] >= len(state["items"]):
        return "generate_order"
    return "evaluate"


def generate_order(state: State) -> dict:
    if not state["approved_items"]:
        return {"order_summary": "Orden vacía — ningún item aprobado."}
    lines = [f"ORDEN DE COMPRA — {len(state['approved_items'])} items:"]
    for item in state["approved_items"]:
        lines.append(f"  • {item['name']}: ${item['price']:.2f} ({item['decision']})")
    lines.append(f"  TOTAL: ${state['total_cost']:.2f}")
    return {"order_summary": "\n".join(lines)}


graph_builder = StateGraph(State)
graph_builder.add_node("evaluate", evaluate_item)
graph_builder.add_node("generate_order", generate_order)
graph_builder.add_edge(START, "evaluate")
graph_builder.add_conditional_edges("evaluate", check_more, {"evaluate": "evaluate", "generate_order": "generate_order"})
graph_builder.add_edge("generate_order", END)

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

config = {"configurable": {"thread_id": "shopping-demo"}}

items = [
    {"name": "Cable USB", "price": 5.99},
    {"name": "Teclado mecánico", "price": 45.00},
    {"name": "Monitor 4K", "price": 350.00},
    {"name": "Mouse pad", "price": 8.50},
]

result = graph.invoke(
    {"items": items, "approved_items": [], "current_index": 0, "total_cost": 0, "order_summary": ""},
    config,
)

approved_count = 0
while graph.get_state(config).next:
    state = graph.get_state(config)
    print(f"  Interrupt — aprobando...")
    result = graph.invoke(Command(resume={"approved": True}), config)
    approved_count += 1

print(f"\n{result['order_summary']}")
print(f"\n({approved_count} aprobaciones humanas requeridas)")
# Output esperado:
# (El Cable USB y Mouse pad se auto-aprueban, teclado y monitor requieren aprobación)
#   Interrupt — aprobando...
#   Interrupt — aprobando...
#
# ORDEN DE COMPRA — 4 items:
#   • Cable USB: $5.99 (auto_approved)
#   • Teclado mecánico: $45.00 (human_approved)
#   • Monitor 4K: $350.00 (human_approved)
#   • Mouse pad: $8.50 (auto_approved)
#   TOTAL: $409.49
#
# (2 aprobaciones humanas requeridas)

Resumen

En esta cápsula aprendiste:

• La decisión de cuándo interrumpir es más importante que la mecánica de cómo interrumpir. Demasiadas interrupciones hacen al agente inútil. Demasiadas pocas lo hacen peligroso. El framework de 4 criterios te da un método sistemático para encontrar el punto exacto
• Los 4 criterios son: Costo, Reversibilidad, Impacto, y Confianza. Costo mide el dinero en juego. Reversibilidad mide si puedes deshacer la acción. Impacto mide cuántas personas o sistemas se afectan. Confianza mide qué tan seguro está el agente. Un solo criterio en rojo es suficiente para escalar
• La función risk_score combina los 4 criterios usando max() para los tres primeros (un criterio crítico domina) y una penalización por baja confianza. El score resultante determina: auto-approve (< 0.3), quick approve (0.3-0.7), o full review (> 0.7)
• El framework se calibra con datos reales. Mide la tasa de aprobación por acción. Si una acción tiene >95% de aprobación, elimina el interrupt. Si tiene <40% de aprobación, el agente no debería intentar esa acción
• Los niveles de autonomía por usuario permiten adaptar la experiencia: usuarios nuevos reciben más interrupts, usuarios expertos operan con más autonomía. Mismo agente, diferentes thresholds
• En industrias reguladas, el compliance tiene prioridad sobre el risk score. Ciertas acciones siempre requieren aprobación humana, independientemente de lo seguras que parezcan

Próxima cápsula: El proyecto del módulo — aplicarás este framework al Research Assistant v3 para crear la v4 con aprobaciones humanas, feedback loop, y edición de estado.

Recursos adicionales

1. LangGraph Human-in-the-Loop — Conceptos oficiales de HITL en LangGraph
2. How to add human-in-the-loop — Guías prácticas de HITL
3. interrupt() Reference — API reference de interrupt
4. LangGraph Command — Uso de Command para resume dinámico
5. Trust and Safety in AI Agents — Paper sobre confianza y seguridad en agentes autónomos

Módulo 9 — LangChain & LangGraph: From Chains to Agents