API Timeout: cómo diagnosticar desconexiones y configurar reintentos seguros en LLM

Dónde ocurren las caídas: fases del ciclo de vida HTTP

Una solicitud a la API de un modelo de IA se divide en cuatro fases técnicas independientes:

[Cliente] --- (1. Connect Timeout) ---> [API Gateway]
[Cliente] --- (2. Write Timeout)   ---> [Envío de Prompt]
[Modelo]  --- (3. Read / TTFT)     ---> [Razonamiento y Respuesta]
[Cliente] <--- (4. Pool Timeout)    --- [Mantenimiento de Conexión]

1. Connect Timeout: Tiempo asignado para establecer la conexión TCP y el protocolo TLS (5–10 segundos recomendados).
2. Write Timeout: Tiempo para enviar el cuerpo de la solicitud (crucial con contextos de más de 100k tokens).
3. Read Timeout: Tiempo de espera de respuesta o de chunks sucesivos en el flujo SSE.
4. Pool Timeout: Tiempo de espera para obtener un socket libre del grupo de conexiones bajo alta concurrencia.

---

Matriz de diagnóstico de tiempos de espera

---

Configuración de timeouts granulares en Python (HTTPX)

El valor por defecto timeout=10.0 suele provocar fallos con modelos de razonamiento. A continuación se muestra una configuración recomendada:

import os
import httpx
from openai import OpenAI

API_KEY = os.environ.get("BETTERTOKEN_API_KEY", "your_api_key_here")

custom_timeout = httpx.Timeout(
    connect=5.0,    # Fallo rápido ante caídas de red
    read=120.0,     # Ventana amplia para razonamiento profundo
    write=10.0,     # Envío del cuerpo de la petición
    pool=10.0       # Obtención de socket en el pool
)

http_client = httpx.Client(
    timeout=custom_timeout,
    limits=httpx.Limits(max_keepalive_connections=50, max_connections=100)
)

client = OpenAI(
    base_url="https://www.bettertoken.ai/v1",
    api_key=API_KEY,
    http_client=http_client
)

response = client.chat.completions.create(
    model="claude-3-7-sonnet-20250219",
    messages=[
        {"role": "system", "content": "You are a senior systems architect."},
        {"role": "user", "content": "Design a high-throughput distributed message broker."}
    ],
    stream=True
)

for chunk in response:
    delta = chunk.choices[0].delta.content or ""
    print(delta, end="", flush=True)
Mostrar más

---

Estabilización de flujos SSE y reintentos seguros

Para evitar gastos duplicados durante desconexiones, los reintentos deben cumplir tres principios:

1. No reintentar si el streaming ya ha comenzado: Si ya se recibieron tokens parciales, volver a enviar la consulta causará un doble cobro.
2. Retroceso exponencial con Full Jitter: Los reintentos deben espaciarse con pausas aleatorias crecientes.
3. Uso de claves de idempotencia: Emplee identificadores únicos para evitar ejecuciones repetidas en procesamiento por lotes.

import time
import random
import httpx

def execute_with_safe_retry(client, model, messages, max_retries=3):
    base_delay = 1.0
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model=model,
                messages=messages,
                stream=True
            )
            return response
        except (httpx.ConnectTimeout, httpx.ReadTimeout, httpx.NetworkError) as err:
            if attempt == max_retries - 1:
                raise err
            sleep_time = random.uniform(0, base_delay * (2 ** attempt))
            time.sleep(sleep_time)

Lista de verificación

• Código de respuesta 200 OK.
• Recepción continua de todos los fragmentos sin errores ChunkedEncodingError.
• Liberación adecuada de sockets tras la sesión.

Consulte todos los métodos de integración en la referencia de API de BetterToken.