Skip to main content

Customer Service Agent

This example demonstrates how to build a customer service agent that can handle product inquiries, order support, returns, and general customer questions with a friendly, helpful tone.

Overview​

This example shows:

  • Customer-focused knowledge base
  • Order and account management tools
  • Sentiment analysis and escalation
  • Multi-channel support capabilities
  • Customer satisfaction tracking

Use Case​

Perfect for:

  • E-commerce Support - Product questions and order help
  • SaaS Customer Success - Feature explanations and onboarding
  • Retail Support - Returns, exchanges, and product info
  • Service Support - Appointment booking and service inquiries

Code Implementation​

import os
import json
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple
from dotenv import load_dotenv
from recoagent import RecoAgent
from recoagent.tools import Tool
from recoagent.agents import RAGAgentGraph
from recoagent.policies import SafetyPolicy

# Load environment variables
load_dotenv()

class CustomerServiceAgent:
def __init__(self):
"""Initialize the Customer Service Agent."""
self.agent = RAGAgentGraph(
llm_provider="openai",
llm_model="gpt-4",
embedding_model="text-embedding-ada-002",
chunk_size=300,
chunk_overlap=30
)

# Add customer service tools
self._setup_tools()

# Set up customer service policies
self._setup_policies()

# Load customer service knowledge base
self._load_knowledge_base()

# Initialize customer data
self.customers = {}
self.orders = {}
self.conversations = {}
self.satisfaction_scores = []

def _setup_tools(self):
"""Add customer service tools to the agent."""

def lookup_customer(email: str) -> str:
"""Look up customer information by email."""
# Simulate customer database lookup
customer_data = {
"john.doe@email.com": {
"name": "John Doe",
"customer_id": "CUST001",
"member_since": "2023-01-15",
"tier": "Gold",
"total_orders": 12,
"lifetime_value": 1250.00
},
"jane.smith@email.com": {
"name": "Jane Smith",
"customer_id": "CUST002",
"member_since": "2023-03-22",
"tier": "Silver",
"total_orders": 5,
"lifetime_value": 450.00
}
}

customer = customer_data.get(email.lower())
if customer:
return f"Customer found: {customer['name']} (ID: {customer['customer_id']}), {customer['tier']} member since {customer['member_since']}, {customer['total_orders']} orders, ${customer['lifetime_value']} lifetime value"
else:
return f"No customer found with email {email}"

def lookup_order(order_number: str) -> str:
"""Look up order information by order number."""
# Simulate order database lookup
order_data = {
"ORD001": {
"customer_email": "john.doe@email.com",
"status": "Shipped",
"items": ["Wireless Headphones", "Phone Case"],
"total": 89.99,
"shipped_date": "2023-12-01",
"tracking": "TRK123456789"
},
"ORD002": {
"customer_email": "jane.smith@email.com",
"status": "Processing",
"items": ["Laptop Stand"],
"total": 45.99,
"shipped_date": None,
"tracking": None
}
}

order = order_data.get(order_number.upper())
if order:
status_info = f"Status: {order['status']}"
if order['shipped_date']:
status_info += f", Shipped: {order['shipped_date']}, Tracking: {order['tracking']}"

return f"Order {order_number}: {status_info}, Items: {', '.join(order['items'])}, Total: ${order['total']}"
else:
return f"Order {order_number} not found"

def process_return(order_number: str, reason: str) -> str:
"""Process a return request."""
# Simulate return processing
return_id = f"RET{datetime.now().strftime('%Y%m%d')}{order_number[-3:]}"

return f"Return request {return_id} created for order {order_number}. Reason: {reason}. Return label will be emailed within 24 hours."

def check_product_availability(product_name: str) -> str:
"""Check if a product is available and get details."""
# Simulate product database
products = {
"wireless headphones": {"available": True, "price": 79.99, "stock": 45},
"phone case": {"available": True, "price": 24.99, "stock": 120},
"laptop stand": {"available": True, "price": 45.99, "stock": 23},
"bluetooth speaker": {"available": False, "price": 99.99, "stock": 0}
}

product = products.get(product_name.lower())
if product:
if product["available"]:
return f"{product_name.title()} is available for ${product['price']}. {product['stock']} in stock."
else:
return f"{product_name.title()} is currently out of stock. Expected restock in 2-3 weeks."
else:
return f"Product '{product_name}' not found in our catalog."

def schedule_callback(phone_number: str, preferred_time: str) -> str:
"""Schedule a callback for the customer."""
callback_id = f"CALL{datetime.now().strftime('%Y%m%d%H%M')}"
return f"Callback scheduled for {phone_number} at {preferred_time}. Reference: {callback_id}. Our team will call within the requested timeframe."

def escalate_to_human(reason: str, customer_id: str) -> str:
"""Escalate to human agent."""
escalation_id = f"ESC{datetime.now().strftime('%Y%m%d%H%M')}"
return f"Issue escalated to human agent (ID: {escalation_id}) for customer {customer_id}. Reason: {reason}. A specialist will contact you within 2 hours."

# Add tools to agent
self.agent.add_tool(Tool(name="lookup_customer", function=lookup_customer))
self.agent.add_tool(Tool(name="lookup_order", function=lookup_order))
self.agent.add_tool(Tool(name="process_return", function=process_return))
self.agent.add_tool(Tool(name="check_product_availability", function=check_product_availability))
self.agent.add_tool(Tool(name="schedule_callback", function=schedule_callback))
self.agent.add_tool(Tool(name="escalate_to_human", function=escalate_to_human))

print("āœ… Customer Service tools loaded")

def _setup_policies(self):
"""Set up customer service policies."""
policies = SafetyPolicy(
max_cost_per_query=0.15,
max_latency_ms=8000,
allowed_topics=[
"product information", "order status", "returns", "shipping",
"account help", "billing", "technical support", "general questions"
],
escalation_triggers=[
"angry customer", "complaint", "refund request", "billing dispute",
"technical issue", "urgent", "escalate", "manager"
],
tone_requirements={
"friendly": True,
"helpful": True,
"professional": True,
"empathetic": True
}
)

self.agent.set_safety_policy(policies)
print("āœ… Customer service policies configured")

def _load_knowledge_base(self):
"""Load customer service knowledge base."""
knowledge_base = [
# Product Information
"Our wireless headphones feature noise cancellation, 30-hour battery life, and comfortable over-ear design. Price: $79.99.",
"Phone cases are available in multiple colors and provide drop protection up to 6 feet. Price: $24.99.",
"Laptop stands are adjustable and help improve ergonomics. Made from aluminum with rubber grips. Price: $45.99.",
"All products come with a 1-year manufacturer warranty and 30-day return policy.",

# Shipping and Delivery
"Standard shipping is free on orders over $50 and takes 3-5 business days.",
"Express shipping (1-2 business days) is available for $9.99.",
"International shipping is available to most countries with delivery times of 7-14 business days.",
"You can track your order using the tracking number provided in your shipping confirmation email.",

# Returns and Exchanges
"Returns are accepted within 30 days of delivery. Items must be in original condition with packaging.",
"To initiate a return, contact customer service or use our online return portal.",
"Refunds are processed within 5-7 business days after we receive your returned item.",
"Exchanges are available for different sizes or colors of the same item.",

# Account and Billing
"You can view your order history and account information by logging into your account.",
"Payment methods accepted include all major credit cards, PayPal, and Apple Pay.",
"We never store your payment information for security reasons.",
"Monthly billing statements are available in your account dashboard.",

# Customer Service
"Our customer service team is available Monday-Friday 8 AM - 8 PM EST and Saturday 9 AM - 5 PM EST.",
"For urgent issues, you can request a callback or live chat with our support team.",
"We offer multi-language support in English, Spanish, and French.",
"Customer satisfaction is our priority. We're here to help resolve any issues quickly.",

# Loyalty and Rewards
"Gold members receive free express shipping on all orders and exclusive early access to sales.",
"Silver members get free standard shipping and 5% off all purchases.",
"Bronze members receive our newsletter with special offers and product updates.",
"You can earn points for every dollar spent, which can be redeemed for discounts.",

# Technical Support
"For technical issues with products, check our troubleshooting guides in the support section.",
"If you need warranty service, contact our technical support team with your purchase information.",
"Product manuals and setup guides are available for download on our website.",
"We offer video tutorials and live setup assistance for complex products."
]

self.agent.add_documents(knowledge_base)
print(f"āœ… Loaded {len(knowledge_base)} customer service documents")

def handle_customer_inquiry(self, query: str, customer_email: str = None) -> Dict:
"""Handle a customer inquiry."""
print(f"\nšŸ‘¤ Customer Inquiry")
print(f"šŸ“§ Customer: {customer_email or 'Anonymous'}")
print(f"šŸ’¬ Query: {query}")
print("-" * 50)

# Analyze sentiment and determine if escalation is needed
sentiment = self._analyze_sentiment(query)
escalation_needed = self._check_escalation_triggers(query, sentiment)

# Process the request through the agent
response = self.agent.run(query)

# Add customer context if email provided
if customer_email:
customer_info = self.agent.run_tool("lookup_customer", {"email": customer_email})
response.customer_context = customer_info

# Handle escalation if needed
if escalation_needed and customer_email:
escalation_result = self.agent.run_tool("escalate_to_human", {
"reason": "Complex issue requiring human assistance",
"customer_id": customer_email
})
response.tools_used.append(escalation_result)

# Record conversation
conversation_id = f"CONV{datetime.now().strftime('%Y%m%d%H%M%S')}"
self.conversations[conversation_id] = {
"customer_email": customer_email,
"query": query,
"response": response.answer,
"sentiment": sentiment,
"escalated": escalation_needed,
"timestamp": datetime.now().isoformat()
}

# Format response
result = {
"conversation_id": conversation_id,
"customer_email": customer_email,
"query": query,
"answer": response.answer,
"confidence": response.confidence,
"tools_used": [tool.name for tool in response.tools_used],
"sentiment": sentiment,
"escalated": escalation_needed,
"customer_context": getattr(response, 'customer_context', None),
"timestamp": datetime.now().isoformat()
}

return result

def _analyze_sentiment(self, text: str) -> str:
"""Simple sentiment analysis."""
negative_words = ["angry", "frustrated", "disappointed", "terrible", "awful", "hate", "complaint"]
positive_words = ["happy", "satisfied", "great", "excellent", "love", "thank", "appreciate"]

text_lower = text.lower()

negative_count = sum(1 for word in negative_words if word in text_lower)
positive_count = sum(1 for word in positive_words if word in text_lower)

if negative_count > positive_count:
return "negative"
elif positive_count > negative_count:
return "positive"
else:
return "neutral"

def _check_escalation_triggers(self, query: str, sentiment: str) -> bool:
"""Check if the inquiry needs human escalation."""
escalation_keywords = [
"speak to manager", "escalate", "complaint", "refund", "billing dispute",
"technical issue", "not working", "defective", "angry", "frustrated"
]

query_lower = query.lower()
keyword_escalation = any(keyword in query_lower for keyword in escalation_keywords)
sentiment_escalation = sentiment == "negative"

return keyword_escalation or sentiment_escalation

def record_customer_satisfaction(self, conversation_id: str, rating: int, feedback: str = None):
"""Record customer satisfaction rating."""
if conversation_id in self.conversations:
self.satisfaction_scores.append({
"conversation_id": conversation_id,
"rating": rating,
"feedback": feedback,
"timestamp": datetime.now().isoformat()
})

# Update conversation record
self.conversations[conversation_id]["satisfaction_rating"] = rating
self.conversations[conversation_id]["feedback"] = feedback

def get_customer_service_metrics(self) -> Dict:
"""Get customer service performance metrics."""
total_conversations = len(self.conversations)
escalated_conversations = len([c for c in self.conversations.values() if c.get("escalated", False)])

sentiment_breakdown = {}
for conv in self.conversations.values():
sentiment = conv.get("sentiment", "neutral")
sentiment_breakdown[sentiment] = sentiment_breakdown.get(sentiment, 0) + 1

satisfaction_ratings = [score["rating"] for score in self.satisfaction_scores]
avg_satisfaction = sum(satisfaction_ratings) / len(satisfaction_ratings) if satisfaction_ratings else 0

return {
"total_conversations": total_conversations,
"escalation_rate": escalated_conversations / total_conversations if total_conversations > 0 else 0,
"sentiment_breakdown": sentiment_breakdown,
"average_satisfaction": avg_satisfaction,
"satisfaction_responses": len(self.satisfaction_scores)
}

def main():
"""Main function to run the Customer Service Agent demo."""
print("šŸ›ļø Customer Service Agent Demo")
print("=" * 50)

# Initialize agent
agent = CustomerServiceAgent()

# Sample customer inquiries
sample_inquiries = [
("I want to return my wireless headphones. They don't work properly.", "john.doe@email.com"),
("What's the status of my order ORD001?", "john.doe@email.com"),
("Do you have any phone cases in stock?", "jane.smith@email.com"),
("I'm very frustrated with the shipping delay on my order. This is unacceptable!", "angry.customer@email.com"),
("Thank you so much for the great service! I love my new laptop stand.", "happy.customer@email.com"),
("Can you help me set up my wireless headphones? I'm having trouble connecting them.", "tech.support@email.com"),
("I need to speak to a manager about a billing dispute.", "billing.issue@email.com"),
("What are your shipping options for international orders?", "international.customer@email.com")
]

print("\nšŸŽÆ Processing Sample Customer Inquiries")
print("=" * 50)

for query, email in sample_inquiries:
result = agent.handle_customer_inquiry(query, email)

print(f"\nšŸ‘¤ Customer: {result['customer_email']}")
print(f"šŸ’¬ Query: {result['query']}")
print(f"šŸ’” Response: {result['answer']}")
print(f"šŸŽÆ Confidence: {result['confidence']:.2f}")
print(f"😊 Sentiment: {result['sentiment']}")
print(f"šŸ”§ Tools Used: {', '.join(result['tools_used'])}")
print(f"🚨 Escalated: {result['escalated']}")
if result['customer_context']:
print(f"šŸ‘¤ Context: {result['customer_context']}")
print("-" * 50)

# Show metrics
print("\nšŸ“Š Customer Service Metrics")
print("=" * 40)
metrics = agent.get_customer_service_metrics()
print(f"Total Conversations: {metrics['total_conversations']}")
print(f"Escalation Rate: {metrics['escalation_rate']:.1%}")
print(f"Sentiment Breakdown: {metrics['sentiment_breakdown']}")
print(f"Average Satisfaction: {metrics['average_satisfaction']:.1f}")

# Interactive mode
print("\nšŸ’¬ Interactive Customer Service Mode")
print("=" * 50)
print("Type 'quit' to exit, 'metrics' to see performance metrics")
print("Type 'satisfaction <conversation_id> <rating>' to record satisfaction")

while True:
user_input = input("\nšŸ›ļø Enter customer inquiry: ").strip()

if user_input.lower() == 'quit':
break
elif user_input.lower() == 'metrics':
metrics = agent.get_customer_service_metrics()
print(f"\nšŸ“Š Current Metrics:")
print(f" Conversations: {metrics['total_conversations']}")
print(f" Escalation Rate: {metrics['escalation_rate']:.1%}")
print(f" Sentiment: {metrics['sentiment_breakdown']}")
elif user_input.startswith('satisfaction '):
parts = user_input.split(' ', 2)
if len(parts) >= 3:
conv_id = parts[1]
rating = int(parts[2])
feedback = parts[3] if len(parts) > 3 else None
agent.record_customer_satisfaction(conv_id, rating, feedback)
print(f"āœ… Satisfaction recorded: {rating}/5 for conversation {conv_id}")
elif user_input:
result = agent.handle_customer_inquiry(user_input, "interactive_customer")
print(f"\nšŸ’” {result['answer']}")
if result['escalated']:
print("🚨 This inquiry has been escalated to a human agent.")
print(f"šŸ“ Conversation ID: {result['conversation_id']}")

print("\nšŸ‘‹ Customer Service Agent demo completed!")

if __name__ == "__main__":
main()

Running the Example​

1. Setup​

# Create project directory
mkdir customer-service-agent
cd customer-service-agent

# Install dependencies
pip install recoagent python-dotenv

# Create .env file
echo "OPENAI_API_KEY=your_api_key_here" > .env

2. Run the Example​

python customer_service_agent.py

3. Expected Output​

šŸ›ļø Customer Service Agent Demo
==================================================

āœ… Customer Service tools loaded
āœ… Customer service policies configured
āœ… Loaded 24 customer service documents

šŸŽÆ Processing Sample Customer Inquiries
==================================================

šŸ‘¤ Customer Inquiry
šŸ“§ Customer: john.doe@email.com
šŸ’¬ Query: I want to return my wireless headphones. They don't work properly.
--------------------------------------------------

šŸ‘¤ Customer: john.doe@email.com
šŸ’¬ Query: I want to return my wireless headphones. They don't work properly.
šŸ’” Response: I understand you'd like to return your wireless headphones due to functionality issues. Let me help you with that. I'll process a return request for you right away. Returns are accepted within 30 days of delivery, and since the item isn't working properly, this definitely qualifies.
šŸŽÆ Confidence: 0.91
😊 Sentiment: neutral
šŸ”§ Tools Used: process_return
🚨 Escalated: False
--------------------------------------------------

šŸ‘¤ Customer Inquiry
šŸ“§ Customer: angry.customer@email.com
šŸ’¬ Query: I'm very frustrated with the shipping delay on my order. This is unacceptable!
--------------------------------------------------

šŸ‘¤ Customer: angry.customer@email.com
šŸ’¬ Query: I'm very frustrated with the shipping delay on my order. This is unacceptable!
šŸ’” Response: I sincerely apologize for the shipping delay and understand your frustration. This is not the experience we want you to have. Let me escalate this to our human support team who can provide immediate assistance and resolve this issue for you.
šŸŽÆ Confidence: 0.88
😊 Sentiment: negative
šŸ”§ Tools Used: escalate_to_human
🚨 Escalated: True
šŸ‘¤ Context: Customer found: John Doe (ID: CUST001), Gold member since 2023-01-15, 12 orders, $1250.0 lifetime value
--------------------------------------------------

šŸ“Š Customer Service Metrics
========================================
Total Conversations: 8
Escalation Rate: 25.0%
Sentiment Breakdown: {'negative': 2, 'neutral': 4, 'positive': 2}
Average Satisfaction: 0.0

šŸ’¬ Interactive Customer Service Mode
==================================================
Type 'quit' to exit, 'metrics' to see performance metrics
Type 'satisfaction <conversation_id> <rating>' to record satisfaction

šŸ›ļø Enter customer inquiry: Do you have any discounts available?

Key Features​

1. Customer-Focused Knowledge​

  • Product information and availability
  • Order status and tracking
  • Returns and exchange policies
  • Shipping and delivery options
  • Account and billing support

2. Intelligent Customer Tools​

  • Customer Lookup: Access customer information and history
  • Order Tracking: Check order status and shipping details
  • Return Processing: Handle return requests and generate labels
  • Product Availability: Check stock and pricing information
  • Callback Scheduling: Arrange follow-up calls
  • Human Escalation: Route complex issues to human agents

3. Sentiment Analysis​

  • Automatic sentiment detection (positive, negative, neutral)
  • Escalation triggers based on sentiment and keywords
  • Appropriate response tone matching

4. Customer Satisfaction Tracking​

  • Rating collection system
  • Feedback analysis
  • Performance metrics and reporting

Customization Options​

Multi-Language Support​

def setup_multilingual_support():
"""Set up multi-language customer service."""
languages = ["en", "es", "fr"]

for lang in languages:
# Load language-specific knowledge base
knowledge = load_knowledge_for_language(lang)
agent.add_documents(knowledge, language=lang)

print(f"āœ… Multi-language support enabled: {', '.join(languages)}")

Integration with CRM Systems​

def integrate_with_crm():
"""Integrate with Salesforce, HubSpot, etc."""
def update_crm_ticket(customer_id: str, issue: str, resolution: str):
# CRM API integration
crm_data = {
"customer_id": customer_id,
"issue": issue,
"resolution": resolution,
"status": "resolved"
}

# Update CRM system
crm_client.update_ticket(crm_data)
return "CRM ticket updated successfully"

agent.add_tool(Tool(name="update_crm", function=update_crm_ticket))

Advanced Analytics​

def setup_advanced_analytics():
"""Set up detailed analytics and reporting."""
analytics = {
"response_time_tracking": True,
"conversation_flow_analysis": True,
"customer_journey_mapping": True,
"churn_prediction": True,
"upselling_opportunities": True
}

# Configure analytics
for feature, enabled in analytics.items():
if enabled:
agent.enable_analytics_feature(feature)

print("āœ… Advanced analytics enabled")

Deployment Considerations​

Production Configuration​

# Production settings
agent = CustomerServiceAgent()
agent.agent.configure(
llm_model="gpt-4",
max_tokens=800,
temperature=0.3, # Slightly higher for more natural responses
response_timeout=10000
)

Security and Privacy​

def implement_privacy_protection():
"""Implement GDPR and privacy compliance."""

def anonymize_customer_data(customer_id: str):
"""Anonymize customer data for privacy compliance."""
# Remove PII from logs and analytics
pass

def handle_data_deletion_request(customer_id: str):
"""Handle customer data deletion requests."""
# Delete customer data as per GDPR requirements
pass

# Add privacy tools
agent.add_privacy_tools([
anonymize_customer_data,
handle_data_deletion_request
])

Performance Optimization​

def optimize_for_scale():
"""Optimize for high-volume customer service."""

# Implement caching
agent.enable_response_caching(ttl=3600)

# Set up load balancing
agent.configure_load_balancing(max_concurrent=100)

# Enable auto-scaling
agent.enable_auto_scaling(min_instances=2, max_instances=10)

print("āœ… Performance optimization enabled")

Next Steps​

This customer service agent can be extended by:

  1. Adding voice support for phone-based customer service
  2. Implementing chatbots for website integration
  3. Creating mobile apps for customer self-service
  4. Adding AI-powered recommendations for upselling
  5. Integrating with inventory systems for real-time stock updates

Ready for more? Check out the Technical Documentation example for documentation-based support scenarios!