""" 訊息路由器 - 根據前綴和內容智能路由訊息 """ import logging from typing import Dict, Any from backend.services.groq_service import GroqService from backend.services.business_query_service import BusinessQueryService from backend.services.pydantic_ai_service import ProductQueryService logger = logging.getLogger(__name__) class MessageRouter: """訊息路由器類""" def __init__(self): self.groq_service = GroqService() self.business_service = BusinessQueryService() self.product_query_service = ProductQueryService() # 路由統計 self.route_stats = { "chat": 0, "search": 0, "product_query": 0, "help": 0, "smart": 0, "error": 0 } def route_message(self, message: str, user_id: str) -> Dict[str, Any]: """ 根據前綴和內容路由訊息 Args: message: 用戶訊息 user_id: 用戶ID Returns: 處理結果字典 """ try: # 清理訊息 clean_message = message.strip() if not clean_message: return self._handle_empty_message(user_id) logger.info(f"路由訊息 - 用戶: {user_id}, 內容: {clean_message[:50]}...") # 1. 聊天模式 - /chat 前綴 if clean_message.startswith('/chat '): self.route_stats["chat"] += 1 return self._handle_chat_mode(clean_message[6:].strip(), user_id) # 2. 商品查詢 - /search 前綴 elif clean_message.startswith('/search '): self.route_stats["search"] += 1 return self._handle_search_mode(clean_message[8:].strip(), user_id) # 3. 幫助指令 elif clean_message.lower() in ['/help', '幫助', 'help', '說明', '指令']: self.route_stats["help"] += 1 return self._handle_help_mode(user_id) # 4. 特殊指令 elif clean_message.lower() in ['選單', 'menu', '功能']: return self._handle_menu_mode(user_id) elif clean_message.lower() in ['統計', 'stats', '路由統計']: return self._handle_stats_mode(user_id) # 5. 智能路由 - 無前綴時進行意圖分析 else: self.route_stats["smart"] += 1 return self._handle_smart_routing(clean_message, user_id) except Exception as e: self.route_stats["error"] += 1 logger.error(f"訊息路由錯誤: {str(e)}") return { "type": "text", "text": "❌ 系統處理訊息時發生錯誤，請稍後再試。", "mode": "error", "success": False, "error": str(e) } def _handle_chat_mode(self, message: str, user_id: str) -> Dict[str, Any]: """處理聊天模式""" try: if not message: return { "type": "text", "text": "💬 請輸入您想聊的內容！\n\n範例：/chat 今天天氣如何？", "mode": "chat", "success": True } if not self.groq_service.is_available(): return { "type": "text", "text": "💬 聊天服務暫時無法使用，請稍後再試。\n\n💡 您可以嘗試使用 /search 進行商品查詢。", "mode": "chat", "success": False, "error": "Groq 服務不可用" } system_prompt = """你是一個友善的客服助手，專門協助用戶解答問題。 請用繁體中文回應，語氣要親切自然。 如果用戶詢問商品或業務相關問題，建議他們使用 /search 指令進行查詢。 回應長度控制在 150 字以內。""" response = self.groq_service.chat_completion(message, system_prompt) return { "type": "text", "text": f"💬 {response}", "mode": "chat", "success": True, "user_id": user_id } except Exception as e: logger.error(f"聊天模式處理錯誤: {str(e)}") return { "type": "text", "text": "💬 聊天服務暫時無法使用，請稍後再試。", "mode": "chat", "success": False, "error": str(e) } def _handle_search_mode(self, message: str, user_id: str) -> Dict[str, Any]: """處理搜尋模式""" try: if not message: return { "type": "text", "text": "🔍 請輸入您要查詢的內容！\n\n範例：\n• /search iPhone 15\n• /search 價格 1000-5000\n• /search 庫存不足的商品", "mode": "search", "success": True } # 使用業務查詢服務 result = self.business_service.process_user_query(message, user_id) # 如果有 Groq 服務，使用它來生成更自然的回應 if self.groq_service.is_available() and result.get("success"): try: natural_response = self.groq_service.generate_business_response(result, message) response_text = f"🔍 {natural_response}" except Exception as e: logger.warning(f"Groq 回應生成失敗，使用原始回應: {str(e)}") response_text = f"🔍 {result['response_message']}" else: response_text = f"🔍 {result['response_message']}" return { "type": "text", "text": response_text, "mode": "search", "success": result["success"], "data": result.get("data", []), "intent": result.get("intent", "unknown"), "confidence": result.get("confidence", 0.0), "user_id": user_id } except Exception as e: logger.error(f"搜尋模式處理錯誤: {str(e)}") return { "type": "text", "text": "🔍 搜尋服務暫時無法使用，請稍後再試。", "mode": "search", "success": False, "error": str(e) } def _handle_smart_routing(self, message: str, user_id: str) -> Dict[str, Any]: """智能路由 - 根據內容判斷意圖""" try: # 1. 優先檢查是否為商品查詢 (使用 Pydantic AI) if self.product_query_service.is_available(): product_intent = self.product_query_service.analyze_query_intent(message) # 降低信心度閾值，讓更多商品查詢被 Pydantic AI 處理 if product_intent["is_product_query"] and product_intent["confidence"] > 0.5: logger.info(f"智能路由 -> Pydantic AI 商品查詢 (信心度: {product_intent['confidence']:.2f})") self.route_stats["product_query"] += 1 return self._handle_product_query_mode(message, user_id) else: logger.info(f"Pydantic AI 意圖分析: 商品查詢={product_intent['is_product_query']}, 信心度={product_intent['confidence']:.2f}") else: logger.warning("Pydantic AI 服務不可用，使用傳統路由") # 2. 進行簡單的關鍵字預篩選（只處理非商品查詢） quick_intent = self._quick_intent_check(message) if quick_intent == "help": # 幫助相關 -> 轉到幫助模式 return self._handle_help_mode(user_id) elif quick_intent == "search": # 傳統搜尋關鍵字 -> 轉到搜尋模式 logger.info(f"快速意圖識別: 搜尋模式 - {message[:30]}...") return self._handle_search_mode(message, user_id) # 使用 Groq 進行深度意圖分析 if self.groq_service.is_available(): try: intent_result = self.groq_service.analyze_intent(message) logger.info(f"Groq 意圖分析: {intent_result}") # 根據意圖和信心度決定路由 if intent_result["intent"] in ["search", "order", "inventory"] and intent_result["confidence"] > 0.6: logger.info(f"智能路由 -> 搜尋模式 (信心度: {intent_result['confidence']})") return self._handle_search_mode(message, user_id) elif intent_result["intent"] == "help" and intent_result["confidence"] > 0.7: return self._handle_help_mode(user_id) else: # 其他情況或低信心度 -> 轉到聊天模式 logger.info(f"智能路由 -> 聊天模式 (意圖: {intent_result['intent']}, 信心度: {intent_result['confidence']})") return self._handle_chat_mode(message, user_id) except Exception as e: logger.warning(f"Groq 意圖分析失敗，使用預設路由: {str(e)}") # 錯誤時預設為聊天模式 return self._handle_chat_mode(message, user_id) else: # Groq 不可用時，根據簡單規則路由 if quick_intent == "search": return self._handle_search_mode(message, user_id) else: return self._handle_chat_mode(message, user_id) except Exception as e: logger.error(f"智能路由處理錯誤: {str(e)}") # 錯誤時預設為聊天模式 return self._handle_chat_mode(message, user_id) def _handle_product_query_mode(self, message: str, user_id: str) -> Dict[str, Any]: """處理 Pydantic AI 商品查詢模式""" try: if not message: return { "type": "text", "text": "🛍️ 請告訴我您想查詢什麼商品！\n\n範例：\n• 是否有推薦貓砂？\n• 查詢狗糧庫存\n• 有什麼寵物用品？", "mode": "product_query", "success": True } if not self.product_query_service.is_available(): return { "type": "text", "text": "🛍️ 商品查詢服務暫時無法使用，請稍後再試。\n\n💡 您可以嘗試使用 /search 進行基本查詢。", "mode": "product_query", "success": False, "error": "Pydantic AI 服務不可用" } # 使用同步版本避免異步複雜性 result = self.product_query_service.process_product_query_sync(message, user_id) if result["success"]: return { "type": "text", "text": f"🛍️ {result['text']}", "mode": "product_query", "success": True, "products_found": result.get("products_found", 0), "has_recommendations": result.get("has_recommendations", False), "user_id": user_id } else: return { "type": "text", "text": f"🛍️ {result.get('text', '商品查詢失敗')}", "mode": "product_query", "success": False, "error": result.get("error"), "user_id": user_id } except Exception as e: logger.error(f"商品查詢模式處理錯誤: {str(e)}") return { "type": "text", "text": "🛍️ 商品查詢服務暫時無法使用，請稍後再試。", "mode": "product_query", "success": False, "error": str(e) } def _quick_intent_check(self, message: str) -> str: """快速意圖檢查 - 基於關鍵字（避免與 Pydantic AI 商品查詢衝突）""" message_lower = message.lower() # 幫助相關關鍵字（優先處理） help_keywords = ['幫助', 'help', '說明', '怎麼用', '指令', '功能', '統計', 'stats', '選單', 'menu'] # 非商品的業務查詢關鍵字（避免與商品查詢衝突） business_keywords = [ '訂單狀態', '訂單查詢', '我的訂單', '交易記錄', '購買記錄', '客戶資料', '會員資料', '帳戶資訊', '銷售報表', '財務報表' ] if any(keyword in message_lower for keyword in help_keywords): return "help" elif any(keyword in message_lower for keyword in business_keywords): return "search" else: # 預設返回 "chat"，讓 Pydantic AI 有機會處理商品查詢 return "chat" def _handle_help_mode(self, user_id: str) -> Dict[str, Any]: """處理幫助模式""" help_text = """🤖 智能客服助手使用指南 📝 指令模式： • /chat [訊息] - 聊天模式 範例：/chat 今天天氣如何？ • /search [查詢] - 商品查詢模式 範例：/search iPhone 15 Pro 範例：/search 價格 1000-5000 • /help - 顯示此說明 🧠 智能模式： 直接輸入訊息，系統會自動判斷是聊天還是查詢！ 🛍️ 商品查詢範例（AI 智能識別）： • "是否有推薦貓砂？" • "有什麼狗糧推薦？" • "查詢寵物用品庫存" • "iPhone 還有庫存嗎？" • "1000到5000的商品有哪些？" 💬 聊天範例： • "你好！" • "今天天氣如何？" • "推薦一些好用的功能" 📋 其他查詢： • "我的訂單狀態如何？" • "低庫存商品有哪些？" 輸入「選單」查看功能選單 輸入「統計」查看使用統計""" return { "type": "text", "text": help_text, "mode": "help", "success": True, "user_id": user_id } def _handle_menu_mode(self, user_id: str) -> Dict[str, Any]: """處理選單模式""" menu_text = """🏪 功能選單 請選擇您需要的功能： 🔍 商品查詢 • /search [商品名稱] • /search 價格 [範圍] 💬 智能聊天 • /chat [您的問題] 📊 常用查詢 • "庫存查詢" • "我的訂單" • "低庫存商品" • "業務統計" ❓ 說明 • /help - 詳細說明 • 統計 - 使用統計 💡 小提示：您也可以直接輸入問題，系統會智能判斷處理方式！""" return { "type": "text", "text": menu_text, "mode": "menu", "success": True, "user_id": user_id } def _handle_stats_mode(self, user_id: str) -> Dict[str, Any]: """處理統計模式""" total_routes = sum(self.route_stats.values()) if total_routes == 0: stats_text = "📊 路由統計\n\n尚無使用記錄" else: stats_text = f"""📊 路由統計 (總計: {total_routes}) 💬 聊天模式: {self.route_stats['chat']} ({self.route_stats['chat']/total_routes*100:.1f}%) 🔍 搜尋模式: {self.route_stats['search']} ({self.route_stats['search']/total_routes*100:.1f}%) 🛍️ 商品查詢: {self.route_stats['product_query']} ({self.route_stats['product_query']/total_routes*100:.1f}%) 🧠 智能路由: {self.route_stats['smart']} ({self.route_stats['smart']/total_routes*100:.1f}%) ❓ 幫助模式: {self.route_stats['help']} ({self.route_stats['help']/total_routes*100:.1f}%) ❌ 錯誤次數: {self.route_stats['error']} ({self.route_stats['error']/total_routes*100:.1f}%) 🤖 Groq 服務: {'✅ 可用' if self.groq_service.is_available() else '❌ 不可用'} 🛍️ Pydantic AI: {'✅ 可用' if self.product_query_service.is_available() else '❌ 不可用'}""" return { "type": "text", "text": stats_text, "mode": "stats", "success": True, "user_id": user_id, "data": self.route_stats } def _handle_empty_message(self, user_id: str) -> Dict[str, Any]: """處理空訊息""" return { "type": "text", "text": "🤔 您似乎沒有輸入任何內容。\n\n輸入 /help 查看使用說明，或直接告訴我您需要什麼幫助！", "mode": "empty", "success": True, "user_id": user_id } def get_route_statistics(self) -> Dict[str, Any]: """取得路由統計資訊""" return { "stats": self.route_stats.copy(), "total": sum(self.route_stats.values()), "groq_available": self.groq_service.is_available() }