Stabilité de l'API intermédiaire : timeouts, retries, files de limites de débit et coupures de streaming

Classez d'abord les erreurs

La première étape n’est pas d’augmenter les tentatives, mais de savoir quoi réessayer. Les échecs courants se divisent en trois catégories selon le traitement :

Catégorie	Symptômes typiques	Traitement
Récupération instantanée	429 (limite), 503 (busy), reset de connexion, timeout	Relance avec backoff exponentiel, limité en nombre
Erreur de requête	400 (input > 100 000 ou corps trop lourd), 403 (content_blocked)	Pas de retry. Corrigez la requête ou retournez l'erreur à l'utilisateur.
Problème de compte	401 (clé invalide), 402 (no_credit)	Pas de retry. Alerte, recharge ou changement de clé.

Ce tableau est à intégrer dans les commentaires du code. Un incident courant survient lorsque le solde est épuisé : l'API commence à retourner systématiquement une erreur 402. Si votre logique de retry ne distingue pas les codes d'état, chaque requête sera réessayée cinq fois, ce qui multiplie artificiellement le trafic par cinq et remplit les journaux de logs d'erreurs.

Timeouts à plusieurs niveaux

Un seul timeout global est insuffisant. Nous recommandons trois niveaux :

• Timeout de connexion : temps pour établir TCP/TLS. 5 à 10 s suffisent généralement. Échouez vite si cela bloque.
• Timeout de lecture : temps d'attente des données. En non-streaming, il doit couvrir la génération complète. En streaming, c'est le délai maximal entre deux chunks (10 à 30 s est raisonnable).
• Temps d'attente global : définissez la limite via asyncio.wait_for ou le proxy, incluant les retries.

Plus le texte est long, plus le non-streaming est lent. Si vous fixez max_tokens à 32 000 avec un timeout de 30 s, vous aurez des timeouts fréquents. Privilégiez le streaming pour les longs textes : feedback instantané et timeout clair. Le paramètre timeout du SDK accepte un nombre ou une config détaillée.

Backoff exponentiel pour 429 et 503

Les principes du backoff sont au nombre de trois : la latence double à chaque échec, une limite supérieure est définie et un jitter aléatoire est ajouté. Sans jitter, un lot de tâches qui échouent simultanément tenteront toutes le retry au même moment, ce qui peut à nouveau surcharger le service après sa récupération. La fonction ci-dessous désactive le mécanisme de retry intégré au SDK pour le gérer nous-mêmes de manière centralisée, et ne tente le retry que pour les erreurs 429, 503 et les erreurs de connexion :

复制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)

Le 503 indique une surcharge temporaire. Un backoff de 1 s à 20 s, max 5 fois, couvre ce cas. Pour le 429 récurrent, le problème est souvent le débit d'envoi, pas le retry (voir section suivante).

File d'attente sous limite de 300 req/min

300 req/min par clé (5 req/s). Risque avec asyncio.gather : épuisez le quota instantanément et recevez des 429.

Solution fiable à deux niveaux : un sémaphore limite le nombre de requêtes en cours (protège mémoire/connexions) et un token bucket limite le débit (répartit les requêtes). Les deux sont nécessaires : sans bucket, un sémaphore sur des requêtes rapides dépasse 300 req/min ; sans sémaphore, des requêtes lentes s'accumulent. Nous fixons le débit à 270 req/min pour laisser 10 % de marge pour les instances multiples et le jitter :

复制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 plusieurs processus ou machines partagent la clé, le batteur mémoire ne suffit pas, car la limite est agrégée par clé. Utilisez un token bucket partagé (Redis) ou un service de file d’attente unique pour les requêtes par lots.

Gestion des coupures de streaming

Le streaming est fragile : connexion longue, nombreux équipements réseau. Un timeout d'inactivité peut couper la connexion. Principe : le contenu reçu est un actif. Ne le perdez pas.

L'implémentation ci-dessous accumule les chunks reçus. En cas d'exception de connexion ou de timeout, elle renvoie le texte accumulé comme message assistant pour demander au modèle de continuer. Il s'agit d'une tentative de récupération à meilleur effort : la suite n'est pas toujours fluide, ce qui convient aux textes longs et aux conversations, mais pas aux sorties nécessitant une structure stricte comme le JSON. En cas d'interruption d'une sortie structurée, il est plus sûr de tout recommencer. Notez qu'un chunk usage est automatiquement joint à la fin du streaming ; le code le récupère également :

复制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)

Le texte déjà envoyé compte comme tokens d'entrée, ajoutant un coût marginal. Limitez le nombre de reprises avec max_resume.

Dégradation, circuit breaker et checklist de mise en production

Les tentatives gèrent les pannes instantanées. Pour les pannes durables, ajoutez un disjoncteur : si 10 échecs surviennent en 1 minute, suspendez les requêtes et utilisez une logique de dégradation (cache, file). Après le délai de refroidissement, envoyez des requêtes de test avant de reprendre.

Prévoyez la dégradation. Pour le chat, affichez un message d'occupation. Pour les lots, écrivez les échecs dans une table et relancez plus tard. Assurez-vous qu'aucune erreur ne soit silencieuse : loggez au moins l'ID de requête.

Checklist avant la mise en production : retry SDK désactivé ? 400/401/402/403 non retryés ? Timeout global incluant les retries ? Contrôle de débit pour les lots ? Gestion des coupures de streaming ? Usage enregistré ? Alerte solde ? Validez ces points avant de stresser le système.

Monitoring et réconciliation via l'usage

Chaque réponse contient l'usage (dernier chunk en streaming). Enregistrer l'usage avec le nom de la fonction est le moyen de monitoring le moins cher. Prix : 0,25 $/M tokens d'entrée, 1,00 $/M tokens de sortie. Estimez le coût ainsi :

复制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}"]
        )

Une fois déployé, il est recommandé de surveiller quotidiennement trois indicateurs : la médiane des tokens d'entrée par fonction, pour détecter un gonflement progressif des prompts ; la proportion des tokens de sortie, car le prix par token de sortie est quatre fois supérieur à celui des tokens d'entrée, ce qui en fait généralement le poste principal de la facture ; et le ratio des erreurs 429 et 503. Une augmentation de ce ratio indique qu'il faut ajuster le débit d'envoi ou la stratégie de retry. Ajoutez une alerte de solde pour notifier le responsable de recharger le compte avant l'apparition d'une erreur 402. Pour plus de détails sur l'intégration, consultez la configuration du framework. Les questions fréquentes sont résumées dans FAQ.