Best practices for handling rate limiting

• Platform configuration solutions (minimal code changes): Server-side queuing, Increase quota limits, PTU, and Batch API.
• Client-side traffic control strategies (client code changes): Four strategies with increasing engineering complexity, from basic retry to adaptive congestion control.
• Architectural fallback solutions (system architecture changes): Model fallback and peak-load shifting using message queues (MQ).

​

Platform rate limiting mechanism

• Minute-level quota limits (RPM / TPM): The maximum number of requests per minute (RPM) and the maximum token usage per minute (TPM).
• Instantaneous frequency limits (RPS / TPS): Maximum requests per second (RPS) and maximum token usage per second (TPS). Dense API calls or token consumption within a single second can trigger rate limiting.
• Growth rate limits (Traffic Burst): A sudden surge in request volume or token usage triggers rate limiting. The threshold adjusts dynamically. Ramp up requests gradually to avoid triggering this limit.

​

Error diagnosis and strategy recommendations

​

Platform configuration solutions

​

Server-side queuing (recommended)

• Non-streaming requests (stream: false): Timeout = original base timeout + Wait-Timeout value.
• Streaming requests (stream: true): Timeout > Wait-Timeout value. Streaming requests start timing after the first chunk, so the initial response timeout just needs to exceed the queuing time.

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,  # Original timeout 120s + queuing wait 30s
)

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

​

Increase quota limits

​

Provisioned throughput unit (PTU)

​

Asynchronous batch processing (Batch API)

​

Client-side traffic control strategies

• Basic retry provides only passive defense.
• Request rate limiting adds active queuing.
• Traffic shaping further introduces token-level control and smooth sending.
• Adaptive congestion control dynamically adjusts the sending rate based on real-time feedback.

​

Throughput performance comparison of each strategy

• Basic retry strategy: Effective under low load. Prone to congestive collapse under high concurrency, causing a sharp drop in throughput.
• Request rate limiting strategy: Strong protection against collapse. However, under mixed workloads with long texts, throughput shows sawtooth-like fluctuations due to the lack of token control.
• Traffic shaping strategy: High stability. Achieves smooth output by sacrificing some peak throughput.
• Adaptive congestion control strategy: Can dynamically converge to a stable, high throughput point under high load, but has cold-start probing overhead.

​

Basic retry strategy

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

• Wait time doubles progressively: For example, 1s, 2s, 4s.... This avoids repeated requests in a short period.
• Add random jitter: A random value (such as 2s +/- 0.5s) spreads retry traffic, preventing a secondary flood (thundering herd effect).

​

Request rate limiting strategy

• RPM token bucket: Limits total requests per minute. Bucket capacity equals the RPM quota; tokens refill at a constant rate. Supports borrowing: if tokens are insufficient, a request borrows from future quota, strictly FIFO.
• Concurrency semaphore: Limits concurrent requests to prevent instantaneous high concurrency from triggering RPS limits.

import time

class TokenBucket:
    """
    Token bucket implementation to control requests per minute (RPM).
    Supports a debt mechanism to ensure first-in, first-out (FIFO) order under high concurrency.
    """
    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:
        """
        Acquires a token.
        If tokens are insufficient, returns the number of seconds to wait (supports debt).
        """
        self._refill()

        # 1. Sufficient tokens: Deduct directly
        if self.tokens >= cost:
            self.tokens -= cost
            return 0.0

        # 2. Insufficient tokens: Calculate wait time and incur debt
        # "Reserves" future tokens for the current request to ensure FIFO order
        deficit = cost - self.tokens
        wait_seconds = deficit / self.refill_rate
        self.tokens -= cost
        return wait_seconds

    def _refill(self):
        """Refills tokens based on elapsed time."""
        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

​

Traffic shaping strategy

• Dual resource control (RPM & TPM): Maintains both RPM and TPM token buckets. All requests must pass quota checks for both dimensions before being sent.
• Pre-deduction for input, post-settlement for output: Output length is unknown before the request. The TPM bucket pre-deducts input tokens when sending and settles actual output tokens after completion. Even if tokens go negative, subsequent requests wait for the count to become positive, naturally smoothing the flow.
• Continuous warm-up: During a cold start, the token issuance rate increases linearly over time, eliminating the risk of initial bursts.
• Smoothing rate limiter (Pacing): Smooths the sending rate by enforcing a minimum interval between requests (pacing), reducing the risk of triggering rate limits.

import time

class TokenBucket:
    """Advanced token bucket that supports a continuous warm-up mechanism."""
    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)

​

Adaptive congestion control strategy

• EBP: Stores the historical highest successful watermark. Calculates probing gain by simulating spring tension based on the distance from current concurrency to the watermark (farther = faster, closer = slower). Adds a small linear thrust to keep exploring even at high saturation.
• TPT congestion awareness: LLM generation time scales with output length — high latency on long texts does not mean congestion. TPT (Time Per Token) filters out content-length noise. Congestion is only declared when TPT degrades significantly.
• Anti-burst rate governor: Regardless of the EBP target, the governor caps concurrency growth acceleration to ensure smooth ramp-up and avoid step changes that trigger growth rate limits.

• Guided probing: Introduces known RPM/TPM quotas as a "guiding upper limit" to avoid frequent collisions caused by blind probing.
• Signal source (RTT → TPT): Native BBR uses RTT. In LLM scenarios, content-length latency dwarfs network jitter. TPT eliminates that interference.
• Response mechanism enhancement (ProbeRTT → Hold): In the face of latency fluctuations, it chooses to maintain the current concurrency level rather than proactively backing off and reducing throughput.
• Hard rate limit response (Packet Loss → 429 Drain): Once a 429 error is triggered, it enters an aggressive Drain state and performs a fast recovery after a cooldown period.

• TPT noise: TPT is estimated as "total latency / total tokens". Total latency includes network round-trip, queuing, and time to first token — susceptible to jitter or long inputs, which can falsely trigger the Hold state.
• Large request starvation: Uses a non-strict FIFO wakeup for scheduling performance. When quotas are scarce, short-token requests may preempt resources, causing long-token requests to wait too long.
• Cold start: Requires a warm-up period to build a statistical model. In low-load or short-lived tasks, throughput may be lower than the first three strategies.

class ElasticCongestionController:
    async def acquire(self):
        """[Admission phase] Check before initiating a request"""
        # 1. SSR slow-start restart: If idle for too long, proactively decay the limit
        #    to prevent burst traffic caused by an outdated watermark.
        if self.is_idle_too_long():
            self.perform_slow_start_restart()

        # 2. Circuit breaker check: If in DRAIN (cooldown) state, force wait.
        if self.state == CongestionState.DRAIN:
            await self.wait_for_cooldown()

        # 3. Dual budget check: Check both concurrency slots and token budget.
        await self.wait_for_budget(request_tokens)

    async def release(self, latency, actual_tokens, error):
        """[Feedback phase] Decision after the request finishes"""
        if error:
            # [Fault response] On rate limit error (429/503): Immediately drain + multiplicative backoff
            self.state = CongestionState.DRAIN
            self.concurrency_limit *= self.backoff_factor  # e.g. 0.7
            return

        # [Normal response] Calculate TPT (Time-Per-Token)
        current_tpt = latency / actual_tokens

        # [Congestion awareness] TPT spike (generation slows down): Enter HOLD to observe
        # Maintain concurrency level, neither backing off nor increasing
        if current_tpt > self.metrics.ema_tpt * 2.0:
            self.state = CongestionState.HOLD
        else:
            # [Steady-state probing] Network is healthy: Perform EBP elastic probing
            self.state = CongestionState.PROBING
            self.update_limit_via_ebp()

​

Architectural fallback solutions

​

Model fallback

• Choose models from different series: Rate limiting is per-model. Use a different model as fallback — for example, fall back from qwen3.6-plus to qwen3.6-flash.
• Trigger fallback only on rate limit errors: Fall back on 429 errors, not all exceptions. Switching models will not fix network timeouts or parameter errors.
• Validate the fallback model in advance: Ensure it supports required features (Function Calling, structured output, etc.) to avoid functional issues after fallback.

import os
import asyncio
from openai import AsyncOpenAI, APIStatusError

# Primary and fallback models (different series, independent quotas)
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:
    """Request with fallback: Automatically switches to the fallback model when the primary model is rate-limited."""
    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 Triggered] {model}, falling back to {FALLBACK_MODEL}")
                continue
            raise
    raise RuntimeError("All models are unavailable")

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

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

​

Peak-load shifting using message queues (MQ)

• Consumer rate control: The consumer side should use the request rate limiting or traffic shaping strategy to consume messages at a steady rate based on the RPM/TPM quota, rather than pulling messages without limits.
• Dead-letter handling: Move messages that fail after multiple retries to a dead-letter queue and trigger an alert. Prevent infinite retries from blocking consumption.
• Back-pressure propagation: When the MQ backlog exceeds a threshold, propagate pressure upstream (for example, return a queuing status) to prevent unbounded queue growth.

​

Production environment considerations

• Adapting to non-text models The strategies above use text models as examples, but the core principles apply to multimodal services (image generation, speech synthesis). The units differ, but the essence is the same: limiting submission rate and processing capacity.
  • Models such as speech recognition are typically constrained by both the number of requests per unit of time (such as RPM) and usage (such as audio duration). The strategies are basically the same as for text models.
  • Models for images and videos are typically constrained by the task submission rate and the number of concurrent tasks. You can use the same approach as the request rate limiting strategy: limit the task submission rate and use a semaphore to control concurrency.
  Regardless of how metrics change, the principle of client-side throttling stays the same. Replace the counter or probing metric with the one for your modality. For specific rules, see Rate limiting.
• Atomicity in concurrent models Example: asyncio uses single-threaded cooperative scheduling, so state modifications are inherently atomic within a single process. Production: In multi-threaded or multi-process environments, ensure concurrency safety of the token bucket and statistics window. Race conditions will break traffic control.
• Distributed rate limiting Example: All traffic control components are in-memory. Production: In multi-instance deployments, each instance throttles independently. Total usage may exceed the limit. Use a centralized counter (such as Redis) to manage usage across all nodes.
• Priority queues and starvation prevention Example: No priority differentiation. The adaptive congestion control strategy uses non-strict FIFO wakeup for scheduling performance. Production: For high/low priority requests, implement a weighted priority queue to guarantee bandwidth for high-priority traffic. Reserve a minimum quota for low-priority queues to prevent starvation.