Mejores prácticas para llamadas a APIs de LLMs en producción

Llamar a una API de LLM en producción no es lo mismo que hacerlo en un notebook. La diferencia entre una demo y un sistema que funciona a las 3 de la mañana está en los detalles.

Control de timeouts

Las APIs de LLM son lentas. Una respuesta puede tardar 30-60 segundos para respuestas largas. Sin timeout configurado, tu aplicación puede quedarse colgada indefinidamente.

import anthropic

client = anthropic.Anthropic()

# Timeout específico para LLMs: más largo que una API normal
message = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "..."}],
    timeout=60.0  # 60 segundos
)

Regla práctica: timeout = max_tokens / velocidad_esperada_TPS * 2 (margen de seguridad).

Reintentos con backoff exponencial

Las APIs de LLM fallan. Rate limits, timeouts transitorios, errores 500 del servidor. Sin reintentos, cada fallo llega al usuario.

import time
import anthropic
from anthropic import RateLimitError, APIStatusError

def call_with_retry(prompt: str, max_retries: int = 3) -> str:
    client = anthropic.Anthropic()
    
    for attempt in range(max_retries):
        try:
            message = client.messages.create(
                model="claude-sonnet-4-5",
                max_tokens=1024,
                messages=[{"role": "user", "content": prompt}]
            )
            return message.content[0].text
            
        except RateLimitError:
            wait = (2 ** attempt) + 1  # 1s, 3s, 5s
            time.sleep(wait)
            
        except APIStatusError as e:
            if e.status_code >= 500:  # Error del servidor, reintentable
                time.sleep(2 ** attempt)
            else:
                raise  # Error del cliente (400), no reintentar
    
    raise Exception("Máximo de reintentos alcanzado")

Streaming para UX en tiempo real

Para aplicaciones de chat, el streaming reduce la latencia percibida radicalmente. El usuario ve respuesta inmediata en lugar de esperar el texto completo.

with client.messages.stream(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": prompt}]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

Para Next.js/React, usa la respuesta como ReadableStream y muestra tokens conforme llegan.

Prompt caching para reducir costos

Si tu system prompt es largo y repetitivo (>1.024 tokens), el prompt caching puede reducir costos hasta un 90% en el contexto repetido.

message = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": system_prompt_largo,
            "cache_control": {"type": "ephemeral"}  # Cache por 5 minutos
        }
    ],
    messages=[{"role": "user", "content": user_message}]
)

El primer token cuesta lo mismo. Las llamadas posteriores dentro de los 5 minutos pagan solo el 10% del precio de los tokens en cache.

Control de costos en tiempo real

Sin monitoreo de tokens, el costo puede dispararse sin que te enteres. Loguea siempre el uso:

response = client.messages.create(...)

# Siempre logguear
print(f"Input tokens: {response.usage.input_tokens}")
print(f"Output tokens: {response.usage.output_tokens}")
print(f"Costo estimado: ${(response.usage.input_tokens * 3 + response.usage.output_tokens * 15) / 1_000_000:.4f}")

Implementa alertas cuando el costo diario supere un umbral.

Validación del output antes de usarlo

No asumas que el modelo devolverá exactamente el formato pedido. Valida:

import json

def parse_json_response(text: str) -> dict:
    # El modelo puede añadir texto antes/después del JSON
    start = text.find('{')
    end = text.rfind('}') + 1
    if start == -1 or end == 0:
        raise ValueError(f"No JSON encontrado en respuesta: {text[:100]}")
    try:
        return json.loads(text[start:end])
    except json.JSONDecodeError as e:
        raise ValueError(f"JSON inválido: {e}")

Observabilidad: lo que debes logguear

Para debuggear en producción necesitas el contexto completo de cada llamada:

import time
import uuid

def traced_call(messages: list, system: str) -> dict:
    trace_id = str(uuid.uuid4())[:8]
    start = time.time()
    
    try:
        response = client.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=1024,
            system=system,
            messages=messages
        )
        
        latency = time.time() - start
        
        # Log estructurado para indexar en tu sistema de logs
        log = {
            "trace_id": trace_id,
            "model": "claude-sonnet-4-5",
            "latency_ms": int(latency * 1000),
            "input_tokens": response.usage.input_tokens,
            "output_tokens": response.usage.output_tokens,
            "stop_reason": response.stop_reason,
        }
        print(json.dumps(log))
        
        return {"ok": True, "text": response.content[0].text}
        
    except Exception as e:
        print(json.dumps({"trace_id": trace_id, "error": str(e)}))
        return {"ok": False, "error": str(e)}

Checklist de producción

• [ ] Timeout configurado (no usar el default del SDK)
• [ ] Reintentos con backoff para errores 429 y 5xx
• [ ] Streaming activado para interfaces de usuario
• [ ] Prompt caching para system prompts largos y repetitivos
• [ ] Logging de tokens por llamada
• [ ] Alertas de costo por día/mes
• [ ] Validación del output antes de procesarlo
• [ ] Circuit breaker para fallos en cascada
• [ ] Variables de entorno para API keys (nunca hardcodeadas)
• [ ] Modelo de fallback si el primario falla