API Timeout: como diagnosticar quedas e configurar tentativas seguras em modelos LLM

Onde ocorrem as quedas: fases do ciclo de requisição HTTP

Uma requisição à API de modelos LLM envolve quatro fases técnicas distintas:

[Cliente] --- (1. Connect Timeout) ---> [API Gateway]
[Cliente] --- (2. Write Timeout)   ---> [Envio de Prompt]
[Modelo]  --- (3. Read / TTFT)     ---> [Raciocínio e Geração]
[Cliente] <--- (4. Pool Timeout)    --- [Manutenção do Socket]

1. Connect Timeout: Tempo para handshake TCP e TLS (recomendado 5–10 segundos).
2. Write Timeout: Tempo para enviar o payload da solicitação (importante para contextos com mais de 100k tokens).
3. Read Timeout: Tempo de espera pela resposta ou entre chunks sucessivos no fluxo SSE.
4. Pool Timeout: Tempo de espera para obter uma conexão disponível no pool sob alta concorrência.

---

Matriz de diagnóstico de timeouts

---

Configuração de timeouts granulares em Python (HTTPX)

O valor padrão timeout=10.0 frequentemente causa falhas em modelos de raciocínio. Veja a configuração recomendada abaixo:

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,    # Falha rápida se a rede estiver inacessível
    read=120.0,     # Janela ampla para processamento de raciocínio
    write=10.0,     # Tempo para envio do payload
    pool=10.0       # Aquisição de socket no 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 mais

---

Estabilização de streams SSE e tentativas seguras

Para evitar custos duplicados durante desconexões, o mecanismo de retentativa deve seguir três regras:

1. Não repetir se o streaming já iniciou: Caso tokens parciais já tenham sido entregues, reenviar a requisição inteira gerará cobrança duplicada.
2. Backoff exponencial com Full Jitter: As novas tentativas devem utilizar intervalos graduais e aleatórios.
3. Chaves de idempotência: Utilize identificadores únicos para evitar duplicação em tarefas em lote.

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)

Validação de estabilidade

• Resposta com status 200 OK.
• Recepção completa de todos os blocos SSE sem erros de codificação.
• Liberação correta dos sockets após o término da transmissão.

Acesse mais guias de integração na referência de API do BetterToken.