Confront LLM Rate Limits Efficiently
Leaky Bucket Rate Limit
The Leaky Bucket algorithm is a rate limiting strategy that allows for a fixed number of requests to be processed at a steady rate. It is particularly useful for applications that handle a large volume of requests and need to maintain performance while preventing abuse. Here’s how it works:
- Bucket Capacity: The algorithm maintains a fixed capacity for requests, which is the maximum number of requests that can be processed at once.
- Request Processing: Each request is added to the bucket, and once the bucket is full, new requests are either dropped or delayed until the bucket is emptied.
- Rate Limiting: The algorithm processes requests at a fixed rate, ensuring that the bucket does not overflow and that requests are processed in a controlled manner.
- Implementation: The Leaky Bucket algorithm can be implemented using various programming languages and libraries, such as Python’s aiolimiter or fastapi with slowapi.
This algorithm is particularly effective for applications that require a steady stream of requests, such as web applications or APIs, where the volume of requests can fluctuate. By using the Leaky Bucket algorithm, developers can ensure that their applications remain responsive and that requests are processed efficiently, even when faced with high traffic volumes.
References:
https://martinlwx.github.io/en/async-and-leaky-bucket-algorithm-batch-llm-api-call/
https://markaicode.com/implement-rate-limiting-prevent-llm-abuse/
Python Sample Middleware for LLM APIs
Here’s a ready‑to‑use, production‑grade retry middleware in Python designed for LLM APIs. It includes:
- Exponential backoff with jitter
- Honor Retry-After headers
- Transparent handling of 429 & 5xx responses + network errors
- RPM (requests/min), TPM (tokens/min), and concurrency limiting
- Async client based on httpx (high-throughput) + a sync requests fallback
- Clean hooks for logging/metrics
- Idempotency key support (headers)
Note: Python library backoff provides an easy solution to use backoff strategy in code, hiding all the implementation complexity (that is described in following paragraphs):
import openai, backoff
@backoff.on_exception(backoff.expo, openai.error.RateLimitError, max_tries=5)
async def call_api (query: str, limit: int = 5) -> str:
"""Semantic search over document chunks."""
return openai.ChatCompletion.create(
model="gpt-4o",
messages=[{"role": "user", "content": "Hello"}],
max_tokens=50
)
response = call_api()
print(response.choices[0].message["content"])
Dependencies
The sample code use the following libraries:
- httpx (for async client)
- requests (optional: for sync fallback)
pip install httpx requests
Features at a glance
- Traffic shaping: Token bucket–style limiters for RPM & TPM, and a semaphore for concurrent calls.
- Robust retries: Exponential backoff + jitter, respects server Retry-After.
- Safety: Retries on 429, 500, 502, 503, 504 and common transient network errors.
- Token-aware: Lets you estimate tokens before send to avoid TPM bursts, you can feed back actual usage after a call.
- Streaming‑safe: Will only retry before streaming starts (once a stream begins, it won’t attempt to resume mid-stream).
- Pluggable: Pass your own estimate_tokens and extract_usage callables.
The Middleware (async, with httpx)
Save as llm_retry.py.
# llm_retry.py
import asyncio
import random
import time
from dataclasses import dataclass, field
from typing import Any, Awaitable, Callable, Dict, Iterable, Optional, Tuple, Union
import httpx
# ---------------------------
# Configuration structures
# ---------------------------
@dataclass
class RetryConfig:
max_attempts: int = 6
base_delay: float = 0.5 # seconds
max_delay: float = 30.0 # seconds
backoff_factor: float = 2.0
jitter: Tuple[float, float] = (0.1, 0.5) # add +/- random jitter to avoid thundering herd
retry_on_statuses: Tuple[int, ...] = (429, 500, 502, 503, 504)
retry_on_exceptions: Tuple[type, ...] = (
httpx.ReadTimeout, httpx.ConnectTimeout, httpx.RemoteProtocolError, httpx.ConnectError
)
respect_retry_after: bool = True
@dataclass
class RateLimitConfig:
rpm: Optional[int] = None # requests per minute
tpm: Optional[int] = None # tokens per minute
concurrent: Optional[int] = None # max in-flight requests
# ---------------------------
# Token Bucket for TPM / RPM
# ---------------------------
class _AsyncTokenBucket:
"""
Token bucket that refills at a steady rate (tokens per second).
Acquire() waits until enough tokens are available.
"""
def __init__(self, capacity: float, refill_per_sec: float):
self.capacity = float(capacity)
self._tokens = float(capacity)
self.refill_per_sec = float(refill_per_sec)
self._last = time.monotonic()
self._lock = asyncio.Lock()
self._not_empty = asyncio.Condition()
async def acquire(self, amount: float = 1.0):
if amount <= 0:
return
async with self._lock:
while True:
now = time.monotonic()
# Refill
elapsed = now - self._last
self._last = now
self._tokens = min(self.capacity, self._tokens + elapsed * self.refill_per_sec)
if self._tokens >= amount:
self._tokens -= amount
return
# Wait until enough tokens accumulate
needed = amount - self._tokens
wait_time = needed / self.refill_per_sec if self.refill_per_sec > 0 else 0.5
wait_time = max(0.01, min(wait_time, 2.0))
# Release lock while sleeping to allow other refills
await asyncio.sleep(wait_time)
# ---------------------------
# Rate limiter wiring
# ---------------------------
class AsyncRateLimiter:
"""
Combines:
- RPM limiter via token bucket
- TPM limiter via token bucket
- Concurrency limiter via semaphore
"""
def __init__(self, cfg: RateLimitConfig):
self._rpm_bucket = _AsyncTokenBucket(cfg.rpm, cfg.rpm / 60.0) if cfg.rpm else None
self._tpm_bucket = _AsyncTokenBucket(cfg.tpm, cfg.tpm / 60.0) if cfg.tpm else None
self._concurrency = asyncio.Semaphore(cfg.concurrent) if cfg.concurrent else None
async def acquire(self, req_cost: float = 1.0, token_cost: float = 0.0):
# Concurrency first to bound parallel pressure
if self._concurrency:
await self._concurrency.acquire()
try:
# Requests/min
if self._rpm_bucket:
await self._rpm_bucket.acquire(req_cost)
# Tokens/min
if self._tpm_bucket and token_cost > 0:
await self._tpm_bucket.acquire(token_cost)
except Exception:
# If acquisition fails, release semaphore (if taken) before raising
if self._concurrency:
self._concurrency.release()
raise
def release(self):
if self._concurrency:
self._concurrency.release()
# ---------------------------
# Retry helpers
# ---------------------------
def _compute_backoff(
attempt: int,
cfg: RetryConfig,
retry_after_s: Optional[float] = None
) -> float:
"""
Compute next sleep duration using exponential backoff + jitter.
If Retry-After is present and respect_retry_after=True, we take max(backoff, retry_after_s).
"""
base = cfg.base_delay * (cfg.backoff_factor ** max(0, attempt - 1))
base = min(base, cfg.max_delay)
jitter = random.uniform(*cfg.jitter) if cfg.jitter else 0.0
delay = min(cfg.max_delay, base + jitter)
if retry_after_s is not None and cfg.respect_retry_after:
delay = max(delay, retry_after_s)
return max(0.0, delay)
def _parse_retry_after(headers: httpx.Headers) -> Optional[float]:
"""
Returns seconds from Retry-After header if present (supports delta-seconds).
RFC 7231 also allows HTTP-date; here we implement delta-seconds for simplicity.
"""
ra = headers.get("Retry-After")
if not ra:
return None
try:
secs = float(ra.strip())
if secs < 0:
return None
return secs
except ValueError:
# If a date is provided, you can parse it and compute delta; omitted for brevity.
return None
# ---------------------------
# The middleware client
# ---------------------------
class LLMRetryClient:
"""
A robust async client with retry, backoff, RPM/TPM/concurrency control.
Parameters
----------
base_url : str
API base URL (e.g., "https://api.openai.com/v1")
headers : Dict[str, str]
Default headers (e.g., Authorization)
retry : RetryConfig
Retry behavior
limits : RateLimitConfig
Rate limiting config
timeout : float
Request timeout (seconds)
client : Optional[httpx.AsyncClient]
Optional custom AsyncClient; if not provided, one is created.
Hooks
-----
estimate_tokens(req) -> int
Optional function to estimate tokens (for TPM limiting) before sending.
extract_usage(resp_json) -> int
Optional function to extract actual token usage after completion to improve future estimates.
on_retry(event_dict)
Optional callback with details on each retry attempt (for logging/metrics).
"""
def __init__(
self,
base_url: str,
headers: Optional[Dict[str, str]] = None,
retry: Optional[RetryConfig] = None,
limits: Optional[RateLimitConfig] = None,
timeout: float = 60.0,
client: Optional[httpx.AsyncClient] = None,
estimate_tokens: Optional[Callable[[Dict[str, Any]], int]] = None,
extract_usage: Optional[Callable[[Dict[str, Any]], int]] = None,
on_retry: Optional[Callable[[Dict[str, Any]], None]] = None,
):
self.base_url = base_url.rstrip("/")
self.headers = headers or {}
self.retry = retry or RetryConfig()
self.limits = limits or RateLimitConfig()
self.timeout = timeout
self._client = client or httpx.AsyncClient(base_url=self.base_url, timeout=self.timeout)
self._own_client = client is None
self._rl = AsyncRateLimiter(self.limits)
self._estimate_tokens = estimate_tokens
self._extract_usage = extract_usage
self._on_retry = on_retry
async def aclose(self):
if self._own_client:
await self._client.aclose()
async def request(
self,
method: str,
path: str,
*,
json: Optional[Dict[str, Any]] = None,
params: Optional[Dict[str, Any]] = None,
headers: Optional[Dict[str, str]] = None,
stream: bool = False,
idempotency_key: Optional[str] = None,
) -> Union[httpx.Response, Dict[str, Any]]:
"""
Make an HTTP request with full retry & rate limiting.
If stream=False, returns parsed JSON (dict).
If stream=True, returns httpx.Response with `aiter_lines()` for the caller to consume.
"""
url = path if path.startswith("http") else f"{self.base_url}/{path.lstrip('/')}"
merged_headers = dict(self.headers)
if headers:
merged_headers.update(headers)
if idempotency_key:
# Supported by several providers (OpenAI: Idempotency-Key; Azure ARM uses Repeatability-*)
merged_headers.setdefault("Idempotency-Key", idempotency_key)
# Estimate token cost for TPM control
est_tokens = 0
if self._estimate_tokens and json is not None:
try:
est_tokens = max(0, int(self._estimate_tokens(json)))
except Exception:
est_tokens = 0 # be safe if estimator fails
attempt = 0
last_exc: Optional[BaseException] = None
while attempt < self.retry.max_attempts:
attempt += 1
# Acquire rate limits before sending
await self._rl.acquire(req_cost=1.0, token_cost=float(est_tokens))
try:
# Send request (status available before streaming starts)
resp = await self._client.request(
method=method.upper(),
url=url,
json=json,
params=params,
headers=merged_headers,
timeout=self.timeout,
)
# If status retriable -> compute backoff and retry
if resp.status_code in self.retry.retry_on_statuses:
retry_after = _parse_retry_after(resp.headers)
if attempt >= self.retry.max_attempts:
# no more retries
self._rl.release()
resp.raise_for_status()
# log/report retry
if self._on_retry:
self._on_retry({
"attempt": attempt,
"status": resp.status_code,
"url": url,
"retry_after": retry_after,
"reason": f"HTTP {resp.status_code}",
})
delay = _compute_backoff(attempt, self.retry, retry_after)
self._rl.release()
await asyncio.sleep(delay)
continue
# Success path:
if stream:
# streaming caller consumes the response
self._rl.release()
return resp
# Non-stream: parse JSON
data = resp.json()
resp.raise_for_status()
# Optional feedback loop for TPM estimation
if self._extract_usage and isinstance(data, dict):
try:
actual_tokens = int(self._extract_usage(data))
# (Optional) could adjust internal estimators; left as hook.
_ = actual_tokens
except Exception:
pass
self._rl.release()
return data
except tuple(self.retry.retry_on_exceptions) as e:
last_exc = e
if attempt >= self.retry.max_attempts:
self._rl.release()
raise
if self._on_retry:
self._on_retry({
"attempt": attempt,
"exception": repr(e),
"url": url,
"reason": "network/timeout",
})
delay = _compute_backoff(attempt, self.retry)
self._rl.release()
await asyncio.sleep(delay)
continue
except httpx.HTTPStatusError as e:
# Non-retriable HTTP
self._rl.release()
raise
except Exception as e:
# Unknown error: do not blindly retry unless you want to
self._rl.release()
raise
# If loop exits, re-raise last exception or generic
if last_exc:
raise last_exc
raise RuntimeError("Retry loop exhausted without returning or raising expected error.")
# ---------------------------
# Utilities for OpenAI-compatible endpoints
# ---------------------------
def default_estimate_tokens_for_chat(payload: Dict[str, Any]) -> int:
"""
Very rough estimate: ~ 4 tokens per word as a conservative upper bound.
For production, plug a real tokenizer (tiktoken) for the target model.
"""
def text_len(s: str) -> int:
return 0 if not s else len(s.split())
tokens = 0
if "messages" in payload:
for m in payload["messages"]:
tokens += 4 * text_len(m.get("content", "")) + 8 # role & formatting overhead
# Also include system/instruction + function schemas if any
if "tools" in payload:
tokens += 200 # rough bump for tool schemas
if "max_tokens" in payload:
tokens += int(payload["max_tokens"])
return tokens
def default_extract_usage_openai(resp_json: Dict[str, Any]) -> int:
"""
Pulls total_tokens from OpenAI-style response if present.
"""
try:
return int(resp_json["usage"]["total_tokens"])
except Exception:
return 0
Example: OpenAI‑compatible Chat Completions (Async)
Save as example_async.py.
# example_async.py
import asyncio
import os
from llm_retry import (
LLMRetryClient, RetryConfig, RateLimitConfig,
default_estimate_tokens_for_chat, default_extract_usage_openai
)
OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY")
async def main():
client = LLMRetryClient(
base_url="https://api.openai.com/v1",
headers={
"Authorization": f"Bearer {OPENAI_API_KEY}",
"Content-Type": "application/json",
},
retry=RetryConfig(
max_attempts=6,
base_delay=0.5,
max_delay=20,
backoff_factor=2.0,
retry_on_statuses=(429, 500, 502, 503, 504),
),
limits=RateLimitConfig(
rpm=300, # set to your account limit
tpm=100_000, # set to your account TPM limit
concurrent=8, # tune per workload
),
estimate_tokens=default_estimate_tokens_for_chat,
extract_usage=default_extract_usage_openai,
on_retry=lambda evt: print("[retry]", evt),
)
try:
payload = {
"model": "gpt-4o-mini",
"messages": [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Summarize the benefits of exponential backoff."}
],
"max_tokens": 200,
"temperature": 0.2,
"stream": False
}
# Non-streaming call (JSON result)
result = await client.request(
"POST",
"/chat/completions",
json=payload,
idempotency_key="example-req-001" # optional but recommended for safe retries
)
print(result["choices"][0]["message"]["content"])
# Streaming example
stream_payload = {**payload, "stream": True}
resp = await client.request(
"POST",
"/chat/completions",
json=stream_payload,
stream=True,
)
async for line in resp.aiter_lines():
if not line:
continue
if line.startswith("data: "):
chunk = line[len("data: "):]
if chunk.strip() == "[DONE]":
break
print(chunk)
finally:
await client.aclose()
if __name__ == "__main__":
asyncio.run(main())
Notes & Best Practices
- Idempotency: Always set an idempotency key for operations you might retry to avoid duplicated charges/side effects where supported.
- Streaming: This middleware retries only before consuming a stream. If a stream drops mid‑flight, surface it to the caller to decide (resume patterns are API‑specific).
- Token estimation: Plug in a real tokenizer (e.g., tiktoken) for accurate TPM limiting. The default estimator is intentionally conservative.
- Per‑deployment limits: If you use Azure OpenAI or multiple model deployments, you can instantiate one client per deployment to distribute load.
- Observability: Use on_retry hook to log metrics (attempt, reason, delay). Add counters for 429s, backoff time, queue depth, and tail latency.
LangChain with Retry
Save as llm_lc_with_retry.py.
# llm_lc_with_retry.py
import os
from typing import Callable, Any, Tuple, Type
from langchain_core.runnables import Runnable
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception
# ----------------------------
# Retry Condition
# ----------------------------
TRANSIENT_ERRORS: Tuple[Type[BaseException], ...] = (
TimeoutError,
ConnectionError,
OSError
)
OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY")
def is_transient_error(exception: BaseException) -> bool:
"""Return True if the exception is considered transient."""
return isinstance(exception, TRANSIENT_ERRORS)
# ----------------------------
# Retry Wrapper
# ----------------------------
def with_retry(
runnable: Runnable,
max_attempts: int = 3,
backoff_min: int = 2,
backoff_max: int = 10
) -> Runnable:
"""
Wraps a LangChain Runnable with retry logic.
Args:
runnable: Any LangChain Runnable (LLM, chain, tool).
max_attempts: Maximum retry attempts.
backoff_min: Minimum backoff in seconds.
backoff_max: Maximum backoff in seconds.
"""
@retry(
stop=stop_after_attempt(max_attempts),
wait=wait_exponential(multiplier=1, min=backoff_min, max=backoff_max),
retry=retry_if_exception(is_transient_error),
reraise=True
)
def _invoke_with_retry(*args, **kwargs) -> Any:
try:
logger.info(f"Invoking {runnable.__class__.__name__} (attempt)...")
return runnable.invoke(*args, **kwargs)
except Exception as e:
logger.warning(f"Error: {e}. Retrying...")
raise
class RetryRunnable(Runnable):
def invoke(self, *args, **kwargs):
return _invoke_with_retry(*args, **kwargs)
return RetryRunnable()
# ----------------------------
# Example Usage
# ----------------------------
if __name__ == "__main__":
from langchain_openai import ChatOpenAI
# Example LLM
llm = ChatOpenAI(model="gpt-5.4-nano", temperature=0)
# Wrap with retry
retry_llm = with_retry(llm, max_attempts=3, backoff_min=2, backoff_max=8)
try:
result = retry_llm.invoke("Explain retry patterns in LangChain")
print("\nFinal Result:", result.content)
except Exception as e:
logger.error(f"Final failure after retries: {e}")