Módulo 2: OpenAI API - Introducción

Error Handling y Retry Strategies

Siguiente →

Descripción de la cápsula

APIs fallan. OpenAI no es excepción. En producción verás:

  • Rate limit errors (429): Demasiados requests
  • Timeouts: API lenta
  • Server errors (500, 503): OpenAI tiene issues
  • Network errors: Tu internet falla

Esta cápsula te enseña a manejar todos estos casos con retry strategies robustas.

Tiempo: 25 minutos
Dificultad: Media-Alta

🎯 Objetivos

  • ✅ Identificar tipos de errores comunes
  • ✅ Implementar exponential backoff
  • ✅ Crear retry logic robusto
  • ✅ Logging de errores

❌ Tipos de Errores Comunes

1. RateLimitError (429)

Cuándo: Excedes requests/min o tokens/min

openai.RateLimitError: Rate limit reached for requests

Causa común:

  • Free tier: >60 requests/min
  • Loop sin delays
  • Spike de tráfico

Solución: Retry con backoff

2. APIError (500, 503)

Cuándo: OpenAI servers tienen problemas

openai.APIError: The server had an error processing your request

Causa:

  • OpenAI outage (raro, pero pasa)
  • Deploy en progreso

Solución: Retry automático

3. Timeout

Cuándo: Request tarda >60s (default timeout)

openai.APITimeoutError: Request timed out

Causa:

  • OpenAI API lenta
  • Network congestion
  • Prompt muy largo

Solución: Aumentar timeout o retry

4. AuthenticationError (401)

Cuándo: API key inválida

openai.AuthenticationError: Incorrect API key provided

Causa:

  • Key incorrecta
  • Key revocada
  • Typo en .env

Solución: Verificar key (NO retry)

5. NetworkError

Cuándo: Sin internet o DNS issues

requests.exceptions.ConnectionError: Failed to establish connection

Causa:

  • Sin internet
  • DNS fail
  • Firewall bloqueando

Solución: Retry con backoff corto

🔄 Retry Strategy: Exponential Backoff

Concepto:

Esperar tiempo creciente entre retries:

  • Retry 1: Espera 1s
  • Retry 2: Espera 2s
  • Retry 3: Espera 4s
  • Retry 4: Espera 8s

Por qué funciona:

  • Si OpenAI está saturado, darle tiempo a recuperarse
  • Evita "stampede" (todos retrying simultáneamente)

Implementación básica:

import time
from openai import OpenAI, RateLimitError, APIError

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

def chat_with_retry(prompt: str, max_retries: int = 3):
    """
    Envía prompt con retry automático.
    
    Args:
        prompt: Mensaje del usuario
        max_retries: Intentos máximos (default: 3)
        
    Returns:
        Respuesta de GPT o None si falla
    """
    
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model="gpt-3.5-turbo",
                messages=[{"role": "user", "content": prompt}]
            )
            return response.choices[0].message.content
            
        except RateLimitError as e:
            if attempt == max_retries - 1:  # Último intento
                print(f"❌ Rate limit después de {max_retries} intentos")
                return None
            
            # Exponential backoff
            wait_time = 2 ** attempt  # 1s, 2s, 4s
            print(f"⚠️ Rate limit, esperando {wait_time}s...")
            time.sleep(wait_time)
            
        except APIError as e:
            if attempt == max_retries - 1:
                print(f"❌ API error después de {max_retries} intentos: {e}")
                return None
            
            wait_time = 2 ** attempt
            print(f"⚠️ API error, reintentando en {wait_time}s...")
            time.sleep(wait_time)
            
        except Exception as e:
            # Errores inesperados (no retry)
            print(f"❌ Error inesperado: {e}")
            return None
    
    return None

# Test
respuesta = chat_with_retry("¿Qué es Python?")
if respuesta:
    print(respuesta)
else:
    print("No se pudo obtener respuesta")

Mejora: Jitter (randomización)

Añadir pequeño random para evitar que todos retries sean simultáneos:

import random

wait_time = (2 ** attempt) + random.uniform(0, 1)

Beneficio: Distribuye carga en OpenAI servers

🛠️ Implementación Avanzada con Tenacity

Librería tenacity (recomendado para producción):

pip install tenacity

Código:

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type
)
from openai import OpenAI, RateLimitError, APIError, APITimeoutError

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

@retry(
    retry=retry_if_exception_type((RateLimitError, APIError, APITimeoutError)),
    wait=wait_exponential(multiplier=1, min=2, max=60),
    stop=stop_after_attempt(5),
    reraise=True
)
def chat_with_tenacity(prompt: str) -> str:
    """
    Envía prompt con retry automático (tenacity).
    
    Retry solo en errores transientes:
    - RateLimitError
    - APIError
    - APITimeoutError
    
    NO retry en:
    - AuthenticationError (key inválida)
    - InvalidRequestError (prompt inválido)
    """
    response = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}],
        timeout=30.0  # 30s timeout
    )
    return response.choices[0].message.content

# Test
try:
    respuesta = chat_with_tenacity("¿Qué es Python?")
    print(respuesta)
except Exception as e:
    print(f"❌ Error después de retries: {e}")

Ventajas:

  • Código más limpio (decorator)
  • Exponential backoff built-in
  • Configurable (min/max wait, max attempts)
  • Solo retry en errores transientes

📊 Logging de Errores

Implementación con logging estándar:

import logging
from datetime import datetime

# Configurar logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('openai_errors.log'),
        logging.StreamHandler()  # También print a consola
    ]
)

def chat_with_logging(prompt: str):
    """Chat con logging de errores."""
    
    try:
        response = client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": prompt}]
        )
        
        # Log éxito
        logging.info(f"Request exitoso | Tokens: {response.usage.total_tokens}")
        
        return response.choices[0].message.content
        
    except RateLimitError as e:
        logging.warning(f"Rate limit error | Prompt: {prompt[:50]}")
        raise
        
    except APIError as e:
        logging.error(f"API error | Status: {e.status_code} | Message: {e.message}")
        raise
        
    except APITimeoutError as e:
        logging.error(f"Timeout error | Prompt length: {len(prompt)}")
        raise
        
    except Exception as e:
        logging.critical(f"Unexpected error | Type: {type(e)} | Message: {e}")
        raise

# Test
try:
    respuesta = chat_with_logging("¿Qué es Python?")
    print(respuesta)
except Exception as e:
    print(f"Error: {e}")

Output en openai_errors.log:

2024-02-15 10:30:15 - INFO - Request exitoso | Tokens: 45
2024-02-15 10:31:22 - WARNING - Rate limit error | Prompt: ¿Qué es Python?
2024-02-15 10:32:45 - ERROR - API error | Status: 503 | Message: Service unavailable

🎯 Estrategia Completa de Error Handling

Código production-ready:

import time
import logging
from typing import Optional
from openai import OpenAI, RateLimitError, APIError, APITimeoutError, AuthenticationError

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

def robust_chat(
    prompt: str,
    max_retries: int = 3,
    initial_wait: float = 1.0
) -> Optional[str]:
    """
    Chat con OpenAI con error handling robusto.
    
    Maneja:
    - Rate limits (retry con backoff)
    - API errors (retry con backoff)
    - Timeouts (retry con timeout mayor)
    - Auth errors (no retry, log crítico)
    - Otros errors (no retry, log)
    
    Returns:
        Respuesta de GPT o None si falla después de retries
    """
    
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model="gpt-3.5-turbo",
                messages=[{"role": "user", "content": prompt}],
                timeout=30.0 + (attempt * 10)  # Aumenta timeout en cada retry
            )
            
            logger.info(f"✅ Request exitoso | Attempt: {attempt + 1} | Tokens: {response.usage.total_tokens}")
            return response.choices[0].message.content
            
        except RateLimitError as e:
            logger.warning(f"⚠️ Rate limit | Attempt: {attempt + 1}/{max_retries}")
            
            if attempt == max_retries - 1:
                logger.error("❌ Rate limit persistente después de retries")
                return None
            
            wait_time = initial_wait * (2 ** attempt)
            logger.info(f"Esperando {wait_time}s antes de retry...")
            time.sleep(wait_time)
            
        except APIError as e:
            logger.warning(f"⚠️ API error {e.status_code} | Attempt: {attempt + 1}/{max_retries}")
            
            if attempt == max_retries - 1:
                logger.error(f"❌ API error persistente: {e.message}")
                return None
            
            wait_time = initial_wait * (2 ** attempt)
            time.sleep(wait_time)
            
        except APITimeoutError as e:
            logger.warning(f"⚠️ Timeout | Attempt: {attempt + 1}/{max_retries}")
            
            if attempt == max_retries - 1:
                logger.error("❌ Timeout persistente")
                return None
            
            # Retry con timeout mayor (ya configurado arriba)
            wait_time = initial_wait
            time.sleep(wait_time)
            
        except AuthenticationError as e:
            logger.critical(f"❌ CRITICAL: Auth error | Verifica API key")
            return None  # NO retry (key inválida)
            
        except Exception as e:
            logger.error(f"❌ Error inesperado: {type(e).__name__} | {e}")
            return None  # NO retry (error desconocido)
    
    return None

# Test
respuesta = robust_chat("¿Qué es Python?")
if respuesta:
    print(respuesta)
else:
    print("No se pudo obtener respuesta después de retries")

🧪 Test de Error Handling

Simular rate limit:

# Forzar rate limit (enviar muchos requests)
for i in range(100):
    respuesta = robust_chat(f"Request {i}")
    # Eventualmente verás rate limit errors y retries

Simular timeout:

# Prompt muy largo (más likely timeout)
long_prompt = "Explica en detalle " * 1000  # 5000+ tokens
respuesta = robust_chat(long_prompt)

📊 Resumen

Conceptos clave:

  1. Tipos de errores:

    • Rate limit (429) → Retry
    • API error (500, 503) → Retry
    • Timeout → Retry con timeout mayor
    • Auth (401) → NO retry, fix key
    • Otros → NO retry, log

  2. Exponential backoff:

    wait_time = 2^attempt  # 1s, 2s, 4s, 8s

  3. Tenacity (recomendado):

    • Decorator @retry
    • Configurable
    • Built-in exponential backoff

  4. Logging:

    • INFO: Request exitoso
    • WARNING: Error con retry
    • ERROR: Error persistente
    • CRITICAL: Auth o bug crítico

Checklist:

  • Implementaste retry con exponential backoff
  • Manejas 3+ tipos de errores (rate limit, API error, timeout)
  • Logging configurado (archivo + consola)
  • Testeaste con rate limit simulado

🔗 Recursos adicionales

  1. OpenAI Error Codes - Docs oficial
  2. Tenacity - Retry library
  3. Python Logging - Docs estándar

➡️ Próximo paso

Siguiente cápsula: 08-proyecto-chatbot-soporte.md

¡Proyecto final del módulo!

Integrarás todo lo aprendido:

  • Setup y API keys
  • Conversaciones con contexto
  • Parameters optimizados
  • Cost tracking
  • Error handling robusto

Construirás un chatbot de soporte técnico production-ready.

Tiempo: 60-90 minutos

Tiempo estimado: 25 minutos
Siguiente: 08-proyecto-chatbot-soporte.md

← AnteriorPricing, Rate Limits y Optimización de CostosSiguiente →Mini-Proyecto: Chatbot de Soporte Técnico