2. Error Handling para LLM APIs

Descripción

Antes de implementar retry o circuit breakers, necesitas entender qué puede salir mal y por qué. Las APIs de LLM tienen una taxonomía de errores diferente a las APIs REST tradicionales: algunos son transitorios (retry los resuelve), algunos son input errors (retry no te ayuda — el mismo input va a dar el mismo error), y algunos son outages que requieren fallback. En esta cápsula vas a implementar el error handling completo para OpenAI: clasificar cada error, decidir si vale la pena reintentar, y propagar la información correcta al resto del sistema.

Taxonomía de errores de LLM APIs

ERROR TRANSITORIO → Retry puede ayudar
├── Timeout (30s, 60s): el modelo tardó demasiado
│   Causa: prompt muy largo, modelo ocupado
│   Estrategia: retry con backoff corto (1-2s)
│
├── RateLimitError (429): demasiadas requests
│   Causa: superaste el límite de RPM o TPM
│   Estrategia: retry con backoff largo (30-60s) o Retry-After header
│
└── ServerError (500, 502, 503): error del lado de OpenAI
    Causa: outage parcial, instancia con problema
    Estrategia: retry con backoff, max 3-5 intentos

ERROR DE INPUT → No retry, cambiar el input
├── TokenLimitExceeded: el input es demasiado largo
│   Causa: prompt + context > context window
│   Estrategia: truncar el input, no retry con mismo input
│
├── BadRequest (400): input inválido
│   Causa: formato incorrecto de mensajes, parámetros inválidos
│   Estrategia: fix el código, no retry
│
└── ContentPolicyViolation: contenido bloqueado por moderación
    Causa: el input o expected output violan las políticas
    Estrategia: aplicar guardrails pre-LLM (Módulo 4), no retry

ERROR PERMANENTE → No retry, no sirve
├── AuthenticationError (401): API key inválida
│   Causa: key expirada, key incorrecta
│   Estrategia: alerta de configuración, no retry
│
└── PermissionError (403): sin acceso al recurso/modelo
    Causa: plan no tiene acceso al modelo solicitado
    Estrategia: cambiar modelo en config, no retry

OUTPUT ERROR → Problema con la respuesta, no con el request
├── InvalidJSON: el LLM devolvió texto en vez de JSON
│   Causa: modelo olvidó el formato, prompt ambiguo
│   Estrategia: retry (con max 1-2 intentos), fallback a default
│
└── TruncatedResponse: finish_reason="length"
    Causa: max_tokens insuficiente para la respuesta completa
    Estrategia: aumentar max_tokens, no retry con mismo config

Errores de la librería de OpenAI

# Las excepciones del cliente openai de Python

from openai import (
    OpenAI,
    # Jerarquía de excepciones:
    APIError,              # Base para todos los errores de API
    APIConnectionError,    # No se pudo conectar (red, DNS)
    APITimeoutError,       # Request timeout
    RateLimitError,        # 429: rate limit excedido
    AuthenticationError,   # 401: API key inválida
    PermissionDeniedError, # 403: sin acceso
    NotFoundError,         # 404: recurso no encontrado
    UnprocessableEntityError,  # 422: input no procesable
    InternalServerError,   # 5xx: error del servidor de OpenAI
    BadRequestError,       # 400: request inválido
)

# APIError tiene atributos útiles:
# e.status_code: int (el HTTP status code)
# e.message: str (el mensaje de error)
# e.request_id: str (el ID del request en OpenAI — útil para soporte)

Función de clasificación de errores

# src/infrastructure/error_classifier.py
from openai import (
    APIError, APITimeoutError, RateLimitError, APIConnectionError,
    AuthenticationError, PermissionDeniedError, BadRequestError,
    InternalServerError, UnprocessableEntityError
)
from enum import Enum
from dataclasses import dataclass
from typing import Optional

class ErrorCategory(Enum):
    TRANSIENT = "transient"           # Retry puede ayudar
    INPUT_ERROR = "input_error"       # Cambiar el input
    AUTH_ERROR = "auth_error"         # Problema de config
    OUTAGE = "outage"                 # Servicio caído
    OUTPUT_ERROR = "output_error"     # Problema con la respuesta
    UNKNOWN = "unknown"               # No clasificado

@dataclass
class ClassifiedError:
    category: ErrorCategory
    should_retry: bool
    retry_after_seconds: Optional[float]  # Si la API indica cuándo reintentar
    user_message: str                     # Mensaje apropiado para el usuario
    log_level: str                        # "warning", "error", "critical"
    original_error: Exception

def classify_error(e: Exception) -> ClassifiedError:
    """
    Clasifica un error de LLM API y determina la estrategia de manejo.
    """
    
    if isinstance(e, APITimeoutError):
        return ClassifiedError(
            category=ErrorCategory.TRANSIENT,
            should_retry=True,
            retry_after_seconds=None,
            user_message="El servicio tardó demasiado. Reintentando...",
            log_level="warning",
            original_error=e
        )
    
    if isinstance(e, RateLimitError):
        # Intentar extraer el Retry-After del header si está disponible
        retry_after = _extract_retry_after(e)
        return ClassifiedError(
            category=ErrorCategory.TRANSIENT,
            should_retry=True,
            retry_after_seconds=retry_after or 30.0,
            user_message="Servicio temporalmente saturado. Reintentando...",
            log_level="warning",
            original_error=e
        )
    
    if isinstance(e, APIConnectionError):
        return ClassifiedError(
            category=ErrorCategory.OUTAGE,
            should_retry=True,
            retry_after_seconds=5.0,
            user_message="No se pudo conectar con el servicio.",
            log_level="error",
            original_error=e
        )
    
    if isinstance(e, AuthenticationError):
        return ClassifiedError(
            category=ErrorCategory.AUTH_ERROR,
            should_retry=False,
            retry_after_seconds=None,
            user_message="Error de configuración del servicio.",
            log_level="critical",  # Alerta inmediata — config rota
            original_error=e
        )
    
    if isinstance(e, PermissionDeniedError):
        return ClassifiedError(
            category=ErrorCategory.AUTH_ERROR,
            should_retry=False,
            retry_after_seconds=None,
            user_message="Sin acceso al modelo solicitado.",
            log_level="error",
            original_error=e
        )
    
    if isinstance(e, BadRequestError):
        # Distinguir entre "prompt demasiado largo" y "request malformado"
        if e.status_code == 400 and "context_length" in str(e.message).lower():
            return ClassifiedError(
                category=ErrorCategory.INPUT_ERROR,
                should_retry=False,
                retry_after_seconds=None,
                user_message="El texto es demasiado largo para procesar.",
                log_level="warning",
                original_error=e
            )
        return ClassifiedError(
            category=ErrorCategory.INPUT_ERROR,
            should_retry=False,
            retry_after_seconds=None,
            user_message="Solicitud inválida.",
            log_level="error",
            original_error=e
        )
    
    if isinstance(e, InternalServerError):
        return ClassifiedError(
            category=ErrorCategory.TRANSIENT,
            should_retry=True,
            retry_after_seconds=10.0,
            user_message="El servicio está experimentando problemas.",
            log_level="error",
            original_error=e
        )
    
    if isinstance(e, APIError):
        # APIError genérico — clasificar por status_code
        if e.status_code == 429:
            return ClassifiedError(
                category=ErrorCategory.TRANSIENT,
                should_retry=True,
                retry_after_seconds=30.0,
                user_message="Servicio temporalmente saturado.",
                log_level="warning",
                original_error=e
            )
        if e.status_code and e.status_code >= 500:
            return ClassifiedError(
                category=ErrorCategory.TRANSIENT,
                should_retry=True,
                retry_after_seconds=10.0,
                user_message="Error del servidor.",
                log_level="error",
                original_error=e
            )
    
    # Error desconocido
    return ClassifiedError(
        category=ErrorCategory.UNKNOWN,
        should_retry=False,
        retry_after_seconds=None,
        user_message="Error inesperado.",
        log_level="error",
        original_error=e
    )

def _extract_retry_after(e: RateLimitError) -> Optional[float]:
    """Extrae el valor de Retry-After si está disponible."""
    try:
        # El header puede estar en la respuesta
        if hasattr(e, "response") and e.response:
            retry_after = e.response.headers.get("Retry-After")
            if retry_after:
                return float(retry_after)
    except (AttributeError, ValueError):
        pass
    return None

OpenAIProvider con clasificación integrada

# src/infrastructure/openai_provider.py (actualizado)
import time
import structlog
from openai import APIError, APITimeoutError, RateLimitError
from src.infrastructure.llm_provider import LLMProvider, LLMProviderError
from src.infrastructure.error_classifier import classify_error, ErrorCategory
from src.logging_config import calculate_cost

log = structlog.get_logger()

class OpenAIProvider:
    def __init__(self, client, model: str, temperature: float,
                 max_tokens: int, seed: int = None):
        self._client = client
        self._model = model
        self._temperature = temperature
        self._max_tokens = max_tokens
        self._seed = seed
    
    def complete(self, messages: list[dict], **kwargs) -> str:
        """
        Realiza una llamada al API de OpenAI.
        
        Clasifica los errores antes de propagarlos, convirtiendo
        excepciones específicas de OpenAI en LLMProviderError
        con información de clasificación.
        """
        start = time.time()
        
        try:
            response = self._client.chat.completions.create(
                model=self._model,
                messages=messages,
                temperature=self._temperature,
                max_tokens=self._max_tokens,
                seed=self._seed,
                **kwargs
            )
            
            duration_ms = (time.time() - start) * 1000
            cost_usd = calculate_cost(
                self._model,
                response.usage.prompt_tokens,
                response.usage.completion_tokens
            )
            
            log.info(
                "llm_call_completed",
                model=self._model,
                input_tokens=response.usage.prompt_tokens,
                output_tokens=response.usage.completion_tokens,
                cost_usd=cost_usd,
                duration_ms=round(duration_ms, 1),
                finish_reason=response.choices[0].finish_reason
            )
            
            # Advertir si la respuesta fue truncada
            if response.choices[0].finish_reason == "length":
                log.warning(
                    "response_truncated_by_max_tokens",
                    max_tokens=self._max_tokens,
                    output_tokens=response.usage.completion_tokens
                )
            
            return response.choices[0].message.content
        
        except Exception as e:
            duration_ms = (time.time() - start) * 1000
            classified = classify_error(e)
            
            # Loguear con el nivel apropiado
            log_fn = getattr(log, classified.log_level)
            log_fn(
                "llm_call_failed",
                error_type=type(e).__name__,
                error_category=classified.category.value,
                should_retry=classified.should_retry,
                model=self._model,
                duration_ms=round(duration_ms, 1)
            )
            
            raise LLMProviderError(
                message=classified.user_message,
                original_error=e,
                category=classified.category,
                should_retry=classified.should_retry,
                retry_after=classified.retry_after_seconds
            )

LLMProviderError enriquecido

# src/infrastructure/llm_provider.py (actualizado)
from src.infrastructure.error_classifier import ErrorCategory
from typing import Optional

class LLMProviderError(Exception):
    """
    Error genérico del LLM provider con información de clasificación.
    
    Encapsula errores específicos de proveedor y añade:
    - category: tipo de error para routing de retry/fallback
    - should_retry: si vale la pena reintentar
    - retry_after: cuántos segundos esperar antes de reintentar
    """
    def __init__(
        self,
        message: str,
        original_error: Exception = None,
        category: ErrorCategory = ErrorCategory.UNKNOWN,
        should_retry: bool = False,
        retry_after: Optional[float] = None
    ):
        super().__init__(message)
        self.original_error = original_error
        self.category = category
        self.should_retry = should_retry
        self.retry_after = retry_after
    
    @property
    def is_transient(self) -> bool:
        return self.category == ErrorCategory.TRANSIENT
    
    @property
    def is_auth_error(self) -> bool:
        return self.category == ErrorCategory.AUTH_ERROR
    
    @property
    def is_input_error(self) -> bool:
        return self.category == ErrorCategory.INPUT_ERROR

Handling de output errors: respuestas malformadas

# src/processing/sentiment_parser.py (con error handling robusto)
import json
import re
import structlog
from typing import Optional

log = structlog.get_logger()

class ParseResult:
    """Resultado del parsing con información de calidad."""
    def __init__(self, data: dict, parse_strategy: str, warnings: list = None):
        self.data = data
        self.parse_strategy = parse_strategy  # "direct", "regex", "fallback"
        self.warnings = warnings or []
        self.is_fallback = parse_strategy == "fallback"

def parse_with_full_error_handling(raw: str, request_id: str = None) -> ParseResult:
    """
    Parsea el output del LLM con múltiples estrategias y logging completo.
    """
    # Estrategia 1: JSON directo
    stripped = raw.strip()
    if stripped.startswith("{"):
        try:
            data = json.loads(stripped)
            return ParseResult(data, "direct")
        except json.JSONDecodeError:
            pass
    
    # Estrategia 2: JSON en markdown code block
    match = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", raw, re.DOTALL)
    if match:
        try:
            data = json.loads(match.group(1))
            log.warning("json_in_markdown", request_id=request_id)
            return ParseResult(data, "markdown_extraction",
                             warnings=["JSON extraído de markdown block"])
        except json.JSONDecodeError:
            pass
    
    # Estrategia 3: Buscar cualquier objeto JSON
    match = re.search(r"\{[^{}]+\}", raw, re.DOTALL)
    if match:
        try:
            data = json.loads(match.group())
            log.warning("json_regex_extracted", request_id=request_id,
                       raw_preview=raw[:100])
            return ParseResult(data, "regex_extraction",
                             warnings=["JSON extraído con regex"])
        except json.JSONDecodeError:
            pass
    
    # Estrategia 4: Fallback total
    log.error(
        "json_parse_completely_failed",
        request_id=request_id,
        raw_preview=raw[:200],
        raw_length=len(raw)
    )
    return ParseResult(
        data={"sentiment": "unknown", "score": 0.0, "confidence": 0.0},
        parse_strategy="fallback",
        warnings=["No se pudo extraer JSON válido — usando fallback"]
    )

Tests del error classifier

# tests/unit/test_error_classifier.py
import pytest
from unittest.mock import MagicMock
from openai import APITimeoutError, RateLimitError, AuthenticationError
from src.infrastructure.error_classifier import classify_error, ErrorCategory

def test_timeout_is_transient_and_retryable():
    error = APITimeoutError.__new__(APITimeoutError)
    result = classify_error(error)
    assert result.category == ErrorCategory.TRANSIENT
    assert result.should_retry is True

def test_rate_limit_has_retry_delay():
    error = MagicMock(spec=RateLimitError)
    error.response = None
    result = classify_error(error)
    assert result.category == ErrorCategory.TRANSIENT
    assert result.should_retry is True
    assert result.retry_after_seconds is not None
    assert result.retry_after_seconds > 0

def test_auth_error_is_not_retryable():
    error = MagicMock(spec=AuthenticationError)
    result = classify_error(error)
    assert result.should_retry is False
    assert result.log_level == "critical"  # Alerta inmediata

def test_server_error_is_transient():
    from openai import InternalServerError
    error = MagicMock(spec=InternalServerError)
    result = classify_error(error)
    assert result.should_retry is True

Ejercicios

Ejercicio 1: Árbol de decisión

Para cada error, dibuja el árbol de decisión (retry/no retry, fallback/no fallback):

1. OpenAI devuelve 500 por primera vez
2. OpenAI devuelve 401
3. El LLM devuelve respuesta JSON inválida por tercera vez seguida
4. El input tiene 150,000 tokens para un modelo con límite de 128K

Ejercicio 2: Completar el classify_error

La función classify_error no maneja UnprocessableEntityError (422). Escribe el bloque que falta:

if isinstance(e, UnprocessableEntityError):
    return ClassifiedError(
        category=???,
        should_retry=???,
        retry_after_seconds=???,
        user_message=???,
        log_level=???,
        original_error=e
    )

if isinstance(e, UnprocessableEntityError):
    return ClassifiedError(
        category=ErrorCategory.INPUT_ERROR,
        should_retry=False,
        retry_after_seconds=None,
        user_message="No se pudo procesar la solicitud.",
        log_level="warning",
        original_error=e
    )

Ejercicio 3: Mejorar el parser de JSON

El parse_with_full_error_handling no maneja el caso donde el LLM responde con un JSON válido pero le faltan campos requeridos (por ejemplo, tiene sentiment pero no score). Escribe una función de validación post-parse:

REQUIRED_FIELDS = {"sentiment", "score", "confidence"}

def validate_parsed_result(data: dict, request_id: str = None) -> ParseResult:
    """Valida que el JSON parseado tenga todos los campos requeridos."""
    missing = REQUIRED_FIELDS - set(data.keys())
    if not missing:
        return ParseResult(data, "validated")
    
    # Rellenar campos faltantes con defaults
    defaults = {"sentiment": "unknown", "score": 0.0, "confidence": 0.0}
    for field in missing:
        data[field] = defaults[field]
    
    log.warning("json_missing_fields", missing=list(missing), request_id=request_id)
    return ParseResult(data, "partial_with_defaults",
                      warnings=[f"Campos faltantes rellenados: {missing}"])

Ejercicio 4: Test para error desconocido

Escribe un test que verifique que classify_error maneja correctamente una excepción completamente desconocida (por ejemplo, un RuntimeError genérico):

def test_unknown_error_is_not_retryable():
    """Errores desconocidos no se reintentan por seguridad."""
    error = RuntimeError("Something completely unexpected")
    result = classify_error(error)
    
    assert result.category == ErrorCategory.UNKNOWN
    assert result.should_retry is False
    assert result.log_level == "error"
    assert result.original_error is error

Troubleshooting

"Mi classify_error no reconoce las excepciones de OpenAI"

Verifica que estás importando del módulo correcto. La librería openai en versión 1.x cambió la jerarquía de excepciones:

# ❌ Incorrecto (v0.x, deprecated)
from openai.error import RateLimitError

# ✅ Correcto (v1.x+)
from openai import RateLimitError

Ejecuta pip show openai para verificar tu versión. Necesitas >=1.0.

"El LLMProviderError no tiene category — me da AttributeError"

Asegúrate de que estás creando el error con todos los campos. Si algún código antiguo hace raise LLMProviderError("msg") sin los argumentos de clasificación, category será ErrorCategory.UNKNOWN por default, pero retry_after será None. Busca todos los raise LLMProviderError en tu código y actualízalos.

"El parser siempre cae al fallback"

Revisa qué está respondiendo el LLM realmente. Agrega un log temporal del raw output:

log.debug("raw_llm_output", raw=raw[:500], length=len(raw))

Causas comunes: el modelo responde con explicación antes del JSON, usa comillas simples en vez de dobles, o incluye trailing commas que no son JSON válido.

"¿Cómo sé si mi error es de la librería openai o de la red?"

APIConnectionError es un error de red (DNS, firewall, proxy). APITimeoutError es un timeout del request HTTP. Ambos son transitorios, pero APIConnectionError es más grave porque puede indicar un problema de infraestructura tuyo (no de OpenAI). Verifica tu conectividad antes de asumir que es culpa de OpenAI.

Resumen

• Clasificar antes de manejar: no todo error merece retry
• Transitorios: timeout, 429, 5xx → retry con backoff
• Input errors: token limit, 400, content policy → fix el input, no retry
• Auth errors: 401, 403 → alerta de configuración, no retry
• Output errors: JSON inválido → retry máx 1 vez, luego fallback
• LLMProviderError enriquecido: transporta la categoría para routing de retry/fallback

Recursos adicionales

1. OpenAI Error Handling Guide — Documentación oficial
2. OpenAI API Reference — Errors — Lista completa de errores
3. HTTP Status Codes — Referencia
4. Retry-After Header — Cómo usarlo