Introduction to Rate Limiting and Retry Strategies with CrewAI
When building AI agent systems with CrewAI, your agents frequently interact with external LLM providers like OpenAI, Anthropic, or local models. These providers impose rate limits to manage traffic, and network conditions can cause transient failures. Without proper rate limiting and retry strategies, your CrewAI applications may crash, waste resources, or get temporarily banned from APIs. This guide walks you through everything you need to know to build resilient CrewAI workflows.
What Is Rate Limiting and Why Does It Matter?
Rate limiting is the practice of controlling the frequency of requests sent to an API or service. In the context of CrewAI, it applies to how often your agents call LLM endpoints, search tools, or third-party APIs. Retry strategies complement rate limiting by gracefully handling failures when they do occur.
Common Rate Limit Types
- Requests Per Minute (RPM): Limits how many API calls you can make in a 60-second window.
- Tokens Per Minute (TPM): Limits the total number of tokens processed per minute.
- Concurrent Requests: Limits how many simultaneous requests can be in flight.
- Daily Quotas: Caps total usage over a 24-hour period.
When these limits are exceeded, APIs typically return HTTP 429 (Too Many Requests) responses. Without handling, CrewAI will raise an exception and your entire crew execution fails — even if only one agent hit the limit.
Why This Matters for CrewAI Specifically
CrewAI crews often run multiple agents in sequence or parallel, each making multiple LLM calls. A single research task might trigger dozens of API requests in seconds. This makes CrewAI applications particularly vulnerable to rate limits because:
- Agents can make recursive or chained calls during reasoning.
- Tool usage (web search, code execution) adds extra API calls.
- Parallel task execution multiplies request volume.
- Long-running crews accumulate requests over time.
Setting Up Basic Rate Limiting in CrewAI
CrewAI uses LiteLLM under the hood for LLM calls, which provides built-in rate limiting support. Let's start with a basic setup.
Configuring LLM with Rate Limits
from crewai import Agent, Task, Crew, Process
from crewai.llms import LLM
# Configure LLM with rate limiting via LiteLLM
llm = LLM(
model="openai/gpt-4o",
temperature=0.7,
max_tokens=1000,
# LiteLLM rate limiting parameters
rate_limit={
"requests": 50, # max requests
"interval": 60, # per 60 seconds
},
max_retries=3, # automatic retries on failure
retry_after=5, # wait 5 seconds between retries
)
researcher = Agent(
role="Research Analyst",
goal="Find comprehensive information about the topic",
backstory="You are an expert researcher with attention to detail.",
llm=llm,
verbose=True,
)
task = Task(
description="Research the latest trends in AI agents.",
expected_output="A detailed report on AI agent trends.",
agent=researcher,
)
crew = Crew(
agents=[researcher],
tasks=[task],
process=Process.sequential,
)
result = crew.kickoff()
print(result)
Implementing Custom Rate Limiters
For more granular control, you can implement a custom rate limiter using the token bucket algorithm. This is useful when you need to coordinate rate limiting across multiple agents or when you want to respect TPM limits alongside RPM limits.
Token Bucket Rate Limiter
import time
import threading
from collections import defaultdict
class TokenBucketRateLimiter:
def __init__(self, rate, capacity):
"""
rate: tokens added per second
capacity: max tokens in bucket
"""
self.rate = rate
self.capacity = capacity
self.tokens = capacity
self.last_update = time.time()
self.lock = threading.Lock()
def acquire(self, tokens=1):
with self.lock:
now = time.time()
elapsed = now - self.last_update
self.tokens = min(
self.capacity,
self.tokens + elapsed * self.rate
)
self.last_update = now
if self.tokens >= tokens:
self.tokens -= tokens
return True
return False
def wait_for_token(self, tokens=1):
while not self.acquire(tokens):
time.sleep(0.1)
# Global rate limiter shared across agents
rate_limiter = TokenBucketRateLimiter(rate=2, capacity=10)
Wrapping CrewAI Tools with Rate Limiting
from crewai.tools import tool
import time
# Shared rate limiter instance
api_rate_limiter = TokenBucketRateLimiter(rate=1, capacity=5)
@tool("Rate Limited Web Search")
def rate_limited_search(query: str) -> str:
"""Search the web for information, respecting rate limits."""
# Wait until a token is available
api_rate_limiter.wait_for_token()
# Simulate API call
try:
# Replace with actual search API call
result = f"Search results for: {query}"
return result
except Exception as e:
return f"Search failed: {str(e)}"
searcher = Agent(
role="Web Searcher",
goal="Find relevant information online",
backstory="You are skilled at finding information on the internet.",
tools=[rate_limited_search],
llm=llm,
)
Building Robust Retry Strategies
Retries are essential for handling transient failures. The key is to use exponential backoff with jitter to avoid thundering herd problems when many agents retry simultaneously.
Exponential Backoff with Jitter
import random
import time
from functools import wraps
def retry_with_backoff(
max_retries=5,
base_delay=1.0,
max_delay=60.0,
exceptions=(Exception,)
):
"""
Decorator that retries a function with exponential backoff and jitter.
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except exceptions as e:
last_exception = e
if attempt == max_retries - 1:
raise
# Calculate delay with exponential backoff
delay = min(
base_delay * (2 ** attempt),
max_delay
)
# Add jitter (random variation)
jitter = random.uniform(0, delay * 0.1)
total_delay = delay + jitter
print(
f"Attempt {attempt + 1} failed: {e}. "
f"Retrying in {total_delay:.2f}s..."
)
time.sleep(total_delay)
raise last_exception
return wrapper
return decorator
Applying Retries to Custom Tools
@tool("Resilient API Caller")
@retry_with_backoff(
max_retries=5,
base_delay=2.0,
exceptions=(ConnectionError, TimeoutError, ValueError)
)
def resilient_api_call(endpoint: str) -> str:
"""Make a resilient API call with automatic retries."""
import requests
response = requests.get(endpoint, timeout=10)
if response.status_code == 429:
raise ValueError("Rate limit hit")
if response.status_code >= 500:
raise ConnectionError(f"Server error: {response.status_code}")
response.raise_for_status()
return response.text
Handling 429 Responses Specifically
Most APIs include a Retry-After header in 429 responses, telling you exactly how long to wait. You should respect this value rather than guessing.
import requests
import time
def call_llm_with_rate_limit_handling(prompt: str, max_retries: int = 5) -> dict:
"""
Call an LLM API with proper 429 handling.
"""
url = "https://api.openai.com/v1/chat/completions"
headers = {
"Authorization": "Bearer YOUR_API_KEY",
"Content-Type": "application/json",
}
payload = {
"model": "gpt-4o",
"messages": [{"role": "user", "content": prompt}],
}
for attempt in range(max_retries):
response = requests.post(url, json=payload, headers=headers)
if response.status_code == 200:
return response.json()
if response.status_code == 429:
# Respect Retry-After header if present
retry_after = response.headers.get("Retry-After")
if retry_after:
wait_time = float(retry_after)
else:
# Exponential backoff fallback
wait_time = min(2 ** attempt, 60)
print(f"Rate limited. Waiting {wait_time}s before retry...")
time.sleep(wait_time)
continue
if response.status_code >= 500:
# Server error, retry with backoff
wait_time = min(2 ** attempt, 60)
print(f"Server error {response.status_code}. Retrying in {wait_time}s...")
time.sleep(wait_time)
continue
# Client error (not 429), don't retry
response.raise_for_status()
raise Exception(f"Max retries ({max_retries}) exceeded")
Using LiteLLM's Built-in Retry and Rate Limit Features
Since CrewAI uses LiteLLM, you can leverage its robust built-in features for rate limiting and retries across all supported providers.
from crewai.llms import LLM
import litellm
# Enable LiteLLM's global retry configuration
litellm.num_retries = 5
litellm.retry_after = 5
# Configure callbacks for rate limit tracking
litellm.success_callback = ["console"]
litellm.failure_callback = ["console"]
# Create LLM with full retry configuration
llm = LLM(
model="anthropic/claude-3-5-sonnet",
temperature=0.5,
max_tokens=2000,
max_retries=5,
retry_after=10,
# Set request timeout
request_timeout=30,
)
# For multiple providers with different limits
openai_llm = LLM(
model="openai/gpt-4o",
max_retries=3,
rate_limit={"requests": 40, "interval": 60},
)
anthropic_llm = LLM(
model="anthropic/claude-3-5-sonnet",
max_retries=3,
rate_limit={"requests": 30, "interval": 60},
)
Coordinating Multiple Agents with Shared Rate Limits
When running crews with multiple agents that share the same API key, you need a centralized rate limiter to prevent combined requests from exceeding limits.
from crewai import Agent, Task, Crew, Process
from concurrent.futures import ThreadPoolExecutor
import threading
class CrewRateLimitManager:
"""
Manages rate limits across all agents in a crew.
"""
def __init__(self, rpm_limit=50, tpm_limit=90000):
self.rpm_limit = rpm_limit
self.tpm_limit = tpm_limit
self.request_times = []
self.token_counts = []
self.lock = threading.Lock()
def check_and_wait(self, estimated_tokens=1000):
with self.lock:
now = time.time()
window = 60 # 60 second window
# Clean old entries
self.request_times = [t for t in self.request_times if now - t < window]
self.token_counts = [(t, c) for t, c in self.token_counts if now - t < window]
# Check RPM
if len(self.request_times) >= self.rpm_limit:
wait_time = window - (now - self.request_times[0])
if wait_time > 0:
print(f"RPM limit reached. Waiting {wait_time:.1f}s...")
time.sleep(wait_time)
# Check TPM
total_tokens = sum(c for _, c in self.token_counts)
if total_tokens + estimated_tokens > self.tpm_limit:
wait_time = window - (now - self.token_counts[0][0])
if wait_time > 0:
print(f"TPM limit approaching. Waiting {wait_time:.1f}s...")
time.sleep(wait_time)
# Record this request
self.request_times.append(time.time())
self.token_counts.append((time.time(), estimated_tokens))
# Create shared manager
rate_manager = CrewRateLimitManager(rpm_limit=40, tpm_limit=80000)
# Use in a custom LLM wrapper
class RateLimitedLLM:
def __init__(self, base_llm, rate_manager):
self.base_llm = base_llm
self.rate_manager = rate_manager
def call(self, messages, **kwargs):
# Estimate tokens (rough: 4 chars per token)
estimated = sum(len(m.get("content", "")) for m in messages) // 4
self.rate_manager.check_and_wait(estimated)
return self.base_llm.call(messages, **kwargs)
Circuit Breaker Pattern for Critical Failures
When an API is consistently failing, continuing to retry wastes resources. A circuit breaker stops attempts after repeated failures, allowing the system to fail fast and recover gracefully.
import time
from enum import Enum
class CircuitState(Enum):
CLOSED = "closed"
OPEN = "open"
HALF_OPEN = "half_open"
class CircuitBreaker:
def __init__(self, failure_threshold=5, recovery_timeout=60):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.failure_count = 0
self.state = CircuitState.CLOSED
self.last_failure_time = None
def record_success(self):
self.failure_count = 0
self.state = CircuitState.CLOSED
def record_failure(self):
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = CircuitState.OPEN
print(f"Circuit breaker OPENED after {self.failure_count} failures")
def can_execute(self):
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
if time.time() - self.last_failure_time > self.recovery_timeout:
self.state = CircuitState.HALF_OPEN
print("Circuit breaker entering HALF_OPEN state")
return True
return False
# HALF_OPEN: allow one test request
return True
# Usage with CrewAI tools
circuit_breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=120)
@tool("Protected API Tool")
def protected_api_call(query: str) -> str:
"""Make an API call protected by a circuit breaker."""
if not circuit_breaker.can_execute():
return "Service temporarily unavailable. Circuit breaker is open."
try:
# Your API call here
result = make_api_request(query)
circuit_breaker.record_success()
return result
except Exception as e:
circuit_breaker.record_failure()
return f"API call failed: {str(e)}"
Monitoring and Logging Rate Limit Events
Visibility into rate limit events is crucial for tuning your strategies. Implement logging to track when limits are hit and how retries perform.
import logging
from datetime import datetime
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('crewai_rate_limits.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger("CrewAI.RateLimiter")
class MonitoredRateLimiter:
def __init__(self, name, rate, capacity):
self.name = name
self.bucket = TokenBucketRateLimiter(rate, capacity)
self.total_requests = 0
self.total_waits = 0
self.total_wait_time = 0
def acquire(self, tokens=1):
self.total_requests += 1
start = time.time()
self.bucket.wait_for_token(tokens)
wait_duration = time.time() - start
if wait_duration > 0.01:
self.total_waits += 1
self.total_wait_time += wait_duration
logger.info(
f"[{self.name}] Rate limited. Waited {wait_duration:.2f}s. "
f"Total waits: {self.total_waits}, "
f"Avg wait: {self.total_wait_time/self.total_waits:.2f}s"
)
def get_stats(self):
return {
"name": self.name,
"total_requests": self.total_requests,
"total_waits": self.total_waits,
"avg_wait_time": (
self.total_wait_time / self.total_waits
if self.total_waits > 0 else 0
),
}
Best Practices
1. Set Conservative Initial Limits
Start with rate limits well below the API provider's maximum. For OpenAI's 500 RPM tier, start at 50-100 RPM and increase gradually. This gives you headroom for unexpected spikes.
2. Always Use Exponential Backoff with Jitter
Fixed delays cause synchronized retry storms when multiple agents fail simultaneously. Jitter spreads retries over time, reducing load on the API.
3. Respect Retry-After Headers
When an API tells you how long to wait, listen. Ignoring this header and retrying sooner can result in longer bans or permanent suspension.
4. Implement Timeouts on Every Call
llm = LLM(
model="openai/gpt-4o",
request_timeout=30, # 30 second timeout
max_retries=3,
)
5. Use Fallback Models
Configure fallback LLMs so that if one provider is rate limited, your crew can switch to another automatically.
from litellm import Router
router = Router(
model_list=[
{
"model_name": "primary",
"litellm_params": {
"model": "openai/gpt-4o",
"api_key": "your-openai-key",
},
},
{
"model_name": "fallback",
"litellm_params": {
"model": "anthropic/claude-3-5-sonnet",
"api_key": "your-anthropic-key",
},
},
],
fallbacks=[{"primary": ["fallback"]}],
num_retries=3,
retry_after=5,
)
# Use router with CrewAI
llm = LLM(
model="primary",
router=router,
)
6. Cache Responses When Possible
For repeated identical queries, cache LLM responses to avoid redundant API calls. This is especially useful during development and testing.
import hashlib
import json
import os
class ResponseCache:
def __init__(self, cache_dir=".crewai_cache"):
self.cache_dir = cache_dir
os.makedirs(cache_dir, exist_ok=True)
def _get_key(self, prompt, model):
content = f"{model}:{prompt}"
return hashlib.md5(content.encode()).hexdigest()
def get(self, prompt, model):
key = self._get_key(prompt, model)
path = os.path.join(self.cache_dir, f"{key}.json")
if os.path.exists(path):
with open(path, 'r') as f:
return json.load(f)
return None
def set(self, prompt, model, response):
key = self._get_key(prompt, model)
path = os.path.join(self.cache_dir, f"{key}.json")
with open(path, 'w') as f:
json.dump(response, f)
cache = ResponseCache()
7. Test Rate Limit Handling
Simulate rate limit errors in your test environment to ensure your retry logic works correctly before deploying to production.
import unittest
from unittest.mock import patch, MagicMock
class TestRateLimitHandling(unittest.TestCase):
def test_retry_on_429(self):
call_count = 0
def mock_api_call(*args, **kwargs):
nonlocal call_count
call_count += 1
if call_count < 3:
raise ValueError("Rate limit hit (429)")
return {"result": "success"}
with patch('your_module.make_api_request', side_effect=mock_api_call):
result = call_llm_with_rate_limit_handling("test prompt")
self.assertEqual(result["result"], "success")
self.assertEqual(call_count, 3)
def test_circuit_breaker_opens(self):
breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=1)
for _ in range(3):
breaker.record_failure()
self.assertEqual(breaker.state, CircuitState.OPEN)
self.assertFalse(breaker.can_execute())
if __name__ == '__main__':
unittest.main()
8. Monitor Token Usage Proactively
Track token consumption across your crew to anticipate when you're approaching limits, rather than waiting for errors.
Complete Example: Production-Ready CrewAI Setup
Here is a complete example combining all the strategies discussed:
import time
import logging
import random
from crewai import Agent, Task, Crew, Process
from crewai.llms import LLM
from crewai.tools import tool
# Logging setup
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("CrewAI.Production")
# Token bucket rate limiter
class TokenBucketRateLimiter:
def __init__(self, rate, capacity):
self.rate = rate
self.capacity = capacity
self.tokens = capacity
self.last_update = time.time()
def wait_for_token(self, tokens=1):
while True:
now = time.time()
elapsed = now - self.last_update
self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
self.last_update = now
if self.tokens >= tokens:
self.tokens -= tokens
return
time.sleep(0.1)
# Circuit breaker
class CircuitBreaker:
def __init__(self, failure_threshold=5, recovery_timeout=60):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.failure_count = 0
self.state = "closed"
self.last_failure_time = 0
def can_execute(self):
if self.state == "open":
if time.time() - self.last_failure_time > self.recovery_timeout:
self.state = "half_open"
return True
return False
return True
def record_success(self):
self.failure_count = 0
self.state = "closed"
def record_failure(self):
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = "open"
logger.warning("Circuit breaker opened")
# Shared instances
llm_rate_limiter = TokenBucketRateLimiter(rate=2, capacity=10)
api_circuit_breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=120)
# Retry decorator
def retry_with_backoff(max_retries=5, base_delay=1.0, max_delay=60.0):
def decorator(func):
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == max_retries - 1:
raise
delay = min(base_delay * (2 ** attempt), max_delay)
delay += random.uniform(0, delay * 0.1)
logger.info(f"Retry {attempt + 1}/{max_retries} in {delay:.2f}s: {e}")
time.sleep(delay)
return wrapper
return decorator
# Configure LLM with built-in rate limiting
llm = LLM(
model="openai/gpt-4o",
temperature=0.7,
max_tokens=1000,
max_retries=3,
retry_after=5,
request_timeout=30,
rate_limit={"requests": 40, "interval": 60},
)
# Rate-limited tool
@tool("Web Search Tool")
@retry_with_backoff(max_retries=4, base_delay=2.0)
def web_search(query: str) -> str:
"""Search the web for information."""
if not api_circuit_breaker.can_execute():
return "Search service temporarily unavailable."
llm_rate_limiter.wait_for_token()
try:
# Simulate search
result = f"Results for: {query}"
api_circuit_breaker.record_success()
return result
except Exception as e:
api_circuit_breaker.record_failure()
raise
# Create agents
researcher = Agent(
role="Research Analyst",
goal="Conduct thorough research on given topics",
backstory="You are a meticulous researcher.",
llm=llm,
tools=[web_search],
verbose=True,
)
writer = Agent(
role="Content Writer",
goal="Write clear, engaging content based on research",
backstory="You are a skilled writer.",
llm=llm,
verbose=True,
)
# Create tasks
research_task = Task(
description="Research the topic of AI agent frameworks.",
expected_output="A summary of key findings about AI agent frameworks.",
agent=researcher,
)
writing_task = Task(
description="Write an article based on the research findings.",
expected_output="A well-structured article about AI agent frameworks.",
agent=writer,
context=[research_task],
)
# Create and run crew
crew = Crew(
agents=[researcher, writer],
tasks=[research_task, writing_task],
process=Process.sequential,
verbose=True,
)
if __name__ == "__main__":
try:
result = crew.kickoff()
print("\n=== Final Result ===")
print(result)
except Exception as e:
logger.error(f"Crew execution failed: {e}")
Conclusion
Rate limiting and retry strategies are not optional add-ons for production CrewAI applications — they are essential components that determine whether your AI agent systems run reliably or fail unpredictably. By combining LiteLLM's built-in rate limiting, custom token bucket limiters for fine-grained control, exponential backoff with jitter for retries, circuit breakers for catastrophic failure protection, and response caching for efficiency, you can build CrewAI crews that gracefully handle API constraints and network instability. Start with conservative limits, monitor your usage patterns, and gradually tune your parameters based on real-world data. The investment in resilient infrastructure pays off every time your crew encounters a rate limit or transient error and recovers automatically instead of crashing the entire workflow.