Alibaba Cloud Model Studio：限流應對最佳實務

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,  # 原逾時 120s + 排隊等待 30s
)

response = client.chat.completions.create(
    model="qwen-plus",
    messages=[{"role": "user", "content": "Hello"}],
    extra_headers={
        "X-DashScope-Wait-Timeout": "30"  # 最大排隊等待 30 秒
    }
)
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"}]
  }'

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": "什麼是指數退避重試？"}],
        max_tokens=1024
    )
    print(response.choices[0].message.content)
except Exception as e:
    print(f"請求失敗: {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"觸發 {type(e).__name__}，等待 {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": "什麼是指數退避重試？"}],
        max_tokens=1024
    )
    print(response.choices[0].message.content)
except Exception as e:
    print(f"請求失敗: {e}")

import time

class TokenBucket:
    """
    令牌桶實現，用於控制每分鐘請求數 (RPM)。
    支援預支 (Debt) 機制，以保證高並發下的先進先出 (FIFO) 順序。
    """
    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:
        """
        申請令牌。
        如果令牌不足，返回需要等待的秒數（支援預支）。
        """
        self._refill()

        # 1. 令牌充足：直接扣除
        if self.tokens >= cost:
            self.tokens -= cost
            return 0.0

        # 2. 令牌不足：計算等待時間並預支
        # 為當前請求"預定"了未來的令牌，確保 FIFO 順序
        deficit = cost - self.tokens
        wait_seconds = deficit / self.refill_rate
        self.tokens -= cost
        return wait_seconds

    def _refill(self):
        """根據流逝時間補充令牌"""
        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)
        # 組件 1: RPM 令牌桶 (控制總量)
        self.rpm_bucket = TokenBucket(
            quota_per_minute=rpm_limit,
            initial_tokens=rpm_limit  # 滿桶啟動，適合輕量線上服務
        )
        # 組件 2: 並發訊號量 (控制瞬間並發)
        self.semaphore = asyncio.Semaphore(max_concurrency)

    async def _execute_request(self, model, messages, max_tokens):
        """執行單個請求：依次通過 RPM 檢查和並發限制。"""
        # 1. RPM 檢查 (先拿令牌)
        wait_seconds = self.rpm_bucket.reserve(1.0)
        if wait_seconds > 0:
            await asyncio.sleep(wait_seconds)
        # 2. 並發檢查 (再拿訊號量)
        async with self.semaphore:
            # 3. 發起 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):
        # 設計考量：為什麼重試也要重新拿 Token？
        # 答：為了安全。如果不重新拿，重試帶來的流量脈衝
        # 可能會瞬間突破 RPM 限制
        return await self._execute_request(model, messages, max_tokens)

import time

class TokenBucket:
    """進階令牌桶，支援勻速預熱 (Continuous Warm-up) 機制。"""
    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):
        # [步驟 1] 雙重准入控制 (Parallel Admission)
        # 同時檢查 RPM 和 TPM，取兩者中較長的等待時間
        wait_rpm = self._rpm_bucket.reserve(1.0)
        # TPM 檢查僅針對輸入 Token 申請額度
        wait_tpm = self._tpm_bucket.reserve(input_tokens)
        admission_wait = max(wait_rpm, wait_tpm)
        if admission_wait > 0:
            await asyncio.sleep(admission_wait)

        # [步驟 2] 擷取並發鎖 (Concurrency Lock)
        async with self._concurrency_semaphore:
            # [步驟 3] 流量整形 (Traffic Shaping)
            # 關鍵：在鎖內進行平滑等待
            # 犧牲部分並發效率，換取發送間隔的精準可控
            smooth_wait = self._smooth_limiter.reserve()
            if smooth_wait > 0:
                await asyncio.sleep(smooth_wait)

            # [步驟 4] 發送請求
            content, actual_usage = await self._send_chat_request(model, prompt, max_tokens)

            # [步驟 5] 輸出 Token 結算
            output_tokens = actual_usage.completion_tokens
            if output_tokens > 0:
                self._tpm_bucket.adjust(-output_tokens)
            return content

class ElasticCongestionController:
    async def acquire(self):
        """[准入階段] 請求發起前的檢查"""
        # 1. SSR 慢啟動重啟：若空閑太久，主動衰減上限
        #    防止過時水位導致的突發流量
        if self.is_idle_too_long():
            self.perform_slow_start_restart()

        # 2. 熔斷檢查：若處於 DRAIN (冷卻) 狀態，強制等待
        if self.state == CongestionState.DRAIN:
            await self.wait_for_cooldown()

        # 3. 雙重預算檢查：同時檢查並發槽位和 Token 預算
        await self.wait_for_budget(request_tokens)

    async def release(self, latency, actual_tokens, error):
        """[反饋階段] 請求結束後的決策"""
        if error:
            # [故障響應] 遇限流錯誤 (429/503)：立即排水 + 乘性回退
            self.state = CongestionState.DRAIN
            self.concurrency_limit *= self.backoff_factor  # e.g. 0.7
            return

        # [正常響應] 計算 TPT (Time-Per-Token)
        current_tpt = latency / actual_tokens

        # [擁塞感知] TPT 突增 (產生變慢)：進入 HOLD 觀察
        # 維持並發水平，不退避也不增長
        if current_tpt > self.metrics.ema_tpt * 2.0:
            self.state = CongestionState.HOLD
        else:
            # [穩態探測] 網路健康：執行 EBP 彈性探測
            self.state = CongestionState.PROBING
            self.update_limit_via_ebp()

def probe_next_limit(self, current_limit, max_known_capacity):
    """
    計算下一個並發上限
    核心公式：Next = Max(彈簧張力, 線性推力) + 調速器平滑
    """
    # 1. 計算物理上限 (Little's Law)
    # 理論上限 = 輸送量 * 延遲 * 緩衝因子
    dynamic_ceiling = self.metrics.tps * self.metrics.avg_latency * 1.2

    # 2. 彈簧邏輯 (Spring Tension)
    # 距離歷史最高水位越遠，張力越大（加速）；越近則越小（減速）
    tension = 1.0 - (current_limit / max_known_capacity)
    spring_target = current_limit * (1.0 + tension * gain)

    # 3. 線性推力 (Additive Thrust)
    # 解決"芝諾悖論"：張力趨近於 0 時，強制疊加微小線性增量
    # 確保系統能突破局部極值，持續探索邊界
    linear_target = current_limit + self.min_additive_step

    raw_target = max(spring_target, linear_target)

    # 4. 防突發調速器 (Rate Governor)
    # 限制並發增長的加速度，防止階梯跳變
    final_limit = self.governor.smooth(raw_target)

    return min(final_limit, dynamic_ceiling)

class CongestionMetrics:
    def update_stats(self, latency, token_count):
        """
        [感應器] 即時更新統計指標
        使用 EMA (指數移動平均) 濾除長尾請求的雜訊
        """
        alpha = 0.2  # 平滑因子

        # 1. 估算單請求大小 (Token Size)
        self.ema_tokens = (1 - alpha) * self.ema_tokens + alpha * token_count

        # 2. 估算 TPT (Time Per Token)
        # 用 TPT 代替 Latency，消除 LLM 產生長度不同帶來的誤差
        instant_tpt = latency / token_count
        self.ema_tpt = (1 - alpha) * self.ema_tpt + alpha * instant_tpt

    def track_inflight(self, estimated_tokens):
        """
        [盲區填充] 修正"響應後才計數"的滯後性
        請求發起瞬間，立即預扣額度
        """
        self.inflight_tokens += estimated_tokens