Alibaba Cloud Model Studio:Praktik terbaik untuk menangani rate limiting

API Model Studio membatasi volume permintaan, penggunaan token, dan laju pertumbuhan. Terapkan strategi berikut untuk memaksimalkan throughput dan menjaga ketersediaan.

API Model Studio menerapkan rate limiting terhadap permintaan, penggunaan token, dan laju pertumbuhan dari waktu ke waktu. Model bahasa besar memiliki latensi tinggi dan dibatasi pada dua dimensi (jumlah permintaan dan volume token). Strategi sederhana seperti "retry on error" tidak efektif dalam skenario ini; Anda memerlukan pengendalian trafik yang dirancang khusus.

Tiga jenis solusi, diurutkan dari biaya implementasi terendah hingga tertinggi:

• 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).

Jika Anda sedang melakukan troubleshooting error 429, buka Diagnosis error dan rekomendasi strategi untuk mengidentifikasi penyebabnya. Untuk error akibat lonjakan trafik, coba server-side queuing terlebih dahulu — hanya memerlukan satu header permintaan.

Mekanisme rate limiting platform

Platform menerapkan rate limiting pada setiap model secara independen di tingkat akun root. Setelah dipicu, layanan biasanya pulih dalam waktu satu menit. Untuk kondisi rate limiting dan penggunaan saat ini per model, lihat 限流 dan Pemantauan model. Tiga jenis aturan rate limiting berlaku:

• 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.

Bagian-bagian berikut mencakup solusi pada tiga tingkatan: konfigurasi platform, pengendalian trafik sisi klien, dan fallback arsitektural.

Diagnosis error dan rekomendasi strategi

Kode error yang sama dapat dipicu oleh dimensi rate limiting yang berbeda. Konkurensi tinggi juga dapat menyebabkan server jenuh, sehingga terjadi timeout. Strategi adaptive congestion control dapat membantu mengurangi hal ini.

Solusi konfigurasi platform

Solusi-solusi ini mengandalkan kemampuan platform dan memerlukan perubahan kode sisi klien yang minimal atau tanpa perubahan.

Server-side queuing (direkomendasikan)

Untuk rate limiting akibat lonjakan trafik, Model Studio menerima waktu tunggu maksimum dalam header permintaan. Server mengantrikan dan mencoba ulang permintaan dalam waktu tersebut hingga mulai diproses atau antrian timeout. Ini secara signifikan meningkatkan tingkat keberhasilan selama lonjakan trafik dibandingkan respons 429 langsung.

Tambahkan bidang X-DashScope-Wait-Timeout ke header permintaan:

Setelah mengaktifkan antrian, sesuaikan timeout klien untuk mempertimbangkan waktu tunggu tambahan:

• 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.

Contoh: Jika timeout dasar asli adalah 120 detik dan Wait-Timeout diatur ke 30 detik, timeout permintaan non-streaming harus diatur ke 150 detik.

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)

curl -X POST "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions" \
  -H "Authorization: Bearer $DASHSCOPE_API_KEY" \
  -H "Content-Type: application/json" \
  -H "X-DashScope-Wait-Timeout: 30" \
  -d '{
    "model": "qwen-plus",
    "messages": [{"role": "user", "content": "Hello"}]
  }'

Tingkatkan batas kuota

Jika kuota default tidak mencukupi, tingkatkan kuota rate limit sementara di Konsol Model Studio. Perubahan berlaku segera. Tersedia di wilayah China (Beijing) dan Singapura.

Skenario: Kuota RPM/TPM default tidak mencukupi karena pertumbuhan bisnis, atau diperlukan peningkatan throughput sementara untuk acara jangka pendek. Lihat Batas laju.

Mudah dikonfigurasi. Evaluasi opsi ini sebelum mencoba strategi sisi klien.

Provisioned throughput unit (PTU)

Layanan PTU menyediakan daya komputasi khusus dan berdedikasi. Layanan ini menghindari persaingan di kolam sumber daya publik dan merupakan solusi utama untuk kebutuhan real-time dengan throughput tinggi.

Gunakan PTU ketika bisnis Anda memiliki kebutuhan throughput yang pasti (seperti komitmen SLA) atau ketika Anda menginginkan throughput tinggi yang stabil tanpa pengendalian trafik sisi klien.

PTU adalah sumber daya berlangganan yang ditagih secara terus-menerus, bahkan saat tidak sepenuhnya digunakan. Evaluasi spesifikasi yang diperlukan berdasarkan beban puncak aktual untuk menghindari pemborosan.

Pemrosesan batch asinkron (Batch API)

Untuk tugas yang tidak memiliki persyaratan real-time ketat (pembersihan data, analitik batch), gunakan Batch API untuk mengirimnya ke pemrosesan batch. Tugas dijalankan selama jam sepi, mengembalikan hasil secara asinkron, dan tidak tunduk pada batas laju online.

Cocok untuk tugas offline yang dapat menerima waktu pengembalian hitungan jam hingga hari: anotasi data, analisis log, ringkasan batch. Biaya Batch API biasanya lebih rendah daripada panggilan API real-time.

Waktu pengembalian tidak dijamin. Tidak cocok untuk layanan yang memerlukan respons segera. Ambil hasil melalui polling atau callback setelah pengiriman.

Strategi pengendalian trafik sisi klien

Ketika solusi platform (seperti server-side queuing dan peningkatan kuota) tidak cukup, tambahkan pengendalian trafik di sisi klien. Prinsip intinya: sebarkan permintaan secara merata dalam jendela waktu untuk menghindari lonjakan. Setelah startup sistem atau periode idle panjang, tingkatkan konkurensi secara bertahap.

Empat strategi diurutkan berdasarkan kompleksitas yang meningkat. Setiap strategi mencakup strategi sebelumnya dan menambahkan fitur baru:

• 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.

Pilih strategi dengan biaya terendah yang memenuhi kebutuhan Anda.

Perbandingan performa throughput setiap strategi

Perbandingan throughput di bawah beban berbeda:

• 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

Cocok untuk pengujian pribadi, skrip lokal, dan tugas latar belakang frekuensi rendah. Tidak ada batas laju pada permintaan keluar. Hanya memicu retry exponential backoff dengan jitter acak pada error 429 atau 5xx.

Tidak ada pengendalian trafik proaktif. Di bawah konkurensi multi-threaded, strategi ini mudah memicu rate limiting dan menyebabkan penumpukan permintaan.

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}")

import time
import random
import openai
from openai import OpenAI

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

def chat_with_retry(client, model, messages, max_tokens):
    attempt = 0
    max_retries = 5
    base_delay = 1
    max_delay = 60

    while attempt <= max_retries:
        try:
            return client.chat.completions.create(
                model=model,
                max_tokens=max_tokens,
                messages=messages
            )
        except RETRYABLE_ERRORS as e:
            attempt += 1
            if attempt > max_retries:
                raise e
            backoff = min(max_delay, base_delay * (2 ** (attempt - 1)))
            sleep_time = backoff + random.uniform(0, 1)
            print(f"Triggered {type(e).__name__}, retrying after {sleep_time:.2f}s...")
            time.sleep(sleep_time)

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}")

Kode ini menggunakan exponential backoff, bukan retry interval tetap. Retry interval tetap menyebabkan semua permintaan gagal dikirim ulang secara simultan, memicu rate limiting lagi. Exponential backoff dengan jitter acak menyebarkan retry:

• 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).

Sistem pulih secara bertahap alih-alih terjebak dalam loop "gagal — retry serentak — gagal lagi".

Strategi pembatasan laju permintaan

Retry pasif saja tidak dapat menangani trafik nyata. Retry yang sering meningkatkan latensi. Strategi ini memperkenalkan pengendalian trafik aktif: periksa dan kendalikan permintaan sebelum mengirimnya, mengatur trafik tidak teratur menjadi antrian yang menghormati batas RPM. Penghalusan aktif menambahkan penundaan antrian kecil yang dapat diprediksi — jauh lebih kecil daripada biaya loop "error — tunggu — retry". Biaya kecil yang diketahui lebih baik daripada biaya besar yang tidak diketahui.

Cocok untuk chatbot dan layanan request-response ringan lainnya yang sensitif terhadap waktu hingga token pertama.

Antrian aktif sisi klien menggunakan dua tingkat pengendalian:

• 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.

Jalankan kedua tingkat ini secara ketat berurutan: akuisisi token RPM terlebih dahulu, lalu akuisisi semaphore. Slot konkurensi langka — hanya alokasikan ke permintaan yang siap dieksekusi. Membalik urutan menyebabkan blokir kepala antrian di bawah beban tinggi: permintaan memegang slot tetapi tidak memiliki token, semua slot terisi, tidak ada yang dikirim. Prinsip: jangan pegang sumber daya langka saat menunggu.

Kode di bawah menginisialisasi token bucket ke penuh (initial_tokens=rpm_limit), cocok untuk layanan online yang perlu memproses permintaan segera saat startup. Jika bucket penuh memicu rate limiting, turunkan nilai awal (misalnya, initial_tokens=0 untuk "start bucket kosong") untuk meningkatkan secara lebih bertahap.

Strategi ini tidak melacak penggunaan token. Tugas teks panjang masih dapat menghabiskan kuota TPM.

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

import asyncio
import openai
from openai import AsyncOpenAI
from tenacity import retry, wait_random_exponential, stop_after_attempt, retry_if_exception_type

class RateLimitedClient:
    def __init__(
        self,
        api_key: str,
        base_url: str = "https://dashscope.aliyuncs.com/compatible-mode/v1",
        rpm_limit: float = 600.0,
        max_concurrency: int = 20
    ):
        self.client = AsyncOpenAI(api_key=api_key, base_url=base_url)
        # Komponen 1: RPM token bucket (mengontrol volume total)
        self.rpm_bucket = TokenBucket(
            quota_per_minute=rpm_limit,
            initial_tokens=rpm_limit  # Mulai dengan bucket penuh, cocok untuk layanan online ringan
        )
        # Komponen 2: Concurrency semaphore (mengontrol konkurensi instan)
        self.semaphore = asyncio.Semaphore(max_concurrency)

    async def _execute_request(self, model, messages, max_tokens):
        """Menjalankan satu permintaan, melewati pemeriksaan RPM dan batas konkurensi secara berurutan."""
        # 1. Pemeriksaan RPM (akuisisi token terlebih dahulu)
        wait_seconds = self.rpm_bucket.reserve(1.0)
        if wait_seconds > 0:
            await asyncio.sleep(wait_seconds)
        # 2. Pemeriksaan konkurensi (akuisisi semaphore berikutnya)
        async with self.semaphore:
            # 3. Lakukan panggilan API
            return await self.client.chat.completions.create(
                model=model,
                messages=messages,
                max_tokens=max_tokens
            )

    @retry(
        wait=wait_random_exponential(min=1, max=60),
        stop=stop_after_attempt(5),
        retry=retry_if_exception_type((
            openai.RateLimitError,
            openai.InternalServerError,
            openai.APIConnectionError
        ))
    )
    async def chat_with_limit(self, model, messages, max_tokens=1024):
        # Pertimbangan desain: Mengapa retry juga perlu mengakuisisi ulang token?
        # Jawab: Untuk keamanan. Tanpa akuisisi ulang, pulsa trafik dari retry
        # dapat langsung melebihi batas RPM.
        return await self._execute_request(model, messages, max_tokens)

Strategi traffic shaping

Dalam skenario batch yang memerlukan throughput tinggi dan stabil (ingesti RAG real-time, analisis dokumen panjang massal), pembatasan laju permintaan memiliki titik buta TPM. Traffic shaping menambahkan kesadaran sumber daya ganda (RPM & TPM) dan mekanisme shaping yang mengubah trafik lonjakan menjadi aliran lancar.

Penyempurnaan dibandingkan pembatasan laju permintaan:

• 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.

Alternatif: Jika penundaan antrian kecil saat startup dapat diterima, gunakan kembali token bucket standar (atur initial_tokens=0) untuk start aman dengan kompleksitas lebih rendah. Token bucket Python di sini hanya untuk demonstrasi. Di produksi, gunakan pustaka pembatasan laju matang seperti SmoothRateLimiter Guava di Java.

Dalam contoh kode, tunggu penghalusan ditempatkan di dalam kunci konkurensi. Beberapa permintaan mungkin bersaing untuk semaphore secara simultan setelah waktu tunggu mereka berakhir, mengelompokkan kembali trafik di pintu keluar. Penghalusan di dalam kunci sedikit mengurangi efisiensi konkurensi tetapi memastikan interval pengiriman yang tepat.

Pipeline traffic shaping lengkap adalah: Perkirakan token input → Admisi ganda (RPM & TPM) → Kunci konkurensi → Traffic shaping → Kirim → Selesaikan token output.

Penghalusan konservatif mengorbankan sebagian konkurensi puncak. Tidak cocok untuk layanan yang memerlukan latensi sangat rendah.

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)

import time

class SmoothRateLimiter:
    def __init__(self, rate_per_minute: float):
        self._min_interval = 60.0 / rate_per_minute
        self._last_operation = time.monotonic()

    def reserve(self) -> float:
        now = time.monotonic()
        elapsed = now - self._last_operation
        wait_time = max(0.0, self._min_interval - elapsed)
        self._last_operation = now + wait_time
        return wait_time

import asyncio

class TrafficShapingClient:
    def __init__(self):
        self._rpm_bucket = TokenBucket(quota_per_minute=600)
        self._tpm_bucket = TokenBucket(quota_per_minute=1_000_000)
        self._smooth_limiter = SmoothRateLimiter(rate_per_minute=600)
        self._concurrency_semaphore = asyncio.Semaphore(20)

    async def _execute_throttled_request(self, model, prompt, max_tokens, input_tokens):
        # [Langkah 1] Kontrol admisi ganda
        # Periksa RPM dan TPM, ambil waktu tunggu yang lebih lama
        wait_rpm = self._rpm_bucket.reserve(1.0)
        # Pemeriksaan TPM hanya meminta kuota untuk token input
        wait_tpm = self._tpm_bucket.reserve(input_tokens)
        admission_wait = max(wait_rpm, wait_tpm)
        if admission_wait > 0:
            await asyncio.sleep(admission_wait)

        # [Langkah 2] Akuisisi kunci konkurensi
        async with self._concurrency_semaphore:
            # [Langkah 3] Traffic shaping
            # Kunci: Lakukan tunggu penghalusan di dalam kunci
            # Mengorbankan sebagian efisiensi konkurensi demi kontrol tepat atas interval pengiriman
            smooth_wait = self._smooth_limiter.reserve()
            if smooth_wait > 0:
                await asyncio.sleep(smooth_wait)

            # [Langkah 4] Kirim permintaan
            content, actual_usage = await self._send_chat_request(model, prompt, max_tokens)

            # [Langkah 5] Selesaikan token output
            output_tokens = actual_usage.completion_tokens
            if output_tokens > 0:
                self._tpm_bucket.adjust(-output_tokens)
            return content

Strategi pengendalian kemacetan adaptif

Cocok untuk beban kerja skala besar dan dinamis: gerbang API, proxy kompleks, sistem multi-penyewa.

Pembatasan laju permintaan dan traffic shaping adalah strategi kuota statis — bekerja baik di bawah beban stabil dan dapat diprediksi. Tetapi di tingkat gerbang, beban downstream terus berubah (permintaan pendek dicampur dengan inferensi dalam), dan ambang batas platform berfluktuasi secara dinamis. Strategi statis kesulitan menyeimbangkan efisiensi dan stabilitas.

Terinspirasi oleh BBR, strategi ini membangun sistem kontrol loop tertutup berdasarkan EBP (Elastic Bandwidth Probing). Strategi ini menggunakan kuota RPM/TPM sebagai batas atas panduan dan secara dinamis menghitung laju pengiriman optimal dari umpan balik real-time (perubahan latensi, sinyal rate limit).

• 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.

Modifikasi utama dari BBR asli untuk model besar:

• 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.

Batasan:

• 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()

def probe_next_limit(self, current_limit, max_known_capacity):
    """
    Hitung batas konkurensi berikutnya
    Rumus inti: Berikutnya = Maks(Tegangan Pegas, Dorongan Aditif) + Penghalusan Governor
    """
    # 1. Hitung batas fisik (Hukum Little)
    # Batas teoretis = Throughput * Latensi * Faktor Buffer
    dynamic_ceiling = self.metrics.tps * self.metrics.avg_latency * 1.2

    # 2. Logika pegas (Tegangan Pegas)
    # Semakin jauh dari watermark historis tertinggi, semakin besar tegangan (mempercepat); semakin dekat, semakin kecil (memperlambat)
    tension = 1.0 - (current_limit / max_known_capacity)
    spring_target = current_limit * (1.0 + tension * gain)

    # 3. Dorongan aditif
    # Menyelesaikan "Paradoks Zeno": Saat tegangan mendekati 0, paksa tambahkan kenaikan linier kecil
    # untuk memastikan sistem dapat keluar dari maksimum lokal dan terus mengeksplorasi batas.
    linear_target = current_limit + self.min_additive_step

    raw_target = max(spring_target, linear_target)

    # 4. Governor laju anti-lonjakan
    # Membatasi percepatan pertumbuhan konkurensi untuk mencegah perubahan langkah.
    final_limit = self.governor.smooth(raw_target)

    return min(final_limit, dynamic_ceiling)

class CongestionMetrics:
    def update_stats(self, latency, token_count):
        """
        [Sensor] Perbarui metrik statistik secara real time
        Gunakan EMA (exponential moving average) untuk menyaring noise dari permintaan ekor panjang
        """
        alpha = 0.2  # Faktor penghalusan

        # 1. Perkirakan ukuran permintaan tunggal (Ukuran Token)
        self.ema_tokens = (1 - alpha) * self.ema_tokens + alpha * token_count

        # 2. Perkirakan TPT (Time Per Token)
        # Gunakan TPT alih-alih Latensi untuk menghilangkan kesalahan akibat panjang generasi model bahasa besar yang berbeda
        instant_tpt = latency / token_count
        self.ema_tpt = (1 - alpha) * self.ema_tpt + alpha * instant_tpt

    def track_inflight(self, estimated_tokens):
        """
        [Mengisi titik buta] Koreksi keterlambatan "menghitung hanya setelah respons"
        Deduksi kuota sejak permintaan dimulai
        """
        self.inflight_tokens += estimated_tokens

Solusi fallback arsitektural

Ketika konfigurasi platform dan pengendalian trafik sisi klien masih tidak dapat memenuhi persyaratan ketersediaan atau throughput puncak, tambahkan mekanisme fallback di tingkat arsitektur.

Model fallback

Ketika model utama tidak dapat merespons karena rate limiting atau masalah layanan, secara otomatis fallback ke model alternatif dengan kuota yang lebih longgar.

• 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())

Model fallback dapat dikombinasikan dengan strategi pengendalian trafik sisi klien. Misalnya, Anda dapat mengintegrasikan logika fallback ke dalam mekanisme retry dari strategi pembatasan laju permintaan. Ketika retry habis dan rate limiting masih dipicu, beralih ke model fallback.

Peak-load shifting menggunakan message queues (MQ)

Untuk layanan backend yang tidak memerlukan respons segera, perkenalkan middleware pesan (RabbitMQ, Kafka) untuk peak-load shifting. Trafik lonjakan masuk ke MQ terlebih dahulu; konsumen menarik dan memproses pada laju stabil yang sesuai dengan kuota rate limit. Ini memisahkan puncak frontend dari panggilan backend.

Cocok untuk bisnis di mana pengguna menerima hasil asinkron: pemrosesan tiket, moderasi konten, anotasi data batch.

Poin desain utama:

• 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

Contoh kode menggunakan loop single-threaded asyncio Python untuk mendemonstrasikan algoritma inti. Sebelum penggunaan produksi skala besar, pertimbangkan hal berikut.

• 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.