Rate Limiting
Rate limiting is essential for LLM applications to prevent abuse, manage costs, ensure fair resource allocation, and stay within API provider quotas. Unlike traditional rate limiting (which primarily protects infrastructure), LLM rate limiting also protects your budget — each API call has a direct monetary cost proportional to token usage.
LLM APIs impose multiple rate limits simultaneously: requests per minute (RPM), tokens per minute (TPM), and concurrent requests. Exceeding these limits results in 429 (Too Many Requests) errors that must be handled gracefully with retry logic and backoff.
This guide covers the major rate limiting strategies — token bucket, sliding window, concurrent request limiting, queue-based throttling — along with backpressure mechanisms, retry algorithms, and production-grade middleware implementations.
The token bucket algorithm allows bursty traffic while enforcing a steady long-term rate. Tokens are added to a bucket at a fixed rate (the refill rate). Each request consumes one or more tokens. If the bucket is empty, the request is rejected or queued. Burst tolerance is determined by the bucket size.
| 1 | import time |
| 2 | import asyncio |
| 3 | from typing import Optional |
| 4 | |
| 5 | class TokenBucket: |
| 6 | def __init__(self, rate: float, burst: int): |
| 7 | self.rate = rate # Tokens per second |
| 8 | self.burst = burst # Maximum bucket capacity |
| 9 | self.tokens = burst # Current tokens |
| 10 | self.last_refill = time.monotonic() |
| 11 | self._lock = asyncio.Lock() |
| 12 | |
| 13 | async def _refill(self): |
| 14 | now = time.monotonic() |
| 15 | elapsed = now - self.last_refill |
| 16 | new_tokens = elapsed * self.rate |
| 17 | self.tokens = min(self.burst, self.tokens + new_tokens) |
| 18 | self.last_refill = now |
| 19 | |
| 20 | async def acquire(self, tokens: int = 1) -> bool: |
| 21 | async with self._lock: |
| 22 | await self._refill() |
| 23 | if self.tokens >= tokens: |
| 24 | self.tokens -= tokens |
| 25 | return True |
| 26 | return False |
| 27 | |
| 28 | async def wait_and_acquire( |
| 29 | self, tokens: int = 1, timeout: Optional[float] = None |
| 30 | ) -> bool: |
| 31 | deadline = time.monotonic() + timeout if timeout else None |
| 32 | while True: |
| 33 | if await self.acquire(tokens): |
| 34 | return True |
| 35 | if deadline and time.monotonic() >= deadline: |
| 36 | return False |
| 37 | # Wait for next token (at most 1/rate seconds) |
| 38 | await asyncio.sleep(1.0 / self.rate) |
| 39 | |
| 40 | # Usage |
| 41 | bucket = TokenBucket(rate=10, burst=20) # 10 req/s, burst up to 20 |
| 42 | |
| 43 | async def rate_limited_call(prompt: str) -> Optional[str]: |
| 44 | if await bucket.acquire(): |
| 45 | return await llm_call(prompt) |
| 46 | return None # Rate limited |
info
The sliding window algorithm tracks request timestamps within a rolling time window (e.g., the last 60 seconds). It provides more precise rate enforcement than token buckets because it does not allow "debt" accumulation — the rate is strictly bounded over any window-sized interval.
| 1 | from collections import deque |
| 2 | import time |
| 3 | |
| 4 | class SlidingWindowRateLimiter: |
| 5 | def __init__(self, max_requests: int, window_seconds: int): |
| 6 | self.max_requests = max_requests |
| 7 | self.window_seconds = window_seconds |
| 8 | self.requests: deque = deque() |
| 9 | |
| 10 | def allow_request(self) -> bool: |
| 11 | now = time.monotonic() |
| 12 | # Remove expired timestamps |
| 13 | while self.requests and self.requests[0] < now - self.window_seconds: |
| 14 | self.requests.popleft() |
| 15 | |
| 16 | if len(self.requests) < self.max_requests: |
| 17 | self.requests.append(now) |
| 18 | return True |
| 19 | return False |
| 20 | |
| 21 | def time_until_available(self) -> float: |
| 22 | if not self.requests: |
| 23 | return 0.0 |
| 24 | if len(self.requests) < self.max_requests: |
| 25 | return 0.0 |
| 26 | # Time until the oldest request expires |
| 27 | oldest = self.requests[0] |
| 28 | wait = (oldest + self.window_seconds) - time.monotonic() |
| 29 | return max(0.0, wait) |
| 30 | |
| 31 | # Multi-tier rate limiter |
| 32 | class MultiTierRateLimiter: |
| 33 | def __init__(self): |
| 34 | self.limiters = { |
| 35 | "rpm": SlidingWindowRateLimiter(60, 60), # 60 RPM |
| 36 | "tpm": SlidingWindowRateLimiter(40000, 60), # 40K TPM |
| 37 | "concurrent": ConcurrentRequestLimiter(10), # 10 concurrent |
| 38 | } |
| 39 | |
| 40 | def allow_request(self, estimated_tokens: int = 1000) -> bool: |
| 41 | if not self.limiters["rpm"].allow_request(): |
| 42 | return False |
| 43 | if not self.limiters["tpm"].allow_request(): |
| 44 | return False |
| 45 | if not self.limiters["concurrent"].allow_request(): |
| 46 | return False |
| 47 | return True |
Concurrent request limits restrict how many LLM calls can be in-flight simultaneously. This is critical because API providers enforce concurrent limits independently of RPM/TPM, and exceeding them causes connection-level errors rather than clean 429 responses.
| 1 | import asyncio |
| 2 | from contextlib import asynccontextmanager |
| 3 | |
| 4 | class ConcurrentRequestLimiter: |
| 5 | def __init__(self, max_concurrent: int): |
| 6 | self.semaphore = asyncio.Semaphore(max_concurrent) |
| 7 | self.active_requests = 0 |
| 8 | |
| 9 | @asynccontextmanager |
| 10 | async def limit(self): |
| 11 | async with self.semaphore: |
| 12 | self.active_requests += 1 |
| 13 | try: |
| 14 | yield |
| 15 | finally: |
| 16 | self.active_requests -= 1 |
| 17 | |
| 18 | async def __call__(self, fn, *args, **kwargs): |
| 19 | async with self.limit(): |
| 20 | return await fn(*args, **kwargs) |
| 21 | |
| 22 | |
| 23 | # Full rate-limited client |
| 24 | class RateLimitedLLMClient: |
| 25 | def __init__( |
| 26 | self, |
| 27 | rpm: int = 60, |
| 28 | tpm: int = 40000, |
| 29 | max_concurrent: int = 10 |
| 30 | ): |
| 31 | self.rpm_limiter = SlidingWindowRateLimiter(rpm, 60) |
| 32 | self.tpm_limiter = SlidingWindowRateLimiter(tpm, 60) |
| 33 | self.concurrent_limiter = ConcurrentRequestLimiter(max_concurrent) |
| 34 | self._queue = asyncio.Queue() |
| 35 | self._worker_task = None |
| 36 | |
| 37 | async def call(self, prompt: str, max_tokens: int = 500) -> str: |
| 38 | estimated_tokens = len(prompt.split()) * 1.3 + max_tokens |
| 39 | await self._wait_for_capacity(estimated_tokens) |
| 40 | async with self.concurrent_limiter.limit(): |
| 41 | response = await actual_llm_call(prompt, max_tokens) |
| 42 | return response |
| 43 | |
| 44 | async def _wait_for_capacity(self, tokens: int, timeout: float = 30.0): |
| 45 | start = time.monotonic() |
| 46 | while time.monotonic() - start < timeout: |
| 47 | if (self.rpm_limiter.allow_request() and |
| 48 | self.tpm_limiter.allow_request()): |
| 49 | return |
| 50 | wait_rpm = self.rpm_limiter.time_until_available() |
| 51 | wait_tpm = self.tpm_limiter.time_until_available() |
| 52 | await asyncio.sleep(min(wait_rpm, wait_tpm, 0.1)) |
| 53 | raise TimeoutError("Rate limit capacity not available") |
When the API returns a 429 (rate limit exceeded) or 5xx (server error), implement retry with exponential backoff and jitter. Never retry immediately — this creates a thundering herd problem that worsens congestion. Add randomness (jitter) to prevent synchronized retry waves.
| 1 | import asyncio |
| 2 | import random |
| 3 | |
| 4 | async def retry_with_backoff( |
| 5 | fn, |
| 6 | max_retries: int = 5, |
| 7 | base_delay: float = 1.0, |
| 8 | max_delay: float = 60.0, |
| 9 | jitter: bool = True |
| 10 | ): |
| 11 | last_exception = None |
| 12 | for attempt in range(max_retries): |
| 13 | try: |
| 14 | return await fn() |
| 15 | except (RateLimitError, ServerError) as e: |
| 16 | last_exception = e |
| 17 | if attempt == max_retries - 1: |
| 18 | raise |
| 19 | |
| 20 | delay = base_delay * (2 ** attempt) |
| 21 | delay = min(delay, max_delay) |
| 22 | |
| 23 | if jitter: |
| 24 | delay = delay * (0.5 + random.random()) |
| 25 | |
| 26 | # Respect Retry-After header if present |
| 27 | retry_after = getattr(e, "retry_after", None) |
| 28 | if retry_after: |
| 29 | delay = max(delay, float(retry_after)) |
| 30 | |
| 31 | print(f"Retry {attempt + 1}/{max_retries} " |
| 32 | f"after {delay:.1f}s: {e}") |
| 33 | await asyncio.sleep(delay) |
| 34 | |
| 35 | raise last_exception |
| 36 | |
| 37 | |
| 38 | # Decorator version |
| 39 | def rate_limit_retry(max_retries=5, base_delay=1.0): |
| 40 | def decorator(fn): |
| 41 | async def wrapper(*args, **kwargs): |
| 42 | return await retry_with_backoff( |
| 43 | lambda: fn(*args, **kwargs), |
| 44 | max_retries=max_retries, |
| 45 | base_delay=base_delay |
| 46 | ) |
| 47 | return wrapper |
| 48 | return decorator |
| 49 | |
| 50 | @rate_limit_retry(max_retries=3) |
| 51 | async def get_completion(prompt: str) -> str: |
| 52 | # This call will automatically retry on 429/5xx |
| 53 | return await client.chat.completions.create(...) |
best practice
Queue-based throttling decouples request submission from rate-limited execution. Requests enter a queue and are processed at the rate allowed by API limits. Backpressure mechanisms prevent the queue from growing unboundedly by rejecting or shedding load when the queue exceeds capacity.
| 1 | import asyncio |
| 2 | from dataclasses import dataclass, field |
| 3 | from typing import Optional |
| 4 | |
| 5 | @dataclass |
| 6 | class QueueItem: |
| 7 | prompt: str |
| 8 | max_tokens: int |
| 9 | future: asyncio.Future = field(default_factory=asyncio.Future) |
| 10 | |
| 11 | class ThrottledQueue: |
| 12 | def __init__(self, rate_limiter, max_queue_size: int = 100): |
| 13 | self.rate_limiter = rate_limiter |
| 14 | self.queue: asyncio.Queue = asyncio.Queue(maxsize=max_queue_size) |
| 15 | self._worker_task = asyncio.create_task(self._worker()) |
| 16 | |
| 17 | async def submit(self, prompt: str, max_tokens: int = 500) -> str: |
| 18 | item = QueueItem(prompt=prompt, max_tokens=max_tokens) |
| 19 | try: |
| 20 | self.queue.put_nowait(item) |
| 21 | except asyncio.QueueFull: |
| 22 | raise BackpressureError("Queue full — try again later") |
| 23 | return await item.future |
| 24 | |
| 25 | async def _worker(self): |
| 26 | while True: |
| 27 | item = await self.queue.get() |
| 28 | # Wait for rate limit capacity |
| 29 | await self.rate_limiter.wait_for_capacity(item.max_tokens) |
| 30 | try: |
| 31 | result = await actual_llm_call(item.prompt, item.max_tokens) |
| 32 | item.future.set_result(result) |
| 33 | except Exception as e: |
| 34 | item.future.set_exception(e) |
| 35 | |
| 36 | # Backpressure-aware client |
| 37 | class BackpressureError(Exception): |
| 38 | pass |
| 39 | |
| 40 | class AdaptiveRateLimiter: |
| 41 | def __init__(self, initial_rpm: int = 60): |
| 42 | self.current_rpm = initial_rpm |
| 43 | self.recent_429s = deque(maxlen=10) |
| 44 | |
| 45 | async def call(self, fn): |
| 46 | while True: |
| 47 | if await self._allow(): |
| 48 | try: |
| 49 | return await fn() |
| 50 | except RateLimitError: |
| 51 | self.recent_429s.append(time.monotonic()) |
| 52 | self.current_rpm = max(10, int(self.current_rpm * 0.8)) |
| 53 | await asyncio.sleep(self._backoff()) |
| 54 | else: |
| 55 | if len(self.recent_429s) == 0: |
| 56 | self.current_rpm = min(1000, int(self.current_rpm * 1.1)) |
| 57 | |
| 58 | def _backoff(self) -> float: |
| 59 | recent = sum( |
| 60 | 1 for t in self.recent_429s |
| 61 | if time.monotonic() - t < 60 |
| 62 | ) |
| 63 | return min(60.0, 2.0 ** recent) |
429 Too Many Requests responses include a Retry-After header indicating how long to wait before retrying. Always respect this header — it reflects the server's current congestion state. Ignoring it or retrying too aggressively can result in IP bans or account suspension.
429 Response Best Practices
warning