Aracı API kararlılığı uygulamaları: Zaman aşımı, yeniden deneme, hız limiti kuyruğu ve akış kopması

Önce hataları üç sınıfa ayırın

İstikrar sorunlarında ilk adım yeniden denemeyi artırmak değil, neyin yeniden denenmesi gerektiğini bilmektir. İşlem türlerine göre yaygın hataları üç sınıfa ayırın:

Kategori	Tipik Belirti	İşlem Yöntemi
Ani ve Kurtarılabilir	429 hız limiti, 503 upstream_busy, bağlantının sıfırlanması, zaman aşımı	Üssel geri çekilme ile yeniden dene, toplam sayıyı sınırla
İstek kendisi hatalı	400 (girdi + max_tokens 100.000'i aştı, istek gövdesi çok büyük), 403 content_blocked	Yeniden deneme yapma, isteği düzelt veya kullanıcıya bildir
Hesap durumu sorunu	401 geçersiz anahtar, 402 no_credit	Yeniden deneme yapma, sorumlu kişiye uyarı gönder, bakiye yükle veya anahtarı değiştir

Bu tabloyu kod yorumu olarak eklemeniz önerilir. Yaygın bir sorun şudur: Bakiye bittikten sonra arayüz kararlı bir şekilde 402 döndürür. Yeniden deneme mantığınız durum kodlarını ayırt etmezse her istek beş kez denenir; bu da trafiği beş katına çıkarır ve logları hatalarla doldurur.

Zaman aşımı katmanlı ayarlanmalı

Tek bir toplam zaman aşımı ayarlamak yeterli değildir. Zaman aşımını üç katmana bölmek önerilir:

• Bağlantı zaman aşımı: TCP ve TLS bağlantısının kurulması için geçen süredir; genellikle 5-10 saniye yeterlidir, bağlantı kurulamazsa hemen hata verilmelidir.
• Okuma zaman aşımı: Yanıt verisi beklemek için geçen süredir. Akış olmayan isteklerde tüm metin bitene kadar beklenir, bu nedenle en uzun çıktıyı kapsamalıdır; akış isteklerinde ise "iki veri bloğu arasındaki en uzun sessizlik" olarak düşünülür, 10-30 saniye daha uygundur.
• Uygulama düzeyinde toplam zaman aşımı: Kendi uygulamanızın tolerans edebileceği maksimum bekleme süresini asyncio.wait_for veya ağ geçidi katmanı ile sağlayın; tüm yeniden denemeleri kapsar.

Çıktı ne kadar uzunsa, akış olmayan istek o kadar uzun sürer. max_tokens değerini 32.000 üst sınırına ayarlayıp toplam zaman aşımını sadece 30 saniye yaparsanız, uzun metin senaryolarında sık sık zaman aşımı alırsınız. Daha güvenli bir yaklaşım, uzun çıktılar için her zaman akış modunu kullanmaktır; bu hem kullanıcıya anlık geri bildirim sağlar hem de zaman aşımının anlamını netleştirir. SDK'daki timeout parametresi bir sayı veya alt yapılandırma nesnesi olarak iletilebilir; kullandığınız sürümün dokümantasyonuna bakın.

429 ve 503 için üssel geri çekilme yeniden denemesi

Geri çekme stratejisinin üç temel kuralı vardır: Gecikme süresi başarısızlık sayısına göre iki katına çıkarılmalı, bir üst sınır belirlenmeli ve rastgele gürültü (jitter) eklenmelidir. Jitter olmadan, aynı anda başarısız olan görevler aynı anda yeniden denenir ve servisi tekrar çökertir. Aşağıdaki fonksiyon SDK'nın yerleşik yeniden denemelerini kapatır; tüm kontrolü kendimiz üstlenir ve yalnızca 429, 503 ve bağlantı hatalarında yeniden dener:

复制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 hatası sunucunun geç olarak meşgul olduğunu gösterir; birkaç saniye sonra yeniden denemeniz önerilir. Yukarıdaki temel gecikme 1 saniye, üst sınır 20 saniye ve maksimum 5 yeniden deneme bu senaryoyu karşılamak için yeterlidir. 429 hatası sıkça alıyorsanız, sorun yeniden deneme değil, gönderen tarafın hız kontrolü yapmamasıdır; lütfen bir sonraki bölüme bakın.

Dakikada 300 hız limiti altında eşzamanlı kuyruk

Her anahtar dakikada 300 istek yapabilir; bu da saniyede ortalama 5 istek anlamına gelir. Toplu işler burada sıkça hata yapar: asyncio.gather ile binlerce isteği aynı anda gönderdiğinizde, ilk birkaç istek limiti anında tüketir ve kalanlar 429 hatası alır.

Güvenli yaklaşım iki katmanlı kontrol gerektirir. Sema, "aynı anda devam eden" istek sayısını sınırlayarak yerel bağlantı ve bellek taşmasını önler; ritim belirleyici (metronom) "gönderme hızını" sınırlayarak istekleri eşit dağıtır. İkisi de şarttır: Sadece sema kullanırsanız, istekler çok hızlıysa dakikada 300'ü aşabilirsiniz; sadece metronom kullanırsanız, istekler yavaşsa devam eden istekler birikir. Örnekte hız dakikada 270 olarak ayarlanmıştır; çoklu örneklere, yeniden denemelere ve saat hatalarına %10 pay bırakır:

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

Birden fazla işlem veya makine aynı anahtarı paylaşıyorsa, bellek düzeyindeki metronom yeterli olmaz; hız limiti anahtara göre birleştirilir. Bu durumda paylaşılan bir belirteç kovanı gerekir; yaygın yöntem Redis ile sayım yapmaktır veya tüm toplu iş isteklerini tek bir kuyruk tüketim servisine yönlendirmektir.

Akış çıktısı kesildiğinde ne yapılır?

Akış bağlantısı normal isteklerden daha kırılgandır; uzun sürer, daha fazla ağ cihazından geçer; herhangi bir bileşenin boşta kalma zaman aşımı bağlantıyı koparabilir. İşleme ilkesi şudur: "Zaten alınan veri bir varlıktır", kopma nedeniyle tüm veriyi atmayın.

Aşağıdaki uygulama, alınan parçaları biriktirir; bağlantı ve zaman aşımı hatalarını yakalayıp mevcut metni bir assistant mesajı olarak tekrar göndererek modelin devam etmesini sağlar. Bu çaba gösteren bir kurtarma yöntemidir; devam eden metin her zaman mükemmel bir geçiş sağlamayabilir; uzun metinler ve sohbetler için uygundur ancak JSON gibi sıkı yapılar için uygun değildir. Yapısal çıktılar kesilirse, tüm isteği baştan yapmak daha güvenlidir. Akış sonlandığında otomatik olarak bir usage parçası da gelir; kodda bu da okunmuştur:

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

Devam yazarken, gönderilen metnin de giriş token olarak sayıldığını unutmayın. Bu nedenle her devam yazma işlemi ek maliyet doğurur; max_resume değerini sınırlamak önemlidir.

Degrade etme, sigorta ve yayına hazırlık kontrol listesi

Yeniden deneme anlık arızaları çözer; ancak arıza birkaç dakika sürerse, yeniden deneme yapmak istekleri biriktirir. Yeniden denemenin üzerine bir sigorta (circuit breaker) katmanı eklemeniz önerilir: Eşik sayısına (örneğin bir dakikada 10 hata) ulaşıldığında dışarıya istek göndermek kısa süreliğine durdurulur; bu sürede degrade mantık çalışır (örneğin önbellek döndür, kullanıcıya tekrar dene de veya işi kuyruğa al). Soğuma süresi bittikten sonra yalnızca az sayıda keşif isteği gönderilir, başarılı olursa tam kapasiteye dönülür.

Degrade etme de önceden planlanmalıdır. Kullanıcıya açık sohbet işlevi nazik bir meşgul mesajı döndürebilir; çevrimdışı toplu işler hatalı girdileri hata tablosuna yazmalı, tüm iş bitince toplu olarak tamamlamalıdır; ana akışta sonsuza kadar beklememelidir. Hangi yöntem seçilirse seçilsin, hataların sessizce yutulmaması, en azından request id içeren bir log satırı bırakılması sağlanmalıdır.

Son olarak yayına çıkmadan önce kontrol listesi: SDK'nın örtük yeniden denemeleri kapatıldı mı ve yalnızca tek bir yer mi kullanılıyor; 400, 401, 402, 403 hatalarında hemen vazgeçiliyor mu; toplam zaman aşımı tüm yeniden denemeleri kapsıyor mu; toplu işler için hız kontrolü var mı yoksa tüm istekler aynı anda mı gönderiliyor; akış kopmaları işleniyor mu; usage kalıcı hale getiriliyor mu; bakiye için uyarı var mı? Tüm maddeler geçtikten sonra stres testi yapın, tersini değil.

usage ile izleme ve muhasebe

Her yanıt usage değerini içerir; akış modunda bu değer son parçada yer alır. Bu değeri fonksiyon adıyla birlikte kalıcı olarak kaydetmek, maliyet açısından en ucuz izleme yöntemidir. Bu sitede giriş fiyatı her milyon token için 0,25 ABD doları, çıkış için ise 1,00 ABD dolarıdır; her istek için maliyeti yerel olarak doğrudan tahmin edebilirsiniz:

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

Kalıcı depolama alanına yazdıktan sonra, her gün üç metriği incelemenizi öneririz: İşlev bazında giriş token medyanı (istem şişmesini tespit etmek için), çıktı token oranı (çıktı fiyatı girişin dört katı olduğu için faturanın büyük kısmını oluşturur) ve 429/503 oranı (oran artarsa gönderim hızı veya yeniden deneme stratejisi ayarlanmalıdır). Bir bakiye uyarısı ekleyerek 402 hatası oluşmadan önce sorumlu kişiyi bilgilendirin. Daha fazla bağlantı detayı için Çerçeve yapılandırması'na, sık sorulan sorular için SSS'ye bakın.