Melhores práticas para lidar com limitação de taxa

• Soluções de configuração da plataforma (alterações mínimas de código): Enfileiramento no servidor, Aumento dos limites de cota, PTU e Batch API.
• Estratégias de controle de tráfego no cliente (alterações no código do cliente): Quatro estratégias com complexidade de engenharia crescente, desde nova tentativa básica até controle de congestionamento adaptativo.
• Soluções de fallback arquitetônico (alterações na arquitetura do sistema): Fallback de modelo e Deslocamento de pico de carga usando filas de mensagens (MQ).

​

Mecanismo de limitação de taxa da plataforma

• Limites de cota por minuto (RPM / TPM): Número máximo de requisições por minuto (RPM) e uso máximo de tokens por minuto (TPM).
• Limites de frequência instantânea (RPS / TPS): Máximo de requisições por segundo (RPS) e máximo de uso de tokens por segundo (TPS). Chamadas densas à API ou consumo de tokens dentro de um único segundo podem acionar a limitação de taxa.
• Limites de taxa de crescimento (Pico de Tráfego): Um aumento repentino no volume de requisições ou no uso de tokens aciona a limitação de taxa. O limiar se ajusta dinamicamente. Aumente as requisições gradualmente para evitar o acionamento desse limite.

​

Diagnóstico de erros e recomendações de estratégia

​

Soluções de configuração da plataforma

​

Enfileiramento no servidor (recomendado)

• Requisições sem streaming (stream: false): Tempo limite = tempo limite base original + valor de Wait-Timeout.
• Requisições com streaming (stream: true): Tempo limite > valor de Wait-Timeout. Requisições com streaming iniciam a contagem de tempo após o primeiro chunk, portanto, o tempo limite de resposta inicial precisa apenas exceder o tempo de enfileiramento.

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)

​

Aumento dos limites de cota

​

Unidade de throughput provisionada (PTU)

​

Processamento em lote assíncrono (Batch API)

​

Estratégias de controle de tráfego no cliente

• Nova tentativa básica oferece apenas defesa passiva.
• Limitação de taxa de requisições adiciona enfileiramento ativo.
• Modelagem de tráfego introduz ainda controle no nível de token e envio suave.
• Controle de congestionamento adaptativo ajusta dinamicamente a taxa de envio com base em feedback em tempo real.

​

Comparação de desempenho de throughput de cada estratégia

• Estratégia de nova tentativa básica: Eficaz sob baixa carga. Propensa a colapso congestivo sob alta concorrência, causando uma queda acentuada no throughput.
• Estratégia de limitação de taxa de requisições: Forte proteção contra colapso. No entanto, sob cargas mistas com textos longos, o throughput apresenta flutuações semelhantes a dentes de serra devido à falta de controle de tokens.
• Estratégia de modelagem de tráfego: Alta estabilidade. Alcança uma saída suave sacrificando parte do throughput de pico.
• Estratégia de controle de congestionamento adaptativo: Converge dinamicamente para um ponto de throughput alto e estável sob carga elevada, mas possui sobrecarga de sondagem na inicialização a frio.

​

Estratégia de nova tentativa básica

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

• Tempo de espera dobra progressivamente: Por exemplo, 1s, 2s, 4s.... Isso evita requisições repetidas em um curto período.
• Adição de variação aleatória: Um valor aleatório (como 2s +/- 0.5s) espalha o tráfego de novas tentativas, prevenindo uma inundação secundária (efeito manada).

​

Estratégia de limitação de taxa de requisições

• Token bucket de RPM: Limita o total de requisições por minuto. A capacidade do bucket é igual à cota de RPM; os tokens são reabastecidos a uma taxa constante. Suporta empréstimo: se os tokens forem insuficientes, uma requisição toma emprestado da cota futura, estritamente em ordem FIFO.
• Semáforo de concorrência: Limita requisições concorrentes para evitar que alta concorrência instantânea acione limites de RPS.

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

​

Estratégia de modelagem de tráfego

• Controle duplo de recursos (RPM & TPM): Mantém buckets de tokens tanto para RPM quanto para TPM. Todas as requisições devem passar pelas verificações de cota de ambas as dimensões antes do envio.
• Pré-dedução para entrada, liquidação posterior para saída: O comprimento da saída é desconhecido antes da requisição. O bucket de TPM pré-deduz tokens de entrada no envio e liquida os tokens reais de saída após a conclusão. Mesmo que os tokens fiquem negativos, requisições subsequentes esperam até que a contagem se torne positiva, suavizando naturalmente o fluxo.
• Warm-up contínuo: Durante uma inicialização a frio, a taxa de emissão de tokens aumenta linearmente ao longo do tempo, eliminando o risco de picos iniciais.
• Limitador de taxa com suavização (Pacing): Suaviza a taxa de envio impondo um intervalo mínimo entre requisições (pacing), reduzindo o risco de acionar limites de taxa.

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)

​

Estratégia de controle de congestionamento adaptativo

• EBP: Armazena a marca d'água histórica de maior sucesso. Calcula o ganho de sondagem simulando a tensão de uma mola com base na distância da concorrência atual até a marca d'água (mais longe = mais rápido, mais perto = mais lento). Adiciona um pequeno impulso linear para continuar explorando mesmo em alta saturação.
• Consciência de congestionamento TPT: O tempo de geração de LLM escala com o comprimento da saída — alta latência em textos longos não significa congestionamento. TPT (Tempo Por Token) filtra o ruído de comprimento de conteúdo. O congestionamento só é declarado quando o TPT degrada significativamente.
• Governador de taxa anti-pico: Independentemente do alvo EBP, o governador limita a aceleração do crescimento da concorrência para garantir uma rampa suave e evitar mudanças abruptas que acionem limites de taxa de crescimento.

• Sondagem guiada: Introduz cotas conhecidas de RPM/TPM como um "limite superior orientador" para evitar colisões frequentes causadas por sondagem cega.
• Fonte de sinal (RTT → TPT): O BBR nativo usa RTT. Em cenários de LLM, a latência de comprimento de conteúdo ofusca o jitter de rede. O TPT elimina essa interferência.
• Aprimoramento do mecanismo de resposta (ProbeRTT → Hold): Diante de flutuações de latência, opta por manter o nível atual de concorrência em vez de recuar proativamente e reduzir o throughput.
• Resposta a limite rígido de taxa (Packet Loss → 429 Drain): Uma vez que um erro 429 é acionado, entra em um estado agressivo de Drenagem e realiza uma recuperação rápida após um período de resfriamento.

• Ruído no TPT: O TPT é estimado como "latência total / total de tokens". A latência total inclui ida e volta de rede, enfileiramento e tempo até o primeiro token — suscetível a jitter ou entradas longas, o que pode acionar falsamente o estado Hold.
• Inanição de requisições grandes: Usa um despertar FIFO não estrito para desempenho de agendamento. Quando as cotas são escassas, requisições de poucos tokens podem preemptar recursos, fazendo com que requisições de muitos tokens esperem demais.
• Inicialização a frio: Requer um período de aquecimento para construir um modelo estatístico. Em tarefas de baixa carga ou curta duração, o throughput pode ser menor que nas três primeiras estratégias.

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

​

Soluções de fallback arquitetônico

​

Fallback de modelo

• Escolha modelos de séries diferentes: A limitação de taxa é por modelo. Use um modelo diferente como fallback — por exemplo, faça fallback de qwen3.6-plus para qwen3.6-flash.
• Acione o fallback apenas em erros de limite de taxa: Faça fallback em erros 429, não em todas as exceções. Trocar modelos não corrigirá tempos limite de rede ou erros de parâmetro.
• Valide o modelo de fallback antecipadamente: Garanta que ele suporte os recursos necessários (Function Calling, saída estruturada, etc.) para evitar problemas funcionais após o 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())

​

Deslocamento de pico de carga usando filas de mensagens (MQ)

• Controle de taxa do consumidor: O lado do consumidor deve usar a estratégia de limitação de taxa de requisições ou modelagem de tráfego para consumir mensagens a uma taxa constante com base na cota de RPM/TPM, em vez de puxar mensagens sem limites.
• Tratamento de mensagens mortas: Mova mensagens que falham após múltiplas tentativas para uma fila de mensagens mortas e acione um alerta. Evite que tentativas infinitas bloqueiem o consumo.
• Propagação de contrapressão: Quando o backlog do MQ excede um limiar, propague a pressão para montante (por exemplo, retorne um status de enfileiramento) para evitar crescimento ilimitado da fila.

​

Considerações para ambiente de produção

• Adaptação para modelos não textuais As estratégias acima usam modelos de texto como exemplos, mas os princípios centrais aplicam-se a serviços multimodais (geração de imagens, síntese de fala). As unidades diferem, mas a essência é a mesma: limitar a taxa de envio e a capacidade de processamento.
  • Modelos como reconhecimento de fala são tipicamente restringidos tanto pelo número de requisições por unidade de tempo (como RPM) quanto pelo uso (como duração do áudio). As estratégias são basicamente as mesmas para modelos de texto.
  • Modelos para imagens e vídeos são tipicamente restringidos pela taxa de envio de tarefas e pelo número de tarefas concorrentes. Use a mesma abordagem da estratégia de limitação de taxa de requisições: limite a taxa de envio de tarefas e use um semáforo para controlar a concorrência.
  Independentemente de como as métricas mudam, o princípio de limitação no lado do cliente permanece o mesmo. Substitua o contador ou métrica de sondagem pelo da sua modalidade. Para regras específicas, consulte Limitação de taxa.
• Atomicidade em modelos concorrentes Exemplo: asyncio usa agendamento cooperativo de thread única, então modificações de estado são inerentemente atômicas dentro de um único processo. Produção: Em ambientes multithread ou multiprocesso, garanta a segurança de concorrência do token bucket e da janela de estatísticas. Condições de corrida quebrarão o controle de tráfego.
• Limitação de taxa distribuída Exemplo: Todos os componentes de controle de tráfego estão em memória. Produção: Em implantações de múltiplas instâncias, cada instância limita independentemente. O uso total pode exceder o limite. Use um contador centralizado (como Redis) para gerenciar o uso em todos os nós.
• Filas prioritárias e prevenção de inanição Exemplo: Sem diferenciação de prioridade. A estratégia de controle de congestionamento adaptativo usa despertar FIFO não estrito para desempenho de agendamento. Produção: Para requisições de alta/baixa prioridade, implemente uma fila prioritária ponderada para garantir largura de banda para tráfego de alta prioridade. Reserve uma cota mínima para filas de baixa prioridade para prevenir inanição.