5. Rate Limiting

Descripción

Rate limiting client-side es la diferencia entre un spike de tráfico que degrada gradualmente tu servicio y uno que lo derriba completamente. Sin rate limiting, 200 usuarios concurrentes hacen 200 requests a OpenAI simultáneamente — exceden el límite de RPM, todos reciben 429, y el retry empeora el problema. Con rate limiting, tus requests se distribuyen en el tiempo, respetando los límites de la API y el budget que tú establezcas. En esta cápsula vas a implementar un token bucket thread-safe, wrapearlo como provider, gestionar límites por modelo, y agregar control de presupuesto diario.

Por qué rate limiting client-side (no solo confiar en los límites del API)

Sin rate limiting client-side:

t=0s:   200 requests llegan simultáneamente a tu app
        → Tu app manda 200 requests a OpenAI
        → OpenAI: límite es 100/min → rechaza 100 con 429
        → Los 100 hacen retry → empeoran el problema
        → Resultado: experiencia degradada, budget quemado, errores visibles

Con rate limiting client-side:

t=0s:   200 requests llegan a tu app
        → Rate limiter: capacidad de 8/s
        → Procesa 8 en t=0s, pone los demás en cola
        → Procesa 8 más en t=1s, 8 en t=2s...
        → Los 200 se procesan en ~25 segundos
        → OpenAI nunca ve más de 8/s → nunca rechaza nada
        → Resultado: todos completan, costo controlado, sin errores 429

Beneficio adicional: budget control
        → Con rate limiting: puedes limitar 10 requests/usuario/día
        → Sin rate limiting: un usuario puede gastar todo tu budget en 1 minuto

Algoritmo Token Bucket explicado

# El token bucket es el algoritmo más común para rate limiting

# Concepto:
#   - Un "bucket" que se llena de tokens a una tasa constante
#   - Cada request consume tokens
#   - Si no hay tokens, el request espera o es rechazado

# Propiedades:
#   - rate: tokens que se añaden por segundo
#   - capacity: tamaño máximo del bucket (burst máximo)
#   - Un bucket lleno = capacidad para un burst
#   - Un bucket vacío = throttling estricto

# Ejemplo: rate=5, capacity=10
#
# t=0s:  bucket=10 (lleno). Usuario A hace 10 requests → bucket=0
# t=0s:  Usuario B hace 1 request → ESPERA (bucket vacío)
# t=1s:  bucket=5 (se llenó con rate=5). Procesa 5 requests pendientes
# t=2s:  bucket=5. Procesa más requests
# ...
# Resultado: el burst inicial de 10 se permite, luego se throttlea a 5/s

# Esto es mejor que rechazar todos: permite bursts cortos pero controla sostenido

TokenBucket thread-safe

# src/infrastructure/rate_limiter.py
import time
import threading
from dataclasses import dataclass, field
from typing import Optional
import structlog

log = structlog.get_logger()

class TokenBucket:
    """
    Token bucket rate limiter, thread-safe.
    
    Permite bursts cortos hasta 'capacity' tokens,
    luego throttlea a 'rate' tokens por segundo.
    """
    
    def __init__(self, rate: float, capacity: int, name: str = "default"):
        """
        rate: tokens por segundo (ej: 8.3 para 500 RPM / 60)
        capacity: tamaño del bucket = burst máximo permitido
        name: identificador para logs
        """
        self.rate = rate
        self.capacity = capacity
        self.name = name
        self._tokens = float(capacity)  # Empieza lleno (permite burst inicial)
        self._last_refill = time.monotonic()
        self._lock = threading.Lock()
        
        # Métricas
        self._total_consumed = 0
        self._total_rejected = 0
        self._total_waited_seconds = 0.0
    
    def _refill(self) -> None:
        """
        Refill el bucket basado en el tiempo transcurrido.
        Debe llamarse dentro del lock.
        """
        now = time.monotonic()
        elapsed = now - self._last_refill
        new_tokens = elapsed * self.rate
        self._tokens = min(self.capacity, self._tokens + new_tokens)
        self._last_refill = now
    
    def consume(self, tokens: int = 1) -> bool:
        """
        Intenta consumir tokens del bucket.
        
        Returns:
            True si se pudo consumir (request permitido)
            False si no hay suficientes tokens (request rechazado)
        """
        with self._lock:
            self._refill()
            if self._tokens >= tokens:
                self._tokens -= tokens
                self._total_consumed += tokens
                return True
            else:
                self._total_rejected += 1
                return False
    
    def consume_or_wait(self, tokens: int = 1, max_wait_seconds: float = 30.0) -> bool:
        """
        Consume tokens, esperando si el bucket está vacío.
        
        Args:
            tokens: tokens a consumir
            max_wait_seconds: máximo tiempo a esperar
        
        Returns:
            True si se consumió (puede haber esperado)
            False si el tiempo de espera excede max_wait_seconds
        """
        deadline = time.monotonic() + max_wait_seconds
        
        while True:
            with self._lock:
                self._refill()
                if self._tokens >= tokens:
                    self._tokens -= tokens
                    self._total_consumed += tokens
                    return True
                
                # Calcular cuánto hay que esperar
                tokens_needed = tokens - self._tokens
                wait_time = tokens_needed / self.rate
            
            if time.monotonic() + wait_time > deadline:
                self._total_rejected += 1
                log.warning(
                    "rate_limit_wait_exceeded",
                    bucket=self.name,
                    wait_time=round(wait_time, 2),
                    max_wait=max_wait_seconds
                )
                return False
            
            self._total_waited_seconds += wait_time
            time.sleep(wait_time)
    
    @property
    def available_tokens(self) -> float:
        """Tokens disponibles actualmente (aproximado)."""
        with self._lock:
            self._refill()
            return self._tokens
    
    def get_metrics(self) -> dict:
        """Métricas para monitoring."""
        return {
            "bucket": self.name,
            "rate_per_second": self.rate,
            "capacity": self.capacity,
            "available_tokens": round(self.available_tokens, 2),
            "total_consumed": self._total_consumed,
            "total_rejected": self._total_rejected,
            "total_waited_seconds": round(self._total_waited_seconds, 2),
            "utilization_percent": round(
                (1 - self.available_tokens / self.capacity) * 100, 1
            )
        }

RateLimitedProvider

# src/infrastructure/rate_limited_provider.py
import time
import structlog
from src.infrastructure.llm_provider import LLMProvider, LLMProviderError
from src.infrastructure.rate_limiter import TokenBucket
from src.infrastructure.error_classifier import ErrorCategory

log = structlog.get_logger()

class RateLimitError(LLMProviderError):
    """El request fue rechazado por el rate limiter client-side."""
    def __init__(self, bucket_name: str, max_wait: float):
        super().__init__(
            message=f"Rate limit exceeded (client-side). Try again in a few seconds.",
            category=ErrorCategory.TRANSIENT,
            should_retry=True,
            retry_after=max_wait / 2  # Sugerir esperar la mitad del max_wait
        )
        self.bucket_name = bucket_name

class RateLimitedProvider:
    """
    Wrapper que aplica rate limiting client-side a cualquier LLMProvider.
    
    Previene exceder los límites de RPM/TPM del API,
    controlando la velocidad de requests salientes.
    """
    
    def __init__(
        self,
        inner: LLMProvider,
        requests_per_minute: int = 60,
        burst_size: int = None,
        max_wait_seconds: float = 30.0,
        model: str = "default"
    ):
        """
        requests_per_minute: límite de requests por minuto
        burst_size: tamaño del burst (default: 20% del RPM)
        max_wait_seconds: máximo tiempo que un request espera en cola
        model: nombre del modelo (para logs)
        """
        self._inner = inner
        self._max_wait = max_wait_seconds
        self._model = model
        
        rate_per_second = requests_per_minute / 60.0
        burst = burst_size or max(1, int(requests_per_minute * 0.2))
        
        self._bucket = TokenBucket(
            rate=rate_per_second,
            capacity=burst,
            name=f"rate_limit_{model}"
        )
        
        log.info(
            "rate_limited_provider_initialized",
            model=model,
            requests_per_minute=requests_per_minute,
            rate_per_second=round(rate_per_second, 2),
            burst_size=burst
        )
    
    def complete(self, messages: list[dict], **kwargs) -> str:
        """
        Realiza la llamada respetando el rate limit.
        Si el bucket está vacío, espera hasta max_wait_seconds.
        Si no puede obtener tokens en ese tiempo, rechaza el request.
        """
        acquired = self._bucket.consume_or_wait(
            tokens=1,
            max_wait_seconds=self._max_wait
        )
        
        if not acquired:
            metrics = self._bucket.get_metrics()
            log.warning(
                "client_rate_limit_rejected",
                model=self._model,
                available_tokens=metrics["available_tokens"],
                utilization=metrics["utilization_percent"]
            )
            raise RateLimitError(self._bucket.name, self._max_wait)
        
        return self._inner.complete(messages, **kwargs)
    
    def get_metrics(self) -> dict:
        return self._bucket.get_metrics()

Rate limiting por modelo (límites específicos)

# src/infrastructure/multi_model_rate_limiter.py
from src.infrastructure.rate_limiter import TokenBucket

# Límites reales de OpenAI (Tier 1, aproximados — verificar siempre los actuales):
# https://platform.openai.com/docs/guides/rate-limits

OPENAI_RATE_LIMITS = {
    "gpt-4o": {
        "rpm": 500,      # Requests Per Minute
        "tpm": 30_000,   # Tokens Per Minute
        "burst": 20,
    },
    "gpt-4o-mini": {
        "rpm": 500,
        "tpm": 200_000,
        "burst": 50,
    },
    "gpt-4": {
        "rpm": 500,
        "tpm": 10_000,
        "burst": 10,
    },
    "gpt-3.5-turbo": {
        "rpm": 3500,
        "tpm": 90_000,
        "burst": 100,
    },
}

class ModelRateLimiter:
    """
    Rate limiter que gestiona límites separados por modelo.
    
    Usa el 80% de los límites para tener margen de seguridad.
    """
    
    SAFETY_MARGIN = 0.8  # Usar solo el 80% del límite
    
    def __init__(self, custom_limits: dict = None):
        limits = {**OPENAI_RATE_LIMITS, **(custom_limits or {})}
        
        self._buckets = {
            model: TokenBucket(
                rate=config["rpm"] / 60.0 * self.SAFETY_MARGIN,
                capacity=config["burst"],
                name=f"rate_{model}"
            )
            for model, config in limits.items()
        }
    
    def acquire(self, model: str, max_wait: float = 30.0) -> bool:
        """
        Intenta adquirir un token para el modelo especificado.
        Espera hasta max_wait segundos.
        """
        bucket = self._buckets.get(model)
        if bucket is None:
            # Modelo desconocido — usar límite conservador
            return True  # O crear un bucket por defecto
        return bucket.consume_or_wait(max_wait_seconds=max_wait)
    
    def get_all_metrics(self) -> dict:
        return {
            model: bucket.get_metrics()
            for model, bucket in self._buckets.items()
        }

# Instancia global (singleton)
model_rate_limiter = ModelRateLimiter()

Budget limiting: controlar el costo total

# src/infrastructure/budget_limiter.py
import threading
from datetime import datetime, date
import structlog

log = structlog.get_logger()

class DailyBudgetExceeded(Exception):
    """El presupuesto diario de la API ha sido excedido."""
    def __init__(self, spent: float, limit: float):
        self.spent = spent
        self.limit = limit
        super().__init__(
            f"Daily budget exceeded: ${spent:.4f} spent of ${limit:.2f} limit"
        )

class DailyBudgetLimiter:
    """
    Controla el gasto diario en APIs de LLM.
    
    Se reinicia automáticamente cada día a medianoche.
    Thread-safe para apps con concurrencia.
    """
    
    def __init__(self, daily_limit_usd: float, warning_threshold: float = 0.8):
        self._limit = daily_limit_usd
        self._warning_threshold = warning_threshold
        self._spent_today = 0.0
        self._current_date = date.today()
        self._lock = threading.Lock()
        self._warning_sent = False
    
    def _reset_if_new_day(self) -> None:
        today = date.today()
        if today != self._current_date:
            log.info(
                "budget_limiter_daily_reset",
                previous_date=str(self._current_date),
                spent_yesterday=round(self._spent_today, 4)
            )
            self._current_date = today
            self._spent_today = 0.0
            self._warning_sent = False
    
    def check_and_record(self, estimated_cost_usd: float) -> None:
        """
        Verifica que el request cabe dentro del budget y lo registra.
        
        Lanza DailyBudgetExceeded si el budget se excedería.
        """
        with self._lock:
            self._reset_if_new_day()
            
            if self._spent_today + estimated_cost_usd > self._limit:
                raise DailyBudgetExceeded(self._spent_today, self._limit)
            
            self._spent_today += estimated_cost_usd
            
            # Warning si nos acercamos al límite
            utilization = self._spent_today / self._limit
            if utilization >= self._warning_threshold and not self._warning_sent:
                log.warning(
                    "daily_budget_warning",
                    spent=round(self._spent_today, 4),
                    limit=self._limit,
                    utilization_percent=round(utilization * 100, 1)
                )
                self._warning_sent = True
    
    @property
    def remaining_budget(self) -> float:
        with self._lock:
            self._reset_if_new_day()
            return max(0, self._limit - self._spent_today)
    
    def get_status(self) -> dict:
        with self._lock:
            self._reset_if_new_day()
            return {
                "limit_usd": self._limit,
                "spent_today_usd": round(self._spent_today, 4),
                "remaining_usd": round(max(0, self._limit - self._spent_today), 4),
                "utilization_percent": round((self._spent_today / self._limit) * 100, 1),
                "date": str(self._current_date)
            }

Ejercicios

Ejercicio 1: Calcular el rate para gpt-4o

OpenAI Tier 1 permite 500 RPM para gpt-4o. ¿Qué rate y capacity usarías en el TokenBucket?

# Límite: 500 RPM
# Con 80% de safety margin: 500 × 0.8 = 400 RPM
# Rate en tokens/segundo: 400 / 60 = 6.67 tokens/s
# Capacity (burst): 20% de 400 = 80, pero limitar razonablemente
# Un burst de 20-30 es práctico para la mayoría de apps

bucket = TokenBucket(
    rate=400 / 60,  # 6.67 tokens/segundo
    capacity=20,     # Burst de 20 requests
    name="gpt_4o"
)

Ejercicio 2: Queue vs Reject

Para cada caso de uso, ¿debería el rate limiter hacer esperar al request (queue) o rechazarlo inmediatamente (reject)?

1. API pública donde el usuario puede esperar 10-15 segundos
2. Webhook de Slack que debe responder en <3 segundos
3. Proceso batch nocturno sin usuario activo esperando

Ejercicio 3: Calcular budget diario

Tu app procesa en promedio 1,000 requests/día con gpt-4o. Cada request usa ~500 input tokens y ~200 output tokens. Con los precios de OpenAI ($5/1M input, $15/1M output), ¿cuál debería ser tu daily_budget_limit_usd?

# Costo por request:
input_cost = 500 / 1_000_000 * 5    # = $0.0025
output_cost = 200 / 1_000_000 * 15  # = $0.003
cost_per_request = 0.0025 + 0.003   # = $0.0055

# Costo diario esperado:
daily_expected = 1000 * 0.0055       # = $5.50

# Budget con margen de seguridad (+50%):
daily_budget_limit_usd = 5.50 * 1.5  # = $8.25

# Redondeado: $10.00 como budget cómodo

Ejercicio 4: Test del TokenBucket

Escribe un test que verifique que un TokenBucket(rate=2, capacity=5) permite 5 requests inmediatos (burst) pero bloquea el 6to:

def test_token_bucket_burst_and_block():
    bucket = TokenBucket(rate=2, capacity=5, name="test")
    
    # Los primeros 5 deben pasar (burst)
    for i in range(5):
        assert bucket.consume(1) is True, f"Request {i+1} debería pasar"
    
    # El 6to debe ser rechazado (bucket vacío)
    assert bucket.consume(1) is False, "Request 6 debería ser rechazado"
    
    # Verificar métricas
    metrics = bucket.get_metrics()
    assert metrics["consumed"] == 5
    assert metrics["rejected"] == 1

Troubleshooting

"Mis requests pasan el rate limiter pero OpenAI sigue dando 429"

Tu safety margin puede ser insuficiente, o estás contando solo requests HTTP pero no tokens. OpenAI tiene límites tanto de RPM como de TPM (tokens por minuto). Si tus requests son largos, puedes estar dentro de RPM pero excediendo TPM. Revisa tus límites en la dashboard de OpenAI y considera implementar un token-based rate limiter además del request-based.

"El TokenBucket parece no rellenar tokens"

Verifica que tu lógica de _refill se ejecuta correctamente. El TokenBucket usa time.time() para calcular cuántos tokens agregar. Si estás en tests y mockeas el tiempo, los tokens no se rellenan. Usa time.monotonic() o inyecta una función de tiempo para tests.

"El DailyBudgetLimiter no se reinicia a medianoche"

Revisa la timezone de tu servidor. date.today() usa la timezone del sistema. Si tu servidor está en UTC pero esperas medianoche en tu zona horaria, el reset ocurre en el momento "incorrecto". Considera usar datetime.now(timezone.utc).date() para consistencia.

"¿Cómo manejo rate limiting cuando tengo múltiples instancias de mi app?"

El TokenBucket en memoria funciona por instancia. Con 3 instancias, cada una permite su propio rate — en total triplicas el tráfico. Para rate limiting distribuido, necesitas Redis (con el patrón INCR + EXPIRE) o un API gateway como Kong/Nginx que centralice el control. Para la mayoría de apps AI con pocas instancias, rate limiting por instancia dividido entre el número de réplicas es suficiente.

Resumen

• Rate limiting client-side: controla la velocidad de requests antes de llegar a la API
• Token bucket: permite bursts cortos, throttlea el flujo sostenido
• safety margin: usar 80% del límite real para tener margen de error
• Queue vs reject: queue para usuarios interactivos con tolerancia a espera; reject rápido para webhooks y tiempo-críticos
• Budget limiting: complementa el rate limiting con control de costo absoluto
• Por modelo: cada modelo tiene sus propios límites — rate limiters separados

Recursos adicionales

1. Token Bucket Algorithm — El algoritmo explicado
2. OpenAI Rate Limits — Límites oficiales
3. Leaky Bucket vs Token Bucket — Comparación de algoritmos
4. Redis Rate Limiting — Rate limiting distribuido para múltiples instancias