Rate Limiting and Circuit Breakers in Distributed AI Systems

Distributed AI systems are inherently complex, handling massive volumes of requests, variable latency from model inference, and dependencies on external services like GPU clusters, databases, or third-party APIs. Without proper safeguards, a single misbehaving component or a sudden traffic surge can cascade into system-wide failure. Two fundamental patterns—rate limiting and circuit breakers—provide essential protection. This post explores their roles, implementation strategies, and practical Python examples tailored for AI workloads.

Why Distributed AI Systems Need These Patterns

Consider a typical AI pipeline: a user sends a prompt, which hits a load balancer, then an API gateway, then an inference service (e.g., a large language model), which may call a vector database or a fine-tuning API. Each component has capacity limits:

• GPU inference servers can handle limited concurrent requests.
• External APIs (e.g., OpenAI, HuggingFace) impose rate limits.
• Database connections are finite.

Without rate limiting, a single abusive client can exhaust resources. Without circuit breakers, a failing downstream service can cause cascading timeouts and resource exhaustion across the entire system.

Rate Limiting: Controlling Request Flow

Rate limiting restricts how many requests a client, user, or service can make in a given time window. It prevents resource starvation and ensures fair access.

Common Algorithms

Algorithm	Pros	Cons
Token Bucket	Smooth burst handling, easy to implement	Memory per bucket
Leaky Bucket	Constant outflow rate, simple	Less flexible for bursts
Fixed Window	Simple, low overhead	Boundary spikes (reset issues)
Sliding Window	Smoother than fixed, accurate	Slightly more complex

For AI systems, token bucket is often preferred because it allows short bursts (e.g., a user sending a batch of prompts) while maintaining a long-term average.

Python Implementation: Token Bucket Rate Limiter

import time
import threading
from collections import defaultdict

class TokenBucket:
    def __init__(self, rate: float, capacity: int):
        """
        rate: tokens per second
        capacity: maximum bucket size (burst)
        """
        self.rate = rate
        self.capacity = capacity
        self.tokens = capacity
        self.last_refill = time.monotonic()
        self.lock = threading.Lock()

    def _refill(self):
        now = time.monotonic()
        elapsed = now - self.last_refill
        self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
        self.last_refill = now

    def consume(self, tokens: int = 1) -> bool:
        with self.lock:
            self._refill()
            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            return False

# Per-user rate limiter for AI inference
class AIInferenceRateLimiter:
    def __init__(self, default_rate: float = 10, default_capacity: int = 20):
        self.buckets = defaultdict(lambda: TokenBucket(default_rate, default_capacity))

    def check_request(self, user_id: str) -> bool:
        return self.buckets[user_id].consume(1)

# Usage
limiter = AIInferenceRateLimiter(rate=5, capacity=10)  # 5 req/s, burst up to 10
if limiter.check_request("user_123"):
    print("Proceed with model inference")
else:
    print("429 Too Many Requests")

AI-Specific Considerations

• Token-level throttling: For LLMs, rate-limit by input/output token count, not just requests. A request generating 4096 tokens is much heavier than one generating 100.
• Priority queues: Premium users get higher rate limits or separate buckets.
• Distributed state: Use Redis or etcd for shared rate limiter state across multiple service instances.

# Redis-based distributed rate limiter (simplified)
import redis
import time

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

def check_rate_limit(user_id: str, max_requests: int, window_seconds: int) -> bool:
    key = f"ratelimit:{user_id}:{int(time.time()) // window_seconds}"
    current = r.get(key)
    if current and int(current) >= max_requests:
        return False
    r.incr(key)
    r.expire(key, window_seconds + 1)
    return True

Circuit Breakers: Fail Fast and Recover Gracefully

A circuit breaker monitors calls to a downstream service (e.g., a GPU inference server, a vector database). When failures exceed a threshold, the circuit opens, and subsequent calls fail immediately (or use a fallback). After a cooldown, it transitions to half-open to test if the service has recovered.

States

Closed (normal) --> Open (failing) --> Half-Open (testing) --> Closed or Open

Python Implementation: Circuit Breaker for AI Inference

import time
import logging
from enum import Enum

class CircuitState(Enum):
    CLOSED = 1
    OPEN = 2
    HALF_OPEN = 3

class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, recovery_timeout: float = 30.0,
                 half_open_max_requests: int = 3):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.half_open_max_requests = half_open_max_requests

        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure_time = None
        self.half_open_requests = 0
        self.lock = threading.Lock()

    def call(self, func, fallback=None, *args, **kwargs):
        with self.lock:
            if self.state == CircuitState.OPEN:
                if time.monotonic() - self.last_failure_time >= self.recovery_timeout:
                    self.state = CircuitState.HALF_OPEN
                    self.half_open_requests = 0
                    logging.info("Circuit breaker entering half-open state")
                else:
                    logging.warning("Circuit breaker open, using fallback")
                    return self._execute_fallback(fallback, *args, **kwargs)

            if self.state == CircuitState.HALF_OPEN:
                if self.half_open_requests >= self.half_open_max_requests:
                    logging.warning("Half-open max requests reached, using fallback")
                    return self._execute_fallback(fallback, *args, **kwargs)
                self.half_open_requests += 1

        # Execute the actual call (outside lock to avoid blocking)
        try:
            result = func(*args, **kwargs)
            with self.lock:
                if self.state == CircuitState.HALF_OPEN:
                    self.state = CircuitState.CLOSED
                    self.failure_count = 0
                    logging.info("Circuit breaker closed after successful half-open call")
                elif self.state == CircuitState.CLOSED:
                    self.failure_count = 0
            return result
        except Exception as e:
            with self.lock:
                self.failure_count += 1
                self.last_failure_time = time.monotonic()
                if self.state == CircuitState.HALF_OPEN:
                    self.state = CircuitState.OPEN
                    logging.warning("Circuit breaker reopened after half-open failure")
                elif self.state == CircuitState.CLOSED and self.failure_count >= self.failure_threshold:
                    self.state = CircuitState.OPEN
                    logging.error(f"Circuit breaker opened after {self.failure_count} failures")
            return self._execute_fallback(fallback, *args, **kwargs)

    def _execute_fallback(self, fallback, *args, **kwargs):
        if fallback:
            return fallback(*args, **kwargs)
        raise Exception("Circuit breaker open and no fallback provided")

# Usage with AI inference call
def call_gpu_inference(prompt: str) -> str:
    # Simulated call to GPU inference server
    import random
    if random.random() < 0.3:  # 30% failure rate
        raise ConnectionError("GPU server timeout")
    return f"Response to: {prompt}"

def fallback_response(prompt: str) -> str:
    return f"Fallback: {prompt[:50]}... (service degraded)"

circuit_breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=10.0)

for i in range(20):
    result = circuit_breaker.call(call_gpu_inference, fallback_response, f"Prompt {i}")
    print(f"Request {i}: {result}")
    time.sleep(0.5)

AI-Specific Circuit Breaker Features

• Failure types: Distinguish between transient errors (timeouts, 503s) and persistent errors (400 bad request). Only count transient errors toward the threshold.
• Slow calls: Treat abnormally slow responses (e.g., >10s for LLM inference) as failures. This prevents resource hoarding.
• **Bulk