Prácticas de estabilidad de la API intermedia: tiempos de espera, reintentos, colas de límite y desconexiones de streaming

Clasifica los errores en tres categorías

El primer paso ante inestabilidad no es añadir reintentos, sino saber qué debe ser reintentado. Clasificamos los fallos comunes en tres categorías según cómo se manejan:

Categoría	Síntoma típico	Solución
Recuperable al instante	429 límite, 503 upstream_busy, conexión reiniciada, tiempo de espera	Reintenta con retroexponencial y limita el número total de intentos
Error en la petición	400 (input + max_tokens > 100.000, cuerpo grande), 403 content_blocked	No reintentes; corrige la petición o avisa al usuario
Problema de cuenta	401 clave inválida, 402 no_credit	No reintentes; alerta al responsable, recarga o cambia la clave

Esta tabla conviene ponerla en los comentarios del código. Un error común es que, tras agotarse el saldo, la API devuelva 402 de forma estable. Si tu lógica de reintento no distingue códigos de estado, reintentará cada petición cinco veces, multiplicando el tráfico por cinco y saturando los logs.

Configura tiempos de espera por capas

No basta con un solo timeout total. Se recomienda dividir el timeout en tres capas:

• Tiempo de conexión: tiempo para establecer TCP y TLS. 5-10 segundos bastan; falla rápido si no conecta.
• Tiempo de lectura: espera de datos. En no-streaming debe cubrir la generación completa; en streaming es el silencio máximo entre bloques (10-30 s).
• Tiempo de espera total de negocio: el máximo que tu negocio tolera. Úsalo con asyncio.wait_for o en la capa de gateway, incluyendo reintentos.

A mayor longitud de salida, mayor tiempo de espera en peticiones no streaming. Si ajustas max_tokens al límite de 32,000 y configuras un tiempo de espera total de solo 30 segundos, los tiempos de espera serán frecuentes en textos largos. Lo más seguro es usar streaming para salidas largas; esto da feedback inmediato y aclara el significado del tiempo de espera. El parámetro timeout del SDK acepta un número o una configuración detallada; consulta la documentación según tu versión.

Retroexponencial para 429 y 503

La espera con retroceso tiene tres aspectos clave: el retraso se duplica con cada fallo, se establece un tope y se añade jitter. Sin jitter, las tareas falladas simultáneamente reintentarán al mismo tiempo, colapsando el servicio. La siguiente función desactiva los reintentos del SDK para controlarlo todo nosotros, y solo reintenta 429, 503 y errores de conexión:

复制import asyncio
import os
import random

from openai import AsyncOpenAI, APIConnectionError, APIStatusError, APITimeoutError

client = AsyncOpenAI(
    base_url="https://api.llmzhongzhuan.com/v1",
    api_key=os.environ["API_KEY"],
    timeout=60.0,
    max_retries=0,  # 重试由下面的函数统一控制
)

RETRY_STATUS = {429, 503}


async def chat_with_retry(messages, max_attempts=5, base_delay=1.0, cap=20.0):
    for attempt in range(1, max_attempts + 1):
        try:
            resp = await client.chat.completions.create(
                model="uncensored", messages=messages, max_tokens=800
            )
            return resp
        except APIStatusError as e:
            if e.status_code not in RETRY_STATUS or attempt == max_attempts:
                raise  # 400/401/402/403 重试没有意义
            reason = f"HTTP {e.status_code}"
        except (APIConnectionError, APITimeoutError) as e:
            if attempt == max_attempts:
                raise
            reason = type(e).__name__
        delay = min(cap, base_delay * 2 ** (attempt - 1))
        delay = random.uniform(0, delay)  # 全抖动，避免所有任务同时醒来
        print(f"第 {attempt} 次失败（{reason}），{delay:.1f}s 后重试")
        await asyncio.sleep(delay)

El 503 indica que el servidor está ocupado. Se recomienda reintentar tras unos segundos. La configuración de 1 segundo de base, 20 segundos de tope y 5 intentos cubre este caso. Si ves 429 frecuentemente, el problema no es el reintento, sino la falta de control de tasa en el emisor. Lee la siguiente sección.

Cola de concurrencia bajo límite de 300 por minuto

Cada clave permite 300 peticiones por minuto, lo que equivale a 5 peticiones por segundo. Las tareas por lotes suelen fallar aquí: usar asyncio.gather para enviar miles de peticiones a la vez agota el cupo con los primeros cientos y devuelve 429 al resto.

La forma fiable es usar dos capas. Un semáforo limita las peticiones simultáneas para no saturar memoria y conexiones. Un limitador de tasa controla la velocidad de emisión. Ambos son necesarios: solo semáforo y peticiones rápidas superas las 300/min; solo limitador y peticiones lentas acumulas peticiones en curso. El ejemplo fija la tasa en 270/min para dejar un 10% de margen para instancias múltiples, reintentos y errores de reloj:

复制import asyncio
import os
import time

from openai import AsyncOpenAI

client = AsyncOpenAI(
    base_url="https://api.llmzhongzhuan.com/v1",
    api_key=os.environ["API_KEY"],
    timeout=60.0,
    max_retries=0,
)

MAX_IN_FLIGHT = 8          # 同时在途的请求数
PER_MINUTE = 270           # 留 10% 余量，低于每分钟 300 次的上限


class Pacer:
    """把请求的发出时刻均匀铺开：每 60/PER_MINUTE 秒放行一个。"""

    def __init__(self, per_minute):
        self.interval = 60.0 / per_minute
        self.next_at = 0.0
        self.lock = asyncio.Lock()

    async def wait(self):
        async with self.lock:
            now = time.monotonic()
            delay = self.next_at - now
            if delay > 0:
                await asyncio.sleep(delay)
            self.next_at = max(now, self.next_at) + self.interval


sem = asyncio.Semaphore(MAX_IN_FLIGHT)
pacer = Pacer(PER_MINUTE)


async def summarize(idx, text):
    async with sem:            # 控制并发
        await pacer.wait()     # 控制速率
        resp = await client.chat.completions.create(
            model="uncensored",
            messages=[{"role": "user", "content": f"用两句话概括：{text}"}],
            max_tokens=120,
        )
        return idx, resp.choices[0].message.content


async def main():
    texts = [f"第 {i} 条工单的正文……" for i in range(1000)]
    tasks = [summarize(i, t) for i, t in enumerate(texts)]
    done = 0
    for coro in asyncio.as_completed(tasks):
        idx, out = await coro
        done += 1
        if done % 100 == 0:
            print(f"已完成 {done} 条")


asyncio.run(main())

Si varios procesos o máquinas comparten una clave, el regulador en memoria no basta. Necesitas un token bucket compartido, por ejemplo con Redis, o una cola de salida unificada para todos los lotes.

Qué hacer tras una interrupción de streaming

El streaming es más frágil: dura más, pasa por más dispositivos y cualquier tiempo de espera de inactividad puede cortarlo. La regla es: el contenido recibido es un activo, no lo descartes.

El siguiente código acumula fragmentos y, tras capturar errores de conexión o tiempo de espera, reenvía el texto como mensaje assistant para continuar. Es una solución best-effort: la continuidad no es perfecta y es ideal para texto largo, no para JSON estricto. Si se corta JSON, es mejor reiniciar. Nota: el último fragmento incluye usage:

复制import asyncio
import os

from openai import AsyncOpenAI, APIConnectionError, APITimeoutError

client = AsyncOpenAI(
    base_url="https://api.llmzhongzhuan.com/v1",
    api_key=os.environ["API_KEY"],
    timeout=30.0,   # 流式场景下，这是相邻数据块之间允许的最长静默
    max_retries=0,
)


async def stream_once(messages):
    parts, usage = [], None
    stream = await client.chat.completions.create(
        model="uncensored", messages=messages, max_tokens=600, stream=True
    )
    try:
        async for chunk in stream:
            if chunk.usage:                      # 最后一个分片携带 usage
                usage = chunk.usage
            if chunk.choices and chunk.choices[0].delta.content:
                parts.append(chunk.choices[0].delta.content)
    except (APIConnectionError, APITimeoutError):
        return "".join(parts), None, False       # 中断：返回已收到的部分
    return "".join(parts), usage, True


async def stream_with_resume(question, max_resume=2):
    messages = [{"role": "user", "content": question}]
    text = ""
    for _ in range(max_resume + 1):
        part, usage, finished = await stream_once(messages)
        text += part
        if finished:
            return text, usage
        # 把已生成部分作为 assistant 消息，请模型接着写
        messages = [
            {"role": "user", "content": question},
            {"role": "assistant", "content": text},
            {"role": "user", "content": "请从上一条回复的结尾处继续，不要重复已写的内容。"},
        ]
    return text, None


if __name__ == "__main__":
    out, usage = asyncio.run(stream_with_resume("用三段话说明为什么日志要带 request id。"))
    print(out)
    print("usage:", usage)

Al reanudar, recuerda que el texto enviado cuenta como tokens de entrada, lo que añade costo. Limita el número de reanudos con max_resume.

Degradación, circuit breaker y checklist pre-lanzamiento

Los reintentos cubren fallos transitorios. Si persisten, acumulas peticiones. Añade un circuit breaker: si hay un umbral de fallos (ej. 10 en 1 minuto), pausa las peticiones y usa lógica degradada (cache, mensaje de espera). Tras el enfriamiento, prueba con peticiones de sondeo antes de restaurar el tráfico total.

Planifica la degradación: chat muestra mensaje; batch guarda fallos para reintentar. No silencies errores; deja un log con request id.

Antes de producción, revisa esta lista: ¿desactivaste los reintentos implícitos del SDK y usas solo los tuyos? ¿Abandonas directamente ante 400, 401, 402, 403? ¿El timeout total incluye los reintentos? ¿El procesamiento por lotes tiene control de tasa? ¿Manejas interrupciones en streaming? ¿Guardas el usage? ¿Tienes alertas de saldo? Aprueba cada punto antes de hacer pruebas de carga.

Monitoreo y conciliación con usage

Cada respuesta incluye usage (en el último fragmento del streaming). Persistirlo junto al nombre de la función es la forma más barata de monitorizar. Nuestro precio es 0,25 USD por millón de tokens de entrada y 1,00 USD de salida. Puedes estimar el costo localmente:

复制import csv
import time

LOG = "usage_log.csv"


def record_usage(feature, resp):
    u = resp.usage
    cost = u.prompt_tokens * 0.25 / 1_000_000 + u.completion_tokens * 1.00 / 1_000_000
    with open(LOG, "a", newline="", encoding="utf-8") as f:
        csv.writer(f).writerow(
            [int(time.time()), feature, u.prompt_tokens, u.completion_tokens, f"{cost:.6f}"]
        )

Tras guardar los datos, revisa diariamente tres métricas: la mediana de tokens de entrada para detectar prompts que crecen; la proporción de tokens de salida (el precio de salida es 4x el de entrada, así que suele ser el mayor gasto); y la proporción de 429 vs 503. Si sube, ajusta la tasa de envío o la estrategia de reintento. Añade una alerta de saldo para avisar antes del 402. Para más detalles, consulta configuración del framework y preguntas frecuentes.