Customer Service Multi-Agent System
Build a comprehensive customer service system with specialized agents for technical support, billing, and general inquiries.
Overview
This example demonstrates how to build a multi-agent customer service system that can:
- Route customer inquiries to specialized agents
- Search knowledge bases for solutions
- Create and manage support tickets
- Handle technical, billing, and general inquiries
- Escalate complex issues through worker hierarchies
Perfect for: Customer support platforms, help desk systems, and automated service centers.
How It Works
Workflow Steps
- Customer Query Reception: Manager agent receives customer inquiry
- Intent Analysis: Manager analyzes the request type (technical, billing, general)
- Agent Delegation: Routes to appropriate specialized worker agent
- Knowledge Search: Worker searches relevant knowledge base
- Solution Generation: Worker formulates response or creates ticket
- Response Delivery: Manager consolidates and delivers final response
Agent Hierarchy
Install AgentRouter
# Install AgentRouter SDK
!pip install agentrouter -q
# Import required libraries
from agentrouter import ManagerAgent, WorkerAgent, tool
from agentrouter.visualization import PipelineInspector
import asyncio
import nest_asyncio
from typing import Dict, Any
from datetime import datetime
import json
nest_asyncio.apply()
Complete Code
# ============================================
# CUSTOMER SERVICE MULTI-AGENT SYSTEM
# ============================================
# Configuration
API_KEY = "your-api-key-here" # Replace with your actual API key
# ============================================
# TOOL DEFINITIONS
# ============================================
@tool(
schema={
"type": "function",
"function": {
"name": "search_knowledge_base",
"description": "Searches internal knowledge base for proven solutions, troubleshooting guides, and documented fixes across all support categories",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search query"
},
"category": {
"type": "string",
"enum": ["technical", "billing", "account", "general"],
"description": "Category to search in"
}
},
"required": ["query"]
}
}
}
)
async def search_knowledge_base(query: str, category: str = "general") -> Dict[str, Any]:
"""Search knowledge base for solutions"""
# Simulated knowledge base
knowledge_base = {
"password reset": {
"solution": "1. Go to login page\n2. Click 'Forgot Password'\n3. Enter your email\n4. Check your inbox for reset link\n5. Create new password",
"category": "account",
"article_id": "KB001"
},
"payment failed": {
"solution": "1. Verify card details are correct\n2. Check card expiration date\n3. Ensure sufficient funds\n4. Try alternative payment method\n5. Contact your bank if issue persists",
"category": "billing",
"article_id": "KB002"
},
"screen flickering": {
"solution": "1. Update graphics drivers\n2. Check display cable connection\n3. Adjust screen refresh rate\n4. Test with different monitor\n5. Check for hardware issues",
"category": "technical",
"article_id": "KB003"
},
"slow performance": {
"solution": "1. Close unnecessary programs\n2. Check for malware\n3. Clear temporary files\n4. Update system drivers\n5. Consider hardware upgrade",
"category": "technical",
"article_id": "KB004"
},
"refund request": {
"solution": "1. Log into your account\n2. Go to Order History\n3. Select the order\n4. Click 'Request Refund'\n5. Provide reason and submit",
"category": "billing",
"article_id": "KB005"
}
}
# Search for matching solutions
for key, value in knowledge_base.items():
if key in query.lower():
return {
"found": True,
"query": query,
"category": category,
"solution": value["solution"],
"article_id": value["article_id"]
}
return {
"found": False,
"query": query,
"category": category,
"message": "No solution found. A support agent will assist you."
}
@tool(
schema={
"type": "function",
"function": {
"name": "create_support_ticket",
"description": "Creates prioritized support tickets with automatic routing to appropriate teams, SLA tracking, and escalation workflows",
"parameters": {
"type": "object",
"properties": {
"customer_id": {
"type": "string",
"description": "Customer ID"
},
"issue": {
"type": "string",
"description": "Issue description"
},
"priority": {
"type": "string",
"enum": ["low", "medium", "high", "critical"],
"description": "Ticket priority"
},
"category": {
"type": "string",
"enum": ["technical", "billing", "account", "general"],
"description": "Issue category"
}
},
"required": ["customer_id", "issue", "priority", "category"]
}
}
}
)
async def create_support_ticket(customer_id: str, issue: str, priority: str, category: str) -> Dict[str, Any]:
"""Create a support ticket"""
ticket_id = f"TKT{datetime.now().strftime('%Y%m%d%H%M%S')}"
return {
"ticket_id": ticket_id,
"customer_id": customer_id,
"issue": issue,
"priority": priority,
"category": category,
"status": "open",
"created_at": datetime.now().isoformat(),
"estimated_response": "2-4 hours" if priority in ["high", "critical"] else "24-48 hours"
}
@tool(
schema={
"type": "function",
"function": {
"name": "check_system_status",
"description": "Monitors real-time health metrics, uptime statistics, and performance indicators for critical infrastructure services",
"parameters": {
"type": "object",
"properties": {
"service": {
"type": "string",
"description": "Service to check (website, api, database, payment)"
}
},
"required": ["service"]
}
}
}
)
async def check_system_status(service: str) -> Dict[str, Any]:
"""Check system status"""
import random
statuses = {
"website": {"status": "operational", "uptime": "99.9%", "response_time": "120ms"},
"api": {"status": "operational", "uptime": "99.95%", "response_time": "45ms"},
"database": {"status": "operational", "uptime": "99.99%", "response_time": "5ms"},
"payment": {"status": "degraded", "uptime": "98.5%", "response_time": "800ms"}
}
service_data = statuses.get(service.lower(), {"status": "unknown", "uptime": "N/A", "response_time": "N/A"})
return {
"service": service,
"status": service_data["status"],
"uptime": service_data["uptime"],
"response_time": service_data["response_time"],
"last_checked": datetime.now().isoformat(),
"next_maintenance": "2025-02-01 02:00 UTC" if service == "database" else None
}
@tool(
schema={
"type": "function",
"function": {
"name": "check_account_status",
"description": "Retrieves comprehensive account information including subscription tier, payment history, support privileges, and active service tickets",
"parameters": {
"type": "object",
"properties": {
"customer_id": {
"type": "string",
"description": "Customer ID"
}
},
"required": ["customer_id"]
}
}
}
)
async def check_account_status(customer_id: str) -> Dict[str, Any]:
"""Check customer account status"""
# Simulated account data
return {
"customer_id": customer_id,
"account_type": "Premium",
"status": "active",
"created_date": "2023-01-15",
"last_login": datetime.now().isoformat(),
"outstanding_balance": 0,
"support_tier": "priority",
"open_tickets": 1
}
# ============================================
# AGENT SYSTEM SETUP
# ============================================
class CustomerServiceSystem:
"""Complete customer service multi-agent system"""
def __init__(self, api_key: str):
self.api_key = api_key
self.manager = None
self.agents = {}
def setup(self):
"""Setup the multi-agent hierarchy"""
# Create Manager Agent
self.manager = ManagerAgent(
name="customer_service_manager",
api_key=self.api_key,
model="usf-mini",
backstory="""I am a senior customer service manager with 15 years of experience
in handling customer inquiries across technical support, billing issues, and
general assistance. I excel at understanding customer needs and routing them
to the right specialists.""",
goal="Resolve customer issues efficiently and ensure customer satisfaction",
instruction="""
1. Analyze customer requests to understand the type of support needed
2. Delegate to appropriate specialized teams (technical, billing, or general)
3. Use available tools to search knowledge base and create tickets
4. Ensure timely and accurate responses to customers
5. Maintain professional and empathetic communication
""",
knowledge_cutoff="15 January 2025",
temperature=0.7,
max_iterations=20
)
# Register manager tools
self.manager.register_tool(search_knowledge_base)
self.manager.register_tool(create_support_ticket)
# Create Level 1 Workers
# Technical Support Worker
tech_support = self.manager.create_worker(
name="Technical_Support",
role="Diagnoses and resolves hardware/software issues through systematic troubleshooting and technical expertise",
backstory="Expert in hardware and software troubleshooting with 10 years experience",
goal="Resolve technical issues quickly and effectively",
instruction="Provide clear technical solutions and troubleshooting steps"
)
tech_support.register_tool(search_knowledge_base)
tech_support.register_tool(check_system_status)
self.agents["tech_support"] = tech_support
# Billing Support Worker
billing_support = self.manager.create_worker(
name="Billing_Support",
role="Processes payment transactions, resolves billing disputes, and manages subscription adjustments with financial accuracy",
backstory="Financial operations expert with deep knowledge of payment systems",
goal="Resolve billing issues and payment problems",
instruction="Handle billing inquiries professionally and accurately"
)
billing_support.register_tool(check_account_status)
billing_support.register_tool(create_support_ticket)
self.agents["billing_support"] = billing_support
# General Support Worker
general_support = self.manager.create_worker(
name="General_Support",
role="Delivers first-contact resolution for common inquiries and routes complex issues to specialized teams",
backstory="Friendly support agent skilled in general assistance",
goal="Provide helpful information and guidance",
instruction="Assist with general inquiries and direct to resources"
)
general_support.register_tool(search_knowledge_base)
general_support.register_tool(create_support_ticket)
self.agents["general_support"] = general_support
# Create Level 2 Workers (under Technical Support)
# Hardware Specialist
hardware_specialist = tech_support.create_worker(
name="Hardware_Specialist",
role="Performs component-level diagnostics, identifies hardware failures, and recommends replacement or repair solutions",
backstory="Specialized in hardware troubleshooting and repairs",
goal="Diagnose and resolve hardware-related issues"
)
hardware_specialist.register_tool(check_system_status)
# Software Specialist
software_specialist = tech_support.create_worker(
name="Software_Specialist",
role="Debugs application errors, optimizes software configurations, and deploys patches for system stability",
backstory="Expert in software installation, configuration, and troubleshooting",
goal="Resolve software-related issues and bugs"
)
software_specialist.register_tool(search_knowledge_base)
# Payment Processor (under Billing Support)
payment_processor = billing_support.create_worker(
name="Payment_Processor",
role="Executes secure payment operations, validates transactions, and resolves gateway-specific processing errors",
backstory="Expert in payment gateway operations and transaction processing",
goal="Ensure smooth payment processing and resolve transaction issues"
)
payment_processor.register_tool(check_account_status)
print("✅ Customer Service System initialized successfully!")
# Use PipelineInspector to visualize the hierarchy
inspector = PipelineInspector(self.manager)
print("\n📊 AGENT HIERARCHY:")
print(inspector.visualize(format='mermaid'))
print("\n🔧 TOOLS REGISTERED:")
# Get all tools from inspector
for agent in inspector.pipeline_data['agents']:
if agent['tools']:
print(f" {agent['name']}:")
for tool_name in agent['tools']:
print(f" • {tool_name}")
async def handle_request(self, customer_message: str, customer_id: str = "CUST001"):
"""Handle a customer service request"""
messages = [
{
"role": "system",
"content": f"Customer ID: {customer_id}\nDate: {datetime.now().strftime('%Y-%m-%d %H:%M')}"
},
{
"role": "user",
"content": customer_message
}
]
print(f"\n👤 Customer ({customer_id}): {customer_message}")
print("🤖 Processing with multi-agent system...")
try:
response = await self.manager.execute(messages)
if response and "choices" in response:
result = response["choices"][0]["message"]["content"]
print(f"\n✅ Response:\n{result}")
return {
"success": True,
"customer_id": customer_id,
"query": customer_message,
"response": result,
"timestamp": datetime.now().isoformat()
}
except Exception as e:
print(f"❌ Error: {str(e)}")
return {
"success": False,
"customer_id": customer_id,
"query": customer_message,
"error": str(e),
"timestamp": datetime.now().isoformat()
}
# ============================================
# EXAMPLE USAGE
# ============================================
async def run_example():
"""Run a single customer service scenario"""
# Initialize system
service = CustomerServiceSystem(API_KEY)
service.setup()
# Test scenario
test_case = {
"customer_id": "CUST001",
"message": "I forgot my password and can't log into my account. Can you help me reset it?"
}
print("\n" + "="*60)
print("🎯 RUNNING CUSTOMER SERVICE SCENARIO")
print("="*60)
result = await service.handle_request(
test_case["message"],
test_case["customer_id"]
)
# Display result
print("\n📋 RESULT:")
print(json.dumps(result, indent=2))
return result
# ============================================
# MAIN EXECUTION
# ============================================
# Run the example
result = await run_example()
Expected Output
✅ Customer Service System initialized successfully!
📊 AGENT HIERARCHY:
graph TB
Customer_Service_Manager["Customer_Service_Manager"]
Technical_Support["Technical_Support"]
Billing_Support["Billing_Support"]
General_Support["General_Support"]
Hardware_Specialist["Hardware_Specialist"]
Software_Specialist["Software_Specialist"]
Payment_Processor["Payment_Processor"]
Customer_Service_Manager --> Technical_Support
Customer_Service_Manager --> Billing_Support
Customer_Service_Manager --> General_Support
Technical_Support --> Hardware_Specialist
Technical_Support --> Software_Specialist
Billing_Support --> Payment_Processor
🔧 TOOLS REGISTERED:
• search_knowledge_base
• create_support_ticket
• check_system_status
• check_account_status
============================================================
🎯 RUNNING CUSTOMER SERVICE SCENARIO
============================================================
👤 Customer (CUST001): I forgot my password and can't log into my account. Can you help me reset it?
🤖 Processing with multi-agent system...
✅ Response:
I can definitely help you reset your password. Here's the step-by-step process:
1. Go to the login page
2. Click 'Forgot Password'
3. Enter your email address
4. Check your inbox for the reset link
5. Create a new password
The reset email should arrive within a few minutes. If you don't see it, please check your spam folder. For security reasons, the link will expire in 24 hours.
📋 RESULT:
{
"success": true,
"customer_id": "CUST001",
"query": "I forgot my password and can't log into my account. Can you help me reset it?",
"response": "I can definitely help you reset your password...",
"timestamp": "2025-01-11T10:15:30.123456"
}
Customization Options
1. Add More Specialized Workers
# Add a Security Specialist
security_specialist = manager.create_worker(
name="Security_Specialist",
role="Manages account security and privacy protection", # REQUIRED: Concise role
backstory="Cybersecurity expert focused on account protection",
goal="Ensure account security and privacy"
)
2. Implement Custom Tools
@tool(schema={...})
def escalate_to_human(issue: str, priority: str):
"""Escalate complex issues to human agents"""
return {"escalated": True, "queue": "human_support", "wait_time": "5 min"}
3. Add Multi-Language Support
@tool(schema={...})
def translate_response(text: str, target_language: str):
"""Translate responses to customer's language"""
# Integration with translation API
return {"translated_text": text, "language": target_language}
Production Considerations
- API Key Security: Use environment variables for API keys
- Rate Limiting: Implement rate limiting for customer requests
- Logging: Add comprehensive logging for audit trails
- Caching: Cache knowledge base searches for performance
- Monitoring: Track response times and success rates
- Fallback: Always have human escalation path