What is AI Agent Infrastructure?
AI agent infrastructure is the backbone that enables autonomous AI agents to operate reliably in production environments. It's not just about building a prompt and calling an LLM — it's the entire technical stack that handles orchestration, state management, tool execution, error recovery, logging, and deployment. For solo developers, this infrastructure determines whether your agent is a fragile prototype or a production-grade system that can serve real users.
Think of it as the operating system for your AI agents. Just as a web application needs a server, database, and monitoring, an AI agent needs a runtime environment that manages:
- Context windows — intelligently compressing and managing conversation history so the LLM never loses track
- Tool calling — safely executing functions, APIs, and external services with proper validation and retry logic
- Memory persistence — storing long-term knowledge across sessions and conversations
- Multi-step reasoning — breaking complex tasks into manageable sub-tasks with checkpoints
- Observability — logging every decision, tool call, and response for debugging and improvement
Without dedicated infrastructure, a solo developer ends up writing ad-hoc glue code that works in demos but crumbles under real-world edge cases. With the right infrastructure, you can focus on your agent's unique capabilities while the framework handles the heavy lifting.
Why It Matters for Solo Developers
As a solo developer, your time is your scarcest resource. Building an AI agent from scratch means you'll spend 80% of your effort on infrastructure concerns — retry logic, state serialization, prompt management, error handling — and only 20% on the actual agent logic that differentiates your product. A solid infrastructure flips this ratio.
Here's what changes when you adopt a proper agent infrastructure:
- Faster iteration cycles — swap out LLM providers, add new tools, or modify agent behavior without rewriting the entire pipeline
- Production reliability — built-in retries, fallbacks, and circuit breakers prevent cascading failures when an API goes down
- Cost control — automatic context compression and token budgeting prevent runaway API bills
- Debugging superpowers — structured logs and traces let you replay exactly what your agent thought and did at each step
- Seamless scaling — move from a single conversation to hundreds of concurrent users without architectural rewrites
The difference between a weekend hackathon project and a product people pay for often comes down to infrastructure. Let's build one.
Core Components: A Practical Breakdown
1. The Agent Runtime Loop
At the heart of every agent infrastructure is a runtime loop that cycles through reasoning, acting, and observing until a goal is met. Here's a minimal but production-ready implementation:
import asyncio
import json
from datetime import datetime
from typing import Any, Dict, List, Optional
class AgentRuntime:
def __init__(
self,
llm_client,
tools: List[Dict[str, Any]],
max_iterations: int = 10,
context_budget: int = 8000
):
self.llm = llm_client
self.tools = tools
self.max_iterations = max_iterations
self.context_budget = context_budget
self.execution_log: List[Dict[str, Any]] = []
async def run(self, task: str, memory: Optional[List[Dict]] = None) -> Dict[str, Any]:
messages = self._build_system_prompt()
if memory:
messages.extend(memory)
messages.append({"role": "user", "content": task})
for iteration in range(self.max_iterations):
# Log the iteration start
step_log = {
"iteration": iteration,
"timestamp": datetime.utcnow().isoformat(),
"messages_snapshot": len(messages)
}
# Call the LLM with tool definitions
response = await self.llm.chat(
messages=messages,
tools=self.tools,
tool_choice="auto"
)
message = response["choices"][0]["message"]
step_log["llm_response"] = message
messages.append(message)
# Check if the agent wants to call tools
if message.get("tool_calls"):
for tool_call in message["tool_calls"]:
tool_result = await self._execute_tool(
tool_call["function"]["name"],
json.loads(tool_call["function"]["arguments"])
)
messages.append({
"role": "tool",
"tool_call_id": tool_call["id"],
"content": json.dumps(tool_result)
})
step_log["tool_results"] = tool_result
# Check for final answer (no tool calls)
elif message.get("content") and not message.get("tool_calls"):
self.execution_log.append(step_log)
return {
"status": "completed",
"answer": message["content"],
"iterations_used": iteration + 1,
"execution_log": self.execution_log
}
# Context budget check
if self._estimate_tokens(messages) > self.context_budget:
messages = await self._compress_context(messages)
self.execution_log.append(step_log)
return {
"status": "max_iterations_reached",
"answer": messages[-1].get("content", "No final answer produced"),
"execution_log": self.execution_log
}
async def _execute_tool(self, name: str, args: Dict[str, Any]) -> Any:
try:
# Find the tool function by name
tool_fn = self._tool_registry.get(name)
if not tool_fn:
return {"error": f"Tool '{name}' not found"}
# Execute with timeout and retry
result = await asyncio.wait_for(
self._retry_wrapper(tool_fn, args),
timeout=30.0
)
return {"success": True, "data": result}
except asyncio.TimeoutError:
return {"error": "Tool execution timed out"}
except Exception as e:
return {"error": str(e), "type": type(e).__name__}
async def _retry_wrapper(self, fn, args, max_retries=3):
for attempt in range(max_retries):
try:
return await fn(**args)
except Exception as e:
if attempt == max_retries - 1:
raise
await asyncio.sleep(2 ** attempt)
def _build_system_prompt(self) -> List[Dict]:
return [{
"role": "system",
"content": (
"You are an autonomous AI agent. Use the provided tools to accomplish the user's task. "
"Plan your steps, execute tools, observe results, and adapt. "
"When you have completed the task, provide a clear final answer without calling additional tools."
)
}]
def _estimate_tokens(self, messages: List[Dict]) -> int:
# Rough estimate: 1 token ≈ 4 characters for English text
total_chars = sum(len(json.dumps(m)) for m in messages)
return total_chars // 4
async def _compress_context(self, messages: List[Dict]) -> List[Dict]:
# Summarize older messages, keep recent ones intact
if len(messages) <= 4:
return messages
older_messages = messages[:-4]
recent_messages = messages[-4:]
summary_prompt = (
"Summarize the following conversation history concisely, "
"preserving all key facts, decisions, and tool results:\n\n" +
"\n".join([f"{m['role']}: {m.get('content', str(m.get('tool_calls', '')))}"
for m in older_messages])
)
summary_response = await self.llm.chat(
messages=[{"role": "user", "content": summary_prompt}],
tools=[],
tool_choice="none"
)
summary = summary_response["choices"][0]["message"]["content"]
return [
messages[0], # Keep system prompt
{"role": "assistant", "content": f"[Context Summary] {summary}"}
] + recent_messages2. Tool Registry with Validation
Tools are your agent's hands. A robust tool registry validates inputs before execution and provides clear error messages back to the LLM so it can self-correct:
from functools import wraps
from typing import Callable, get_type_hints
import inspect
class ToolRegistry:
def __init__(self):
self._tools: Dict[str, Callable] = {}
self._schemas: Dict[str, Dict] = {}
def register(self, name: str, description: str, parameters: Dict[str, Any]):
"""Decorator that registers a function as an agent tool with schema validation."""
def decorator(fn: Callable):
@wraps(fn)
async def wrapper(**kwargs):
# Validate required parameters
required = [p for p, schema in parameters.get("properties", {}).items()
if p in parameters.get("required", [])]
missing = [p for p in required if p not in kwargs]
if missing:
raise ValueError(f"Missing required parameters: {missing}")
# Type coercion and validation
validated_args = {}
type_hints = get_type_hints(fn)
for param_name, param_schema in parameters.get("properties", {}).items():
if param_name in kwargs:
value = kwargs[param_name]
expected_type = param_schema.get("type")
if expected_type == "integer" and isinstance(value, float):
value = int(value)
elif expected_type == "number" and isinstance(value, str):
try:
value = float(value)
except ValueError:
raise TypeError(f"Parameter '{param_name}' must be a number")
validated_args[param_name] = value
return await fn(**validated_args)
# Store the tool
self._tools[name] = wrapper
self._schemas[name] = {
"type": "function",
"function": {
"name": name,
"description": description,
"parameters": parameters
}
}
return wrapper
return decorator
def get_tool_schemas(self) -> List[Dict]:
return list(self._schemas.values())
def get_tool(self, name: str) -> Optional[Callable]:
return self._tools.get(name)
# Usage example
tool_registry = ToolRegistry()
@tool_registry.register(
name="search_knowledge_base",
description="Search the company knowledge base for relevant information",
parameters={
"type": "object",
"properties": {
"query": {"type": "string", "description": "Search query string"},
"max_results": {"type": "integer", "description": "Maximum number of results (default: 5)"}
},
"required": ["query"]
}
)
async def search_knowledge_base(query: str, max_results: int = 5):
# Simulate a vector database search
await asyncio.sleep(0.5)
return {
"results": [
{"title": "Result 1", "relevance": 0.95, "content": "..."},
{"title": "Result 2", "relevance": 0.82, "content": "..."}
][:max_results],
"total_found": 42
}3. Persistent Memory Layer
Stateless agents forget everything between sessions. A memory layer stores facts, user preferences, and past interactions so your agent builds relationships over time:
import sqlite3
import json
from datetime import datetime
from typing import List, Dict, Optional
class MemoryStore:
def __init__(self, db_path: str = "agent_memory.db"):
self.conn = sqlite3.connect(db_path)
self.conn.row_factory = sqlite3.Row
self._init_schema()
def _init_schema(self):
self.conn.executescript("""
CREATE TABLE IF NOT EXISTS memories (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL,
memory_type TEXT NOT NULL CHECK(memory_type IN ('fact', 'preference', 'conversation', 'decision')),
content TEXT NOT NULL,
importance REAL DEFAULT 0.5,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
last_accessed TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
access_count INTEGER DEFAULT 0
);
CREATE INDEX IF NOT EXISTS idx_memories_user
ON memories(user_id, memory_type);
CREATE INDEX IF NOT EXISTS idx_memories_importance
ON memories(importance DESC);
CREATE TABLE IF NOT EXISTS conversation_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL,
session_id TEXT NOT NULL,
messages TEXT NOT NULL, -- JSON array of messages
summary TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
""")
def store(self, user_id: str, content: str, memory_type: str = "fact", importance: float = 0.5):
cursor = self.conn.execute(
"""INSERT INTO memories (user_id, memory_type, content, importance)
VALUES (?, ?, ?, ?)""",
(user_id, memory_type, content, importance)
)
self.conn.commit()
return cursor.lastrowid
def retrieve(self, user_id: str, query: str = None, memory_type: str = None, limit: int = 10) -> List[Dict]:
sql = """SELECT * FROM memories WHERE user_id = ?"""
params = [user_id]
if memory_type:
sql += " AND memory_type = ?"
params.append(memory_type)
if query:
sql += " AND content LIKE ?"
params.append(f"%{query}%")
sql += " ORDER BY importance DESC, last_accessed DESC LIMIT ?"
params.append(limit)
cursor = self.conn.execute(sql, params)
results = [dict(row) for row in cursor.fetchall()]
# Update access metadata
if results:
ids = [r['id'] for r in results]
self.conn.execute(
f"""UPDATE memories
SET last_accessed = CURRENT_TIMESTAMP,
access_count = access_count + 1
WHERE id IN ({','.join('?' * len(ids))})""",
ids
)
self.conn.commit()
return results
def save_conversation(self, user_id: str, session_id: str, messages: List[Dict], summary: str = None):
self.conn.execute(
"""INSERT INTO conversation_history (user_id, session_id, messages, summary)
VALUES (?, ?, ?, ?)""",
(user_id, session_id, json.dumps(messages), summary)
)
self.conn.commit()
def load_conversation(self, session_id: str) -> Optional[Dict]:
cursor = self.conn.execute(
"SELECT * FROM conversation_history WHERE session_id = ? ORDER BY id DESC LIMIT 1",
(session_id,)
)
row = cursor.fetchone()
if row:
result = dict(row)
result['messages'] = json.loads(result['messages'])
return result
return None
def consolidate(self, user_id: str):
"""Extract key facts from recent conversations and store as memories."""
conversations = self.conn.execute(
"""SELECT messages FROM conversation_history
WHERE user_id = ?
ORDER BY created_at DESC LIMIT 5""",
(user_id,)
).fetchall()
# In production, you'd send this to an LLM for extraction
# For now, we flag that consolidation is needed
return len(conversations)4. Observability and Logging
When your agent does something unexpected at 3 AM, structured logs are your only lifeline. Here's a logging system that captures every decision point:
import logging
import structlog
from typing import Dict, Any
import uuid
# Configure structured logging
structlog.configure(
processors=[
structlog.stdlib.filter_by_level,
structlog.stdlib.add_logger_name,
structlog.stdlib.add_log_level,
structlog.stdlib.PositionalArgumentsFormatter(),
structlog.processors.TimeStamper(fmt="iso"),
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
structlog.processors.JSONRenderer()
],
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),
cache_logger_on_first_use=True,
)
logger = structlog.get_logger()
class AgentObserver:
def __init__(self, agent_id: str):
self.agent_id = agent_id
self.trace_id = str(uuid.uuid4())
self.step_counter = 0
def log_llm_call(self, messages: List[Dict], response: Dict, latency_ms: float):
self.step_counter += 1
logger.info(
"llm_call",
agent_id=self.agent_id,
trace_id=self.trace_id,
step=self.step_counter,
input_tokens=self._count_tokens(messages),
output_tokens=self._count_tokens([response]),
latency_ms=latency_ms,
model=response.get("model", "unknown"),
finish_reason=response["choices"][0].get("finish_reason")
)
def log_tool_execution(self, tool_name: str, args: Dict, result: Any, success: bool, latency_ms: float):
self.step_counter += 1
logger.info(
"tool_execution",
agent_id=self.agent_id,
trace_id=self.trace_id,
step=self.step_counter,
tool_name=tool_name,
arguments=args,
success=success,
latency_ms=latency_ms,
result_preview=str(result)[:500] if success else str(result)
)
def log_error(self, error_type: str, message: str, context: Dict[str, Any]):
logger.error(
"agent_error",
agent_id=self.agent_id,
trace_id=self.trace_id,
step=self.step_counter,
error_type=error_type,
error_message=message,
context=context
)
def log_completion(self, result: Dict[str, Any], total_steps: int, total_cost: float):
logger.info(
"task_complete",
agent_id=self.agent_id,
trace_id=self.trace_id,
total_steps=total_steps,
total_cost_usd=total_cost,
status=result.get("status"),
answer_preview=str(result.get("answer", ""))[:200]
)
def _count_tokens(self, messages: List[Dict]) -> int:
# In production, use a proper tokenizer (tiktoken or similar)
total_chars = sum(len(str(m.get("content", ""))) for m in messages)
return total_chars // 4 # Rough heuristicBuilding Your First Agent: A Complete Example
Let's assemble all the components into a working agent that can research topics, write content, and save results. This is a production-ready foundation you can extend:
import asyncio
import os
from typing import Dict, Any, List
# Assume the classes above are imported
from agent_runtime import AgentRuntime, ToolRegistry, MemoryStore, AgentObserver
async def main():
# Initialize infrastructure
memory = MemoryStore("my_agent_memory.db")
observer = AgentObserver(agent_id="research-assistant-01")
registry = ToolRegistry()
# Define tools
@registry.register(
name="web_search",
description="Search the web for current information on any topic",
parameters={
"type": "object",
"properties": {
"query": {"type": "string", "description": "What to search for"},
"num_results": {"type": "integer", "description": "Number of results (1-10)"}
},
"required": ["query"]
}
)
async def web_search(query: str, num_results: int = 5):
# In production, call SerpAPI, Google API, or Brave Search
await asyncio.sleep(1.0)
return {
"results": [
{"title": f"Result for {query}", "url": "https://example.com", "snippet": "..."}
for _ in range(min(num_results, 5))
]
}
@registry.register(
name="save_document",
description="Save a document to the user's workspace",
parameters={
"type": "object",
"properties": {
"title": {"type": "string", "description": "Document title"},
"content": {"type": "string", "description": "Document content"},
"format": {"type": "string", "enum": ["markdown", "text", "html"]}
},
"required": ["title", "content"]
}
)
async def save_document(title: str, content: str, format: str = "markdown"):
# In production, save to S3, filesystem, or database
filename = f"{title.lower().replace(' ', '_')}.{format}"
await asyncio.sleep(0.3)
return {"saved": True, "filename": filename, "size_bytes": len(content)}
# Initialize the runtime with an LLM client
# In production, use OpenAI, Anthropic, or an open-source model
from openai import AsyncOpenAI
llm_client = AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"])
# Wrap the client to match our expected interface
class LLMWrapper:
def __init__(self, client):
self.client = client
async def chat(self, messages, tools=None, tool_choice="auto"):
kwargs = {
"model": "gpt-4-turbo",
"messages": messages,
"temperature": 0.7,
}
if tools:
kwargs["tools"] = tools
kwargs["tool_choice"] = tool_choice
start = asyncio.get_event_loop().time()
response = await self.client.chat.completions.create(**kwargs)
latency = (asyncio.get_event_loop().time() - start) * 1000
# Convert to dict format our runtime expects
return {
"choices": [{
"index": 0,
"message": {
"role": "assistant",
"content": response.choices[0].message.content,
"tool_calls": [
{
"id": tc.id,
"function": {
"name": tc.function.name,
"arguments": tc.function.arguments
}
} for tc in (response.choices[0].message.tool_calls or [])
] if response.choices[0].message.tool_calls else None
},
"finish_reason": response.choices[0].finish_reason
}],
"model": response.model,
"usage": {
"prompt_tokens": response.usage.prompt_tokens,
"completion_tokens": response.usage.completion_tokens
}
}
llm = LLMWrapper(llm_client)
# Create the agent runtime
agent = AgentRuntime(
llm_client=llm,
tools=registry.get_tool_schemas(),
max_iterations=15,
context_budget=16000
)
agent._tool_registry = registry._tools # Wire up tool execution
# Load user context from memory
user_id = "user_123"
past_memories = memory.retrieve(user_id, limit=5)
memory_context = [{
"role": "system",
"content": f"Relevant past context: {json.dumps(past_memories)}"
}] if past_memories else []
# Run the agent
task = "Research the latest developments in quantum computing and write a brief summary document"
observer.log_llm_call(
[{"role": "user", "content": task}],
{"choices": [{"finish_reason": "started"}]},
0
)
result = await agent.run(task, memory=memory_context)
observer.log_completion(result, result.get("iterations_used", 0), 0.05)
# Save conversation to memory
if result["status"] == "completed":
memory.store(user_id, f"Researched quantum computing: {result['answer'][:100]}", "fact", 0.8)
print(f"Agent completed in {result['iterations_used']} steps")
print(f"Answer: {result['answer']}")
if __name__ == "__main__":
asyncio.run(main())Deploying to Production
Moving from local development to production requires containerization, API exposure, and proper environment management. Here's a production-ready deployment setup:
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Copy agent code
COPY agent_runtime/ ./agent_runtime/
COPY tools/ ./tools/
COPY main.py .
# Create non-root user for security
RUN useradd -m -s /bin/bash agent && chown -R agent:agent /app
USER agent
# Health check endpoint
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s \
CMD curl -f http://localhost:8080/health || exit 1
EXPOSE 8080
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8080"]Now the FastAPI application that wraps your agent as a REST API:
# main.py - Production API Server
from fastapi import FastAPI, HTTPException, BackgroundTasks
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from typing import Optional, Dict, Any
import asyncio
import uuid
app = FastAPI(title="AI Agent API", version="1.0.0")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
# In-memory task tracking (use Redis/DB in production)
task_store: Dict[str, Dict[str, Any]] = {}
class TaskRequest(BaseModel):
task: str = Field(..., description="The task for the agent to perform")
user_id: str = Field(..., description="Unique user identifier")
context: Optional[Dict[str, Any]] = Field(default={})
max_iterations: Optional[int] = Field(default=15, ge=1, le=50)
class TaskResponse(BaseModel):
task_id: str
status: str
message: str
class TaskResult(BaseModel):
task_id: str
status: str
answer: Optional[str]
iterations_used: int
execution_log: Optional[list]
@app.on_event("startup")
async def startup():
# Initialize agent infrastructure
# In production, load configuration from environment variables
app.state.agent = None # Will be initialized lazily per request or pooled
app.state.memory = MemoryStore("/data/agent_memory.db")
@app.post("/agent/run", response_model=TaskResponse)
async def run_agent_task(request: TaskRequest, background_tasks: BackgroundTasks):
task_id = str(uuid.uuid4())
task_store[task_id] = {"status": "pending", "created_at": datetime.utcnow().isoformat()}
background_tasks.add_task(execute_agent_task, task_id, request)
return TaskResponse(
task_id=task_id,
status="accepted",
message="Task queued for execution"
)
async def execute_agent_task(task_id: str, request: TaskRequest):
try:
task_store[task_id]["status"] = "running"
# Initialize agent (in production, use a pool)
agent = await initialize_agent()
result = await agent.run(request.task)
task_store[task_id].update({
"status": "completed",
"answer": result.get("answer"),
"iterations_used": result.get("iterations_used", 0),
"execution_log": result.get("execution_log", [])
})
except Exception as e:
task_store[task_id].update({
"status": "failed",
"error": str(e)
})
@app.get("/agent/result/{task_id}", response_model=TaskResult)
async def get_task_result(task_id: str):
if task_id not in task_store:
raise HTTPException(status_code=404, detail="Task not found")
task_data = task_store[task_id]
return TaskResult(
task_id=task_id,
status=task_data["status"],
answer=task_data.get("answer"),
iterations_used=task_data.get("iterations_used", 0),
execution_log=task_data.get("execution_log")
)
@app.get("/health")
async def health_check():
return {"status": "healthy", "timestamp": datetime.utcnow().isoformat()}
# WebSocket endpoint for real-time agent streaming
from fastapi import WebSocket
@app.websocket("/agent/stream/{user_id}")
async def agent_stream(websocket: WebSocket, user_id: str):
await websocket.accept()
try:
while True:
task = await websocket.receive_text()
agent = await initialize_agent()
# Stream each step
# This requires modifying AgentRuntime to yield intermediate steps
await websocket.send_json({"type": "status", "message": "Agent started"})
result = await agent.run(task)
await websocket.send_json({
"type": "complete",
"answer": result.get("answer"),
"steps": result.get("iterations_used")
})
except Exception as e:
await websocket.send_json({"type": "error", "message": str(e)})
finally:
await websocket.close()Best Practices for Solo Developers
Start with the Simplest Agent That Works
Resist the urge to build a generalized agent framework from day one. Start with a single tool, a single use case, and a hardcoded system prompt. Get it working end-to-end for one real user before abstracting anything. The fastest path to production is a focused agent that does one thing exceptionally well.
Treat Your System Prompt Like Production Code
Version control your prompts. Test them with a suite of representative inputs. Monitor how changes affect completion quality and token usage. A one-word change in a system prompt can increase costs by 30% or break a critical reasoning path. Store prompts in separate files, not buried in your Python code.
# prompts/research_agent/v1.system.txt
You are a precise research assistant. Follow these rules strictly:
1. For any research task, first use web_search to gather current information
2. Analyze results critically — note dates, sources, and potential biases
3. If information is conflicting, report all sides and explain the disagreement
4. When writing documents, use clear structure: introduction, body, conclusion
5. Never fabricate information. If you cannot find something, say so honestly
6. Keep responses concise but thorough — prefer quality over quantity
# prompts/research_agent/v2.system.txt
You are a precise research assistant with enhanced reasoning...
# In your code:
def load_prompt(version: str = "v1") -> str:
prompt_path = Path(f"prompts/research_agent/{version}.system.txt")
if not prompt_path.exists():
raise ValueError(f"Prompt version {version} not found")
return prompt_path.read_text()Implement Graceful Degradation
External APIs will fail. LLM providers will rate-limit you. Your agent must continue operating at reduced capability rather than crashing entirely. Implement fallback chains: if the primary search API fails, try a backup; if the LLM times out, retry with a simpler model; if context overflows, compress aggressively rather than dropping messages.
class ResilientToolExecutor:
def __init__(self, primary_tools: Dict, fallback_tools: Dict = None):
self.primary = primary_tools
self.fallback = fallback_tools or {}
self.circuit_breaker = {name: {"failures": 0, "open_until": None}
for name in primary_tools}
async def execute(self, tool_name: str, args: Dict) -> Any:
# Check circuit breaker
breaker = self.circuit_breaker.get(tool_name)
if breaker and breaker["open_until"]:
if datetime.utcnow() < breaker["open_until"]:
# Circuit is open — try fallback
if tool_name in self.fallback:
return await self._try_tool(self.fallback[tool_name], args)
raise Exception(f"Circuit open for {tool_name}, no fallback available")
# Try primary tool
try:
result = await self._try_tool(self.primary[tool_name], args)
breaker["failures"] = 0 # Reset on success
return result
except Exception as e:
breaker["failures"] += 1
if breaker["failures"] >= 3:
breaker["open_until"] = datetime.utcnow() + timedelta(seconds=30)
# Try fallback
if tool_name in self.fallback:
return await self._try_tool(self.fallback[tool_name], args)
raise
async def _try_tool(self, tool_fn, args, timeout=10):
return await asyncio.wait_for(tool_fn(**args), timeout=timeout)Test Your Agent Like You'd Test Software
Write unit tests for individual tools, integration tests for tool chains, and end-to-end tests for complete tasks. Use a deterministic LLM mock for reproducible tests. Test edge cases: empty search results, malformed API responses, tasks that require zero tools, and tasks that exceed the iteration budget.
import pytest
from unittest.mock import AsyncMock
@pytest.fixture
def mock_llm():
client = AsyncMock()
client.chat.return_value = {
"choices": [{
"index": 0,
"message": {
"role": "assistant",
"content": "Here is the research summary you requested...",
"tool_calls": None
},
"fin