AI System for Processing Incoming Customer Requests
An AI request processing system automatically classifies, prioritizes, and routes incoming requests across all channels (calls, chats, email, messengers) without manual sorting by operators.
Omnichannel
Architecture
from abc import ABC, abstractmethod
from dataclasses import dataclass
@dataclass
class IncomingRequest:
id: str
channel: str # voice | chat | email | telegram | whatsapp
raw_content: str # transcript or text
metadata: dict
customer_id: str = None
class RequestProcessor(ABC):
@abstractmethod
async def process(self, request: IncomingRequest) -> dict:
pass
class UnifiedRequestOrchestrator:
def __init__(self):
self.processors = {
"voice": VoiceRequestProcessor(),
"chat": ChatRequestProcessor(),
"email": EmailRequestProcessor(),
}
self.classifier = RequestClassifier()
self.router = RequestRouter()
async def handle(self, request: IncomingRequest) -> dict:
# 1. Classify intent and sentiment
classification = await self.classifier.classify(request)
# 2. Check: need immediate response? (complaint, VIP, SLA)
priority = self.calculate_priority(request, classification)
# 3. Route to appropriate handler
return await self.router.route(request, classification, priority)
AI Request
Classifier
CLASSIFICATION_SCHEMA = {
"type": "object",
"properties": {
"intent": {
"type": "string",
"enum": ["order_inquiry", "complaint", "technical_support",
"billing", "general_info", "cancellation", "compliment"]
},
"urgency": {"type": "string", "enum": ["critical", "high", "medium", "low"]},
"sentiment": {"type": "string", "enum": ["positive", "neutral", "negative", "angry"]},
"entities": {
"type": "object",
"properties": {
"order_id": {"type": "string"},
"product_name": {"type": "string"}
}
},
"summary": {"type": "string"},
"requires_human": {"type": "boolean"}
}
}
async def classify_request(text: str) -> dict:
response = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "system",
"content": f"Classify the customer request. JSON format."
}, {"role": "user", "content": text}],
response_format={"type": "json_object"}
)
return json.loads(response.choices[0].message.content)
SLA
Prioritization
PRIORITY_RULES = {
("critical", "angry"): {"score": 100, "max_wait_sec": 60},
("high", "negative"): {"score": 80, "max_wait_sec": 180},
("medium", "neutral"): {"score": 50, "max_wait_sec": 600},
("low", "positive"): {"score": 20, "max_wait_sec": 1800},
}
def calculate_sla(intent: str, sentiment: str, is_vip: bool) -> dict:
base = PRIORITY_RULES.get((urgency, sentiment),
{"score": 40, "max_wait_sec": 900})
if is_vip:
base["score"] += 30
base["max_wait_sec"] //= 2
return base
Timeline: classifier + router — 2–3 weeks. Full omnichannel system — 2–3 months.