Praktik stabilitas API perantara: timeout, retry, antrian batas laju, dan pemutusan streaming

Klasifikasikan error menjadi tiga kategori

Langkah pertama untuk stabilitas bukanlah menambah retry, melainkan mengetahui apa yang harus di-retry. Berdasarkan cara penanganan, kegagalan umum dapat dibagi menjadi tiga kategori:

Kategori	Gejala Khas	Cara Penanganan
Pemulihan Instan	429 batas laju, 503 upstream_busy, koneksi direset, timeout	Retry dengan backoff eksponensial, batasi jumlah total
Masalah pada permintaan itu sendiri	400 (jumlah token konteks melebihi 100.000, body permintaan terlalu besar), 403 content_blocked	Jangan di-retry, perbaiki permintaan atau beri tahu pengguna
Masalah status akun	401 kunci tidak valid, 402 no_credit	Jangan di-retry, beri alarm ke penanggung jawab, isi saldo atau ganti kunci

Tabel ini sebaiknya ditulis sebagai komentar dalam kode. Satu insiden umum adalah: setelah saldo habis, endpoint mulai mengembalikan status 402 secara stabil, dan logika retry Anda tidak membedakan kode status, sehingga setiap permintaan di-retry lima kali. Hal ini secara sia-sia memperbesar traffic jaringan hingga lima kali lipat, sekaligus membuat log penuh dengan error merah.

Atur timeout secara berlapis

Menetapkan satu timeout total saja tidak cukup. Disarankan membagi timeout menjadi tiga lapisan:

• Timeout koneksi: Waktu untuk membangun koneksi TCP dan TLS, biasanya 5 hingga 10 detik sudah cukup. Jika gagal terhubung, sebaiknya segera gagal.
• Timeout baca: waktu tunggu hingga data respons diterima. Untuk permintaan non-streaming, Anda harus menunggu seluruh teks selesai digenerate sebelum respons dikembalikan, sehingga nilai ini harus mencakup durasi output terpanjang; untuk permintaan streaming, ini berubah menjadi "diam terpanjang antara dua blok data", dengan rentang 10 hingga 30 detik yang lebih masuk akal.
• Timeout total bisnis: durasi tunggu maksimum yang dapat ditoleransi oleh bisnis Anda sendiri, dijamin menggunakan asyncio.wait_for atau lapisan gateway, yang mencakup semua proses retry.

Semakin panjang output, semakin lama waktu permintaan non-streaming. Jika Anda mengatur max_tokens ke batas atas 32.000 namun hanya menetapkan timeout total 30 detik, Anda akan sering mengalami timeout pada skenario teks panjang. Cara yang lebih aman adalah menggunakan streaming untuk output panjang, yang memberikan umpan balik instan kepada pengguna dan membuat arti timeout menjadi jelas. Parameter timeout di SDK dapat berupa angka atau konfigurasi terpecah, silakan periksa dokumentasi sesuai versi yang Anda gunakan.

Retry backoff eksponensial untuk 429 dan 503

Ada tiga poin kunci dalam backoff: delay berlipat ganda seiring bertambahnya jumlah kegagalan, menetapkan batas atas, dan menambahkan jitter acak. Tanpa jitter, tugas yang gagal secara bersamaan akan melakukan retry pada waktu yang sama, sehingga dapat kembali membebani layanan yang baru saja pulih. Fungsi di bawah ini menonaktifkan retry bawaan SDK, mengontrolnya secara terpusat, dan hanya melakukan retry untuk 429, 503, serta error koneksi:

复制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 berarti server sibuk sementara. Disarankan melakukan retry setelah beberapa detik. Delay dasar 1 detik, batas atas 20 detik, dan maksimal 5 kali percobaan sudah cukup untuk skenario ini. Untuk 429, jika Anda melihatnya muncul sering, masalahnya bukan pada retry, melainkan pada pengirim yang tidak mengontrol laju. Lihat bagian berikutnya.

Antrian konkurensi di bawah batas laju 300 per menit

Setiap kunci API memiliki batas 300 permintaan per menit, yang setara dengan rata-rata 5 permintaan per detik. Tugas batch paling sering mengalami masalah di sini: menggunakan asyncio.gather untuk mengirim ribuan permintaan sekaligus akan menghabiskan kuota dalam hitungan detik, dan sisanya akan menerima 429.

Pendekatan yang andal melibatkan dua lapisan kontrol. Semaphore membatasi jumlah "permintaan yang sedang berjalan" untuk mencegah koneksi lokal dan memori meledak; rate limiter membatasi "laju pengiriman" untuk menyebarkan permintaan secara merata. Keduanya saling melengkapi: jika hanya menggunakan semaphore, jika satu permintaan selesai sangat cepat, jumlah permintaan per menit masih bisa melebihi 300; jika hanya menggunakan rate limiter, jika permintaan berlangsung lama, jumlah permintaan yang sedang berjalan akan terus menumpuk. Contoh ini menetapkan laju pada 270 permintaan per menit, memberikan margin 10% untuk multi-instance, retry, dan kesalahan jam:

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

Jika beberapa proses atau mesin berbagi kunci yang sama, rate limiter berbasis memori di atas tidak cukup, karena pembatasan laju dihitung secara agregat berdasarkan kunci. Pada situasi ini, diperlukan token bucket yang berbagi, dengan praktik umum menggunakan Redis untuk penghitungan, atau secara sederhana mengarahkan semua permintaan batch melalui layanan konsumen antrian terpisah yang menjadi satu pintu keluar terpadu.

Apa yang harus dilakukan jika streaming terputus

Koneksi streaming lebih rapuh daripada permintaan biasa: waktunya lebih lama, melewati lebih banyak perangkat jaringan, dan timeout idle di mana saja dapat memutusnya. Prinsip penanganannya adalah "konten yang sudah diterima adalah aset", jangan dibuang hanya karena terputus.

Implementasi di bawah ini mengakumulasi fragmen yang telah diterima, menangkap exception koneksi dan timeout, lalu menggunakan teks yang sudah ada sebagai pesan assistant untuk melakukan permintaan ulang, sehingga model dapat melanjutkan penulisan. Ini adalah upaya perbaikan best-effort; sambungan teks hasil penulisan ulang mungkin tidak sempurna, cocok untuk teks panjang dan percakapan, tetapi tidak cocok untuk output dengan struktur ketat seperti JSON. Jika output terstruktur terputus, cara yang lebih aman adalah melakukan pengulangan keseluruhan. Perhatikan bahwa saat streaming berakhir, akan ada fragmen usage yang dilampirkan secara otomatis; kode ini juga mengambil fragmen tersebut:

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

Saat melanjutkan, ingat bahwa teks yang sudah dikirim juga dihitung sebagai token input, sehingga setiap kelanjutan memiliki biaya tambahan. Membatasi jumlah kali max_resume dapat dilakukan sangat diperlukan.

Degradasi, pemutus sirkuit, dan pemeriksaan mandiri sebelum peluncuran

Retry menangani gangguan instan. Jika gangguan berlangsung beberapa menit, melanjutkan retry hanya akan menumpuk permintaan. Disarankan menambahkan lapisan pemutus sirkuit di atas retry: jika kegagalan berturut-turut mencapai ambang batas (misalnya 10 kegagalan dalam satu menit), hentikan pengiriman permintaan untuk sementara waktu. Selama periode ini, jalankan logika degradasi, seperti mengembalikan hasil cache, memberi tahu pengguna untuk mencoba lagi nanti, atau mengembalikan tugas ke antrian untuk diproses nanti. Setelah waktu pendinginan berakhir, izinkan hanya beberapa permintaan probe; jika berhasil, kembalikan ke kapasitas penuh.

Degradasi juga harus direncanakan sebelumnya. Untuk fitur obrolan pengguna, kembalikan pesan sibuk yang ramah; untuk tugas batch offline, tulis entri yang gagal ke tabel kegagalan dan jalankan ulang secara terpusat setelah proses selesai, bukan menunggu tanpa batas di alur utama. Dalam kasus apa pun, pastikan kegagalan tidak tertelan secara diam-diam, setidaknya tinggalkan satu baris log dengan request id.

Berikut adalah daftar periksa pra-peluncuran: Apakah retry implisit SDK sudah dinonaktifkan dan hanya menyisakan satu titik kontrol Anda sendiri? Apakah 400, 401, 402, 403 langsung diabaikan? Apakah timeout total mencakup semua retry? Apakah batch memiliki kontrol laju dan bukan konkurensi sekaligus? Apakah streaming menangani pemutusan? Apakah usage disimpan ke disk? Apakah ada alarm untuk saldo? Luluskan item-item ini satu per satu sebelum melakukan tes tekanan, bukan sebaliknya.

Gunakan usage untuk pemantauan dan rekonsiliasi

Setiap respons menyertakan usage, dan saat streaming, usage muncul di fragmen terakhir. Menyimpannya bersama nama fungsi ke penyimpanan adalah metode monitoring dengan biaya paling murah. Harga input di situs ini adalah 0.25 USD per juta token, dan output adalah 1.00 USD, sehingga Anda dapat memperkirakan biaya setiap permintaan secara 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}"]
        )

Setelah deployment, Anda disarankan memeriksa tiga metrik setiap hari: median token input per fitur untuk mendeteksi apakah prompt mengembang secara diam-diam; rasio token output, karena harga output empat kali lipat input, sehingga ini biasanya menjadi komponen terbesar tagihan; serta rasio 429 dan 503. Jika rasio meningkat, berarti kecepatan pengiriman atau strategi retry perlu disesuaikan. Tambahkan juga peringatan saldo; beri tahu penanggung jawab untuk melakukan isi saldo sebelum kode 402 muncul. Detail lebih lanjut tentang integrasi dapat merujuk ke konfigurasi kerangka kerja, sedangkan pertanyaan umum dirangkum di pertanyaan umum.