AI Customer Service Automation

def triage_support_ticket(ticket_text, customer_tier="standard"): """Classify and prioritize incoming support tickets.""" prompt = f"""Analyze the following customer support inquiry and provide structured triage data. Customer Tier: {customer_tier} Inquiry: {ticket_text} Classify into exactly one category: - BILLING: Payment issues, refunds, invoicing - TECHNICAL: Bugs, errors, integration issues - ACCOUNT: Login, permissions, profile changes - SALES: Upgrade, pricing, feature questions - GENERAL: Policy questions, feedback, other Then determine: - Urgency: CRITICAL (service down, data loss) | HIGH (major feature broken) | MEDIUM (partial issue) | LOW (question/feedback) - Sentiment: FRUSTRATED | CONCERNED | NEUTRAL | HAPPY - Complexity: SIMPLE (FAQ answer) | MODERATE (troubleshooting) | COMPLEX (engineering required) - Recommended action: AUTO_REPLY | AGENT_HANDLE | ESCALATE_ENGINEERING Output as JSON: {{ "category": "...", "urgency": "...", "sentiment": "...", "complexity": "...", "recommended_action": "...", "priority_score": 1-100, "suggested_response_template": "brief suggested reply approach", "key_entities": ["product_name", "error_code", etc] }}""" return call_llm_api(prompt, temperature=0.1, response_format="json")

import faiss import numpy as np class AI SupportAgent: def __init__(self, api_key, knowledge_base): self.api_key = api_key self.knowledge_base = knowledge_base self.conversation_history = {} self.escalation_threshold = 0.3 def retrieve_context(self, query, top_k=3): """Retrieve relevant knowledge base articles.""" # Use embedding model to find relevant docs # Simplified: keyword matching for illustration relevant = [] for doc in self.knowledge_base: score = self.compute_relevance(query, doc) relevant.append((score, doc)) relevant.sort(reverse=True) return [doc for _, doc in relevant[:top_k]] def generate_response(self, session_id, user_message): """Generate contextual support response.""" # Get conversation history history = self.conversation_history.get(session_id, []) # Retrieve knowledge context_docs = self.retrieve_context(user_message) context_text = "\n\n".join([f"Article: {d['title']}\n{d['content']}" for d in context_docs]) # Build prompt messages = [ {"role": "system", "content": f"""You are a helpful, professional customer support agent. Use the following knowledge base articles to answer the customer's question. If you cannot answer from the provided context, say so clearly and offer to escalate to a human agent. Knowledge Base: {context_text} Guidelines: - Be concise but thorough - Use the customer's name if known - Offer specific next steps - Never make up policy details not in the knowledge base"""} ] # Add conversation history for msg in history[-6:]: messages.append(msg) messages.append({"role": "user", "content": user_message}) # Call API response = requests.post( "https://tokenease.io/v1/chat/completions", headers={"Authorization": f"Bearer {self.api_key}"}, json={ "model": "deepseek", "messages": messages, "temperature": 0.4, "max_tokens": 800 }, timeout=30 ) reply = response.json()["choices"][0]["message"]["content"] # Update history history.append({"role": "user", "content": user_message}) history.append({"role": "assistant", "content": reply}) self.conversation_history[session_id] = history[-20:] # Keep last 20 messages # Check if escalation needed confidence = self.assess_confidence(reply, context_docs) if confidence < self.escalation_threshold: reply += "\n\n[This conversation has been flagged for agent review. A specialist will follow up shortly.]" return reply

def analyze_interaction_sentiment(conversation_text): """Analyze sentiment and detect escalation risks.""" prompt = f"""Analyze the following customer service conversation for sentiment and risk signals. Conversation: {conversation_text} Provide: 1. Overall sentiment score (-1.0 to +1.0) 2. Sentiment trajectory (improving/stable/declining) 3. Frustration indicators detected 4. Churn risk level (LOW/MEDIUM/HIGH/CRITICAL) 5. Key complaints or pain points 6. Satisfaction drivers (what went well) 7. Recommended agent intervention: NONE | EMPATHY | MANAGER | RETENTION Output as JSON.""" return call_llm_api(prompt, temperature=0.2, response_format="json") def batch_sentiment_monitor(interactions, alert_threshold=-0.5): """Monitor batch of interactions and alert on negative trends.""" alerts = [] for interaction in interactions: result = analyze_interaction_sentiment(interaction['text']) if result['sentiment_score'] < alert_threshold: alerts.append({ "ticket_id": interaction['id'], "sentiment": result['sentiment_score'], "risk": result['churn_risk'], "action": result['recommended_intervention'] }) return alerts

def quality_check_response(response_text, customer_query, knowledge_base): """Validate AI response against quality criteria.""" prompt = f"""Evaluate the following AI-generated support response against quality standards. Customer Query: {customer_query} AI Response: {response_text} Checklist: 1. ACCURACY: Does the response correctly answer the question? (YES/NO/PARTIAL) 2. HALLUCINATION: Does it make claims not supported by facts? (YES/NO) 3. EMPATHY: Does it acknowledge the customer's situation appropriately? (YES/NO) 4. ACTIONABILITY: Does it provide clear next steps? (YES/NO) 5. TONE: Is the tone professional and appropriate? (YES/NO) 6. SAFETY: Does it contain any harmful, discriminatory, or inappropriate content? (YES/NO) Score each 0-10. Overall pass threshold: 7/10 average. If any safety issue: AUTO_FAIL. Output: {{ "passed": true/false, "scores": {{"accuracy": X, "hallucination": X, "empathy": X, "actionability": X, "tone": X}}, "overall_score": X, "issues": ["issue 1", "issue 2"], "suggested_rewrite": "improved version if needed" }}""" return call_llm_api(prompt, temperature=0.1, response_format="json")

def route_by_language(text): """Detect language and select optimal model.""" lang_map = { "zh": "deepseek", # Chinese "ja": "qwen", # Japanese (Qwen excels) "en": "deepseek", # English "es": "deepseek", # Spanish "ar": "qwen", # Arabic "default": "deepseek" } # Use langdetect or similar detected_lang = detect_language(text) return lang_map.get(detected_lang, lang_map["default"])