Introduction to Rate Limiting and Retries in Pydantic AI
When building production-grade applications with Large Language Models (LLMs), two challenges inevitably surface: API rate limits and transient failures. LLM providers like OpenAI, Anthropic, and Google enforce strict rate limits to protect their infrastructure, and network hiccups, timeouts, or overloaded servers can cause requests to fail unpredictably. Pydantic AI, a powerful agent framework built on top of Pydantic, provides elegant mechanisms to handle both scenarios gracefully.
This guide walks you through everything you need to know about implementing robust rate limiting and retry strategies in your Pydantic AI applications, from basic concepts to advanced production patterns.
Understanding Rate Limiting
What Is Rate Limiting?
Rate limiting is a mechanism used by API providers to control the volume of requests a client can make within a specific time window. LLM providers typically enforce several types of limits:
- Requests Per Minute (RPM): Maximum number of API calls allowed per minute
- Tokens Per Minute (TPM): Maximum number of tokens (input + output) processed per minute
- Concurrent Requests: Maximum number of simultaneous in-flight requests
- Daily Quotas: Total requests or tokens allowed per day
When you exceed these limits, the API responds with a 429 Too Many Requests status code, often accompanied by a Retry-After header indicating how long to wait before retrying.
Why Rate Limiting Matters
Ignoring rate limits leads to cascading failures in your application. A burst of 429 errors can degrade user experience, break batch processing pipelines, and in worst cases, result in temporary API access suspension. Proactively managing your request rate ensures:
- Consistent application performance under load
- Cost predictability and budget control
- Graceful degradation during traffic spikes
- Compliance with provider terms of service
Understanding Retry Strategies
What Are Retry Strategies?
Retry strategies define how your application responds to transient failures. Instead of immediately failing when a request errors out, a retry strategy attempts the request again after a calculated delay. The key components of a retry strategy include:
- Max retries: The upper bound on retry attempts
- Backoff algorithm: How delay increases between retries
- Jitter: Random variation added to delays to prevent thundering herd problems
- Retry conditions: Which errors warrant a retry versus an immediate failure
Common Backoff Algorithms
The most widely used backoff strategies are:
- Fixed delay: Same wait time between each retry (simple but inefficient)
- Linear backoff: Delay increases linearly (e.g., 1s, 2s, 3s, 4s)
- Exponential backoff: Delay doubles each time (e.g., 1s, 2s, 4s, 8s) — the industry standard
- Exponential backoff with jitter: Adds randomness to exponential backoff to distribute load
Setting Up Pydantic AI
Before diving into rate limiting and retries, let's set up a basic Pydantic AI project. Install the required packages:
pip install pydantic-ai tenacity httpx
Here's a minimal Pydantic AI agent to work with throughout this tutorial:
from pydantic_ai import Agent
from pydantic import BaseModel, Field
class ResearchResult(BaseModel):
summary: str = Field(description="A concise summary of the topic")
key_points: list[str] = Field(description="Key bullet points")
confidence: float = Field(description="Confidence score 0-1", ge=0, le=1)
research_agent = Agent(
model="openai:gpt-4o",
result_type=ResearchResult,
system_prompt=(
"You are a research assistant. Provide structured, "
"accurate summaries with confidence scores."
),
)
async def run_research(topic: str) -> ResearchResult:
result = await research_agent.run(topic)
return result.data
This basic setup works fine for development, but in production, you'll quickly hit rate limits or encounter transient errors. Let's build resilience into this foundation.
Implementing Rate Limiting in Pydantic AI
Client-Side Rate Limiting with a Token Bucket
The most effective approach to client-side rate limiting is the token bucket algorithm. It allows bursts of requests while maintaining an average rate over time. Here's a reusable async rate limiter:
import asyncio
import time
from collections import deque
class AsyncTokenBucketRateLimiter:
def __init__(
self,
rate: float,
capacity: float,
tokens_per_minute: int | None = None,
):
"""
Args:
rate: Tokens replenished per second
capacity: Maximum burst size
tokens_per_minute: Optional TPM limit for additional control
"""
self.rate = rate
self.capacity = capacity
self.tokens = capacity
self.last_update = time.monotonic()
self._lock = asyncio.Lock()
# Token-based limiting (TPM)
self.tpm_limit = tokens_per_minute
self.token_usage_window: deque = deque() # (timestamp, token_count)
async def acquire(self, estimated_tokens: int = 0) -> None:
async with self._lock:
now = time.monotonic()
elapsed = now - self.last_update
self.tokens = min(
self.capacity,
self.tokens + elapsed * self.rate,
)
self.last_update = now
# Wait for request token availability
if self.tokens < 1:
wait_time = (1 - self.tokens) / self.rate
await asyncio.sleep(wait_time)
self.tokens = 0
else:
self.tokens -= 1
# Wait for TPM token availability if configured
if self.tpm_limit and estimated_tokens > 0:
await self._check_tpm(estimated_tokens)
async def _check_tpm(self, estimated_tokens: int) -> None:
now = time.monotonic()
# Remove entries older than 60 seconds
while (
self.token_usage_window
and now - self.token_usage_window[0][0] > 60
):
self.token_usage_window.popleft()
current_usage = sum(t for _, t in self.token_usage_window)
if current_usage + estimated_tokens > self.tpm_limit:
wait_time = 60 - (now - self.token_usage_window[0][0])
if wait_time > 0:
await asyncio.sleep(wait_time)
self.token_usage_window.append((now, estimated_tokens))
# Create a shared rate limiter instance
# OpenAI GPT-4o typical limits: 500 RPM, 30K TPM (Tier 1)
rate_limiter = AsyncTokenBucketRateLimiter(
rate=500 / 60, # ~8.3 requests per second
capacity=10, # Allow short bursts
tokens_per_minute=25000, # Stay under TPM limit
)
Integrating the Rate Limiter with Pydantic AI
Now let's wrap our agent calls with the rate limiter. We'll create a resilient wrapper function:
import logging
logger = logging.getLogger(__name__)
async def run_research_rate_limited(topic: str) -> ResearchResult:
# Estimate tokens: rough heuristic of ~4 chars per token
estimated_tokens = len(topic) // 4 + 1000 # Include output estimate
await rate_limiter.acquire(estimated_tokens=estimated_tokens)
try:
result = await research_agent.run(topic)
return result.data
except Exception as e:
logger.error(f"Research failed for topic '{topic}': {e}")
raise
Using a Custom HTTP Client for Rate Limiting
For deeper integration, you can configure Pydantic AI to use a custom HTTP client that enforces rate limiting at the transport layer. This approach catches all outgoing requests automatically:
import httpx
from pydantic_ai.models.openai import OpenAIModel
class RateLimitedTransport(httpx.AsyncBaseTransport):
def __init__(
self,
wrapped_transport: httpx.AsyncBaseTransport,
rate_limiter: AsyncTokenBucketRateLimiter,
):
self.wrapped = wrapped_transport
self.rate_limiter = rate_limiter
async def handle_async_request(
self, request: httpx.Request
) -> httpx.Response:
# Extract estimated token usage from request body if available
body = request.content
estimated_tokens = len(body) // 4 if body else 500
await self.rate_limiter.acquire(estimated_tokens=estimated_tokens)
return await self.wrapped.handle_async_request(request)
# Build the rate-limited model
base_transport = httpx.AsyncHTTPTransport()
limited_transport = RateLimitedTransport(base_transport, rate_limiter)
limited_client = httpx.AsyncClient(transport=limited_transport)
rate_limited_model = OpenAIModel(
"gpt-4o",
http_client=limited_client,
)
rate_limited_agent = Agent(
model=rate_limited_model,
result_type=ResearchResult,
system_prompt="You are a research assistant.",
)
Implementing Retry Strategies with Tenacity
Basic Retry with Tenacity
The tenacity library is the gold standard for retry logic in Python. It integrates cleanly with async code and Pydantic AI. Here's a basic retry decorator:
from tenacity import (
retry,
stop_after_attempt,
wait_exponential,
retry_if_exception_type,
before_sleep_log,
)
import httpx
logger = logging.getLogger(__name__)
@retry(
stop=stop_after_attempt(5),
wait=wait_exponential(multiplier=1, min=1, max=60),
retry=retry_if_exception_type((
httpx.HTTPStatusError,
httpx.ConnectError,
httpx.ReadTimeout,
httpx.WriteTimeout,
ConnectionError,
TimeoutError,
)),
before_sleep=before_sleep_log(logger, logging.WARNING),
reraise=True,
)
async def run_research_with_retry(topic: str) -> ResearchResult:
result = await research_agent.run(topic)
return result.data
Handling 429 Rate Limit Responses Specifically
When you receive a 429 response, the provider often includes a Retry-After header. A smart retry strategy respects this value instead of using a generic backoff:
from tenacity import (
retry,
stop_after_attempt,
wait_exponential_jitter,
retry_if_exception_type,
RetryCallState,
)
import httpx
import asyncio
class RateLimitError(Exception):
"""Raised when API returns 429 Too Many Requests."""
def __init__(self, retry_after: float | None = None):
self.retry_after = retry_after
super().__init__(f"Rate limited. Retry after: {retry_after}s")
def smart_wait(retry_state: RetryCallState) -> float:
"""Use Retry-After header if available, otherwise exponential backoff."""
exception = retry_state.outcome.exception() if retry_state.outcome else None
if isinstance(exception, RateLimitError) and exception.retry_after:
# Add small jitter to Retry-After to avoid synchronized retries
jitter = asyncio.get_event_loop().time() % 0.5
return exception.retry_after + jitter
# Exponential backoff with jitter: 2^attempt * 1 + random(0, 1)
return wait_exponential_jitter(
initial=1, max=60, exp_base=2
)(retry_state)
@retry(
stop=stop_after_attempt(7),
wait=smart_wait,
retry=retry_if_exception_type((
RateLimitError,
httpx.ConnectError,
httpx.ReadTimeout,
TimeoutError,
)),
reraise=True,
)
async def run_research_smart_retry(topic: str) -> ResearchResult:
try:
result = await research_agent.run(topic)
return result.data
except httpx.HTTPStatusError as e:
if e.response.status_code == 429:
retry_after = e.response.headers.get("Retry-After")
retry_after_float = float(retry_after) if retry_after else None
raise RateLimitError(retry_after=retry_after_float)
raise
Combining Rate Limiting and Retries
The most robust approach combines proactive rate limiting (to avoid hitting limits) with reactive retries (to handle unexpected failures). Here's a complete production-ready pattern:
import logging
from tenacity import retry, stop_after_attempt, wait_exponential_jitter
from tenacity import retry_if_exception_type, before_sleep_log
import httpx
logger = logging.getLogger(__name__)
class ResilientAgentRunner:
def __init__(
self,
agent: Agent,
rate_limiter: AsyncTokenBucketRateLimiter,
max_retries: int = 5,
):
self.agent = agent
self.rate_limiter = rate_limiter
self.max_retries = max_retries
async def run(
self,
prompt: str,
estimated_tokens: int = 2000,
) -> ResearchResult:
@retry(
stop=stop_after_attempt(self.max_retries),
wait=wait_exponential_jitter(initial=1, max=60, exp_base=2),
retry=retry_if_exception_type((
httpx.HTTPStatusError,
httpx.ConnectError,
httpx.ReadTimeout,
httpx.WriteTimeout,
ConnectionError,
TimeoutError,
)),
before_sleep=before_sleep_log(logger, logging.WARNING),
reraise=True,
)
async def _execute():
await self.rate_limiter.acquire(
estimated_tokens=estimated_tokens
)
result = await self.agent.run(prompt)
return result.data
return await _execute()
# Usage
runner = ResilientAgentRunner(
agent=research_agent,
rate_limiter=rate_limiter,
max_retries=5,
)
async def main():
topics = [
"quantum computing basics",
"renewable energy trends",
"machine learning interpretability",
]
results = await asyncio.gather(
*[runner.run(topic) for topic in topics],
return_exceptions=True,
)
for topic, result in zip(topics, results):
if isinstance(result, Exception):
logger.error(f"Failed for '{topic}': {result}")
else:
print(f"Topic: {topic}")
print(f"Summary: {result.summary}")
print(f"Confidence: {result.confidence}")
print("---")
asyncio.run(main())
Advanced Patterns
Circuit Breaker for Cascading Failure Prevention
When an LLM provider experiences an extended outage, retrying every request wastes resources and delays failure detection. A circuit breaker pattern trips after consecutive failures, fast-failing subsequent requests for a cooldown period:
import time
from enum import Enum
class CircuitState(Enum):
CLOSED = "closed" # Normal operation
OPEN = "open" # Failing fast, not sending requests
HALF_OPEN = "half_open" # Testing if service recovered
class CircuitBreaker:
def __init__(
self,
failure_threshold: int = 5,
recovery_timeout: float = 60.0,
half_open_max_calls: int = 3,
):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.half_open_max_calls = half_open_max_calls
self.state = CircuitState.CLOSED
self.failure_count = 0
self.last_failure_time = 0.0
self.half_open_calls = 0
self._lock = asyncio.Lock()
async def call(self, func, *args, **kwargs):
async 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_calls = 0
logger.info("Circuit breaker entering HALF_OPEN state")
else:
raise Exception("Circuit breaker is OPEN - failing fast")
if (
self.state == CircuitState.HALF_OPEN
and self.half_open_calls >= self.half_open_max_calls
):
raise Exception("Circuit breaker HALF_OPEN - too many test calls")
try:
if self.state == CircuitState.HALF_OPEN:
self.half_open_calls += 1
result = await func(*args, **kwargs)
async with self._lock:
if self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.CLOSED
self.failure_count = 0
logger.info("Circuit breaker recovered - CLOSED")
return result
except Exception as e:
async with self._lock:
self.failure_count += 1
self.last_failure_time = time.monotonic()
if self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.OPEN
logger.warning("Circuit breaker re-opened from HALF_OPEN")
elif self.failure_count >= self.failure_threshold:
self.state = CircuitState.OPEN
logger.warning(
f"Circuit breaker OPENED after "
f"{self.failure_count} failures"
)
raise
# Integrate with the resilient runner
circuit_breaker = CircuitBreaker(
failure_threshold=5,
recovery_timeout=60.0,
)
async def run_with_circuit_breaker(topic: str) -> ResearchResult:
return await circuit_breaker.call(runner.run, topic)
Fallback Models for High Availability
For mission-critical applications, configure fallback models so that if your primary provider is unavailable, requests route to an alternative:
from pydantic_ai import Agent
class FallbackAgentRunner:
def __init__(self, agent_configs: list[dict], rate_limiter, max_retries=3):
self.agents: list[Agent] = []
for config in agent_configs:
agent = Agent(
model=config["model"],
result_type=ResearchResult,
system_prompt=config.get(
"system_prompt", "You are a research assistant."
),
)
self.agents.append(agent)
self.rate_limiter = rate_limiter
self.max_retries = max_retries
self.current_agent_idx = 0
async def run(self, prompt: str) -> ResearchResult:
last_exception = None
for idx in range(len(self.agents)):
agent_idx = (self.current_agent_idx + idx) % len(self.agents)
agent = self.agents[agent_idx]
try:
@retry(
stop=stop_after_attempt(self.max_retries),
wait=wait_exponential_jitter(initial=1, max=30),
reraise=True,
)
async def _try_agent():
await self.rate_limiter.acquire()
result = await agent.run(prompt)
return result.data
result = await _try_agent()
self.current_agent_idx = agent_idx # Stick with working agent
return result
except Exception as e:
last_exception = e
logger.warning(
f"Agent {agent_idx} failed, trying next: {e}"
)
continue
raise Exception(
f"All agents failed. Last error: {last_exception}"
)
# Configure with multiple providers
fallback_runner = FallbackAgentRunner(
agent_configs=[
{"model": "openai:gpt-4o", "system_prompt": "You are a research assistant."},
{"model": "anthropic:claude-sonnet-4", "system_prompt": "You are a research assistant."},
{"model": "google-gla:gemini-1.5-pro", "system_prompt": "You are a research assistant."},
],
rate_limiter=rate_limiter,
max_retries=3,
)
Concurrency Control with Semaphores
When processing batches of requests, limit concurrency to stay within provider limits. An asyncio.Semaphore works perfectly for this:
async def process_batch_concurrent(
prompts: list[str],
max_concurrent: int = 5,
) -> list[ResearchResult]:
semaphore = asyncio.Semaphore(max_concurrent)
results = []
async def process_one(prompt: str) -> ResearchResult:
async with semaphore:
return await runner.run(prompt)
tasks = [process_one(p) for p in prompts]
raw_results = await asyncio.gather(*tasks, return_exceptions=True)
for prompt, result in zip(prompts, raw_results):
if isinstance(result, Exception):
logger.error(f"Failed for '{prompt}': {result}")
results.append(None)
else:
results.append(result)
return results
# Process 50 prompts with max 5 concurrent requests
prompts = [f"Explain {concept}" for concept in [
"neural networks", "blockchain", "CRISPR", "dark matter",
"microservices", "GraphQL", "Kubernetes", "WebAssembly",
] * 6] # 48 prompts
results = await process_batch_concurrent(prompts, max_concurrent=5)
successful = sum(1 for r in results if r is not None)
print(f"Processed {successful}/{len(prompts)} successfully")
Best Practices
Rate Limiting Best Practices
- Stay below published limits: Target 80-90% of your provider's stated limits to leave headroom for variance in token counts and timing.
- Track actual token usage: Use response metadata to track real token consumption and adjust your rate limiter dynamically rather than relying on estimates.
- Use separate limiters per provider: Different providers have different limits. Don't share a single rate limiter across multiple LLM backends.
- Monitor and alert: Log rate limit events and set up alerts when you approach thresholds so you can upgrade your tier or optimize usage proactively.
- Consider request prioritization: For mixed workloads, implement priority queues so user-facing requests take precedence over background batch jobs.
Retry Strategy Best Practices
- Only retry idempotent or safe operations: LLM completions are generally safe to retry since they don't mutate state, but be cautious with agents that have side effects.
- Always use jitter: Without jitter, multiple clients retry simultaneously after a provider outage, creating a secondary spike. Jitter spreads retries over time.
- Set reasonable max retry limits: 3-5 retries is typically sufficient. More retries waste resources and delay user-facing error responses.
- Cap maximum wait time: Always set an upper bound on backoff delay (e.g., 60 seconds) to prevent excessively long waits.
- Distinguish retryable from non-retryable errors: Authentication errors (401), bad requests (400), and content policy violations should fail immediately, not retry.
- Log retry attempts: Use
before_sleep_logandafter_logfrom tenacity to maintain visibility into retry behavior.
Observability Best Practices
import structlog
from tenacity import after_log
# Configure structured logging
structlog.configure(
processors=[
structlog.processors.TimeStamper(fmt="iso"),
structlog.processors.add_log_level,
structlog.processors.JSONRenderer(),
],
)
log = structlog.get_logger()
@retry(
stop=stop_after_attempt(5),
wait=wait_exponential_jitter(initial=1, max=60),
retry=retry_if_exception_type((httpx.HTTPStatusError, httpx.ConnectError)),
before_sleep=before_sleep_log(logging.getLogger(), logging.WARNING),
after=after_log(logging.getLogger(), logging.INFO),
reraise=True,
)
async def observable_run(prompt: str) -> ResearchResult:
log.info("agent_run_start", prompt_length=len(prompt))
start_time = time.monotonic()
try:
result = await research_agent.run(prompt)
duration = time.monotonic() - start_time
log.info(
"agent_run_success",
duration_ms=round(duration * 1000, 2),
usage=getattr(result, "usage", None),
)
return result.data
except Exception as e:
duration = time.monotonic() - start_time
log.error(
"agent_run_failed",
duration_ms=round(duration * 1000, 2),
error=str(e),
error_type=type(e).__name__,
)
raise
Dynamic Rate Limit Adjustment
In production, your rate limits may change based on your provider tier or time of day. Build flexibility into your rate limiter to adjust dynamically:
class DynamicRateLimiter(AsyncTokenBucketRateLimiter):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._adjustment_lock = asyncio.Lock()
async def adjust_limits(
self,
new_rate: float | None = None,
new_capacity: float | None = None,
new_tpm: int | None = None,
):
async with self._adjustment_lock:
if new_rate is not None:
self.rate = new_rate
logger.info(f"Rate adjusted to {new_rate} tokens/sec")
if new_capacity is not None:
self.capacity = new_capacity
self.tokens = min(self.tokens, new_capacity)
logger.info(f"Capacity adjusted to {new_capacity}")
if new_tpm is not None:
self.tpm_limit = new_tpm
logger.info(f"TPM limit adjusted to {new_tpm}")
# Example: Adjust limits based on time of day
async def adaptive_limits(limiter: DynamicRateLimiter):
while True:
hour = time.localtime().tm_hour
if 9 <= hour < 17: # Business hours - higher traffic
await limiter.adjust_limits(new_rate=400/60, new_tpm=20000)
else: # Off hours - lower traffic
await limiter.adjust_limits(new_rate=200/60, new_tpm=10000)
await asyncio.sleep(300) # Check every 5 minutes
Testing Your Rate Limiting and Retry Logic
Testing resilience patterns is critical. Here's how to simulate failures and verify your retry logic works correctly:
import pytest
from unittest.mock import AsyncMock, patch, MagicMock
@pytest.mark.asyncio
async def test_retry_on_transient_failure():
call_count = 0
async def flaky_agent_run(prompt):
nonlocal call_count
call_count += 1
if call_count < 3:
raise httpx.ConnectError("Simulated connection error")
mock_result = MagicMock()
mock_result.data = ResearchResult(
summary="Test summary",
key_points=["point 1"],
confidence=0.9,
)
return mock_result
with patch.object(research_agent, "run", side_effect=flaky_agent_run):
result = await run_research_smart_retry("test topic")
assert call_count == 3
assert result.summary == "Test summary"
@pytest.mark.asyncio
async def test_rate_limiter_enforces_limit():
fast_limiter = AsyncTokenBucketRateLimiter(
rate=2, # 2 requests per second
capacity=2,
)
start = time.monotonic()
for _ in range(5):
await fast_limiter.acquire()
elapsed = time.monotonic() - start
# 5 requests at 2/sec with capacity 2 means ~1.5 seconds minimum
assert elapsed >= 1.0, f"Rate limit not enforced, elapsed: {elapsed}s"
@pytest.mark.asyncio
async def test_circuit_breaker_opens_after_failures():
breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=1.0)
async def always_fail():
raise Exception("Service down")
# First 3 calls should fail and open the circuit
for _ in range(3):
with pytest.raises(Exception, match="Service down"):
await breaker.call(always_fail)
# 4th call should fail fast with circuit open
with pytest.raises(Exception, match="Circuit breaker is OPEN"):
await breaker.call(always_fail)
Putting It All Together: A Complete Production Example
Here's a complete, production-ready module that combines all the patterns discussed:
import asyncio
import logging
import time
from collections import deque
import httpx
from pydantic import BaseModel, Field
from pydantic_ai import Agent
from tenacity import (
retry,
stop_after_attempt,
wait_exponential_jitter,
retry_if_exception_type,
before_sleep_log,
)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# --- Models ---
class ResearchResult(BaseModel):
summary: str = Field(description="Concise summary")
key_points: list[str] = Field(description="Key points")
confidence: float = Field(ge=0, le=1)
# --- Rate Limiter ---
class AsyncTokenBucketRateLimiter:
def __init__(self, rate: float, capacity: float, tpm_limit: int | None = None):
self.rate = rate
self.capacity = capacity
self.tokens = capacity
self.last_update = time.monotonic()
self.tpm_limit = tpm_limit
self.token_window: deque = deque()
self._lock = asyncio.Lock()
async def acquire(self, estimated_tokens: int = 1000) -> None:
async with self._lock:
now = time.monotonic()
elapsed = now - self.last_update
self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
self.last_update = now
if self.tokens < 1:
wait = (1 - self.tokens) / self.rate
await asyncio.sleep(wait)
self.tokens = 0
else:
self.tokens -= 1
if self.tpm_limit:
while self.token_window and now - self.token_window[0][0] > 60:
self.token_window.popleft()
usage = sum(t for _, t in self.token_window)
if usage + estimated_tokens > self.tpm_limit:
wait = 60 - (now - self.token_window[0][0])
if wait > 0:
await asyncio.sleep(wait)
self.token_window.append((now, estimated_tokens))
# --- Circuit Breaker ---
class CircuitState:
CLOSED = "closed"
OPEN = "open"
HALF_OPEN = "half_open"
class CircuitBreaker:
def __init__(self, failure_threshold=5, recovery_timeout=60.0):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.state = CircuitState.CLOSED
self.failures = 0
self.last_failure = 0.0
self._lock = asyncio.Lock()
async def call(self, func, *args, **kwargs):
async with self._lock:
if self.state == CircuitState.OPEN:
if time.monotonic() - self.last_failure > self.recovery_timeout:
self.state = CircuitState.HALF_OPEN
else:
raise Exception("Circuit breaker OPEN")
try:
result = await func(*args, **kwargs)
async with self._lock:
if self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.CLOSED
self.failures = 0
return result
except Exception:
async with self._lock:
self.failures += 1
self.last_failure = time.monotonic()
if self.failures >= self.failure_threshold:
self.state = CircuitState.OPEN
raise
# --- Production Runner ---
class ProductionAgentRunner:
def __init__(
self,
agent: Agent,
rate_limiter: AsyncTokenBucketRateLimiter,
max_retries: int = 5,
max_concurrent: int = 5,
):
self.agent = agent
self.rate_limiter = rate_limiter
self.max_retries = max_retries
self.semaphore = asyncio.Semaphore(max_concurrent)
self.circuit_breaker = CircuitBreaker(
failure_threshold=5,
recovery_timeout=60.0,
)
async def run(self, prompt: str, estimated_tokens: int = 2000) -> ResearchResult:
async with self.semaphore:
return await self.circuit_breaker.call(
self._run_with_retry, prompt, estimated_tokens
)
async def _run_with_retry(
self, prompt: str, estimated_tokens: int
) -> ResearchResult:
@retry(
stop=stop_after_attempt(self.max_retries),
wait=wait_exponential_jitter(initial=1, max=60, exp_base=2),
retry=retry_if_exception_type((
httpx.HTTPStatusError,
httpx.ConnectError,
httpx.ReadTimeout,
httpx.WriteTimeout,
ConnectionError,
TimeoutError,
)),
before_sleep=before_sleep_log(logger, logging.WARNING),
reraise=True,
)
async def _execute():
await self.rate_limiter.acquire(estimated_tokens)
result = await self.agent.run(prompt)
return result.data
return await _execute()
async def run_batch(
self, prompts: list[str]
) -> list[ResearchResult | None]:
tasks = [self.run(p) for p in prompts]
results = await asyncio.gather(*tasks, return_exceptions=True)
return [
r if not isinstance(r, Exception) else None
for r in results
]
# --- Initialization ---
agent = Agent(
model="openai:gpt-4o",
result_type=ResearchResult,
system_prompt="You are a research assistant providing structured summaries.",
)
rate_limiter = AsyncTokenBucketRateLimiter(
rate=400 / 60,
capacity=10,
tpm_limit=25000,
)
runner = ProductionAgentRunner(
agent=agent,
rate_limiter=rate_limiter,
max_retries=5,
max_concurrent=5,
)
async def main():
topics = [
"quantum computing",
"renewable energy",
"machine learning",
"blockchain",
"biotechnology",
]
results = await runner.run_batch(topics)
for topic, result in zip(topics, results):
if result:
print(f"✓ {topic}: {result.summary[:80]}...")
else:
print(f"✗ {topic}: FAILED")
if __name__ == "__main__":
asyncio.run(main())
Conclusion
Building resilient Pydantic AI applications requires a multi-layered approach to handling the realities of LLM API constraints. Proactive rate limiting with token bucket algorithms prevents you from hitting provider limits in the first place, while reactive retry strategies with exponential backoff and jitter gracefully handle the transient failures that inevitably occur. Adding circuit breakers prevents cascading failures during extended outages, fallback models ensure high availability across providers, and concurrency control via semaphores keeps batch workloads within safe bounds. By combining these patterns with strong observability through structured logging and comprehensive testing, you can build LLM-powered applications that remain reliable and responsive even under heavy load and adverse conditions. Start with the basic rate limiter and retry decorator, then progressively add circuit breakers, fallbacks, and dynamic adjustments as your application's scale and reliability requirements grow.