Praktik terbaik untuk menangani rate limiting

• Solusi konfigurasi platform (perubahan kode minimal): Server-side queuing, Tingkatkan batas kuota, PTU, dan Batch API.
• Strategi pengendalian trafik sisi klien (perubahan kode klien): Empat strategi dengan kompleksitas rekayasa yang meningkat, mulai dari retry dasar hingga adaptive congestion control.
• Solusi fallback arsitektural (perubahan arsitektur sistem): Model fallback dan peak-load shifting menggunakan message queues (MQ).

​

Mekanisme rate limiting platform

• Batas kuota tingkat menit (RPM / TPM): Jumlah maksimum permintaan per menit (RPM) dan penggunaan token maksimum per menit (TPM).
• Batas frekuensi instan (RPS / TPS): Maksimum permintaan per detik (RPS) dan penggunaan token maksimum per detik (TPS). Panggilan API padat atau konsumsi token dalam satu detik dapat memicu rate limiting.
• Batas laju pertumbuhan (Traffic Burst): Lonjakan tiba-tiba dalam volume permintaan atau penggunaan token memicu rate limiting. Ambang batasnya menyesuaikan secara dinamis. Tingkatkan permintaan secara bertahap untuk menghindari pemicuan batas ini.

​

Diagnosis error dan rekomendasi strategi

​

Solusi konfigurasi platform

​

Server-side queuing (direkomendasikan)

• Permintaan non-streaming (stream: false): Timeout = timeout dasar asli + nilai Wait-Timeout.
• Permintaan streaming (stream: true): Timeout > nilai Wait-Timeout. Permintaan streaming mulai menghitung waktu setelah chunk pertama, sehingga timeout respons awal hanya perlu melebihi waktu antrian.

import os
from openai import OpenAI

client = OpenAI(
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    timeout=150.0,  # Timeout asli 120s + antrian tunggu 30s
)

response = client.chat.completions.create(
    model="qwen-plus",
    messages=[{"role": "user", "content": "Hello"}],
    extra_headers={
        "X-DashScope-Wait-Timeout": "30"  # Waktu tunggu antrian maksimum: 30 detik
    }
)
print(response.choices[0].message.content)

​

Tingkatkan batas kuota

​

Provisioned throughput unit (PTU)

​

Pemrosesan batch asinkron (Batch API)

​

Strategi pengendalian trafik sisi klien

• Retry dasar hanya memberikan perlindungan pasif.
• Pembatasan laju permintaan menambahkan antrian aktif.
• Traffic shaping lebih lanjut memperkenalkan kontrol tingkat token dan pengiriman lancar.
• Adaptive congestion control secara dinamis menyesuaikan laju pengiriman berdasarkan umpan balik real-time.

​

Perbandingan performa throughput setiap strategi

• Strategi retry dasar: Efektif di bawah beban rendah. Rentan terhadap keruntuhan kemacetan di bawah konkurensi tinggi, menyebabkan penurunan tajam dalam throughput.
• Strategi pembatasan laju permintaan: Perlindungan kuat terhadap keruntuhan. Namun, di bawah beban campuran dengan teks panjang, throughput menunjukkan fluktuasi seperti gergaji karena kurangnya kontrol token.
• Strategi traffic shaping: Stabilitas tinggi. Mencapai output lancar dengan mengorbankan sebagian throughput puncak.
• Strategi adaptive congestion control: Dapat secara dinamis konvergen ke titik throughput tinggi yang stabil di bawah beban tinggi, tetapi memiliki overhead probing cold-start.

​

Strategi retry dasar

import openai
from openai import OpenAI
from tenacity import (
    retry,
    stop_after_attempt,
    wait_random_exponential,
    retry_if_exception_type
)

RETRYABLE_ERRORS = (
    openai.RateLimitError,
    openai.InternalServerError,
    openai.APIConnectionError,
)

@retry(
    wait=wait_random_exponential(min=1, max=60),
    stop=stop_after_attempt(6),
    retry=retry_if_exception_type(RETRYABLE_ERRORS)
)
def chat_with_retry(client, model, messages, max_tokens):
    return client.chat.completions.create(
        model=model,
        max_tokens=max_tokens,
        messages=messages
    )

client = OpenAI(
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    api_key="YOUR_DASHSCOPE_API_KEY"
)

try:
    response = chat_with_retry(
        client=client,
        model="qwen-plus",
        messages=[{"role": "user", "content": "What is exponential backoff retry?"}],
        max_tokens=1024
    )
    print(response.choices[0].message.content)
except Exception as e:
    print(f"Request failed: {e}")

• Waktu tunggu berlipat ganda secara progresif: Misalnya, 1s, 2s, 4s.... Ini menghindari permintaan berulang dalam periode singkat.
• Tambahkan jitter acak: Nilai acak (seperti 2s +/- 0.5s) menyebarkan trafik retry, mencegah banjir sekunder (efek thundering herd).

​

Strategi pembatasan laju permintaan

• RPM token bucket: Membatasi total permintaan per menit. Kapasitas bucket sama dengan kuota RPM; token diisi ulang pada laju konstan. Mendukung peminjaman: jika token tidak mencukupi, permintaan meminjam dari kuota masa depan, FIFO ketat.
• Concurrency semaphore: Membatasi permintaan konkuren untuk mencegah konkurensi instan tinggi memicu batas RPS.

import time

class TokenBucket:
    """
    Implementasi token bucket untuk mengontrol permintaan per menit (RPM).
    Mendukung mekanisme utang untuk memastikan urutan first-in, first-out (FIFO) di bawah konkurensi tinggi.
    """
    def __init__(self, quota_per_minute: float, initial_tokens: float = 0.0):
        self.capacity = quota_per_minute
        self.tokens = initial_tokens
        self.refill_rate = quota_per_minute / 60.0
        self.last_refill = time.monotonic()

    def reserve(self, cost: float = 1.0) -> float:
        """
        Mengakuisisi token.
        Jika token tidak mencukupi, mengembalikan jumlah detik yang harus ditunggu (mendukung utang).
        """
        self._refill()

        # 1. Token mencukupi: Kurangi langsung
        if self.tokens >= cost:
            self.tokens -= cost
            return 0.0

        # 2. Token tidak mencukupi: Hitung waktu tunggu dan timbulkan utang
        # "Mereservasi" token masa depan untuk permintaan saat ini untuk memastikan urutan FIFO
        deficit = cost - self.tokens
        wait_seconds = deficit / self.refill_rate
        self.tokens -= cost
        return wait_seconds

    def _refill(self):
        """Mengisi ulang token berdasarkan waktu yang berlalu."""
        now = time.monotonic()
        elapsed = now - self.last_refill
        if elapsed > 0:
            self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
            self.last_refill = now

​

Strategi traffic shaping

• Kontrol sumber daya ganda (RPM & TPM): Memelihara kedua bucket token RPM dan TPM. Semua permintaan harus lulus pemeriksaan kuota untuk kedua dimensi sebelum dikirim.
• Deduksi awal untuk input, penyelesaian akhir untuk output: Panjang output tidak diketahui sebelum permintaan. Bucket TPM mendeduksi token input saat mengirim dan menyelesaikan token output aktual setelah selesai. Bahkan jika token menjadi negatif, permintaan berikutnya menunggu hingga jumlahnya positif, secara alami menghaluskan aliran.
• Pemanasan kontinu: Selama cold start, laju penerbitan token meningkat secara linear seiring waktu, menghilangkan risiko lonjakan awal.
• Smoothing rate limiter (Pacing): Menghaluskan laju pengiriman dengan menerapkan interval minimum antarpermintaan (pacing), mengurangi risiko memicu batas laju.

import time

class TokenBucket:
    """Token bucket lanjutan yang mendukung mekanisme pemanasan kontinu."""
    def __init__(self, quota_per_minute: float, warmup_seconds: float = 0.0):
        self.capacity = quota_per_minute
        self.tokens = 0.0
        self.target_refill_rate = quota_per_minute / 60.0
        self.warmup_seconds = warmup_seconds
        self.start_time = time.monotonic()
        self.last_update_time = self.start_time
        self.cumulative_generated = 0.0

    def _get_cumulative_tokens(self, t: float) -> float:
        if t <= 0:
            return 0.0
        R = self.target_refill_rate
        T = self.warmup_seconds
        if T <= 0:
            return R * t
        if t <= T:
            return (R / (2 * T)) * (t ** 2)
        else:
            warmup_total = (R * T) / 2.0
            return warmup_total + R * (t - T)

    def _get_time_for_cumulative_tokens(self, target_cumulative: float) -> float:
        if target_cumulative <= 0:
            return 0.0
        R = self.target_refill_rate
        T = self.warmup_seconds
        if T <= 0:
            return target_cumulative / R
        warmup_total = (R * T) / 2.0
        if target_cumulative <= warmup_total:
            return ((2 * T * target_cumulative) / R) ** 0.5
        else:
            return (target_cumulative - warmup_total) / R + T

    def reserve(self, cost: float = 1.0) -> float:
        now = time.monotonic()
        relative_now = now - self.start_time
        current_cumulative = self._get_cumulative_tokens(relative_now)
        new_tokens = current_cumulative - self.cumulative_generated
        self.tokens = min(self.capacity, self.tokens + new_tokens)
        self.cumulative_generated = current_cumulative
        self.last_update_time = now
        if self.tokens >= cost:
            self.tokens -= cost
            return 0.0
        deficit = cost - self.tokens
        self.tokens -= cost
        target_cumulative = self.cumulative_generated + deficit
        target_time = self._get_time_for_cumulative_tokens(target_cumulative)
        wait_seconds = target_time - relative_now
        return max(0.0, wait_seconds)

    def adjust(self, amount: float):
        self.tokens = min(self.capacity, self.tokens + amount)

​

Strategi pengendalian kemacetan adaptif

• EBP: Menyimpan watermark historis tertinggi yang berhasil. Menghitung gain probing dengan mensimulasikan tegangan pegas berdasarkan jarak dari konkurensi saat ini ke watermark (lebih jauh = lebih cepat, lebih dekat = lebih lambat). Menambahkan dorongan linier kecil untuk terus mengeksplorasi bahkan pada saturasi tinggi.
• Kesadaran kemacetan TPT: Waktu generasi model bahasa besar meningkat seiring panjang output — latensi tinggi pada teks panjang bukan berarti kemacetan. TPT (Time Per Token) menyaring noise panjang konten. Kemacetan hanya dinyatakan ketika TPT menurun secara signifikan.
• Pengatur laju anti-lonjakan: Terlepas dari target EBP, pengatur membatasi percepatan pertumbuhan konkurensi untuk memastikan ramp-up lancar dan menghindari perubahan langkah yang memicu batas laju pertumbuhan.

• Probing terpandu: Memperkenalkan kuota RPM/TPM yang diketahui sebagai "batas atas panduan" untuk menghindari tabrakan berulang akibat probing buta.
• Sumber sinyal (RTT → TPT): BBR asli menggunakan RTT. Dalam skenario model bahasa besar, latensi panjang konten jauh lebih besar daripada fluktuasi jaringan. TPT menghilangkan gangguan tersebut.
• Peningkatan mekanisme respons (ProbeRTT → Hold): Menghadapi fluktuasi latensi, memilih mempertahankan tingkat konkurensi saat ini alih-alih mundur proaktif dan mengurangi throughput.
• Respons batas laju keras (Packet Loss → 429 Drain): Setelah error 429 dipicu, masuk ke keadaan Drain agresif dan melakukan pemulihan cepat setelah periode cooldown.

• Noise TPT: TPT diperkirakan sebagai "latensi total / total token". Latensi total mencakup round-trip jaringan, antrian, dan waktu hingga token pertama — rentan terhadap jitter atau input panjang, yang dapat memicu salah keadaan Hold.
• Starvation permintaan besar: Menggunakan wakeup FIFO non-ketat untuk performa penjadwalan. Ketika kuota langka, permintaan token pendek dapat mendahului sumber daya, menyebabkan permintaan token panjang menunggu terlalu lama.
• Cold start: Memerlukan periode pemanasan untuk membangun model statistik. Dalam tugas beban rendah atau berumur pendek, throughput mungkin lebih rendah daripada tiga strategi pertama.

class ElasticCongestionController:
    async def acquire(self):
        """[Fase admisi] Periksa sebelum memulai permintaan"""
        # 1. SSR slow-start restart: Jika idle terlalu lama, turunkan paksa batas
        #    untuk mencegah trafik lonjakan akibat watermark usang.
        if self.is_idle_too_long():
            self.perform_slow_start_restart()

        # 2. Pemeriksaan circuit breaker: Jika dalam keadaan DRAIN (cooldown), paksa tunggu.
        if self.state == CongestionState.DRAIN:
            await self.wait_for_cooldown()

        # 3. Pemeriksaan anggaran ganda: Periksa slot konkurensi dan anggaran token.
        await self.wait_for_budget(request_tokens)

    async def release(self, latency, actual_tokens, error):
        """[Fase umpan balik] Keputusan setelah permintaan selesai"""
        if error:
            # [Respons kesalahan] Pada error rate limit (429/503): Segera drain + backoff multiplikatif
            self.state = CongestionState.DRAIN
            self.concurrency_limit *= self.backoff_factor  # misalnya 0.7
            return

        # [Respons normal] Hitung TPT (Time-Per-Token)
        current_tpt = latency / actual_tokens

        # [Kesadaran kemacetan] Lonjakan TPT (generasi melambat): Masuk HOLD untuk mengamati
        # Pertahankan tingkat konkurensi, tidak mundur maupun meningkatkan
        if current_tpt > self.metrics.ema_tpt * 2.0:
            self.state = CongestionState.HOLD
        else:
            # [Probing steady-state] Jaringan sehat: Lakukan probing elastis EBP
            self.state = CongestionState.PROBING
            self.update_limit_via_ebp()

​

Solusi fallback arsitektural

​

Model fallback

• Pilih model dari seri berbeda: Rate limiting berlaku per model. Gunakan model berbeda sebagai fallback — misalnya, fallback dari qwen3.6-plus ke qwen3.6-flash.
• Picu fallback hanya pada error rate limit: Fallback pada error 429, bukan semua exception. Mengganti model tidak akan memperbaiki timeout jaringan atau error parameter.
• Validasi model fallback terlebih dahulu: Pastikan model tersebut mendukung fitur yang diperlukan (Function Calling, structured output, dll.) untuk menghindari masalah fungsional setelah fallback.

import os
import asyncio
from openai import AsyncOpenAI, APIStatusError

# Model utama dan fallback (seri berbeda, kuota independen)
PRIMARY_MODEL = "qwen3.6-plus"
FALLBACK_MODEL = "qwen3.6-flash"

client = AsyncOpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

async def chat_with_fallback(messages: list) -> str:
    """Permintaan dengan fallback: Secara otomatis beralih ke model fallback ketika model utama dibatasi lajunya."""
    for model in [PRIMARY_MODEL, FALLBACK_MODEL]:
        try:
            response = await client.chat.completions.create(
                model=model,
                messages=messages
            )
            return response.choices[0].message.content
        except APIStatusError as e:
            if e.status_code == 429 and model == PRIMARY_MODEL:
                print(f"[Rate Limit Dipicu] {model}, fallback ke {FALLBACK_MODEL}")
                continue
            raise
    raise RuntimeError("Semua model tidak tersedia")

async def main():
    result = await chat_with_fallback(
        messages=[{"role": "user", "content": "Hello"}]
    )
    print(result)

if __name__ == "__main__":
    asyncio.run(main())

​

Peak-load shifting menggunakan message queues (MQ)

• Pengendalian laju konsumen: Sisi konsumen harus menggunakan strategi pembatasan laju permintaan atau traffic shaping untuk mengonsumsi pesan pada laju stabil berdasarkan kuota RPM/TPM, alih-alih menarik pesan tanpa batas.
• Penanganan dead-letter: Pindahkan pesan yang gagal setelah beberapa kali retry ke dead-letter queue dan picu alert. Cegah retry tak terbatas menghambat konsumsi.
• Propagasi back-pressure: Ketika backlog MQ melebihi ambang batas, propagasikan tekanan ke hulu (misalnya, kembalikan status antrian) untuk mencegah pertumbuhan antrian tak terbatas.

​

Pertimbangan lingkungan produksi

• Adaptasi ke model non-teks Strategi di atas menggunakan model teks sebagai contoh, tetapi prinsip intinya berlaku untuk layanan multimodal (generasi gambar, sintesis suara). Satuannya berbeda, tetapi esensinya sama: membatasi laju pengiriman dan kapasitas pemrosesan.
  • Model seperti pengenalan suara biasanya dibatasi oleh jumlah permintaan per satuan waktu (seperti RPM) dan penggunaan (seperti durasi audio). Strateginya pada dasarnya sama dengan model teks.
  • Model untuk gambar dan video biasanya dibatasi oleh laju pengiriman tugas dan jumlah tugas konkuren. Anda dapat menggunakan pendekatan yang sama seperti strategi pembatasan laju permintaan: batasi laju pengiriman tugas dan gunakan semaphore untuk mengontrol konkurensi.
  Terlepas dari bagaimana metrik berubah, prinsip throttling sisi klien tetap sama. Ganti penghitung atau metrik probing dengan yang sesuai untuk modalitas Anda. Untuk aturan spesifik, lihat 限流.
• Atomicity dalam model konkuren Contoh: asyncio menggunakan penjadwalan kooperatif single-threaded, sehingga modifikasi state secara inheren atomik dalam satu proses. Produksi: Di lingkungan multi-threaded atau multi-proses, pastikan keamanan konkurensi bucket token dan jendela statistik. Race condition akan merusak pengendalian trafik.
• Pembatasan laju terdistribusi Contoh: Semua komponen pengendalian trafik berada di memori. Produksi: Dalam penyebaran multi-instans, setiap instans melakukan throttling secara independen. Total penggunaan mungkin melebihi batas. Gunakan penghitung terpusat (seperti Redis) untuk mengelola penggunaan di semua node.
• Antrian prioritas dan pencegahan starvation Contoh: Tidak ada diferensiasi prioritas. Strategi adaptive congestion control menggunakan wakeup FIFO non-ketat untuk performa penjadwalan. Produksi: Untuk permintaan prioritas tinggi/rendah, implementasikan antrian prioritas berbobot untuk menjamin bandwidth bagi trafik prioritas tinggi. Cadangkan kuota minimum untuk antrian prioritas rendah untuk mencegah starvation.