What Is an AI-Powered Customer Service Department?
An AI-powered customer service department is a system where artificial intelligence agents handle incoming customer inquiries, perform actions on behalf of customers, and escalate complex issues to human agents when necessary. Unlike simple chatbots that merely pattern-match responses, modern AI agents combine large language models (LLMs) with tool-calling capabilities, memory, and decision-making logic to autonomously resolve customer issues end-to-end.
Think of it as a digital workforce where each AI agent is a specialized worker: one agent handles order tracking, another manages returns and refunds, a third troubleshoots technical issues, and a triage agent routes conversations to the right specialist. These agents don't just generate text — they call APIs, query databases, update CRM records, send emails, and perform any action a human agent would take, all within a unified orchestration framework.
The core components include:
- LLM-powered reasoning — the brain that understands customer intent and decides what to do
- Tool suite — APIs and functions the agent can invoke (look up orders, process refunds, check inventory)
- Memory and context — conversation history and persistent customer data across sessions
- Orchestration layer — routing, prioritization, and escalation logic
- Human handoff protocol — seamless transfer to human agents with full context preservation
Why It Matters
Customer service is one of the highest-ROI application areas for AI agents today. Here's why:
- 24/7 instant response — AI agents never sleep, eliminating queue wait times across time zones
- Cost reduction at scale — companies like Klarna report AI agents handling work equivalent to 700 full-time employees, resolving most inquiries in under 2 minutes versus 11 minutes for humans
- Consistent quality — agents follow your exact policies, brand voice, and compliance requirements every single time
- Human agent elevation — instead of burning out on repetitive "where is my order" queries, human agents focus on high-value, complex, and emotionally sensitive cases where empathy truly matters
- Data-driven insights — every interaction is structured and analyzed, surfacing product issues, documentation gaps, and customer sentiment trends automatically
Architecture Overview
Before diving into code, let's understand the high-level architecture. A production-grade AI customer service department typically follows a layered design:
┌─────────────────────────────────────────────────────┐
│ Customer Touchpoints │
│ (Web Chat, Email, SMS, WhatsApp, Voice, Social) │
└─────────────────┬───────────────────────────────────┘
│
┌─────────────────▼───────────────────────────────────┐
│ Triage & Classification Agent │
│ • Intent recognition & sentiment analysis │
│ • Urgency scoring │
│ • Routing to specialist agent or human queue │
└─────────────────┬───────────────────────────────────┘
│
┌────────────┼────────────┬──────────────────┐
▼ ▼ ▼ ▼
┌─────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────────┐
│ Order │ │ Returns │ │ Technical│ │ General Inquiry │
│ Agent │ │ Agent │ │ Support │ │ Agent │
│ (tracks │ │ (RMA, │ │ Agent │ │ (FAQ, account, │
│ orders, │ │ refunds) │ │ (debug) │ │ billing) │
│ shipping)│ │ │ │ │ │ │
└────┬─────┘ └────┬─────┘ └────┬─────┘ └────────┬─────────┘
│ │ │ │
└────────────┴────────────┴─────────────────┘
│
┌─────────────────▼───────────────────────────────────┐
│ Tool & Integration Layer │
│ Order DB │ CRM │ Payment Gateway │ Inventory │ KB │
└─────────────────────────────────────────────────────┘
│
┌─────────────────▼───────────────────────────────────┐
│ Human Escalation Queue │
│ • Full conversation transcript + agent actions │
│ • Suggested resolution based on AI analysis │
│ • Priority-based routing to available agents │
└─────────────────────────────────────────────────────┘
Building Your First AI Customer Service Agent
Let's build a working triage and order-support agent using Python. We'll use OpenAI's function-calling capabilities as our LLM backend, but the patterns apply to any provider (Anthropic, Gemini, open-source models via Ollama/vLLM). We'll structure the project for extensibility so you can add more specialist agents later.
Step 1: Project Structure and Dependencies
Create a new project directory with the following structure:
customer-service-ai/
├── agents/
│ ├── __init__.py
│ ├── base_agent.py # Abstract agent class
│ ├── triage_agent.py # Intent classification & routing
│ └── order_agent.py # Order lookup, tracking, modifications
├── tools/
│ ├── __init__.py
│ ├── order_tools.py # Functions to query order database
│ ├── customer_tools.py # CRM lookup functions
│ └── notification_tools.py # Email/SMS sending
├── memory/
│ ├── __init__.py
│ └── conversation_store.py # Persistent conversation history
├── orchestration/
│ ├── __init__.py
│ └── dispatcher.py # Main orchestrator
├── config.py # API keys, settings
├── main.py # Entry point
└── requirements.txt
Install dependencies:
pip install openai pydantic sqlite-utils redis flask fastapi uvicornStep 2: Define the Tool Suite
Tools are the functions your AI agent can call. Each tool needs a clear name, description, and parameter schema so the LLM knows when and how to invoke it. Start with order-related tools:
# tools/order_tools.py
from typing import Optional, List, Dict, Any
from datetime import datetime
import sqlite_utils
from pydantic import BaseModel, Field
# ---- Database setup (in production, use your actual order DB) ----
db = sqlite_utils.Database("orders.db")
# Ensure our demo table exists with sample data
def seed_demo_data():
db.execute("""
CREATE TABLE IF NOT EXISTS orders (
order_id TEXT PRIMARY KEY,
customer_id TEXT,
email TEXT,
status TEXT,
total REAL,
items TEXT,
tracking_number TEXT,
shipping_address TEXT,
created_at TEXT,
estimated_delivery TEXT
)
""")
# Insert sample data if table is empty
count = db.execute("SELECT COUNT(*) FROM orders").fetchone()[0]
if count == 0:
sample_orders = [
("ORD-1001", "CUST-42", "jane@example.com", "delivered", 89.99,
'[{"name":"Wireless Headphones","sku":"WH-200","qty":1}]',
"UPS-1Z999999", "123 Main St, Austin, TX 78701",
"2024-01-15T10:00:00", "2024-01-18T18:00:00"),
("ORD-1002", "CUST-42", "jane@example.com", "in_transit", 142.50,
'[{"name":"USB-C Hub","sku":"UH-100","qty":1},{"name":"HDMI Cable","sku":"HC-50","qty":2}]',
"FEDEX-88888888", "123 Main St, Austin, TX 78701",
"2024-02-20T14:30:00", "2024-02-25T18:00:00"),
("ORD-1003", "CUST-99", "bob@example.com", "pending", 299.99,
'[{"name":"Mechanical Keyboard","sku":"MK-700","qty":1}]',
None, "456 Oak Ave, Denver, CO 80205",
"2024-03-01T09:00:00", "2024-03-07T18:00:00"),
("ORD-1004", "CUST-99", "bob@example.com", "cancelled", 49.99,
'[{"name":"Mouse Pad","sku":"MP-50","qty":2}]',
None, "456 Oak Ave, Denver, CO 80205",
"2024-03-10T11:00:00", None),
]
db.execute_many(
"INSERT INTO orders VALUES (?,?,?,?,?,?,?,?,?,?)",
sample_orders
)
db.commit()
seed_demo_data()
# ---- Tool parameter schemas using Pydantic (for OpenAI function calling) ----
class OrderLookupParams(BaseModel):
order_id: Optional[str] = Field(None, description="Exact order ID, e.g. ORD-1001")
customer_id: Optional[str] = Field(None, description="Customer ID to find all their orders")
email: Optional[str] = Field(None, description="Customer email to look up orders")
class CancelOrderParams(BaseModel):
order_id: str = Field(..., description="The order ID to cancel")
reason: Optional[str] = Field("customer_request", description="Reason for cancellation")
class ModifyShippingParams(BaseModel):
order_id: str = Field(..., description="Order ID to modify")
new_address: str = Field(..., description="Updated shipping address")
new_city: str = Field(..., description="City for the new address")
new_state: str = Field(..., description="State abbreviation, e.g. TX")
new_zip: str = Field(..., description="ZIP code")
# ---- Tool implementations (these get called by the LLM) ----
def lookup_orders(order_id: str = None, customer_id: str = None, email: str = None) -> Dict[str, Any]:
"""Find orders by order ID, customer ID, or email. Returns matching orders with full details."""
if order_id:
rows = db.execute("SELECT * FROM orders WHERE order_id = ?", [order_id]).fetchall()
elif customer_id:
rows = db.execute("SELECT * FROM orders WHERE customer_id = ? ORDER BY created_at DESC", [customer_id]).fetchall()
elif email:
rows = db.execute("SELECT * FROM orders WHERE email = ? ORDER BY created_at DESC", [email]).fetchall()
else:
return {"error": "Must provide at least one search parameter: order_id, customer_id, or email"}
if not rows:
return {"message": "No orders found matching your criteria.", "orders": []}
orders = []
for row in rows:
orders.append({
"order_id": row[0],
"customer_id": row[1],
"email": row[2],
"status": row[3],
"total": row[4],
"items": row[5],
"tracking_number": row[6],
"shipping_address": row[7],
"created_at": row[8],
"estimated_delivery": row[9]
})
return {"orders": orders, "count": len(orders)}
def cancel_order(order_id: str, reason: str = "customer_request") -> Dict[str, Any]:
"""Cancel a pending order. Only orders with 'pending' status can be cancelled."""
row = db.execute("SELECT * FROM orders WHERE order_id = ?", [order_id]).fetchone()
if not row:
return {"error": f"Order {order_id} not found."}
status = row[3]
if status != "pending":
return {"error": f"Order {order_id} is in '{status}' status. Only pending orders can be cancelled.", "current_status": status}
# Update status
db.execute("UPDATE orders SET status = 'cancelled' WHERE order_id = ?", [order_id])
db.commit()
return {
"success": True,
"order_id": order_id,
"new_status": "cancelled",
"reason": reason,
"refund_amount": row[4],
"message": f"Order {order_id} has been cancelled. A refund of ${row[4]:.2f} will be processed within 3-5 business days."
}
def modify_shipping_address(order_id: str, new_address: str, new_city: str, new_state: str, new_zip: str) -> Dict[str, Any]:
"""Update the shipping address for an order that hasn't shipped yet."""
row = db.execute("SELECT * FROM orders WHERE order_id = ?", [order_id]).fetchone()
if not row:
return {"error": f"Order {order_id} not found."}
status = row[3]
if status in ("delivered", "cancelled", "returned"):
return {"error": f"Cannot modify shipping for order in '{status}' status."}
full_address = f"{new_address}, {new_city}, {new_state} {new_zip}"
db.execute("UPDATE orders SET shipping_address = ? WHERE order_id = ?", [full_address, order_id])
db.commit()
return {
"success": True,
"order_id": order_id,
"old_address": row[7],
"new_address": full_address,
"message": f"Shipping address for {order_id} has been updated successfully."
}
Step 3: Build the Base Agent Class
Create a reusable agent base class that handles the LLM conversation loop, tool calling, and context management:
# agents/base_agent.py
from typing import List, Dict, Any, Callable, Optional
from openai import OpenAI
import json
class BaseAgent:
"""
Abstract agent that manages an LLM conversation with tool-calling capabilities.
Subclass this for each specialist agent (triage, order support, returns, etc.)
"""
def __init__(
self,
name: str,
system_prompt: str,
tools: List[Dict[str, Any]],
tool_functions: Dict[str, Callable],
model: str = "gpt-4o",
temperature: float = 0.3,
max_turns: int = 10
):
self.name = name
self.system_prompt = system_prompt
self.tools = tools
self.tool_functions = tool_functions
self.model = model
self.temperature = temperature
self.max_turns = max_turns
self.client = OpenAI()
def run(self, user_message: str, conversation_history: Optional[List[Dict]] = None) -> Dict[str, Any]:
"""
Execute the agent loop: send message to LLM, process any tool calls,
return final response and conversation history.
"""
messages = [
{"role": "system", "content": self.system_prompt}
]
if conversation_history:
messages.extend(conversation_history)
messages.append({"role": "user", "content": user_message})
turn_count = 0
actions_taken = []
while turn_count < self.max_turns:
turn_count += 1
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=self.tools,
tool_choice="auto",
temperature=self.temperature
)
assistant_message = response.choices[0].message
# Check if the model wants to call a tool
if assistant_message.tool_calls:
# Add the assistant's tool call request to conversation
messages.append({
"role": "assistant",
"content": assistant_message.content or "",
"tool_calls": [
{
"id": tc.id,
"function": {
"name": tc.function.name,
"arguments": tc.function.arguments
},
"type": "function"
} for tc in assistant_message.tool_calls
]
})
# Execute each tool call and add results
for tool_call in assistant_message.tool_calls:
func_name = tool_call.function.name
try:
args = json.loads(tool_call.function.arguments)
except json.JSONDecodeError:
args = {}
if func_name in self.tool_functions:
try:
result = self.tool_functions[func_name](**args)
actions_taken.append({
"tool": func_name,
"arguments": args,
"result": result
})
except Exception as e:
result = {"error": str(e)}
actions_taken.append({
"tool": func_name,
"arguments": args,
"error": str(e)
})
else:
result = {"error": f"Tool '{func_name}' not available"}
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result)
})
# Continue loop so the LLM can process tool results
continue
# No tool calls — this is the final response
final_content = assistant_message.content or ""
messages.append({"role": "assistant", "content": final_content})
return {
"response": final_content,
"conversation_history": messages,
"actions_taken": actions_taken,
"turns_used": turn_count,
"agent_name": self.name
}
# Max turns exceeded
return {
"response": "I apologize, but I'm having trouble resolving your request. Let me connect you with a human agent who can help.",
"conversation_history": messages,
"actions_taken": actions_taken,
"turns_used": turn_count,
"agent_name": self.name,
"escalation_reason": "max_turns_exceeded"
}
Step 4: Create the Triage Agent
The triage agent classifies intent and routes the customer to the appropriate specialist. It doesn't solve problems itself — it's the dispatcher:
# agents/triage_agent.py
from agents.base_agent import BaseAgent
from typing import Dict, List, Optional
# Triage agent has one tool: the decision to escalate or route
def route_to_specialist(
intent: str,
reason: str,
urgency: str = "normal",
customer_id: str = None,
order_id: str = None
) -> Dict:
"""
Route this conversation to the appropriate specialist agent or human queue.
This function is called by the triage agent to hand off the conversation.
intent: one of 'order_tracking', 'returns_refunds', 'technical_support',
'billing', 'account_management', 'general_inquiry', 'human_escalation'
urgency: 'low', 'normal', 'high', 'critical'
"""
valid_intents = [
'order_tracking', 'returns_refunds', 'technical_support',
'billing', 'account_management', 'general_inquiry', 'human_escalation'
]
if intent not in valid_intents:
return {"error": f"Invalid intent. Must be one of: {', '.join(valid_intents)}"}
return {
"routing_target": intent,
"urgency": urgency,
"reason": reason,
"customer_id": customer_id,
"order_id": order_id,
"message": f"Routing to {intent} specialist with {urgency} priority."
}
class TriageAgent(BaseAgent):
"""
First point of contact. Classifies customer intent, assesses urgency,
and routes to the appropriate specialist agent or human queue.
"""
def __init__(self):
system_prompt = """You are a warm, professional customer service triage specialist for Acme Corp,
an e-commerce company. Your job is to:
1. Greet the customer warmly
2. Understand their issue by asking clarifying questions if needed
3. Determine the PRIMARY intent category
4. Assess urgency (critical = service outage, lost funds, safety; high = order stuck,
wrong item; normal = tracking, returns; low = general questions)
5. Route them using the route_to_specialist function
IMPORTANT: You must ALWAYS call route_to_specialist once you've gathered enough information.
Do NOT try to solve the customer's problem yourself — your job is routing only.
Intent categories:
- order_tracking: "Where is my order?", tracking numbers, delivery estimates, shipping status
- returns_refunds: Returns, exchanges, refunds, damaged items, wrong items received
- technical_support: Website/app issues, login problems, payment gateway errors
- billing: Charges, invoices, payment methods, subscription questions
- account_management: Password resets, profile updates, account deletion
- general_inquiry: Product questions, stock availability, store hours, policies
- human_escalation: Customer explicitly demands a human, or situation is emotionally charged/complex
If a customer is angry, frustrated, or explicitly demands a human agent, use human_escalation."""
tools = [{
"type": "function",
"function": {
"name": "route_to_specialist",
"description": "Route the conversation to the appropriate specialist agent based on customer intent",
"parameters": {
"type": "object",
"properties": {
"intent": {
"type": "string",
"enum": [
"order_tracking", "returns_refunds", "technical_support",
"billing", "account_management", "general_inquiry", "human_escalation"
],
"description": "The primary intent category for this customer inquiry"
},
"reason": {
"type": "string",
"description": "Brief summary of what the customer needs"
},
"urgency": {
"type": "string",
"enum": ["low", "normal", "high", "critical"],
"description": "Urgency level based on impact to customer"
},
"customer_id": {
"type": "string",
"description": "Customer ID if identified during conversation"
},
"order_id": {
"type": "string",
"description": "Order ID if mentioned by customer"
}
},
"required": ["intent", "reason", "urgency"]
}
}
}]
tool_functions = {"route_to_specialist": route_to_specialist}
super().__init__(
name="TriageAgent",
system_prompt=system_prompt,
tools=tools,
tool_functions=tool_functions,
model="gpt-4o",
temperature=0.3,
max_turns=5
)
Step 5: Create the Order Support Specialist Agent
This agent handles order tracking, modifications, and cancellations with full tool access:
# agents/order_agent.py
from agents.base_agent import BaseAgent
from tools.order_tools import lookup_orders, cancel_order, modify_shipping_address
class OrderAgent(BaseAgent):
"""
Specialist agent for order-related inquiries: tracking, status checks,
shipping modifications, and cancellations.
"""
def __init__(self):
system_prompt = """You are an Order Support Specialist at Acme Corp. Your job is to help customers
with everything related to their orders. You have access to real-time order data.
Your capabilities:
- Look up orders by order ID, customer ID, or email address
- Check order status (pending, in_transit, delivered, cancelled)
- Provide tracking numbers and shipping carrier information
- Modify shipping addresses for orders that haven't shipped yet
- Cancel pending orders and explain the refund process
- Estimate delivery dates based on order data
Guidelines:
- Always verify the customer's identity before sharing order details. Ask for their email or
customer ID if they haven't provided it.
- When an order is in_transit, share the tracking number and estimated delivery date.
- If an order is delayed beyond the estimated delivery, apologize and offer to investigate.
- For cancellation requests, confirm with the customer before executing (since it's irreversible).
- If you cannot resolve the issue (e.g., order already delivered and customer claims they didn't
receive it), suggest escalating to a human agent for investigation.
- Be empathetic but efficient. Use the tools to get facts before responding.
- Format monetary amounts with dollar signs and two decimal places.
- When listing multiple orders, present them in a clear, organized way."""
tools = [
{
"type": "function",
"function": {
"name": "lookup_orders",
"description": "Find orders by order_id, customer_id, or email. Returns matching orders with status, tracking, items, and shipping details.",
"parameters": {
"type": "object",
"properties": {
"order_id": {
"type": "string",
"description": "Exact order ID (e.g., ORD-1001) to look up a specific order"
},
"customer_id": {
"type": "string",
"description": "Customer ID (e.g., CUST-42) to find all orders for a customer"
},
"email": {
"type": "string",
"description": "Customer email address to look up their orders"
}
}
}
}
},
{
"type": "function",
"function": {
"name": "cancel_order",
"description": "Cancel a pending order. Only works for orders in 'pending' status. Returns confirmation and refund details.",
"parameters": {
"type": "object",
"properties": {
"order_id": {
"type": "string",
"description": "The order ID to cancel (e.g., ORD-1003)"
},
"reason": {
"type": "string",
"description": "Reason for cancellation, e.g., 'customer_request', 'duplicate_order', 'found_better_price'"
}
},
"required": ["order_id"]
}
}
},
{
"type": "function",
"function": {
"name": "modify_shipping_address",
"description": "Update the shipping address for an order that hasn't been delivered yet. Cannot modify delivered or cancelled orders.",
"parameters": {
"type": "object",
"properties": {
"order_id": {
"type": "string",
"description": "The order ID to modify"
},
"new_address": {
"type": "string",
"description": "Street address (e.g., '789 Pine Street')"
},
"new_city": {
"type": "string",
"description": "City name"
},
"new_state": {
"type": "string",
"description": "Two-letter state abbreviation (e.g., 'CA', 'TX')"
},
"new_zip": {
"type": "string",
"description": "ZIP code (e.g., '94102')"
}
},
"required": ["order_id", "new_address", "new_city", "new_state", "new_zip"]
}
}
}
]
tool_functions = {
"lookup_orders": lookup_orders,
"cancel_order": cancel_order,
"modify_shipping_address": modify_shipping_address
}
super().__init__(
name="OrderAgent",
system_prompt=system_prompt,
tools=tools,
tool_functions=tool_functions,
model="gpt-4o",
temperature=0.2,
max_turns=8
)
Step 6: Build the Orchestration Layer
The orchestrator ties everything together. It receives a customer message, runs the triage agent, inspects the routing decision, and invokes the appropriate specialist agent:
# orchestration/dispatcher.py
from typing import Dict, List, Optional
from agents.triage_agent import TriageAgent
from agents.order_agent import OrderAgent
class CustomerServiceDispatcher:
"""
Main orchestrator for the AI customer service department.
Routes conversations through triage → specialist → resolution,
with human escalation as a fallback.
"""
def __init__(self):
self.triage_agent = TriageAgent()
# Specialist agents registry — add more as you build them
self.specialist_agents = {
"order_tracking": OrderAgent(),
# "returns_refunds": ReturnsAgent(), # Build next
# "technical_support": TechSupportAgent(),
# "billing": BillingAgent(),
# "account_management": AccountAgent(),
# "general_inquiry": GeneralAgent(),
}
# Conversation store (in production, use Redis or a DB)
self.conversations: Dict[str, List[Dict]] = {}
def handle_message(
self,
customer_message: str,
conversation_id: str,
customer_email: Optional[str] = None
) -> Dict:
"""
Process a customer message end-to-end.
Returns a dict with:
- response: The final text response for the customer
- routing_path: Which agents handled this conversation
- actions_taken: All tool calls made during processing
- escalation_required: Boolean indicating if human handoff is needed
"""
# Retrieve or initialize conversation history
history = self.conversations.get(conversation_id, [])
routing_path = []
all_actions = []
escalation_required = False
final_response = ""
# ---- Phase 1: Triage ----
triage_result = self.triage_agent.run(customer_message, history)
routing_path.append("triage")
# Check if triage resulted in a tool call (route_to_specialist)
routing_info = None
for action in triage_result.get("actions_taken", []):
if action["tool"] == "route_to_specialist":
routing_info = action["result"]
break
if routing_info is None:
# Triage didn't route — likely max turns exceeded
escalation_required = True
self.conversations[conversation_id] = triage_result["conversation_history"]
return {
"response": triage_result["response"],
"routing_path": routing_path,
"actions_taken": all_actions,
"escalation_required": True,
"escalation_reason": "Triage agent could not classify intent"
}
intent = routing_info.get("routing_target", "general_inquiry")
urgency = routing_info.get("urgency", "normal")
# ---- Phase 2: Check for human escalation ----
if intent == "human_escalation" or urgency == "critical":
escalation_required = True
# Build a comprehensive handoff context
self.conversations[conversation_id] = triage_result["conversation_history"]
return {
"response": "I understand this requires personal attention. Let me connect you with a human agent right away. "
"I've prepared a full summary of our conversation so you won't have to repeat anything.",
"routing_path": routing_path,
"actions_taken": all_actions,
"escalation_required": True,
"escalation_reason": routing_info.get("reason", "Customer requested human agent"),
"handoff_context": {
"intent": intent,
"urgency": urgency,
"summary": routing_info.get("reason", ""),
"customer_id": routing_info.get("customer_id"),
"order_id": routing_info.get("order_id"),
"conversation_history": triage_result["conversation_history"]
}
}
# ---- Phase 3: Route to specialist ----
specialist = self.specialist_agents.get(intent)
if specialist is None:
# No specialist agent for this intent yet — provide helpful response
fallback_response = (
f"I understand you're reaching out about {intent.replace('_', ' ')}. "
"While I don't have a dedicated specialist for that right now, let me help as best I can. "
"For complex issues, I can connect you with a human agent — just let me know."
)
self.conversations[conversation_id] = triage_result["conversation_history"]
return {
"response": fallback_response,
"routing_path": routing_path,
"actions_taken": all_actions,
"escalation_required": False,
"note": f"No specialist agent registered for intent: {intent}"
}
# Run the specialist agent with the original customer message
specialist_result = specialist.run(customer_message)
routing_path.append(intent)
all_actions.extend(triage_result.get("actions_taken", []))
all_actions.extend(specialist_result.get("actions_taken", []))
# Check if specialist hit max turns (needs escalation)
if "escalation_reason" in specialist_result:
escalation_required = True
# Store updated conversation history
self.conversations[conversation_id] = specialist_result["conversation_history"]
return {
"response": specialist_result["response"],
"routing_path": routing_path,
"actions_taken": all_actions,
"escalation_required": escalation_required,
"escalation_reason": specialist_result.get("escalation_reason"),
"agent_name": specialist_result.get("agent_name")
}
def get_conversation_history(self, conversation_id: str) -> List[Dict]:
"""Retrieve full conversation history for human handoff or audit."""
return self.conversations.get(conversation_id, [])
def clear_conversation(self, conversation_id: str):
"""Clear conversation after resolution."""
if conversation_id in self.conversations:
del self.conversations[conversation_id]
Step 7: Create the Entry Point and Test
Wire everything together in a runnable script that demonstrates a few customer interactions:
# main.py
from orchestration.dispatcher import CustomerServiceDispatcher
def main():
dispatcher = CustomerServiceDispatcher()
# ---- Test Scenario 1: Order tracking ----
print("=" * 60)
print("SCENARIO 1: Customer wants to track their order")
print("=" * 60)
result = dispatcher.handle_message(
customer_message="Hi, I ordered some headphones a few days ago and I want to know where they are. "
"My email is jane@example.com.",
conversation_id="conv-001"
)
print(f"Response: {result['response']}")
print(f"Routing path: {result['routing_path']}")
print(f"Actions taken: {len(result['actions_taken'])}")
print(f"Escalation required: {result['escalation_