Reliability
Handle failures gracefully with retry, timeout, fallback, caching, and context limits.
Retry with backoff
Automatically retry failed requests with exponential backoff:
agent = Agent(retries=3)
# On failure: wait 1s, retry, wait 2s, retry, wait 4s, retry
Useful for:
- Rate limit errors (429)
- Temporary API outages
- Network issues
Timeout
Set maximum time for each request:
agent = Agent(timeout=30.0) # 30 seconds max
Raises asyncio.TimeoutError if exceeded.
Context limit
Auto-trim old messages to manage context window:
agent = Agent(max_messages=20)
# Keeps system prompt + last 19 messages
Essential for:
- Long conversations
- Avoiding context overflow errors
- Controlling costs
Fallback provider
Automatic failover to another provider:
agent = Agent(
provider="openai",
fallback="mistral", # If OpenAI fails, try Mistral
)
The fallback activates when:
- Primary provider returns an error
- All retries are exhausted (if
retriesis set)
You can combine with retries:
agent = Agent(
provider="openai",
fallback="mistral",
retries=2, # Retry OpenAI twice, then try Mistral
)
Caching
Cache responses to avoid repeated API calls:
from pure_agents import Agent, clear_cache
agent = Agent(cache=True)
# First call hits API
result1 = await agent.run("What is 2+2?")
# Same prompt returns cached response instantly
result2 = await agent.run("What is 2+2?")
# Clear the cache if needed
clear_cache()
Cache keys are based on:
- Provider and model
- Messages (prompts)
- Tools configuration
Useful for:
- Development and testing
- Repeated queries
- Cost reduction
Tool timeout
Per-tool execution limits:
@tool(timeout=10)
async def slow_api(query: str) -> str:
"""Call slow external API with 10s limit."""
return await external_service.call(query)
agent = Agent(tools=[slow_api])
Raises asyncio.TimeoutError if exceeded.
Combined usage
agent = Agent(
provider="openai",
fallback="mistral",
retries=2,
timeout=60.0,
max_messages=50,
cache=True,
)
Debug mode
See retry attempts with debug=True:
agent = Agent(retries=3, debug=True)
# Prints: [Retry 1] Connection error. Waiting 1s...
Error handling
from asyncio import TimeoutError
try:
result = await agent.run("Complex task")
except TimeoutError:
print("Request timed out")
except Exception as e:
print(f"All retries failed: {e}")