""" Triage engine — calls OpenAI to replicate the n8n AI Agent logic. In production, the Streamlit app posts to the n8n webhook; this module runs locally for the interactive demo and HuggingFace Space. """ import json import random from datetime import datetime, timedelta # ── Mock order database ──────────────────────────────────────────────────────── MOCK_ORDERS = { "ORD-789": { "product": "Premium Wireless Headphones", "status": "delayed", "amount": 149.99, "eta": (datetime.now() + timedelta(days=5)).strftime("%Y-%m-%d"), "carrier": "FedEx", "tracking": "FX123456789", "delay_reason": "weather disruption at hub", }, "ORD-456": { "product": "Smart Watch Series 3", "status": "in_transit", "amount": 299.00, "eta": (datetime.now() + timedelta(days=2)).strftime("%Y-%m-%d"), "carrier": "UPS", "tracking": "1Z999AA10123456784", "delay_reason": None, }, "ORD-321": { "product": "Laptop Stand Pro", "status": "delivered", "amount": 49.99, "eta": "Delivered", "carrier": "USPS", "tracking": "9400111899223974657", "delay_reason": None, }, "ORD-654": { "product": "Mechanical Keyboard RGB", "status": "processing", "amount": 89.99, "eta": (datetime.now() + timedelta(days=7)).strftime("%Y-%m-%d"), "carrier": "DHL", "tracking": "Pending", "delay_reason": None, }, "ORD-987": { "product": "Gaming Mouse X500", "status": "delayed", "amount": 79.99, "eta": (datetime.now() + timedelta(days=3)).strftime("%Y-%m-%d"), "carrier": "FedEx", "tracking": "FX987654321", "delay_reason": "high volume at sorting facility", }, } SUPPORTED_LANGUAGES = [ "Spanish", "French", "German", "Japanese", "Portuguese", "Italian", "Dutch", "Korean", "Chinese", "Arabic", "Hindi", "Russian", "Turkish", "Polish", "Swedish", "Danish", "Norwegian", "Finnish", "Greek", "Hebrew", "Thai", "Vietnamese", "Indonesian", "Malay", "Czech", "Hungarian", "Romanian", "Bulgarian", "Croatian", "Slovak", ] # ── Demo output (no API key) ─────────────────────────────────────────────────── DEMO_OUTPUTS = { "TKT-001": { "detected_language": "Spanish", "language_code": "es", "order_id": "ORD-789", "order_status": "delayed — weather disruption at hub, ETA 2025-01-28", "sentiment": "very_angry", "sentiment_score": 0.94, "native_reply": ( "Estimado Carlos García,\n\n" "Le pedimos disculpas sinceramente por los inconvenientes causados con su pedido #ORD-789. " "Entendemos completamente su frustración y lamentamos profundamente este retraso inaceptable.\n\n" "Hemos verificado que su pedido se encuentra en tránsito pero ha experimentado un retraso " "debido a interrupciones climáticas en nuestro centro logístico. La nueva fecha estimada " "de entrega es el 28 de enero de 2025.\n\n" "Como compensación por este inconveniente, le ofrecemos:\n" "• Envío gratuito en su próximo pedido\n" "• Descuento del 15% aplicado automáticamente\n\n" "Nuestro equipo de servicio al cliente de habla hispana se pondrá en contacto con usted " "en las próximas 2 horas para resolver esto personalmente.\n\n" "Atentamente,\nEquipo de Soporte Global" ), "suggested_status": "open", "english_summary": "Customer Carlos García is very angry about a 3-week delay on order ORD-789 (Premium Wireless Headphones). Order is delayed due to weather disruption. Customer demands immediate refund. Escalation recommended.", }, } # ── OpenAI triage ────────────────────────────────────────────────────────────── def _call_openai(ticket_id, subject, body, name, email, order_info, api_key): try: from openai import OpenAI except ImportError: return None client = OpenAI(api_key=api_key) order_context = "" if order_info: order_context = f"\n\nOrder Status Retrieved:\n{json.dumps(order_info, indent=2)}" schema = { "detected_language": "string — full language name e.g. Spanish", "language_code": "string — ISO 639-1 code e.g. es", "order_id": "string — extracted order ID or empty string", "order_status": "string — human readable status from order data, or 'Not found'", "sentiment": "string — one of: neutral, mild_concern, upset, very_angry", "sentiment_score": "number — 0.0 to 1.0", "native_reply": "string — complete professional reply in the customer's detected language", "suggested_status": "string — zendesk status: solved or open", "english_summary": "string — 1-2 sentence summary in English for the support team", } prompt = f"""You are an elite multi-lingual customer support AI for a global e-commerce company. Ticket ID: {ticket_id} Subject: {subject} Customer: {name} ({email}) Message: {body}{order_context} Analyze this ticket and return ONLY a valid JSON object matching this schema: {json.dumps(schema, indent=2)} Rules: - detected_language: Full name of the language the customer wrote in - sentiment must be exactly: neutral, mild_concern, upset, or very_angry - native_reply must be written entirely in the customer's detected language, be empathetic, professional, and reference real order details if available - If order is delayed or missing, acknowledge it and offer compensation - english_summary is for the internal English-speaking team """ response = client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": prompt}], response_format={"type": "json_object"}, temperature=0.3, ) return json.loads(response.choices[0].message.content) def _extract_order_id(text: str) -> str: import re patterns = [ r'#?(ORD[-_]?\d+)', r'order\s+(?:number|#|id)?[\s:]*([A-Z0-9][-A-Z0-9]{3,})', r'\b([A-Z]{2,4}-\d{3,})\b', ] for p in patterns: m = re.search(p, text, re.IGNORECASE) if m: raw = m.group(1).upper().replace('_', '-') if not raw.startswith('ORD-'): raw = 'ORD-' + raw.lstrip('ORD').lstrip('-') return raw return "" def _demo_result(ticket_id: str, body: str) -> dict: if ticket_id in DEMO_OUTPUTS: return DEMO_OUTPUTS[ticket_id] order_id = _extract_order_id(body) order_info = MOCK_ORDERS.get(order_id, {}) # Language heuristics lang_map = [ (["hola", "gracias", "pedido", "enojado", "quiero", "reembolso", "semanas"], "Spanish", "es"), (["bonjour", "commande", "merci", "livraison"], "French", "fr"), (["bestellung", "bitte", "danke", "lieferung", "enttäuschend"], "German", "de"), (["注文", "配送", "届いて", "確認"], "Japanese", "ja"), (["pedido", "chegou", "urgente", "obrigado"], "Portuguese", "pt"), (["ordine", "grazie", "consegna", "spedizione"], "Italian", "it"), ] body_lower = body.lower() detected_language, language_code = "English", "en" for words, lang, code in lang_map: if any(w in body_lower for w in words): detected_language, language_code = lang, code break # Sentiment heuristics angry_words = ["enojado", "angry", "furious", "inaceptable", "reembolso", "wut", "angry", "terrible", "horrible", "enttäuschend", "sofort"] upset_words = ["disappointed", "unhappy", "not happy", "upset", "still waiting", "urgente", "please resolve"] mild_words = ["wondering", "checking", "voudrais", "möchte", "confirmar", "savoir"] if any(w in body_lower for w in angry_words) and ("!" in body or "NOW" in body.upper() or "JETZT" in body.upper()): sentiment = "very_angry" score = round(random.uniform(0.85, 0.97), 2) elif any(w in body_lower for w in angry_words): sentiment = "upset" score = round(random.uniform(0.65, 0.84), 2) elif any(w in body_lower for w in upset_words): sentiment = "upset" score = round(random.uniform(0.55, 0.75), 2) elif any(w in body_lower for w in mild_words): sentiment = "mild_concern" score = round(random.uniform(0.3, 0.54), 2) else: sentiment = "neutral" score = round(random.uniform(0.05, 0.29), 2) order_status = "Not found" if order_info: order_status = f"{order_info['status'].replace('_',' ').title()}" if order_info.get("delay_reason"): order_status += f" — {order_info['delay_reason']}, ETA {order_info['eta']}" else: order_status += f", ETA {order_info['eta']}" native_replies = { "es": f"Estimado/a cliente,\n\nLamentamos los inconvenientes con su pedido{' ' + order_id if order_id else ''}. Estado actual: {order_status}.\nNuestro equipo se pondrá en contacto con usted pronto.\n\nAtentamente,\nSoporte Global", "fr": f"Cher(e) client(e),\n\nNous sommes désolés pour les désagréments concernant votre commande{' ' + order_id if order_id else ''}. Statut actuel: {order_status}.\nNotre équipe vous contactera très prochainement.\n\nCordialement,\nSupport Global", "de": f"Liebe/r Kunde/Kundin,\n\nWir entschuldigen uns für die Unannehmlichkeiten mit Ihrer Bestellung{' ' + order_id if order_id else ''}. Aktueller Status: {order_status}.\nUnser Team wird sich bald bei Ihnen melden.\n\nMit freundlichen Grüßen,\nGlobaler Support", "ja": f"お客様へ,\n\nご注文{(' ' + order_id) if order_id else ''}に関するご不便をおかけして申し訳ございません。現在の状況: {order_status}。\n担当者よりご連絡いたします。\n\nよろしくお願いいたします,\nグローバルサポート", "pt": f"Caro(a) cliente,\n\nLamentamos os inconvenientes com o seu pedido{' ' + order_id if order_id else ''}. Status atual: {order_status}.\nNossa equipe entrará em contato em breve.\n\nAtenciosamente,\nSuporteGlobal", "en": f"Dear Customer,\n\nWe sincerely apologize for any inconvenience with your order{' ' + order_id if order_id else ''}. Current status: {order_status}.\nOur team will reach out to you shortly.\n\nBest regards,\nGlobal Support", } native_reply = native_replies.get(language_code, native_replies["en"]) summaries = { "very_angry": f"Customer is very angry about order {order_id or 'N/A'} ({order_status}). Immediate attention required.", "upset": f"Customer is upset about order {order_id or 'N/A'} ({order_status}). Needs prompt resolution.", "mild_concern": f"Customer has mild concerns about order {order_id or 'N/A'} ({order_status}). Standard follow-up.", "neutral": f"Customer is checking on order {order_id or 'N/A'} ({order_status}). Routine inquiry.", } return { "detected_language": detected_language, "language_code": language_code, "order_id": order_id, "order_status": order_status, "sentiment": sentiment, "sentiment_score": score, "native_reply": native_reply, "suggested_status": "open" if sentiment == "very_angry" else "solved", "english_summary": summaries[sentiment], } def run_triage(ticket_id: str, subject: str, body: str, name: str, email: str, api_key: str = None, demo_mode: bool = False) -> dict: order_id = _extract_order_id(body) order_info = MOCK_ORDERS.get(order_id) if demo_mode or not api_key: result = _demo_result(ticket_id, body) else: result = _call_openai(ticket_id, subject, body, name, email, order_info, api_key) if result is None: result = _demo_result(ticket_id, body) if order_info and not result.get("order_status"): result["order_status"] = order_info["status"] return result