5. Rate Limiting

Description

Client-side rate limiting is the difference between a traffic spike that gradually degrades your service and one that takes it down completely. Without rate limiting, 200 concurrent users make 200 requests to OpenAI simultaneously — they exceed the RPM limit, they all get 429, and retry makes the problem worse. With rate limiting, your requests spread out over time, respecting the API's limits and the budget you set. In this capsule you'll implement a thread-safe token bucket, wrap it as a provider, manage per-model limits, and add daily budget control.

Why client-side rate limiting (not just trusting the API's limits)

Without client-side rate limiting:

t=0s:   200 requests arrive simultaneously at your app
        → Your app sends 200 requests to OpenAI
        → OpenAI: limit is 100/min → rejects 100 with 429
        → The 100 retry → they make the problem worse
        → Result: degraded experience, budget burned, visible errors

With client-side rate limiting:

t=0s:   200 requests arrive at your app
        → Rate limiter: capacity of 8/s
        → Processes 8 at t=0s, queues the rest
        → Processes 8 more at t=1s, 8 at t=2s...
        → The 200 are processed in ~25 seconds
        → OpenAI never sees more than 8/s → never rejects anything
        → Result: everyone completes, cost controlled, no 429 errors

Additional benefit: budget control
        → With rate limiting: you can limit 10 requests/user/day
        → Without rate limiting: one user can spend your entire budget in 1 minute

Token Bucket algorithm explained

# The token bucket is the most common algorithm for rate limiting

# Concept:
#   - A "bucket" that fills with tokens at a constant rate
#   - Each request consumes tokens
#   - If there are no tokens, the request waits or is rejected

# Properties:
#   - rate: tokens added per second
#   - capacity: maximum bucket size (maximum burst)
#   - A full bucket = capacity for a burst
#   - An empty bucket = strict throttling

# Example: rate=5, capacity=10
#
# t=0s:  bucket=10 (full). User A makes 10 requests → bucket=0
# t=0s:  User B makes 1 request → WAITS (empty bucket)
# t=1s:  bucket=5 (filled at rate=5). Processes 5 pending requests
# t=2s:  bucket=5. Processes more requests
# ...
# Result: the initial burst of 10 is allowed, then throttled to 5/s

# This is better than rejecting all: it allows short bursts but controls sustained load

Thread-safe TokenBucket

# 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.
    
    Allows short bursts up to 'capacity' tokens,
    then throttles to 'rate' tokens per second.
    """
    
    def __init__(self, rate: float, capacity: int, name: str = "default"):
        """
        rate: tokens per second (e.g., 8.3 for 500 RPM / 60)
        capacity: bucket size = maximum burst allowed
        name: identifier for logs
        """
        self.rate = rate
        self.capacity = capacity
        self.name = name
        self._tokens = float(capacity)  # Starts full (allows initial burst)
        self._last_refill = time.monotonic()
        self._lock = threading.Lock()
        
        # Metrics
        self._total_consumed = 0
        self._total_rejected = 0
        self._total_waited_seconds = 0.0
    
    def _refill(self) -> None:
        """
        Refill the bucket based on elapsed time.
        Must be called within the 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:
        """
        Try to consume tokens from the bucket.
        
        Returns:
            True if it could consume (request allowed)
            False if there aren't enough tokens (request rejected)
        """
        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, waiting if the bucket is empty.
        
        Args:
            tokens: tokens to consume
            max_wait_seconds: maximum time to wait
        
        Returns:
            True if it consumed (it may have waited)
            False if the wait time exceeds 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
                
                # Calculate how long to wait
                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:
        """Currently available tokens (approximate)."""
        with self._lock:
            self._refill()
            return self._tokens
    
    def get_metrics(self) -> dict:
        """Metrics for 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):
    """The request was rejected by the client-side rate limiter."""
    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  # Suggest waiting half of max_wait
        )
        self.bucket_name = bucket_name

class RateLimitedProvider:
    """
    Wrapper that applies client-side rate limiting to any LLMProvider.
    
    Prevents exceeding the API's RPM/TPM limits,
    controlling the speed of outgoing requests.
    """
    
    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: request limit per minute
        burst_size: burst size (default: 20% of RPM)
        max_wait_seconds: maximum time a request waits in queue
        model: model name (for 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:
        """
        Make the call respecting the rate limit.
        If the bucket is empty, wait up to max_wait_seconds.
        If it can't get tokens in that time, reject the 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 per model (specific limits)

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

# Real OpenAI limits (Tier 1, approximate — always verify the current ones):
# 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 that manages separate limits per model.
    
    Uses 80% of the limits to have a safety margin.
    """
    
    SAFETY_MARGIN = 0.8  # Use only 80% of the limit
    
    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:
        """
        Try to acquire a token for the specified model.
        Wait up to max_wait seconds.
        """
        bucket = self._buckets.get(model)
        if bucket is None:
            # Unknown model — use a conservative limit
            return True  # Or create a default bucket
        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()
        }

# Global instance (singleton)
model_rate_limiter = ModelRateLimiter()

Budget limiting: controlling the total cost

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

log = structlog.get_logger()

class DailyBudgetExceeded(Exception):
    """The API's daily budget has been exceeded."""
    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:
    """
    Controls daily spending on LLM APIs.
    
    Resets automatically every day at midnight.
    Thread-safe for concurrent apps.
    """
    
    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:
        """
        Verify that the request fits within the budget and record it.
        
        Raises DailyBudgetExceeded if the budget would be exceeded.
        """
        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 if we're getting close to the limit
            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)
            }

Exercises

Exercise 1: Calculate the rate for gpt-4o

OpenAI Tier 1 allows 500 RPM for gpt-4o. What rate and capacity would you use in the TokenBucket?

# Limit: 500 RPM
# With 80% safety margin: 500 × 0.8 = 400 RPM
# Rate in tokens/second: 400 / 60 = 6.67 tokens/s
# Capacity (burst): 20% of 400 = 80, but cap it reasonably
# A burst of 20-30 is practical for most apps

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

Exercise 2: Queue vs Reject

For each use case, should the rate limiter make the request wait (queue) or reject it immediately (reject)?

1. Public API where the user can wait 10-15 seconds
2. Slack webhook that must respond in <3 seconds
3. Nightly batch process with no active user waiting

Exercise 3: Calculate daily budget

Your app processes on average 1,000 requests/day with gpt-4o. Each request uses ~500 input tokens and ~200 output tokens. With OpenAI's prices ($5/1M input, $15/1M output), what should your daily_budget_limit_usd be?

# Cost per 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

# Expected daily cost:
daily_expected = 1000 * 0.0055       # = $5.50

# Budget with safety margin (+50%):
daily_budget_limit_usd = 5.50 * 1.5  # = $8.25

# Rounded: $10.00 as a comfortable budget

Exercise 4: TokenBucket test

Write a test that verifies that a TokenBucket(rate=2, capacity=5) allows 5 immediate requests (burst) but blocks the 6th:

def test_token_bucket_burst_and_block():
    bucket = TokenBucket(rate=2, capacity=5, name="test")
    
    # The first 5 must pass (burst)
    for i in range(5):
        assert bucket.consume(1) is True, f"Request {i+1} should pass"
    
    # The 6th must be rejected (empty bucket)
    assert bucket.consume(1) is False, "Request 6 should be rejected"
    
    # Verify metrics
    metrics = bucket.get_metrics()
    assert metrics["consumed"] == 5
    assert metrics["rejected"] == 1

Troubleshooting

"My requests pass the rate limiter but OpenAI still returns 429"

Your safety margin may be insufficient, or you're only counting HTTP requests but not tokens. OpenAI has limits for both RPM and TPM (tokens per minute). If your requests are long, you may be within RPM but exceeding TPM. Review your limits in the OpenAI dashboard and consider implementing a token-based rate limiter in addition to the request-based one.

"The TokenBucket seems not to refill tokens"

Verify that your _refill logic runs correctly. The TokenBucket uses time.time() to calculate how many tokens to add. If you're in tests and you mock time, tokens don't refill. Use time.monotonic() or inject a time function for tests.

"The DailyBudgetLimiter doesn't reset at midnight"

Check your server's timezone. date.today() uses the system timezone. If your server is in UTC but you expect midnight in your time zone, the reset happens at the "wrong" moment. Consider using datetime.now(timezone.utc).date() for consistency.

"How do I handle rate limiting when I have multiple instances of my app?"

The in-memory TokenBucket works per instance. With 3 instances, each one allows its own rate — in total you triple the traffic. For distributed rate limiting, you need Redis (with the INCR + EXPIRE pattern) or an API gateway like Kong/Nginx that centralizes the control. For most AI apps with few instances, per-instance rate limiting divided by the number of replicas is enough.

Summary

• Client-side rate limiting: controls the speed of requests before reaching the API
• Token bucket: allows short bursts, throttles sustained flow
• safety margin: use 80% of the real limit to have a margin of error
• Queue vs reject: queue for interactive users with tolerance for waiting; fast reject for webhooks and time-critical paths
• Budget limiting: complements rate limiting with absolute cost control
• Per model: each model has its own limits — separate rate limiters

Additional resources

1. Token Bucket Algorithm — The algorithm explained
2. OpenAI Rate Limits — Official limits
3. Leaky Bucket vs Token Bucket — Comparison of algorithms
4. Redis Rate Limiting — Distributed rate limiting for multiple instances