Stabiliteit van proxy-API: time-outs, retry's, rate limit-wachtrijen en streamingonderbrekingen

Classificeer fouten eerst in drie categorieën

De eerste stap bij stabiliteit is niet meer retryen, maar weten wat je wel of niet moet retryen. We verdelen veelvoorkomende fouten in drie categorieën:

Categorie	Typisch gedrag	Oplossing
Tijdelijk herstelbaar	429 rate limit, 503 upstream_busy, verbinding verbroken, time-out	Exponentiële backoff met retry, beperk het totale aantal pogingen
Fout in het verzoek zelf	400 (input plus max_tokens > 100.000, request body too large), 403 content_blocked	Geen retry, corrigeer het verzoek of retourneer de fout aan de gebruiker
Accountstatus	401 ongeldige sleutel, 402 geen tegoed	Geen retry, alarmeer de verantwoordelijke, laad op of vervang de sleutel

Schrijf deze tabel bij voorkeur in code-comments. Een veelvoorkomend probleem: na verbruik van het saldo retourneert de API stabiel 402. Zonder statuscode-check retryt je code vijf keer per verzoek, wat het verkeer met een factor vijf verhoogt en de logs rood kleurt.

Stel time-outs gelaagd in

Eén globale time-out is vaak onvoldoende. Verdeel de time-out in drie lagen:

• Connectie-timeout: Tijd voor TCP/TLS-handshake. 5 tot 10 seconden is meestal voldoende; faal snel als het niet lukt.
• Lees time-out: tijd om op responsdata te wachten. Voor niet-streaming moet dit de volledige generatietijd dekken; voor streaming is een stilte van 10 tot 30 seconden tussen chunks realistischer.
• Algemene business-timeout: De maximale wachttijd die jouw applicatie aankan. Zorg voor afdwinging via asyncio.wait_for of de gateway, inclusief alle retry's.

Langere outputs duren langer bij niet-streaming. Als je max_tokens op 32.000 zet met een totale timeout van 30 seconden, krijg je veel time-outs. Gebruik streaming voor directe feedback en duidelijke time-outs. Gebruik de timeout parameter.

Exponentiële backoff voor 429 en 503

Drie regels voor backoff: verdubbel de vertraging bij elke fout, stel een maximum in en voeg jitter toe. Zonder jitter proberen alle gefaalde taken het tegelijk opnieuw, wat de herstelde service opnieuw kan overbelasten. De onderstaande functie schakelt SDK-retry uit en beheert dit zelf, alleen voor 429, 503 en connectiefouten:

复制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 betekent dat de server tijdelijk bezet is. Een retry na enkele seconden is gepast. De basisvertraging van 1 seconde, een maximum van 20 seconden en maximaal 5 pogingen dekken dit scenario. Bij frequente 429-fouten ligt de oorzaak vaak aan de zijkant van de afzender (geen rate limiting), zie de volgende sectie.

Concurrentiewachtrij bij 300 verzoeken per minuut

300 verzoeken per minuut komt neer op gemiddeld 5 per seconde. Batchverwerkingen struikelen hier vaak over: asyncio.gather stuurt duizenden verzoeken tegelijk, verbruikt de limiet direct en krijgt voor de rest 429.

Een veilige aanpak vereist twee lagen. Een semafork beperkt het aantal gelijktijdige verzoeken om geheugen en verbindingen te beschermen. Een rate limiter regelt de uitstroomsnelheid. Alleen een semafork is niet genoeg als verzoeken snel zijn; alleen een limiter is niet genoeg als verzoeken lang duren. We stellen 270 verzoeken/min in voor 10% marge voor instance-overhead en 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())

Als meerdere processen of machines dezelfde sleutel delen, werkt een lokale rate limiter niet meer. De limiet geldt per sleutel. Gebruik een gedeelde token bucket (bijv. via Redis) of stuur alle batchverzoeken via een centrale wachtrij-service.

Omgaan met onderbroken streaming

Streamingverbindingen zijn kwetsbaarder: ze duren langer en passeren meer netwerkapparatuur. Een idle-timeout kan de verbinding verbreken. Regel: ontvangen data is waardevol. Gooi niets weg bij een onderbreking.

De onderstaande implementatie accumuleert ontvangen fragmenten. Bij een connectie- of time-outfout vraag je de model voort met de bestaande tekst als assistant-bericht. Dit is een 'best-effort' oplossing, geschikt voor lange teksten en chats, maar minder voor strikte structuren zoals JSON. Bij JSON is een volledige heraanvraag veiliger. Let op: de laatste streamingfragment bevat usage-data:

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

Bij het doorzetten (continuation) tellen verzonden tokens mee als input, wat extra kosten veroorzaakt. Het beperken van max_resume is daarom noodzakelijk.

Degradatie, circuit breaker en pre-flight checks

Retry's lossen tijdelijke fouten op. Bij aanhoudende fouten leidt meer retryen tot ophoping. Voeg een circuit breaker toe: bij een drempel (bijv. 10 fouten per minuut) pauzeer je het uitsturen van verzoeken en val je terug op een degradatielogica (cache, 'probeer het later'-bericht, of terugplaatsen in wachtrij). Na de cooldown-tijd test je met een paar verzoeken voordat je de volledige belasting hervat.

Plan degradatie van tevoren. Voor chat-interfaces is een vriendelijk bericht gepast; voor offline batchjobs schrijf je fouten naar een tabel voor latere verwerking in plaats van oneindig wachten in de hoofdflow. Zorg altijd dat fouten niet stil verdwijnen; log minimaal één regel met de request id.

Checklist voor productie: geen SDK-retries behalve één; geef op bij 400/401/402/403; totale timeout inclusief retries; rate limiting bij batch; streaming onderbrekingen; usage opgeslagen; saldo-alarm.

Monitoring en reconciliatie via usage

Elke respons bevat usage-data (bij streaming in het laatste fragment). Log dit samen met de functienaam voor goedkope monitoring. Onze prijzen zijn $0,25 per miljoen input-tokens en $1,00 per miljoen output-tokens. Je kunt de kosten per verzoek lokaal schatten:

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

Na opslag: controleer dagelijks mediane input tokens, output token ratio (4x prijs) en 429/503 ratio. Voeg een saldo-alarm toe voor 402. Zie frameworkconfiguratie en FAQ.