Praxis zur Stabilität der Proxy-API: Timeouts, Wiederholungen, Ratenlimits und Streaming-Unterbrechungen

Fehler zuerst in drei Kategorien einteilen

Der erste Schritt bei Stabilitätsproblemen ist nicht mehr zu versuchen, sondern zu wissen, was du wiederholen sollst. Wir unterteilen häufige Fehler in drei Kategorien:

Kategorie	Typisches Verhalten	Behandlung
Kurzfristig wiederherstellbar	429 Ratenlimit, 503 upstream_busy, Verbindungsreset, Timeout	Exponentiellen Backoff mit Wiederholung, Anzahl begrenzen
Die Anfrage ist fehlerhaft	400 (Eingabe plus max_tokens über 100.000, zu großer Request-Body), 403 content_blocked	Nicht wiederholen, Anfrage korrigieren oder Fehler anzeigen
Kontostatus	401 ungültiger Schlüssel, 402 no_credit	Nicht wiederholen, Alarm auslösen, Guthaben aufladen oder Schlüssel wechseln

Diese Tabelle gehört am besten in die Code-Kommentare. Ein häufiger Fehler: Nach dem Ende des Guthabens gibt die API stabil 402 zurück. Wenn die Retry-Logik den Statuscode nicht unterscheidet, wird jede Anfrage fünfmal wiederholt. Das verdünnt den Traffic nicht, sondern verstärkt ihn um das Fünffache, und die Logs füllen sich rot.

Timeouts schichtenweise setzen

Ein einzelnes Timeout reicht nicht. Unterteile in drei Schichten:

• Connect-Timeout: Zeit für TCP/TLS. 5–10 Sekunden reichen meist. Bei Misserfolg schnell fehlschlagen.
• Lese-Timeout: Die Zeit, die auf Antwortdaten gewartet wird. Bei nicht-streamenden Anfragen muss die Wartezeit die Generierung des gesamten Textes abdecken; bei Stream-Anfragen ist die längste Stille zwischen zwei Datenblöcken entscheidend. Hier sind 10 bis 30 Sekunden sinnvoller.
• Gesamt-Timeout: Deine maximale Wartezeit inklusive aller Wiederholungen. Stelle dies mit asyncio.wait_for oder im Gateway sicher.

Je länger die Ausgabe, desto länger dauert die nicht-streamende Anfrage. Wenn du max_tokens auf das Maximum von 32.000 stellst und nur ein Gesamt-Timeout von 30 Sekunden setzt, wirst du bei langen Texten häufig Timeouts erleben. Eine sicherere Lösung: Lange Ausgaben immer streamen. Das gibt dem Nutzer sofortiges Feedback und macht die Bedeutung von Timeouts klar. Der Parameter timeout im SDK kann eine Zahl oder eine detaillierte Konfiguration enthalten. Siehe die Dokumentation deiner Version.

Exponentieller Backoff bei 429 und 503

Die wichtigsten Punkte für Backoff: Die Verzögerung verdoppelt sich mit jeder Fehlschlagszahl, es gibt eine Obergrenze und zufälliges Jitter wird hinzugefügt. Ohne Jitter versuchen alle gleichzeitig fehlgeschlagenen Aufgaben zur gleichen Zeit erneut, was den gerade wiederhergestellten Dienst sofort wieder überlastet. Die folgende Funktion deaktiviert die Standard-Wiederholungen des SDK, sodass wir die Steuerung selbst übernehmen. Sie wiederholt nur bei 429, 503 und Verbindungsfehlern:

复制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 bedeutet vorübergehende Überlastung. Warte einige Sekunden. Die Basisverzögerung von 1 Sekunde, Obergrenze 20 Sekunden und max. 5 Versuche decken dies ab. Bei häufigen 429 liegt das Problem am Sender, nicht an der Wiederholung.

Parallele Anfragen bei Ratenlimit von 300/min

300 Anfragen pro Minute entsprechen 5 pro Sekunde. Batch-Jobs scheitern oft hier: asyncio.gather sendet tausende Anfragen gleichzeitig. Die ersten verbrauchen das Limit, alle anderen erhalten 429.

Eine zuverlässige Lösung nutzt zwei Kontrollstufen. Ein Semaphore begrenzt die Anzahl der „gleichzeitig laufenden“ Anfragen, um lokale Verbindungen und Speicher nicht zu überlasten. Ein Taktgeber begrenzt die „Ausgaberate“, um die Anfragen gleichmäßig zu verteilen. Beide sind notwendig: Nur mit einem Semaphore kann es bei schnellen Anfragen trotzdem zu mehr als 300 Anfragen pro Minute kommen; nur mit einem Taktgeber häufen sich die laufenden Anfragen bei langsamen Anfragen immer mehr an. Im Beispiel wird die Rate auf 270 Anfragen pro Minute festgelegt, um 10 % Puffer für mehrere Instanzen, Wiederholungen und Taktungen zu lassen:

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

Bei mehreren Maschinen mit einem Schlüssel reicht der lokale Taktgeber nicht. Da das Limit pro Schlüssel aggregiert wird, nutze einen gemeinsamen Token-Bucket (z. B. Redis) oder einen zentralen Queue-Consumer.

Umgang mit Streaming-Unterbrechungen

Streaming ist anfälliger: Lange Laufzeit und viele Netzwerkknoten erhöhen das Risiko. Der Grundsatz: Empfangene Daten sind Assets. Verwerfe sie nicht bei Abbruch.

Der folgende Code sammelt Fragmente und sendet den bisherigen Text als assistant-Nachricht zur Fortsetzung. Dies ist ein Best-Effort-Ansatz für Texte und Dialoge. Für JSON ist ein Neuanfang sicherer. Achte darauf, dass usage im letzten Chunk enthalten ist:

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

Beim Fortsetzen zählt gesendeter Text als Eingabe-Token. Jede Fortsetzung kostet also etwas. Es ist notwendig, die Anzahl der Fortsetzungen auf max_resume zu begrenzen.

Degradation, Circuit Breaker und Pre-Flight-Check

Wiederholungen beheben nur kurzfristige Fehler. Bei anhaltenden Ausfällen stapeln weitere Wiederholungen nur. Füge einen Circuit Breaker hinzu: Bei Schwellwert (z. B. 10 Fehler/Minute) Pause einlegen und Degradation aktivieren (Cache, Benutzerhinweis oder Warteschlange). Nach der Kühlphase nur Testanfragen senden.

Plane Degradation im Voraus. Chat-Funktionen zeigen eine freundliche Nachricht. Batch-Jobs schreiben Fehler in eine Tabelle und holen sie später nach. Stelle sicher, dass Fehler nicht stumm verschluckt werden. Logge mindestens eine Zeile mit request id.

Checkliste vor dem Go-Live: SDK-Wiederholungen deaktiviert? 400/401/402/403 werden ignoriert? Gesamt-Timeout inkl. Wiederholungen? Batch-Ratenlimit aktiv? Streaming-Unterbrechungen behandelt? usage gespeichert? Guthaben-Alarm aktiv? Bestehe den Check, bevor du Lasttests startest.

Monitoring und Abrechnung via usage

Jede Antwort enthält usage (beim Streaming im letzten Chunk). Speichere usage mit dem Funktionsnamen zur kostengünstigen Überwachung. Unsere Preise: 0,25 USD pro Million Input-Token, 1,00 USD pro Million Output-Token. So schätzt du die Kosten pro Anfrage lokal:

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

Nach dem Speichern der Daten solltest du täglich drei Metriken prüfen: den Median der Input-Token pro Funktion, um zu erkennen, ob Prompts unbemerkt wachsen; das Verhältnis der Output-Token, da die Ausgabepreise viermal höher sind als die für Input und dies meist den größten Teil der Rechnung ausmacht; sowie das Verhältnis von 429 zu 503. Ein Anstieg dieses Verhältnisses zeigt, dass die Sende-Rate oder die Retry-Strategie angepasst werden muss. Füge eine Warnung für das Guthaben hinzu, die den Verantwortlichen warnt, bevor 402 erscheint. Weitere Details zur Anbindung findest du in Framework-Konfiguration, häufige Fragen sind in FAQ zusammengefasst.