7. Multi-Model Structured Output

Descripción

En sistemas de producción robustos, rara vez dependes de un solo proveedor de LLM. Los proveedores tienen diferentes precios, fortalezas, límites de rate y disponibilidad. Además, cada uno retorna datos en formatos ligeramente distintos.

En esta cápsula aprenderás: el patrón Adapter para múltiples proveedores, normalización de outputs a un schema canónico, fallback con retry, gestión de rate limits, y estrategias de selección de proveedor basadas en costo y rendimiento.

Por qué Multi-Model en producción

Patrón Adapter: Interfaz Unificada

El patrón Adapter crea una interfaz común para múltiples proveedores, ocultando las diferencias de API detrás de una abstracción.

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Any
import json

@dataclass
class LLMResponse:
    """Respuesta normalizada de cualquier proveedor."""
    contenido: str
    tokens_input: int
    tokens_output: int
    modelo: str
    proveedor: str
    latencia_ms: float
    metadata: dict = None
    
    @property
    def costo_estimado_usd(self) -> float:
        """Calcula costo estimado basado en proveedor y modelo."""
        precios = {
            "openai": {
                "gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
                "gpt-4o": {"input": 0.0025, "output": 0.01}
            },
            "anthropic": {
                "claude-3-5-haiku-20241022": {"input": 0.0008, "output": 0.004},
                "claude-3-5-sonnet-20241022": {"input": 0.003, "output": 0.015}
            }
        }
        
        proveedor_precios = precios.get(self.proveedor, {})
        modelo_precios = proveedor_precios.get(self.modelo, {"input": 0.001, "output": 0.002})
        
        costo = (
            (self.tokens_input / 1000) * modelo_precios["input"] +
            (self.tokens_output / 1000) * modelo_precios["output"]
        )
        return round(costo, 6)

class LLMAdapter(ABC):
    """Interfaz base para todos los adaptadores de LLM."""
    
    @abstractmethod
    def completar(self, system: str, user: str, **kwargs) -> LLMResponse:
        """Genera una completación."""
        pass
    
    @abstractmethod
    def completar_json(self, system: str, user: str, **kwargs) -> dict:
        """Genera una completación en formato JSON."""
        pass
    
    @abstractmethod
    def health_check(self) -> bool:
        """Verifica que el proveedor está disponible."""
        pass
    
    def extraer_entidades(self, texto: str) -> dict:
        """Extrae entidades del texto. Implementación genérica."""
        system = """
Extrae entidades del texto. Responde ÚNICAMENTE en JSON con esta estructura:
{
    "personas": ["nombre1", "nombre2"],
    "organizaciones": ["org1", "org2"],
    "lugares": ["lugar1", "lugar2"],
    "fechas": ["fecha1"],
    "otros": ["entidad1"]
}
Si no hay entidades de un tipo, usa lista vacía.
"""
        return self.completar_json(system=system, user=texto)

Adaptador OpenAI

import time
from openai import OpenAI, RateLimitError, APIConnectionError

class OpenAIAdapter(LLMAdapter):
    """Adaptador para la API de OpenAI."""
    
    def __init__(
        self, 
        modelo: str = "gpt-4o-mini",
        api_key: str | None = None,
        timeout: int = 30
    ):
        self.cliente = OpenAI(api_key=api_key) if api_key else OpenAI()
        self.modelo = modelo
        self.timeout = timeout
    
    def completar(self, system: str, user: str, **kwargs) -> LLMResponse:
        """
        Genera completación con OpenAI.
        
        Args:
            system: System prompt
            user: User message
            **kwargs: Parámetros adicionales (temperature, max_tokens, etc.)
        
        Returns:
            LLMResponse normalizado
        """
        inicio = time.time()
        
        response = self.cliente.chat.completions.create(
            model=self.modelo,
            messages=[
                {"role": "system", "content": system},
                {"role": "user", "content": user}
            ],
            **kwargs
        )
        
        latencia_ms = (time.time() - inicio) * 1000
        
        return LLMResponse(
            contenido=response.choices[0].message.content,
            tokens_input=response.usage.prompt_tokens,
            tokens_output=response.usage.completion_tokens,
            modelo=self.modelo,
            proveedor="openai",
            latencia_ms=latencia_ms,
            metadata={
                "finish_reason": response.choices[0].finish_reason,
                "response_id": response.id
            }
        )
    
    def completar_json(self, system: str, user: str, **kwargs) -> dict:
        """Genera completación y parsea como JSON."""
        kwargs["response_format"] = {"type": "json_object"}
        response = self.completar(system=system, user=user, **kwargs)
        return json.loads(response.contenido)
    
    def completar_con_schema(
        self, 
        system: str, 
        user: str, 
        schema: type,
        **kwargs
    ) -> dict:
        """
        Usa Structured Outputs de OpenAI para garantizar schema.
        
        Args:
            schema: Clase Pydantic para el output
        """
        # OpenAI Structured Outputs con Pydantic
        response = self.cliente.beta.chat.completions.parse(
            model=self.modelo,
            messages=[
                {"role": "system", "content": system},
                {"role": "user", "content": user}
            ],
            response_format=schema,
            **kwargs
        )
        return response.choices[0].message.parsed
    
    def health_check(self) -> bool:
        """Verifica disponibilidad de OpenAI."""
        try:
            self.cliente.models.retrieve(self.modelo)
            return True
        except Exception:
            return False

Adaptador Anthropic

import anthropic
import time

class AnthropicAdapter(LLMAdapter):
    """Adaptador para la API de Anthropic."""
    
    def __init__(
        self,
        modelo: str = "claude-3-5-haiku-20241022",
        api_key: str | None = None,
        max_tokens: int = 1024
    ):
        self.cliente = anthropic.Anthropic(api_key=api_key) if api_key else anthropic.Anthropic()
        self.modelo = modelo
        self.max_tokens_default = max_tokens
    
    def completar(self, system: str, user: str, **kwargs) -> LLMResponse:
        """
        Genera completación con Anthropic.
        
        Nota: Anthropic usa 'max_tokens' como parámetro requerido.
        """
        inicio = time.time()
        
        max_tokens = kwargs.pop("max_tokens", self.max_tokens_default)
        
        message = self.cliente.messages.create(
            model=self.modelo,
            max_tokens=max_tokens,
            system=system,
            messages=[{"role": "user", "content": user}],
            **kwargs
        )
        
        latencia_ms = (time.time() - inicio) * 1000
        
        return LLMResponse(
            contenido=message.content[0].text,
            tokens_input=message.usage.input_tokens,
            tokens_output=message.usage.output_tokens,
            modelo=self.modelo,
            proveedor="anthropic",
            latencia_ms=latencia_ms,
            metadata={
                "stop_reason": message.stop_reason,
                "message_id": message.id
            }
        )
    
    def completar_json(self, system: str, user: str, **kwargs) -> dict:
        """
        Genera completación JSON con Anthropic.
        
        Anthropic no tiene response_format JSON nativo, 
        pero es muy bueno siguiendo instrucciones JSON en el system prompt.
        """
        system_json = f"{system}\n\nIMPORTANTE: Responde ÚNICAMENTE con JSON válido. Sin texto adicional."
        
        response = self.completar(system=system_json, user=user, **kwargs)
        
        # Limpiar posibles prefijos
        contenido = response.contenido.strip()
        
        # Extraer JSON si está en bloque de código
        import re
        match = re.search(r"```(?:json)?\s*([\s\S]+?)\s*```", contenido)
        if match:
            contenido = match.group(1)
        
        return json.loads(contenido)
    
    def health_check(self) -> bool:
        """Verifica disponibilidad de Anthropic."""
        try:
            # Hacer llamada mínima
            self.cliente.messages.create(
                model=self.modelo,
                max_tokens=5,
                messages=[{"role": "user", "content": "test"}]
            )
            return True
        except Exception:
            return False

Normalización de Outputs

Diferentes modelos usan nombres diferentes para las mismas entidades. La normalización garantiza un schema canónico.

from pydantic import BaseModel, field_validator
from typing import Any

class EntidadesExtraidas(BaseModel):
    """Schema canónico para entidades extraídas."""
    personas: list[str] = []
    organizaciones: list[str] = []
    lugares: list[str] = []
    fechas: list[str] = []
    otros: list[str] = []
    
    @field_validator("*", mode="before")
    @classmethod
    def asegurar_lista(cls, v: Any) -> list:
        """Convierte valores no-lista a lista."""
        if v is None:
            return []
        if isinstance(v, str):
            return [v] if v.strip() else []
        if not isinstance(v, list):
            return list(v)
        return [str(item).strip() for item in v if item]

# Mapeo de nombres de campos por proveedor
FIELD_ALIASES = {
    "personas": ["personas", "people", "persons", "nombre", "nombres", "individuals"],
    "organizaciones": ["organizaciones", "organizations", "orgs", "companies", "empresas"],
    "lugares": ["lugares", "places", "locations", "cities", "locaciones", "ubicaciones"],
    "fechas": ["fechas", "dates", "timestamps", "times", "datas"],
    "otros": ["otros", "others", "misc", "entities", "entidades", "other_entities"]
}

def normalizar_entidades(data: dict) -> EntidadesExtraidas:
    """
    Normaliza output de cualquier proveedor al schema canónico.
    
    Args:
        data: Dict con entidades en cualquier formato
    
    Returns:
        EntidadesExtraidas normalizado
    """
    normalizado = {}
    
    for campo_canonico, aliases in FIELD_ALIASES.items():
        for alias in aliases:
            if alias in data:
                valor = data[alias]
                # Si es string, convertir a lista
                if isinstance(valor, str):
                    normalizado[campo_canonico] = [valor] if valor.strip() else []
                else:
                    normalizado[campo_canonico] = valor
                break
        
        # Si ningún alias encontrado, usar lista vacía
        if campo_canonico not in normalizado:
            normalizado[campo_canonico] = []
    
    return EntidadesExtraidas(**normalizado)

# Tests de normalización
casos_normalizacion = [
    # OpenAI típico
    {"personas": ["Juan Pérez"], "organizations": ["Google"], "places": ["Ciudad de México"]},
    # Anthropic típico  
    {"people": ["María García"], "orgs": ["Microsoft"], "locations": ["Monterrey"], "dates": ["2024-01-15"]},
    # Formato alternativo
    {"individuals": ["Carlos López"], "companies": ["Meta"], "cities": ["Guadalajara"]},
]

for caso in casos_normalizacion:
    normalizado = normalizar_entidades(caso)
    print(f"Input: {list(caso.keys())}")
    print(f"Normalizado: personas={normalizado.personas}, orgs={normalizado.organizaciones}")
    print()

Fallback con Retry y Backoff

import time
import random
import logging
from typing import Callable, TypeVar

logger = logging.getLogger(__name__)
T = TypeVar("T")

class RetryConfig:
    """Configuración de política de retry."""
    
    def __init__(
        self,
        max_intentos: int = 3,
        delay_base: float = 1.0,
        delay_max: float = 60.0,
        backoff_factor: float = 2.0,
        jitter: bool = True,
        excepciones_reintentables: tuple = (Exception,)
    ):
        self.max_intentos = max_intentos
        self.delay_base = delay_base
        self.delay_max = delay_max
        self.backoff_factor = backoff_factor
        self.jitter = jitter
        self.excepciones_reintentables = excepciones_reintentables
    
    def calcular_delay(self, intento: int) -> float:
        """Calcula delay con backoff exponencial y jitter opcional."""
        delay = min(
            self.delay_base * (self.backoff_factor ** (intento - 1)),
            self.delay_max
        )
        if self.jitter:
            delay *= (0.5 + random.random() * 0.5)  # Jitter del 50%
        return delay

def con_retry(func: Callable[..., T], config: RetryConfig | None = None, **kwargs) -> T:
    """
    Ejecuta función con retry y backoff exponencial.
    
    Args:
        func: Función a ejecutar
        config: Configuración de retry (usa defaults si None)
        **kwargs: Argumentos para pasar a func
    
    Returns:
        Resultado de la función
    
    Raises:
        El último error si todos los intentos fallan
    """
    config = config or RetryConfig()
    ultimo_error = None
    
    for intento in range(1, config.max_intentos + 1):
        try:
            return func(**kwargs)
        except config.excepciones_reintentables as e:
            ultimo_error = e
            
            if intento == config.max_intentos:
                logger.error(f"Todos los {config.max_intentos} intentos fallaron")
                raise
            
            delay = config.calcular_delay(intento)
            logger.warning(
                f"Intento {intento}/{config.max_intentos} falló: {e}. "
                f"Reintentando en {delay:.2f}s..."
            )
            time.sleep(delay)
    
    raise ultimo_error

def extraer_con_fallback(
    texto: str,
    adapters: list[LLMAdapter],
    task: str = "entidades",
    retry_config: RetryConfig | None = None
) -> dict:
    """
    Extrae información con fallback automático entre proveedores.
    
    Args:
        texto: Texto a procesar
        adapters: Lista de adaptadores en orden de preferencia
        task: Tarea a ejecutar
        retry_config: Configuración de retry por proveedor
    
    Returns:
        Resultado del primer proveedor exitoso
    """
    config = retry_config or RetryConfig(max_intentos=2, delay_base=0.5)
    errores = []
    
    for adapter in adapters:
        nombre_proveedor = f"{adapter.proveedor if hasattr(adapter, 'proveedor') else type(adapter).__name__}"
        
        try:
            logger.info(f"Intentando con {nombre_proveedor}...")
            
            def llamar():
                if task == "entidades":
                    return adapter.extraer_entidades(texto)
                else:
                    raise ValueError(f"Task no reconocida: {task}")
            
            resultado = con_retry(llamar, config)
            
            logger.info(f"✅ {nombre_proveedor} exitoso")
            return resultado
            
        except Exception as e:
            error_info = {
                "proveedor": nombre_proveedor,
                "error": str(e),
                "tipo": type(e).__name__
            }
            errores.append(error_info)
            logger.warning(f"❌ {nombre_proveedor} falló: {e}")
            continue
    
    raise RuntimeError(
        f"Todos los proveedores fallaron. Errores: {json.dumps(errores, indent=2)}"
    )

Router Inteligente de Proveedores

En lugar de always usar fallback, un router inteligente selecciona el mejor proveedor según la tarea.

from enum import Enum

class TareaLLM(Enum):
    CLASIFICACION_RAPIDA = "clasificacion_rapida"
    ANALISIS_COMPLEJO = "analisis_complejo"
    EXTRACCION_DATOS = "extraccion_datos"
    GENERACION_TEXTO = "generacion_texto"
    ANALISIS_CODIGO = "analisis_codigo"

@dataclass
class ProviderConfig:
    """Configuración de un proveedor para routing."""
    nombre: str
    adapter: LLMAdapter
    costo_relativo: float  # 1.0 = baseline
    velocidad_relativa: float  # 1.0 = baseline
    tareas_optimas: list[TareaLLM]
    disponible: bool = True

class LLMRouter:
    """
    Router inteligente que selecciona el mejor proveedor según contexto.
    """
    
    def __init__(self, providers: list[ProviderConfig]):
        self.providers = providers
        self._stats: dict[str, dict] = {p.nombre: {"exitos": 0, "errores": 0} for p in providers}
    
    def seleccionar_proveedor(
        self,
        tarea: TareaLLM,
        optimizar_por: str = "balance"  # "costo", "velocidad", "calidad", "balance"
    ) -> ProviderConfig:
        """
        Selecciona el mejor proveedor disponible para la tarea.
        
        Args:
            tarea: Tipo de tarea a ejecutar
            optimizar_por: Criterio de optimización
        
        Returns:
            ProviderConfig del proveedor seleccionado
        """
        # Filtrar proveedores disponibles
        disponibles = [p for p in self.providers if p.disponible]
        
        if not disponibles:
            raise RuntimeError("No hay proveedores disponibles")
        
        # Priorizar proveedores con tarea óptima
        optimos = [p for p in disponibles if tarea in p.tareas_optimas]
        candidatos = optimos if optimos else disponibles
        
        # Ordenar según criterio
        if optimizar_por == "costo":
            return min(candidatos, key=lambda p: p.costo_relativo)
        elif optimizar_por == "velocidad":
            return max(candidatos, key=lambda p: p.velocidad_relativa)
        elif optimizar_por == "balance":
            # Score compuesto: bajo costo + alta velocidad
            def score_balance(p: ProviderConfig) -> float:
                return p.velocidad_relativa / p.costo_relativo
            return max(candidatos, key=score_balance)
        else:
            return candidatos[0]
    
    def ejecutar(
        self,
        system: str,
        user: str,
        tarea: TareaLLM = TareaLLM.CLASIFICACION_RAPIDA,
        optimizar_por: str = "balance",
        fallback: bool = True
    ) -> LLMResponse:
        """
        Ejecuta llamada LLM con routing y fallback automático.
        """
        proveedor_principal = self.seleccionar_proveedor(tarea, optimizar_por)
        
        try:
            response = proveedor_principal.adapter.completar(system=system, user=user)
            self._stats[proveedor_principal.nombre]["exitos"] += 1
            return response
            
        except Exception as e:
            self._stats[proveedor_principal.nombre]["errores"] += 1
            logger.warning(f"Proveedor principal {proveedor_principal.nombre} falló: {e}")
            
            if not fallback:
                raise
            
            # Intentar con otros proveedores
            otros_disponibles = [
                p for p in self.providers 
                if p.nombre != proveedor_principal.nombre and p.disponible
            ]
            
            for proveedor_fallback in otros_disponibles:
                try:
                    response = proveedor_fallback.adapter.completar(system=system, user=user)
                    self._stats[proveedor_fallback.nombre]["exitos"] += 1
                    logger.info(f"Fallback exitoso con {proveedor_fallback.nombre}")
                    return response
                except Exception as e2:
                    self._stats[proveedor_fallback.nombre]["errores"] += 1
                    continue
            
            raise RuntimeError("Todos los proveedores fallaron")
    
    def stats(self) -> dict:
        """Retorna estadísticas de uso por proveedor."""
        resultado = {}
        for nombre, stats in self._stats.items():
            total = stats["exitos"] + stats["errores"]
            resultado[nombre] = {
                **stats,
                "tasa_exito": stats["exitos"] / total if total > 0 else 0
            }
        return resultado

# Configuración del router
def crear_router() -> LLMRouter:
    openai_adapter = OpenAIAdapter(modelo="gpt-4o-mini")
    anthropic_adapter = AnthropicAdapter(modelo="claude-3-5-haiku-20241022")
    
    providers = [
        ProviderConfig(
            nombre="openai_mini",
            adapter=openai_adapter,
            costo_relativo=1.0,
            velocidad_relativa=1.0,
            tareas_optimas=[
                TareaLLM.CLASIFICACION_RAPIDA,
                TareaLLM.EXTRACCION_DATOS,
                TareaLLM.ANALISIS_CODIGO
            ]
        ),
        ProviderConfig(
            nombre="anthropic_haiku",
            adapter=anthropic_adapter,
            costo_relativo=1.5,  # Ligeramente más caro
            velocidad_relativa=0.9,  # Ligeramente más lento
            tareas_optimas=[
                TareaLLM.ANALISIS_COMPLEJO,
                TareaLLM.GENERACION_TEXTO
            ]
        )
    ]
    
    return LLMRouter(providers)

Extracción Estructurada con Schema Canónico

from pydantic import BaseModel, Field
from openai import OpenAI
import anthropic
import json

# Schema canónico para extracción
class PersonaExtraida(BaseModel):
    nombre: str
    cargo: str | None = None
    organizacion: str | None = None

class NoticiaBusiness(BaseModel):
    titulo: str
    fecha: str | None = None
    personas: list[PersonaExtraida] = Field(default_factory=list)
    organizaciones: list[str] = Field(default_factory=list)
    monto_usd: float | None = None
    tipo_evento: str | None = None
    resumen: str

def extraer_noticia_openai(texto: str) -> NoticiaBusiness:
    """Extrae datos de noticia usando OpenAI con JSON mode."""
    client = OpenAI()
    
    system = """
Extrae información estructurada de la noticia. Responde en JSON con este schema exacto:
{
    "titulo": "string",
    "fecha": "YYYY-MM-DD o null",
    "personas": [{"nombre": "string", "cargo": "string o null", "organizacion": "string o null"}],
    "organizaciones": ["org1", "org2"],
    "monto_usd": número o null,
    "tipo_evento": "adquisicion|financiamiento|alianza|lanzamiento|otro|null",
    "resumen": "2-3 oraciones"
}
"""
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": system},
            {"role": "user", "content": texto}
        ],
        response_format={"type": "json_object"}
    )
    
    data = json.loads(response.choices[0].message.content)
    return NoticiaBusiness(**data)

def extraer_noticia_anthropic(texto: str) -> NoticiaBusiness:
    """Extrae datos de noticia usando Anthropic."""
    client = anthropic.Anthropic()
    
    system = """
Extrae información estructurada de noticias de negocios.
Siempre responde ÚNICAMENTE con JSON válido, sin explicaciones.
Schema requerido: titulo, fecha (YYYY-MM-DD o null), personas (lista de objetos con nombre/cargo/organizacion), 
organizaciones (lista strings), monto_usd (número o null), tipo_evento, resumen.
"""
    
    message = client.messages.create(
        model="claude-3-5-haiku-20241022",
        max_tokens=1024,
        system=system,
        messages=[{"role": "user", "content": texto}]
    )
    
    raw = message.content[0].text.strip()
    
    # Limpiar código markdown si está presente
    import re
    match = re.search(r"```(?:json)?\s*([\s\S]+?)\s*```", raw)
    if match:
        raw = match.group(1)
    
    data = json.loads(raw)
    return NoticiaBusiness(**data)

def extraer_noticia_multi(texto: str) -> tuple[NoticiaBusiness, str]:
    """
    Extrae datos con fallback multi-provider.
    
    Returns:
        Tuple (resultado, proveedor_usado)
    """
    providers = [
        ("openai", extraer_noticia_openai),
        ("anthropic", extraer_noticia_anthropic)
    ]
    
    errores = []
    for nombre, extractor in providers:
        try:
            resultado = extractor(texto)
            return resultado, nombre
        except Exception as e:
            errores.append(f"{nombre}: {str(e)[:100]}")
            logger.warning(f"Extractor {nombre} falló: {e}")
    
    raise RuntimeError(f"Todos los extractores fallaron: {'; '.join(errores)}")

# Test
noticia = """
Monterrey, 15 de marzo de 2024. La startup mexicana Fintech XYZ anunció hoy 
una ronda de financiamiento Serie A de 25 millones de dólares, liderada por 
SoftBank Latin America Fund. El CEO de la compañía, Carlos Martínez, declaró
que los fondos se usarán para expandir operaciones a Colombia y Chile.
Acme Capital también participó en la ronda.
"""

try:
    resultado, proveedor = extraer_noticia_multi(noticia)
    print(f"✅ Extraído con: {proveedor}")
    print(f"Título: {resultado.titulo}")
    print(f"Monto: ${resultado.monto_usd:,.0f} USD")
    print(f"Personas: {[p.nombre for p in resultado.personas]}")
    print(f"Organizaciones: {resultado.organizaciones}")
    print(f"Tipo: {resultado.tipo_evento}")
except RuntimeError as e:
    print(f"❌ Error: {e}")

Métricas y Observabilidad

from datetime import datetime
from collections import defaultdict
import statistics

class LLMMetricsCollector:
    """Recolector de métricas de uso de LLMs."""
    
    def __init__(self):
        self._llamadas: list[dict] = []
        self._por_proveedor: dict[str, list] = defaultdict(list)
    
    def registrar(self, response: LLMResponse, exito: bool = True, error: str | None = None):
        """Registra una llamada al LLM."""
        entrada = {
            "timestamp": datetime.utcnow().isoformat(),
            "proveedor": response.proveedor,
            "modelo": response.modelo,
            "tokens_input": response.tokens_input,
            "tokens_output": response.tokens_output,
            "latencia_ms": response.latencia_ms,
            "costo_usd": response.costo_estimado_usd,
            "exito": exito,
            "error": error
        }
        
        self._llamadas.append(entrada)
        self._por_proveedor[response.proveedor].append(entrada)
    
    def resumen(self) -> dict:
        """Genera resumen de métricas."""
        if not self._llamadas:
            return {"total_llamadas": 0}
        
        exitos = [l for l in self._llamadas if l["exito"]]
        
        return {
            "total_llamadas": len(self._llamadas),
            "tasa_exito": len(exitos) / len(self._llamadas),
            "costo_total_usd": sum(l["costo_usd"] for l in self._llamadas),
            "tokens_totales": sum(l["tokens_input"] + l["tokens_output"] for l in self._llamadas),
            "latencia_promedio_ms": statistics.mean(l["latencia_ms"] for l in self._llamadas),
            "latencia_p95_ms": statistics.quantiles(
                [l["latencia_ms"] for l in self._llamadas], n=20
            )[18] if len(self._llamadas) >= 2 else 0,
            "por_proveedor": {
                proveedor: {
                    "llamadas": len(llamadas),
                    "costo_usd": sum(l["costo_usd"] for l in llamadas),
                    "latencia_promedio_ms": statistics.mean(l["latencia_ms"] for l in llamadas)
                }
                for proveedor, llamadas in self._por_proveedor.items()
            }
        }

# Uso del collector
metrics = LLMMetricsCollector()

# Integrar con el adapter
class MonitoredOpenAIAdapter(OpenAIAdapter):
    def __init__(self, *args, metrics_collector: LLMMetricsCollector | None = None, **kwargs):
        super().__init__(*args, **kwargs)
        self.metrics = metrics_collector
    
    def completar(self, system: str, user: str, **kwargs) -> LLMResponse:
        try:
            response = super().completar(system=system, user=user, **kwargs)
            if self.metrics:
                self.metrics.registrar(response, exito=True)
            return response
        except Exception as e:
            if self.metrics:
                # Crear respuesta dummy para métricas de error
                dummy = LLMResponse("", 0, 0, self.modelo, "openai", 0)
                self.metrics.registrar(dummy, exito=False, error=str(e))
            raise

Troubleshooting

1. Schemas diferentes entre proveedores

Síntoma: OpenAI retorna "organizations" pero Anthropic retorna "organizaciones".

# Solución robusta: Normalización flexible con múltiples aliases
def normalizar_flexible(data: dict, schema_clase: type) -> BaseModel:
    """
    Normaliza dict a schema Pydantic con tolerancia a variaciones.
    Intenta múltiples nombres de campo antes de usar default.
    """
    campos = schema_clase.model_fields
    normalizado = {}
    
    for campo, field_info in campos.items():
        # Intentar el campo exacto primero
        if campo in data:
            normalizado[campo] = data[campo]
            continue
        
        # Intentar aliases del campo
        aliases_campo = FIELD_ALIASES.get(campo, [campo])
        encontrado = False
        
        for alias in aliases_campo:
            if alias in data:
                normalizado[campo] = data[alias]
                encontrado = True
                break
        
        # Usar default si no se encontró
        if not encontrado and field_info.default is not None:
            normalizado[campo] = field_info.default
    
    return schema_clase.model_validate(normalizado)

2. Rate limits y 429 errors

from openai import RateLimitError
import time

class RateLimitHandler:
    """Maneja rate limits con backoff exponencial."""
    
    def __init__(self, max_intentos: int = 5):
        self.max_intentos = max_intentos
    
    def ejecutar_con_rate_limit(self, func, *args, **kwargs):
        """Ejecuta función manejando rate limits automáticamente."""
        for intento in range(self.max_intentos):
            try:
                return func(*args, **kwargs)
            except RateLimitError as e:
                if intento == self.max_intentos - 1:
                    raise
                
                # Extraer tiempo de espera del error si está disponible
                retry_after = getattr(e, "retry_after", None)
                if retry_after:
                    wait_time = float(retry_after)
                else:
                    wait_time = (2 ** intento) + (0.1 * random.random())
                
                logger.warning(f"Rate limit hit. Esperando {wait_time:.1f}s...")
                time.sleep(wait_time)
        
        raise RuntimeError("Rate limit no resuelto")

# Para Anthropic, headers incluyen rate limit info
def extraer_rate_limit_info(response_headers: dict) -> dict:
    """Extrae información de rate limits de headers."""
    return {
        "requests_limit": response_headers.get("anthropic-ratelimit-requests-limit"),
        "requests_remaining": response_headers.get("anthropic-ratelimit-requests-remaining"),
        "tokens_limit": response_headers.get("anthropic-ratelimit-tokens-limit"),
        "tokens_remaining": response_headers.get("anthropic-ratelimit-tokens-remaining"),
    }

3. Costos inesperadamente altos

# Implementar presupuesto máximo por sesión o usuario
class BudgetGuard:
    """Controla el presupuesto máximo de llamadas LLM."""
    
    def __init__(self, budget_usd: float):
        self.budget_usd = budget_usd
        self._gastado = 0.0
    
    def verificar(self, costo_estimado_usd: float):
        """Verifica que hay presupuesto disponible."""
        if self._gastado + costo_estimado_usd > self.budget_usd:
            raise ValueError(
                f"Presupuesto excedido. Gastado: ${self._gastado:.4f}, "
                f"Límite: ${self.budget_usd:.4f}"
            )
    
    def registrar_gasto(self, costo_usd: float):
        """Registra un gasto."""
        self._gastado += costo_usd
    
    @property
    def disponible(self) -> float:
        return max(0, self.budget_usd - self._gastado)

# Estimación de costo antes de llamar
def estimar_costo_llamada(
    system: str,
    user: str,
    modelo: str = "gpt-4o-mini",
    max_output_tokens: int = 500
) -> float:
    """Estima el costo de una llamada antes de ejecutarla."""
    try:
        import tiktoken
        enc = tiktoken.encoding_for_model(modelo)
        tokens_input = len(enc.encode(system + user))
    except Exception:
        tokens_input = len((system + user).split()) * 1.3
    
    precios_por_1k = {
        "gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
        "gpt-4o": {"input": 0.0025, "output": 0.01},
        "claude-3-5-haiku-20241022": {"input": 0.0008, "output": 0.004},
    }
    
    precios = precios_por_1k.get(modelo, {"input": 0.001, "output": 0.002})
    
    costo = (
        (tokens_input / 1000) * precios["input"] +
        (max_output_tokens / 1000) * precios["output"]
    )
    
    return round(costo, 6)

Ejercicios

Ejercicio 1: Implementar Adapter para Google Gemini

Crea un GeminiAdapter que implemente la misma interfaz LLMAdapter usando la API de Google Generative AI.

# pip install google-generativeai

class GeminiAdapter(LLMAdapter):
    """Adaptador para Google Gemini API."""
    
    def __init__(self, modelo: str = "gemini-1.5-flash", api_key: str | None = None):
        import google.generativeai as genai
        
        if api_key:
            genai.configure(api_key=api_key)
        # Si no hay api_key, usa GOOGLE_API_KEY del environment
        
        self.genai = genai
        self.modelo_nombre = modelo
        self.model = genai.GenerativeModel(modelo)
    
    def completar(self, system: str, user: str, **kwargs) -> LLMResponse:
        import time
        inicio = time.time()
        
        # Gemini combina system y user en el mensaje
        prompt_completo = f"{system}\n\n{user}"
        
        response = self.model.generate_content(
            prompt_completo,
            generation_config=self.genai.types.GenerationConfig(
                max_output_tokens=kwargs.get("max_tokens", 1024),
                temperature=kwargs.get("temperature", 0.7)
            )
        )
        
        latencia_ms = (time.time() - inicio) * 1000
        
        return LLMResponse(
            contenido=response.text,
            tokens_input=response.usage_metadata.prompt_token_count,
            tokens_output=response.usage_metadata.candidates_token_count,
            modelo=self.modelo_nombre,
            proveedor="google",
            latencia_ms=latencia_ms
        )
    
    def completar_json(self, system: str, user: str, **kwargs) -> dict:
        system_json = f"{system}\n\nResponde ÚNICAMENTE con JSON válido, sin texto adicional."
        response = self.completar(system=system_json, user=user, **kwargs)
        
        contenido = response.contenido.strip()
        import re
        match = re.search(r"```(?:json)?\s*([\s\S]+?)\s*```", contenido)
        if match:
            contenido = match.group(1)
        
        return json.loads(contenido)
    
    def health_check(self) -> bool:
        try:
            models = [m for m in self.genai.list_models()]
            return any(self.modelo_nombre in m.name for m in models)
        except Exception:
            return False

# Uso
# gemini_adapter = GeminiAdapter(modelo="gemini-1.5-flash")
# resultado = gemini_adapter.extraer_entidades("Google anunció...")

Ejercicio 2: Fallback con preferencia y logging de métricas

Implementa extraer_con_fallback_v2 que registre qué proveedor se usó, tokens consumidos y costo.

from dataclasses import dataclass

@dataclass
class ResultadoConMeta:
    datos: dict
    proveedor_usado: str
    tokens_total: int
    costo_usd: float
    intentos: int
    errores_previos: list[str]

def extraer_con_fallback_v2(
    texto: str,
    adapters: list[LLMAdapter],
    metrics: LLMMetricsCollector | None = None
) -> ResultadoConMeta:
    """
    Extrae datos con fallback y tracking de métricas.
    """
    errores_previos = []
    
    for i, adapter in enumerate(adapters):
        proveedor = type(adapter).__name__.replace("Adapter", "").lower()
        
        try:
            # Llamada real al adapter
            system = """Extrae entidades. Responde en JSON: 
{"personas": [], "organizaciones": [], "lugares": [], "fechas": []}"""
            
            inicio = time.time()
            data = adapter.completar_json(system=system, user=texto)
            latencia = (time.time() - inicio) * 1000
            
            # Crear LLMResponse para métricas
            response_mock = LLMResponse(
                contenido=json.dumps(data),
                tokens_input=len(texto.split()) * 2,  # Estimado
                tokens_output=len(json.dumps(data).split()),
                modelo=adapter.modelo if hasattr(adapter, 'modelo') else "unknown",
                proveedor=proveedor,
                latencia_ms=latencia
            )
            
            if metrics:
                metrics.registrar(response_mock, exito=True)
            
            return ResultadoConMeta(
                datos=data,
                proveedor_usado=proveedor,
                tokens_total=response_mock.tokens_input + response_mock.tokens_output,
                costo_usd=response_mock.costo_estimado_usd,
                intentos=i + 1,
                errores_previos=errores_previos
            )
            
        except Exception as e:
            error_msg = f"{proveedor}: {str(e)[:100]}"
            errores_previos.append(error_msg)
            logger.warning(f"Adapter {proveedor} falló: {e}")
    
    raise RuntimeError(f"Todos los adapters fallaron: {errores_previos}")

# Test
metrics_collector = LLMMetricsCollector()

openai_adapter = OpenAIAdapter()
anthropic_adapter = AnthropicAdapter()

adapters = [openai_adapter, anthropic_adapter]

texto = "Tim Cook, CEO de Apple, presentó el nuevo iPhone en Cupertino el 15 de septiembre."
resultado = extraer_con_fallback_v2(texto, adapters, metrics_collector)

print(f"Proveedor: {resultado.proveedor_usado}")
print(f"Intentos: {resultado.intentos}")
print(f"Costo: ${resultado.costo_usd:.6f}")
print(f"Datos: {resultado.datos}")

print("\nMétricas globales:")
print(json.dumps(metrics_collector.resumen(), indent=2))

Ejercicio 3: Router que se adapta a rate limits

Implementa un router que, cuando detecta rate limit en un proveedor, lo marca temporalmente como no disponible por N segundos.

import time
from openai import RateLimitError as OpenAIRateLimitError

class AdaptiveRouter:
    """Router que aprende de rate limits y los evita temporalmente."""
    
    def __init__(self, providers: list[ProviderConfig]):
        self.providers = {p.nombre: p for p in providers}
        self._cooldown_hasta: dict[str, float] = {}
        self._cooldown_default_segundos = 60
    
    def _esta_disponible(self, nombre: str) -> bool:
        """Verifica si el proveedor está fuera de cooldown."""
        cooldown_hasta = self._cooldown_hasta.get(nombre, 0)
        return time.time() > cooldown_hasta
    
    def _aplicar_cooldown(self, nombre: str, segundos: int | None = None):
        """Pone al proveedor en cooldown."""
        segundos = segundos or self._cooldown_default_segundos
        self._cooldown_hasta[nombre] = time.time() + segundos
        logger.warning(f"Proveedor {nombre} en cooldown por {segundos}s")
    
    def ejecutar(self, system: str, user: str) -> LLMResponse:
        """Ejecuta con routing adaptativo."""
        disponibles = [
            p for nombre, p in self.providers.items()
            if self._esta_disponible(nombre) and p.disponible
        ]
        
        if not disponibles:
            raise RuntimeError("No hay proveedores disponibles fuera de cooldown")
        
        for proveedor in disponibles:
            try:
                return proveedor.adapter.completar(system=system, user=user)
            except OpenAIRateLimitError as e:
                # Extraer retry-after si está disponible
                retry_after = getattr(e, 'retry_after', 60)
                self._aplicar_cooldown(proveedor.nombre, retry_after)
            except Exception as e:
                logger.error(f"{proveedor.nombre} error: {e}")
                # Cooldown corto para errores no-rate-limit
                self._aplicar_cooldown(proveedor.nombre, 10)
        
        raise RuntimeError("Todos los proveedores fallaron o están en cooldown")
    
    def estado(self) -> dict:
        """Estado actual de disponibilidad."""
        ahora = time.time()
        return {
            nombre: {
                "disponible": ahora > self._cooldown_hasta.get(nombre, 0),
                "cooldown_restante_s": max(0, self._cooldown_hasta.get(nombre, 0) - ahora)
            }
            for nombre in self.providers
        }

Ejercicio 4: Comparar outputs de dos proveedores con evaluador LLM

Implementa una función que ejecute el mismo query con dos proveedores y use un tercer LLM para evaluar cuál respuesta es mejor.

from openai import OpenAI

client = OpenAI()

def evaluar_con_llm(
    query: str,
    respuesta_a: str,
    respuesta_b: str,
    criterio: str = "precisión y claridad"
) -> dict:
    """
    Usa LLM como juez para evaluar dos respuestas.
    
    Returns:
        dict con ganador ("A", "B", "EMPATE") y justificación
    """
    evaluador_system = """
Eres un evaluador experto e imparcial. 
Tu tarea es comparar dos respuestas y determinar cuál es mejor.
Sé objetivo y basa tu evaluación en los criterios especificados.
Responde ÚNICAMENTE en JSON: {"ganador": "A|B|EMPATE", "justificacion": "string", "puntuacion_a": 0-10, "puntuacion_b": 0-10}
"""
    
    evaluador_user = f"""
Query: {query}

Criterio de evaluación: {criterio}

Respuesta A:
{respuesta_a}

Respuesta B:
{respuesta_b}

Evalúa objetivamente cuál respuesta es mejor según el criterio dado.
"""
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": evaluador_system},
            {"role": "user", "content": evaluador_user}
        ],
        response_format={"type": "json_object"},
        temperature=0
    )
    
    return json.loads(response.choices[0].message.content)

# Comparación real
query = "¿Cuáles son las mejores prácticas para manejar errores en FastAPI?"
system = "Eres un experto en FastAPI y Python. Responde de forma concisa y práctica."

openai_adapter = OpenAIAdapter()
anthropic_adapter = AnthropicAdapter()

resp_openai = openai_adapter.completar(system=system, user=query)
resp_anthropic = anthropic_adapter.completar(system=system, user=query)

evaluacion = evaluar_con_llm(
    query=query,
    respuesta_a=resp_openai.contenido,
    respuesta_b=resp_anthropic.contenido,
    criterio="precisión técnica, ejemplos de código, y claridad"
)

print(f"Ganador: {evaluacion['ganador']}")
print(f"OpenAI score: {evaluacion['puntuacion_a']}/10")
print(f"Anthropic score: {evaluacion['puntuacion_b']}/10")
print(f"Justificación: {evaluacion['justificacion']}")

Resumen

Recursos adicionales

1. OpenAI API Reference
2. Anthropic API Reference
3. Google Generative AI Python SDK
4. Design Patterns: Adapter
5. LiteLLM - Proxy unificado para múltiples proveedores
6. OpenAI Structured Outputs