Quickstart Guide
Get up and running with AgentRouter in just a few minutes! This guide will walk you through installation, basic setup, and creating your first multi-agent system.
Prerequisites
- Python 3.8 or higher
- pip package manager
- An API key for your chosen LLM provider
Installation
Install AgentRouter using pip:
pip install agentrouter
Your First Agent System
Let's create a simple customer service system with a manager and worker agents.
Step 1: Import Required Components
from agentrouter import ManagerAgent, WorkerAgent, tool
import asyncio
Step 2: Create a Manager Agent
The manager agent orchestrates the entire system. It acts as the Planning Agent, analyzing messages and deciding when to issue Tool and Agent calls:
manager = ManagerAgent(
name="customer_service_manager",
api_key="your-api-key-here", # Replace with your actual API key
model="usf-mini", # Or any compatible model
backstory="I am a senior customer service manager with 10 years of experience",
goal="Resolve customer issues efficiently and professionally",
instruction="Analyze customer requests and delegate to appropriate teams",
temperature=0.7,
max_iterations=30
)
Step 3: Create Worker Agents
Workers handle specialized tasks:
# Technical Support Worker
tech_support = manager.create_worker(
name="Technical_Support",
role="Handles technical support for hardware and software issues", # REQUIRED: Concise role
backstory="Expert in hardware and software troubleshooting",
goal="Resolve technical issues quickly"
)
# Billing Support Worker
billing_support = manager.create_worker(
name="Billing_Support",
role="Manages billing inquiries and payment processing", # REQUIRED: Clear role
backstory="Expert in payment processing and billing systems",
goal="Handle billing inquiries and payment issues"
)
Step 4: Define and Register Tools
Tools are Python functions that agents can use:
@tool(
schema={
"type": "function",
"function": {
"name": "search_knowledge_base",
"description": "Search the knowledge base for solutions",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search query"
},
"category": {
"type": "string",
"enum": ["technical", "billing", "general"],
"description": "Category to search in"
}
},
"required": ["query"]
}
}
}
)
def search_knowledge_base(query: str, category: str = "general"):
"""Search for solutions in the knowledge base"""
# Simulated search
solutions = {
"password reset": "1. Click 'Forgot Password'\n2. Check email\n3. Follow reset link",
"payment failed": "1. Verify card details\n2. Check available balance\n3. Try another payment method",
"screen flickering": "1. Update graphics drivers\n2. Check cable connections\n3. Adjust refresh rate"
}
for key, solution in solutions.items():
if key in query.lower():
return {"found": True, "solution": solution}
return {"found": False, "message": "No solution found. Please contact support."}
# Register tools with workers
tech_support.register_tool(search_knowledge_base)
billing_support.register_tool(search_knowledge_base)
Step 5: Execute Tasks
Now you can process customer requests:
async def handle_customer_request():
# Customer request
messages = [
{
"role": "user",
"content": "I forgot my password and need help resetting it"
}
]
# Execute with the manager
response = await manager.execute(messages)
# Print the response
print(response["choices"][0]["message"]["content"])
return response
# Run the async function
response = asyncio.run(handle_customer_request())
Complete Example
Here's the complete code in one place:
from agentrouter import ManagerAgent, tool
import asyncio
# Initialize Manager
manager = ManagerAgent(
name="customer_service_manager",
api_key="your-api-key-here",
model="usf-mini",
backstory="Senior customer service manager with 10 years experience",
goal="Resolve customer issues efficiently",
max_iterations=30
)
# Create Workers
tech_support = manager.create_worker(
name="Technical_Support",
role="Handles technical support for hardware and software issues" # REQUIRED: Worker role
)
# Define Tools
@tool(
schema={
"type": "function",
"function": {
"name": "search_kb",
"description": "Search knowledge base",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string"}
},
"required": ["query"]
}
}
}
)
def search_kb(query: str):
# Implement your search logic
return {"results": f"Found solutions for: {query}"}
# Register Tools
tech_support.register_tool(search_kb)
# Execute Task
async def main():
messages = [
{"role": "user", "content": "My laptop screen is flickering"}
]
response = await manager.execute(messages)
print(response["choices"][0]["message"]["content"])
# Run
asyncio.run(main())
Configuration Options
AgentRouter provides extensive configuration options:
manager = ManagerAgent(
name="my_manager",
api_key="your-api-key",
model="usf-mini",
# Execution Configuration
max_iterations=30, # Maximum workflow iterations (3-50)
# Network Configuration
api_timeout=60.0, # API request timeout (5-300 seconds)
worker_timeout=60.0, # Worker execution timeout (5-300 seconds)
max_retries=3, # Maximum retry attempts (0-10)
retry_delay=1.0, # Initial retry delay (0.1-60 seconds)
retry_multiplier=2.0, # Exponential backoff multiplier (1-10)
retry_max_wait=60.0, # Maximum wait between retries (1-300 seconds)
# Agent Configuration
temperature=0.7, # LLM temperature (0-1)
max_tokens=2000, # Maximum response tokens
provider="openai", # LLM provider
)
Worker Inheritance
Worker agents inherit configuration from their parent unless explicitly overridden:
# Worker inherits parent's API key and model
worker = manager.create_worker(
name="My_Worker",
role="Specialist",
# Inherits: api_key, model, timeouts, retries from manager
)
# Override specific settings
worker = manager.create_worker(
name="Independent_Worker",
role="Specialist",
api_key="different-api-key", # Use different API key
model="usf-mini", # Use different model
max_iterations=5 # Custom iteration limit
)
Common Patterns
Pattern 1: Shared Workers
from agentrouter import WorkerAgent
# Create a standalone worker
shared_diagnostics = WorkerAgent(
name="Diagnostics",
api_key="api-key",
role="Performs system diagnostics and troubleshooting" # REQUIRED: Clear role
)
# Attach to multiple managers
manager1.attach_worker(shared_diagnostics)
manager2.attach_worker(shared_diagnostics)
Pattern 2: Nested Workers
# Create multi-level hierarchy
level1_worker = manager.create_worker(
name="L1_Worker",
role="Coordinates team operations and task delegation" # REQUIRED
)
level2_worker = level1_worker.create_worker(
name="L2_Worker",
role="Handles specialized subtasks and reporting" # REQUIRED
)
level3_worker = level2_worker.create_worker(
name="L3_Worker",
role="Executes specific technical operations" # REQUIRED
)
Pattern 3: Tool Chains
# Multiple tools working together
@tool(schema={...})
def get_user_data(user_id: str):
return {"name": "John", "account": "premium"}
@tool(schema={...})
def check_eligibility(account_type: str):
return {"eligible": account_type == "premium"}
worker.register_tool(get_user_data)
worker.register_tool(check_eligibility)
Troubleshooting
Common Issues
-
API Key Errors
# Ensure API key is set correctly
manager = ManagerAgent(
api_key=os.getenv("API_KEY") # Use environment variable
) -
Timeout Issues
# Increase timeouts for long-running tasks
manager = ManagerAgent(
api_timeout=120.0,
worker_timeout=150.0
) -
Max Iterations Reached
# Increase max_iterations for complex workflows
manager = ManagerAgent(
max_iterations=50 # Maximum allowed
)