Fallback strategy

Tu cliente actual tiene un punto único de falla: si OpenAI cae, tu app cae. En producción real, eso es inaceptable. La solución es fallback automático: cuando un provider falla, intenta el siguiente automáticamente.

Pero "intentar el siguiente" no es uniforme — depende del tipo de error. Si fue rate limit, vale la pena esperar y reintentar. Si fue API key inválida, no — esa nunca va a funcionar. Si fue timeout, salta al siguiente provider rápido. Esta cápsula te enseña a hacerlo bien.

Al terminar vas a poder:

• Implementar fallback automático con la lista de providers definida en config.fallback
• Distinguir errores transitorios (retry sirve) de permanentes (retry no sirve)
• Aplicar un circuit breaker simple para no martillear providers caídos
• Reportar errores claros cuando todos los providers fallan

Modelo mental: cuándo retry, cuándo siguiente, cuándo abortar

Tres tipos de errores con tres respuestas distintas:

La diferencia importa: hacer retry en AuthError quema requests inútilmente. Hacer retry en RateLimitError con la misma fuerza no funciona porque el rate sigue activo.

Circuit breaker en una frase

Si un provider falla N veces seguidas, marca ese provider como "abierto" por un tiempo (e.g., 60s). Durante ese tiempo, saltas el provider sin intentarlo — porque sabes que está caído. Después del tiempo, le das otra oportunidad.

Beneficio: cuando OpenAI tiene un outage, no pagas timeout × cada request × cada usuario. Saltas directo al fallback.

Implementación: ampliando UnifiedClient

Agrega esto a unified_ai_client/client.py:

# unified_ai_client/client.py
import time
import logging
from dataclasses import dataclass, field
from pathlib import Path
import yaml

from .models import Message, ChatResponse, ClientConfig
from .factory import ProviderFactory
from .adapters.base import BaseAdapter
from .exceptions import (
    ConfigError,
    AuthError,
    RateLimitError,
    TimeoutError,
    ProviderError,
    AllProvidersFailedError,
)

logger = logging.getLogger(__name__)


@dataclass
class CircuitState:
    """Estado del circuit breaker para un provider."""
    fallos_consecutivos: int = 0
    abierto_hasta: float = 0  # epoch seconds


class UnifiedClient:
    """Cliente unificado con fallback automático y circuit breaker."""

    # Cuántos fallos antes de abrir el circuito
    UMBRAL_CIRCUITO = 3
    # Por cuánto tiempo (segundos) ignorar un provider tras alcanzar el umbral
    DURACION_CIRCUITO_S = 60

    def __init__(self, config: ClientConfig):
        self.config = config
        if config.primary not in config.providers:
            raise ConfigError(f"Primary '{config.primary}' no existe en providers")
        for fb in config.fallback:
            if fb not in config.providers:
                raise ConfigError(f"Fallback '{fb}' no existe en providers")

        # Lista ordenada: primary primero, después fallbacks
        self.adapters: list[BaseAdapter] = [
            ProviderFactory.crear(config.providers[config.primary])
        ] + [
            ProviderFactory.crear(config.providers[name]) for name in config.fallback
        ]

        # Circuit breaker state por provider name
        self.circuit: dict[str, CircuitState] = {
            a.name: CircuitState() for a in self.adapters
        }

    @classmethod
    def from_yaml(cls, path: str | Path) -> "UnifiedClient":
        with open(path) as f:
            return cls(ClientConfig(**yaml.safe_load(f)))

    @classmethod
    def from_dict(cls, data: dict) -> "UnifiedClient":
        return cls(ClientConfig(**data))

    # ===========================================
    # Public API
    # ===========================================
    def chat(
        self,
        prompt: str,
        *,
        system: str | None = None,
        max_tokens: int = 256,
        temperature: float = 0.7,
    ) -> ChatResponse:
        messages: list[Message] = []
        if system:
            messages.append(Message(role="system", content=system))
        messages.append(Message(role="user", content=prompt))
        return self.chat_with_messages(messages, max_tokens=max_tokens, temperature=temperature)

    def chat_with_messages(
        self,
        messages: list[Message],
        max_tokens: int = 256,
        temperature: float = 0.7,
    ) -> ChatResponse:
        errores: dict[str, Exception] = {}

        for adapter in self.adapters:
            if self._circuito_abierto(adapter.name):
                logger.info(f"Saltando '{adapter.name}': circuit breaker abierto")
                errores[adapter.name] = ProviderError(
                    adapter.name, "Circuit breaker abierto (provider considerado caído)"
                )
                continue

            try:
                return self._intentar_con_retry(
                    adapter, messages, max_tokens, temperature
                )
            except AuthError as e:
                # Permanente: NO intentar otros si la auth está mal globalmente
                # Pero puede ser que solo este provider tenga problema; sigamos.
                logger.error(f"Auth error en '{adapter.name}': {e}")
                errores[adapter.name] = e
                self._registrar_fallo(adapter.name)
                continue
            except (RateLimitError, TimeoutError, ProviderError) as e:
                logger.warning(f"'{adapter.name}' falló: {e}. Probando siguiente.")
                errores[adapter.name] = e
                self._registrar_fallo(adapter.name)
                continue

        # Si llegamos acá, ningún adapter funcionó
        raise AllProvidersFailedError(errores)

    # ===========================================
    # Internos: retry y circuit breaker
    # ===========================================
    def _intentar_con_retry(
        self,
        adapter: BaseAdapter,
        messages: list[Message],
        max_tokens: int,
        temperature: float,
        max_retries: int = 2,
    ) -> ChatResponse:
        """Intenta un adapter con retry para errores transitorios."""
        for intento in range(max_retries + 1):
            try:
                response = adapter.chat(messages, max_tokens, temperature)
                self._registrar_exito(adapter.name)
                return response
            except RateLimitError:
                if intento < max_retries:
                    wait = 2 ** intento  # 1s, 2s, 4s
                    logger.info(f"Rate limit en '{adapter.name}', esperando {wait}s")
                    time.sleep(wait)
                    continue
                raise
            except AuthError:
                # Permanente: no retry
                raise
            except (TimeoutError, ProviderError):
                # No retry mismo provider; salta al siguiente
                raise

        raise RuntimeError("unreachable")  # type narrow

    def _circuito_abierto(self, provider_name: str) -> bool:
        state = self.circuit[provider_name]
        return state.abierto_hasta > time.time()

    def _registrar_fallo(self, provider_name: str) -> None:
        state = self.circuit[provider_name]
        state.fallos_consecutivos += 1
        if state.fallos_consecutivos >= self.UMBRAL_CIRCUITO:
            state.abierto_hasta = time.time() + self.DURACION_CIRCUITO_S
            logger.warning(
                f"Circuit abierto para '{provider_name}' por {self.DURACION_CIRCUITO_S}s "
                f"({state.fallos_consecutivos} fallos consecutivos)"
            )

    def _registrar_exito(self, provider_name: str) -> None:
        # Éxito resetea el contador y cierra el circuito
        self.circuit[provider_name] = CircuitState()

Cómo funciona en práctica

Caso 1 — Primary funciona normal:

chat("...") → OpenAI ✓ → return ChatResponse

Caso 2 — Primary tiene rate limit transitorio:

chat("...") → OpenAI 429 → wait 1s → retry OpenAI ✓ → return ChatResponse

Caso 3 — Primary cae, fallback funciona:

chat("...") → OpenAI timeout → retry no aplica (timeout)
            → siguiente: OpenRouter ✓ → return ChatResponse

Caso 4 — Primary cae repetidamente (circuit breaker):

1ª request: OpenAI timeout → OpenRouter ✓
2ª request: OpenAI timeout → OpenRouter ✓
3ª request: OpenAI timeout (circuit ABRE para OpenAI) → OpenRouter ✓
4ª request (siguientes 60s): salta OpenAI → OpenRouter ✓ (sin tocar OpenAI)
65s después: OpenAI se intenta de nuevo, si OK cierra circuit

Caso 5 — Todos los providers fallan:

chat("...") → OpenAI fail → OpenRouter fail → Ollama fail
            → AllProvidersFailedError(errores={...})

Verificación con ejemplo realista

Config con 3 providers en cadena:

# examples/clients_fallback.yaml
primary: openai-mini
fallback:
  - openrouter-mistral
  - ollama-mistral

providers:
  openai-mini:
    name: openai-mini
    type: openai
    model: gpt-4o-mini
    api_key_env: OPENAI_API_KEY
    price_input_per_1m: 0.15
    price_output_per_1m: 0.60

  openrouter-mistral:
    name: openrouter-mistral
    type: openai_compatible
    model: mistralai/mistral-7b-instruct
    api_key_env: OPENROUTER_API_KEY
    base_url: https://openrouter.ai/api/v1
    price_input_per_1m: 0.07
    price_output_per_1m: 0.07

  ollama-mistral:
    name: ollama-mistral
    type: openai_compatible
    model: mistral
    base_url: http://localhost:11434/v1
    api_key_env: OLLAMA_DUMMY_KEY

Script de prueba:

# examples/test_fallback.py
import logging
from unified_ai_client import UnifiedClient, AllProvidersFailedError

logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")

client = UnifiedClient.from_yaml("examples/clients_fallback.yaml")

# Caso normal
try:
    r = client.chat("Di hola en 5 palabras", max_tokens=20)
    print(f"\n→ OK desde {r.provider}: {r.text}")
except AllProvidersFailedError as e:
    print(f"\n✗ Todos fallaron: {e}")

Forzar fallo del primary para verificar fallback:

Cambia OPENAI_API_KEY a un valor inválido y vuelve a correr:

OPENAI_API_KEY=invalid python examples/test_fallback.py

Deberías ver:

ERROR ... : Auth error en 'openai-mini': ...
WARNING ... : 'openai-mini' falló: ...
→ OK desde openrouter-mistral: Hola desde la otra orilla.

Política configurable: fallback opt-in por request

A veces quieres deshabilitar el fallback para una request específica (debugging, A/B testing un provider). Agrega un parámetro:

def chat(
    self,
    prompt: str,
    *,
    system: str | None = None,
    max_tokens: int = 256,
    temperature: float = 0.7,
    use_fallback: bool = True,
) -> ChatResponse:
    ...
    return self.chat_with_messages(
        messages,
        max_tokens=max_tokens,
        temperature=temperature,
        use_fallback=use_fallback,
    )

def chat_with_messages(
    self,
    messages: list[Message],
    max_tokens: int = 256,
    temperature: float = 0.7,
    use_fallback: bool = True,
) -> ChatResponse:
    adapters = self.adapters if use_fallback else [self.adapters[0]]
    # ... resto igual, iterando sobre `adapters` en lugar de `self.adapters`

Uso:

# Default: usa fallback
r = client.chat("Hola")

# Sin fallback: solo intenta primary
r = client.chat("Hola", use_fallback=False)

Trampas comunes

Trampa 1 — "Retry forever en rate limit." Sin cap, te quedas atorado. Mi código usa max_retries=2 con backoff exponencial. Para tu caso real, considera más sofisticado (jitter, max time total).

Trampa 2 — "Circuit breaker que nunca se cierra." Sin _registrar_exito reseteando el counter, el circuito queda abierto para siempre tras 3 fallos. Verifica que tu lógica de éxito sí lo cierra.

Trampa 3 — "Fallback de OpenAI → OpenAI." Si tu fallback es el mismo provider (e.g., diferente modelo), un outage de OpenAI tumba ambos. El fallback debe ser de un proveedor distinto para ser realmente útil.

Trampa 4 — "Mi log spamea durante outage largo." 3 providers × 100 requests × outage de 1 hora = 30,000 líneas de log. Con circuit breaker baja drásticamente, pero verifica que el log de "circuit abierto" sea INFO o DEBUG, no WARNING por request, solo cuando se abre/cierra.

Trampa 5 — "Probé el fallback solo en demo, no en tests." Demos prueban el happy path. Necesitas tests unitarios que mockean fallos del primary y verifican que fallback se invoca (cápsula 07).

Ejercicio

Modifica el código para:

1. Hacer el UMBRAL_CIRCUITO y DURACION_CIRCUITO_S configurables por instancia (no constantes de clase)
2. Agregar un método client.reset_circuit(provider_name) para forzar resetear un circuit (útil para tests y debugging)
3. Agregar un método client.circuit_status() que devuelve un dict con el estado de cada circuit

class UnifiedClient:
    def __init__(
        self,
        config: ClientConfig,
        umbral_circuito: int = 3,
        duracion_circuito_s: int = 60,
    ):
        # ... resto igual
        self.umbral_circuito = umbral_circuito
        self.duracion_circuito_s = duracion_circuito_s
        # ...

    def _registrar_fallo(self, provider_name: str) -> None:
        state = self.circuit[provider_name]
        state.fallos_consecutivos += 1
        if state.fallos_consecutivos >= self.umbral_circuito:
            state.abierto_hasta = time.time() + self.duracion_circuito_s
            # ...

    def reset_circuit(self, provider_name: str) -> None:
        if provider_name not in self.circuit:
            raise ValueError(f"Provider desconocido: {provider_name}")
        self.circuit[provider_name] = CircuitState()

    def circuit_status(self) -> dict[str, dict]:
        now = time.time()
        return {
            name: {
                "fallos_consecutivos": state.fallos_consecutivos,
                "abierto": state.abierto_hasta > now,
                "segundos_hasta_cerrar": max(0, state.abierto_hasta - now),
            }
            for name, state in self.circuit.items()
        }


# Uso
client = UnifiedClient.from_yaml("clients.yaml", umbral_circuito=5, duracion_circuito_s=120)
print(client.circuit_status())
client.reset_circuit("openai-mini")

Resumen

Aprendiste:

• ✅ Iterar sobre primary + fallbacks con manejo de errores diferenciado por tipo
• ✅ Retry con backoff para errores transitorios (rate limit)
• ✅ Saltar provider para errores permanentes (auth, config)
• ✅ Circuit breaker simple que evita martillear providers caídos
• ✅ AllProvidersFailedError con detalle de qué falló en cada uno
• ✅ Fallback opt-in por request (para debugging y A/B testing)

Checkpoint: si forzaste el fallo del primary y tu cliente cayó automáticamente al fallback sin tu intervención, estás listo.

Siguiente cápsula

05 — Cost optimization. Hasta acá fallback corrige errores. Vamos a usar la misma estructura para routing por prioridad: dado un perfil ("cost-first" / "quality-first" / "balanced"), elegir el provider óptimo automáticamente sin que el código de aplicación tenga que saber cuál.

Recursos

1. Martin Fowler — Circuit Breaker pattern — el patrón explicado.
2. Tenacity — librería Python para retry/backoff serio.
3. pybreaker — circuit breaker production-ready en Python.
4. Exponential Backoff and Jitter (AWS) — por qué jitter importa para retry distribuido.