Stabilità API proxy: timeout, retry, code rate limit e streaming

Classifica gli errori in tre categorie

Il primo passo per la stabilità non è aggiungere retry, ma sapere cosa ritentare. Classifichiamo i fallimenti comuni in tre categorie in base alla gestione:

Categoria	Sintomo tipico	Gestione
Recuperabile istantaneamente	429 rate limit, 503 upstream_busy, connessione resettata, timeout	Retry con backoff esponenziale, limitando il numero totale di tentativi
Errore nella richiesta	400 (input + max_tokens > 100.000, payload troppo grande), 403 content_blocked	Nessun retry; correggi la richiesta o restituisci l'errore all'utente
Problema di stato account	401 chiave non valida, 402 no_credit	Nessun retry; invia un alert al responsabile, ricarica o cambia chiave

Questa tabella dovrebbe essere nei commenti del codice. Un incidente comune: dopo l'esaurimento del saldo, l'API restituisce stabilmente 402. Se il retry non distingue gli status code, ogni richiesta viene ritentata 5 volte, amplificando il traffico di 5 volte e saturando i log.

Imposta i timeout a livelli

Un unico timeout totale non basta. Suddividilo in tre livelli:

• Timeout di connessione: tempo per stabilire TCP e TLS. Di solito bastano 5-10s; fallisci rapidamente se non riesci a connetterti.
• Timeout di lettura: tempo di attesa dei dati. Per le richieste non streaming, deve coprire la generazione completa; per lo streaming, diventa il silenzio massimo tra due chunk (10-30s è ragionevole).
• Timeout business totale: il massimo tempo di attesa tollerato dal tuo business. Gestiscilo con asyncio.wait_for o a livello di gateway, includendo tutti i retry.

Più lungo è l'output, più a lungo impiega una richiesta non streaming. Se imposti max_tokens a 32.000 con un timeout totale di 30s, avrai molti timeout su testi lunghi. Usa lo streaming per output lunghi: dà feedback immediato e chiarisce il significato del timeout. Il parametro timeout dell'SDK accetta un numero o una configurazione dettagliata.

Retry con backoff esponenziale per 429 e 503

Tre regole per il backoff: raddoppia il ritardo per ogni fallimento, imposta un limite massimo, aggiungi jitter casuale. Senza jitter, i task falliti simultaneamente ritenteranno tutti insieme, sovraccaricando il servizio appena recuperato. La funzione disabilita i retry SDK e gestisce tutto, ritentando solo per 429, 503 e errori di connessione:

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

503 significa server occupato: ritenta dopo pochi secondi. I parametri (1s base, 20s max, 5 tentativi) coprono questo scenario. Per i 429 frequenti, il problema non è il retry ma la mancanza di rate limiting sul mittente.

Code di parallel requests sotto il rate limit di 300 richieste/min

300 richieste/min per chiave equivalgono a 5 al secondo. I batch sono critici: usare asyncio.gather per inviare migliaia di richieste consuma il budget in pochi secondi, generando 429 per il resto.

Soluzione affidabile a due livelli: un semaforo limita i parallel requests in corso (proteggendo connessioni e memoria), un rate limiter controlla la frequenza. Entrambi sono necessari: solo il semaforo può superare i 300/min se le richieste sono veloci; solo il rate limiter accumula richieste in corso se sono lente. Impostiamo il rate a 270/min (10% di margine per istanze multiple, retry e clock skew):

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

Se più processi o macchine condividono la chiave, il rate limiter in memoria non basta, poiché il conteggio è aggregato per chiave. Usa un token bucket condiviso (es. Redis) o instrada tutti i batch attraverso un servizio di coda unico che gestisce l'uscita.

Gestione delle interruzioni nello streaming

Le connessioni streaming sono più fragili: durano a lungo, attraversano molti dispositivi e un idle timeout può chiuderle. Principio: i dati già ricevuti sono un asset. Non scartarli tutto.

L'implementazione accumula i chunk ricevuti. Catturando errori di connessione/timeout, invia il testo esistente come messaggio assistant per la continuazione. È un best-effort: la continuità non è perfetta, adatta per testi lunghi ma non per strutture rigide come JSON. Per JSON, è meglio rigenerare tutto. Nota: l'ultimo chunk streaming include sempre l'usage, che viene estratto nel codice:

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

Quando continui, ricorda che il testo già inviato conta come token di input, quindi ogni continuazione ha un costo aggiuntivo: è necessario limitare il numero di max_resume.

Degradazione, circuit breaker e checklist pre-lancio

I retry risolvono guasti transienti. Se il guasto dura minuti, i retry accumulano richieste. Aggiungi un circuit breaker: dopo una soglia di fallimenti (es. 10/min), sospendi le richieste per un periodo, eseguendo logica di degradazione (cache, messaggio utente, coda differita). Al termine del cooldown, invia solo richieste di sondaggio; se passano, ripristina il traffico completo.

Pensa anche al degrado. Per le chat utente puoi restituire un messaggio di stato occupato; per i job batch offline, scrivi le voci fallite in un table degli errori e riprocessale in batch alla fine, senza bloccare il flusso principale. In ogni caso, assicurati che i fallimenti non vengano inghiottiti silenziosamente: lascia almeno un log con il request id e lo stato busy.

Checklist pre-lancio: hai disabilitato i retry SDK e gestito i retry centralmente? Ignori 400/401/402/403? Il timeout totale include tutti i retry? I batch hanno rate limiting? Lo streaming gestisce le interruzioni? L'usage è salvato? Hai alert sul saldo? Passa la checklist prima del load test.

Monitoraggio e riconciliazione tramite usage

Ogni risposta include usage (ultimo chunk nello streaming). Salvare usage e nome della funzione è il metodo di monitoraggio più economico. Prezzi del sito: $0.25/1M input token, $1.00/1M output token. Puoi stimare il costo locale così:

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

Dopo il salvataggio su disco, controlla ogni giorno tre metriche: la mediana dei token di input per funzione (per rilevare un'espansione nascosta dei prompt), la percentuale di token di output (il costo per output è quattro volte quello per input, quindi è la voce principale della fattura) e il rapporto tra 429 e 503 (un aumento indica che il tasso di invio o la strategia di retry va aggiustato dopo landing). Aggiungi un alert sul saldo per avvisare il responsabile di ricaricare prima che appaia il 402. Per i dettagli di integrazione, consulta framework configuration; per i dubbi, vai su FAQ.