Fallas en cascada

Descripción

Los errores en flujos de trabajo multiagente se propagan sin control entre agentes dependientes, amplificando un único punto de falla hasta convertirlo en interrupciones a nivel de sistema, corrupción de datos o consumo desbocado de recursos.

Riesgo

Panorama del riesgo

Los sistemas multiagente crean cadenas de dependencia donde la salida de un agente alimenta la entrada del siguiente. Cuando un agente falla — ya sea por un error del modelo, un tiempo de espera agotado en una herramienta, agotamiento de recursos o una entrada adversaria — esa falla se propaga aguas abajo. A diferencia de los sistemas distribuidos tradicionales, que cuentan con décadas de patrones de tolerancia a fallas ya establecidos, los flujos de trabajo agénticos a menudo carecen de interruptores de circuito (circuit breakers), límites de reintento o estrategias de respaldo. Una sola falla de agente puede degenerar en una interrupción a nivel de sistema en la que docenas de agentes quedan atrapados en bucles de reintento, consumiendo recursos y produciendo resultados corruptos.

Superficie de ataque

Amplificación de errores: un agente produce una salida malformada; los agentes aguas abajo fallan al intentar analizarla, generando cada uno sus propias salidas de error que confunden aún más al sistema
Tormentas de reintentos: múltiples agentes reintentan operaciones fallidas simultáneamente, saturando recursos compartidos (bases de datos, API, endpoints de modelos)
Agotamiento de recursos: un agente atrapado en un bucle de fallas consume todo el cómputo, la memoria o la cuota de API disponibles, privando de recursos a otros agentes
Propagación de la 'píldora envenenada': un artefacto de datos corrupto pasa a través de un pipeline de agentes, y cada agente agrega su propia capa de corrupción hasta que el resultado final es completamente poco confiable
Interbloqueos (deadlocks): dependencias circulares entre agentes provocan una espera mutua que nunca se resuelve
Estampida (thundering herd): todos los agentes de un pool se recuperan de una falla simultáneamente, creando un pico de carga que dispara otra falla

Impacto en el negocio

Interrupción a nivel de sistema: la falla de un solo agente se propaga en cascada hasta derribar todo un flujo de trabajo multiagente
Costos de recursos: las tormentas de reintentos y el agotamiento de recursos generan costos inesperados de cómputo y de API
Integridad de los datos: los resultados corruptos se propagan a través del pipeline antes de que alguien detecte la falla original
Violaciones de SLA: los retrasos en cascada provocan que los tiempos de procesamiento de extremo a extremo superen los acuerdos de nivel de servicio
Diagnóstico difícil: la causa raíz queda sepultada bajo capas de efectos en cascada, extendiendo el tiempo medio de resolución

Escenarios de ataque

Un endpoint de servicio de modelo experimenta latencia elevada, causando que el primer agente de un pipeline agote el tiempo de espera. El orquestador reintenta la solicitud 5 veces sin backoff. Los 5 reintentos llegan al endpoint, ya sobrecargado, simultáneamente. Mientras tanto, 3 agentes aguas abajo que esperan la salida del primer agente también agotan su tiempo de espera y comienzan sus propios bucles de reintento. En minutos, el sistema genera 50 veces la carga normal sobre el endpoint del modelo, haciendo que colapse por completo y derribando todos los flujos de trabajo de agentes que dependen de él.

Un agente de ingesta de datos encuentra un archivo CSV malformado y produce una salida parcial y corrupta en lugar de fallar limpiamente. El siguiente agente del pipeline — un agente de validación de datos — procesa los datos corruptos y marca algunas filas como inválidas, pero deja pasar el resto. Un agente de análisis aguas abajo produce resultados incorrectos a partir de los datos corruptos. El agente de generación de informes final presenta estos resultados a las partes interesadas como un análisis fidedigno. La causa raíz (un CSV malformado) no se descubre hasta que las decisiones de negocio basadas en el informe ya se ejecutaron.

Un atacante envía deliberadamente una solicitud que hace que un agente entre en un bucle de razonamiento infinito. El agente en bucle consume todo su presupuesto de tokens y luego solicita más. El orquestador asigna recursos adicionales. Las salidas del bucle del agente disparan bucles similares en agentes pares. En una hora, todo el pool de agentes queda consumido por bucles infinitos y se agota la cuota de API del mes.

Mitigaciones

Interruptores de circuito (circuit breakers)

Implementar interruptores de circuito en cada límite agente-a-agente y agente-a-servicio:

Rastrear las tasas de falla en ventanas de tiempo deslizantes (p. ej., una tasa de falla >50% en los últimos 60 segundos dispara el interruptor)
Cuando el circuito se abre, devolver de inmediato una respuesta de respaldo en lugar de reenviar solicitudes al servicio fallido
Después de un período de enfriamiento, permitir el paso de una única solicitud de prueba para verificar la recuperación (estado semiabierto)
Alertar a los operadores cuando los circuitos se disparen y exigir revisión manual antes de restablecerlos si la falla fue causada por corrupción de datos

Manejadores de respaldo (fallback)

Definir el comportamiento en modo degradado para cada agente del pipeline:

Cada agente debe declarar qué hace cuando sus dependencias no están disponibles: devolver resultados en caché, proporcionar una respuesta predeterminada o declinar la tarea de forma controlada
Implementar degradación escalonada: primero intentar un modelo de respaldo, luego resultados en caché, luego un respaldo estático y, por último, un mensaje de error claro
Nunca dejar que un agente produzca resultados basura silenciosamente — debe producir una salida validada o fallar explícitamente
Probar las rutas de respaldo con regularidad; los respaldos no probados fallan cuando más se necesitan

Límites de radio de impacto

Contener las fallas para prevenir un impacto a nivel de sistema:

Establecer presupuestos de recursos por agente: máximo de tokens, máximo de llamadas a API, tiempo de ejecución máximo, memoria máxima
Implementar mamparos (bulkheads): pools de agentes separados para flujos de trabajo críticos frente a no críticos, de modo que una falla en un pool no prive de recursos al otro
Usar colas con prioridad para garantizar que las tareas de agentes de alta prioridad se procesen incluso durante la contención de recursos
Limitar los intentos de reintento con backoff exponencial y jitter para prevenir estampidas (thundering herds)

Verificaciones de salud

Detectar las fallas de forma temprana antes de que se propaguen en cascada:

Implementar sondas de vitalidad (¿el agente está en ejecución?) y de disposición (¿el agente está listo para aceptar trabajo?) para todos los agentes
Monitorear la latencia del pipeline de extremo a extremo, no solo la latencia de cada agente individual — un agente lento en medio del flujo causa retrasos invisibles aguas abajo
Rastrear las tasas de error por agente y por cada conexión agente-a-agente en el grafo de dependencias
Implementar transacciones sintéticas que ejerciten el pipeline completo según un calendario para detectar la degradación antes que los usuarios

Puertas de validación de salida

Prevenir que los resultados corruptos se propaguen:

Definir esquemas de salida para cada agente y validar la salida contra el esquema antes de pasarla aguas abajo
Implementar sumas de verificación o validación basada en contenido para los artefactos de datos que se pasan entre agentes
Usar un detector de 'píldoras envenenadas' que identifique salidas con características anómalas (nulos inesperados, truncamiento, valores sin sentido)

Ejemplos de código

Patrón de interruptor de circuito (circuit breaker)

import time
import threading
from enum import Enum
from dataclasses import dataclass, field
from typing import Callable, Any, Optional
from collections import deque


class EstadoCircuito(Enum):
    CERRADO = "closed"      # Operación normal
    ABIERTO = "open"          # Fallando, rechazar solicitudes
    SEMIABIERTO = "half_open"  # Probando recuperación


@dataclass
class ConfiguracionInterruptorCircuito:
    umbral_falla: float = 0.5  # 50% de tasa de falla dispara el interruptor
    ventana_segundos: int = 60
    segundos_enfriamiento: int = 30
    llamadas_minimas_en_ventana: int = 5  # Mínimo de llamadas antes de evaluar
    reintentos_maximos: int = 3
    segundos_tiempo_espera: float = 30.0


class InterruptorCircuito:
    """Previene fallas en cascada entre agentes y servicios."""

    def __init__(self, nombre: str, config: ConfiguracionInterruptorCircuito, respaldo: Optional[Callable] = None):
        self.nombre = nombre
        self._config = config
        self._respaldo = respaldo
        self._estado = EstadoCircuito.CERRADO
        self._registro_llamadas: deque = deque()  # (timestamp, exito: bool)
        self._ultima_falla_timestamp = 0.0
        self._lock = threading.Lock()
        self._registro_cambio_estado: list[dict] = []

    @property
    def estado(self) -> EstadoCircuito:
        with self._lock:
            if self._estado == EstadoCircuito.ABIERTO:
                # Verificar si el enfriamiento ya transcurrió
                if time.time() - self._ultima_falla_timestamp >= self._config.segundos_enfriamiento:
                    self._transicionar(EstadoCircuito.SEMIABIERTO)
            return self._estado

    def llamar(self, fn: Callable, *args, **kwargs) -> dict:
        """Ejecuta una función a través del interruptor de circuito."""
        estado_actual = self.estado

        if estado_actual == EstadoCircuito.ABIERTO:
            if self._respaldo:
                return {
                    "status": "fallback",
                    "circuito": self.nombre,
                    "resultado": self._respaldo(*args, **kwargs),
                }
            return {
                "status": "rejected",
                "circuito": self.nombre,
                "razon": "Circuito abierto — servicio no disponible",
            }

        try:
            resultado = fn(*args, **kwargs)
            self._registrar_exito()
            return {"status": "success", "circuito": self.nombre, "resultado": resultado}

        except Exception as e:
            self._registrar_falla()
            if self._respaldo:
                return {
                    "status": "fallback",
                    "circuito": self.nombre,
                    "error": str(e),
                    "resultado": self._respaldo(*args, **kwargs),
                }
            return {"status": "error", "circuito": self.nombre, "error": str(e)}

    def _registrar_exito(self):
        with self._lock:
            self._registro_llamadas.append((time.time(), True))
            self._recortar_ventana()
            if self._estado == EstadoCircuito.SEMIABIERTO:
                self._transicionar(EstadoCircuito.CERRADO)

    def _registrar_falla(self):
        with self._lock:
            ahora = time.time()
            self._registro_llamadas.append((ahora, False))
            self._ultima_falla_timestamp = ahora
            self._recortar_ventana()
            tasa_falla = self._tasa_falla_actual()
            total = len(self._registro_llamadas)
            if total >= self._config.llamadas_minimas_en_ventana and tasa_falla >= self._config.umbral_falla:
                self._transicionar(EstadoCircuito.ABIERTO)

    def _tasa_falla_actual(self) -> float:
        if not self._registro_llamadas:
            return 0.0
        fallas = sum(1 for _, exito in self._registro_llamadas if not exito)
        return fallas / len(self._registro_llamadas)

    def _recortar_ventana(self):
        corte = time.time() - self._config.ventana_segundos
        while self._registro_llamadas and self._registro_llamadas[0][0] < corte:
            self._registro_llamadas.popleft()

    def _transicionar(self, nuevo_estado: EstadoCircuito):
        estado_anterior = self._estado
        self._estado = nuevo_estado
        self._registro_cambio_estado.append({
            "circuito": self.nombre,
            "desde": estado_anterior.value,
            "hasta": nuevo_estado.value,
            "timestamp": time.time(),
            "tasa_falla": self._tasa_falla_actual(),
        })

    def obtener_salud(self) -> dict:
        with self._lock:
            self._recortar_ventana()
            return {
                "circuito": self.nombre,
                "estado": self._estado.value,
                "llamadas_en_ventana": len(self._registro_llamadas),
                "tasa_falla": round(self._tasa_falla_actual(), 3),
                "ultima_falla": self._ultima_falla_timestamp,
            }

Requisitos de evidencia

Registros de transición de estado del interruptor de circuito con timestamps y tasas de falla
Panel de verificación de salud de agentes que muestra vitalidad, disposición y tasas de error por agente
Informes de análisis posterior a fallas en cascada con análisis de causa raíz y mediciones del radio de impacto
Auditoría de consumo de recursos que muestra los presupuestos por agente y el uso real
Resultados de monitoreo de transacciones sintéticas que validan la salud del pipeline de extremo a extremo
Alertas de detección de tormentas de reintentos con métricas de efectividad de la limitación