""" Groq API 服務 提供快速的 AI 推理功能 """ from groq import Groq from backend.config import settings import logging import json import re logger = logging.getLogger(__name__) class GroqService: """Groq AI 服務類""" def __init__(self): self.api_key = settings.GROQ_API_KEY self.model = getattr(settings, 'GROQ_MODEL', 'qwen/qwen3-32b') if not self.api_key: logger.warning("GROQ_API_KEY 未設定，Groq 服務將無法使用") self.client = None else: self.client = Groq(api_key=self.api_key) def is_available(self) -> bool: """檢查 Groq 服務是否可用""" return self.client is not None def chat_completion(self, message: str, system_prompt: str = None, temperature: float = 0.6) -> str: """ 聊天完成 Args: message: 用戶訊息 system_prompt: 系統提示詞 temperature: 溫度參數 (0.0-1.0) Returns: AI 回應內容 """ if not self.is_available(): raise Exception("Groq 服務不可用，請檢查 API Key 設定") try: messages = [] if system_prompt: messages.append({"role": "system", "content": system_prompt}) messages.append({"role": "user", "content": message}) completion = self.client.chat.completions.create( model=self.model, messages=messages, temperature=temperature, max_completion_tokens=4096, top_p=0.95, reasoning_effort="default", stop=None, ) return completion.choices[0].message.content except Exception as e: logger.error(f"Groq API 調用失敗: {str(e)}") raise Exception(f"AI 服務暫時無法使用: {str(e)}") def analyze_intent(self, message: str) -> dict: """ 分析用戶訊息意圖 Args: message: 用戶訊息 Returns: 意圖分析結果 """ if not self.is_available(): return { "intent": "unknown", "confidence": 0.0, "entities": {}, "error": "Groq 服務不可用" } system_prompt = """你是一個意圖分析助手，分析用戶訊息的意圖。 請分析用戶訊息並回傳 JSON 格式： { "intent": "search|chat|help|order|inventory|unknown", "confidence": 0.0-1.0, "entities": { "product": "商品名稱", "action": "動作類型", "price_range": "價格範圍", "category": "商品分類" }, "reasoning": "分析原因" } 意圖類型說明： - search: 商品查詢、搜尋相關 - chat: 一般聊天、問候、閒聊 - help: 求助、說明、指令 - order: 訂單查詢、訂單相關 - inventory: 庫存查詢、庫存相關 - unknown: 無法確定意圖 請用繁體中文回應，並確保回傳有效的 JSON 格式。""" try: response = self.chat_completion(message, system_prompt, temperature=0.3) return self._parse_intent_response(response) except Exception as e: logger.error(f"意圖分析失敗: {str(e)}") return { "intent": "unknown", "confidence": 0.0, "entities": {}, "error": str(e) } def _parse_intent_response(self, response: str) -> dict: """解析意圖分析回應""" try: # 嘗試提取 JSON 部分 json_match = re.search(r'\{.*\}', response, re.DOTALL) if json_match: json_str = json_match.group() result = json.loads(json_str) # 驗證必要欄位 if "intent" not in result: result["intent"] = "unknown" if "confidence" not in result: result["confidence"] = 0.5 if "entities" not in result: result["entities"] = {} # 確保信心度在有效範圍內 result["confidence"] = max(0.0, min(1.0, float(result["confidence"]))) return result else: # 如果沒有找到 JSON，嘗試簡單解析 return self._fallback_intent_parsing(response) except json.JSONDecodeError as e: logger.warning(f"JSON 解析失敗: {str(e)}, 回應內容: {response}") return self._fallback_intent_parsing(response) except Exception as e: logger.error(f"意圖回應解析錯誤: {str(e)}") return { "intent": "unknown", "confidence": 0.0, "entities": {}, "error": f"解析錯誤: {str(e)}" } def _fallback_intent_parsing(self, response: str) -> dict: """備用的意圖解析方法""" response_lower = response.lower() # 簡單的關鍵字匹配 if any(keyword in response_lower for keyword in ['查詢', '搜尋', '找', '商品', '產品']): return { "intent": "search", "confidence": 0.6, "entities": {}, "reasoning": "關鍵字匹配: 搜尋相關" } elif any(keyword in response_lower for keyword in ['訂單', 'order']): return { "intent": "order", "confidence": 0.6, "entities": {}, "reasoning": "關鍵字匹配: 訂單相關" } elif any(keyword in response_lower for keyword in ['庫存', 'inventory', '存貨']): return { "intent": "inventory", "confidence": 0.6, "entities": {}, "reasoning": "關鍵字匹配: 庫存相關" } elif any(keyword in response_lower for keyword in ['幫助', 'help', '說明', '指令']): return { "intent": "help", "confidence": 0.8, "entities": {}, "reasoning": "關鍵字匹配: 幫助相關" } else: return { "intent": "chat", "confidence": 0.4, "entities": {}, "reasoning": "預設為聊天模式" } def generate_business_response(self, query_result: dict, original_message: str) -> str: """ 根據業務查詢結果生成自然回應 Args: query_result: 業務查詢結果 original_message: 原始用戶訊息 Returns: 自然語言回應 """ if not self.is_available(): return self._fallback_business_response(query_result) system_prompt = f"""你是一個友善的客服助手，需要根據查詢結果為用戶生成自然的回應。 用戶原始訊息：{original_message} 查詢結果： - 成功: {query_result.get('success', False)} - 資料筆數: {len(query_result.get('data', []))} - 意圖: {query_result.get('intent', 'unknown')} 請用繁體中文生成一個友善、自然的回應，包含查詢結果的摘要。 如果有具體資料，請整理成易讀的格式。 回應長度控制在 200 字以內。""" try: # 準備查詢結果摘要 data_summary = self._prepare_data_summary(query_result.get('data', [])) full_prompt = f"{system_prompt}\n\n資料摘要：\n{data_summary}" response = self.chat_completion(original_message, full_prompt, temperature=0.4) return response except Exception as e: logger.error(f"生成業務回應失敗: {str(e)}") return self._fallback_business_response(query_result) def _prepare_data_summary(self, data: list) -> str: """準備資料摘要""" if not data: return "沒有找到相關資料" summary_lines = [] for i, item in enumerate(data[:5]): # 最多顯示 5 筆 if isinstance(item, dict): # 提取重要欄位 name = item.get('name', item.get('product_name', '')) price = item.get('price', item.get('unit_price', '')) stock = item.get('stock', item.get('quantity', '')) line_parts = [] if name: line_parts.append(f"名稱: {name}") if price: line_parts.append(f"價格: ${price}") if stock: line_parts.append(f"庫存: {stock}") if line_parts: summary_lines.append(f"{i+1}. {', '.join(line_parts)}") if len(data) > 5: summary_lines.append(f"... 還有 {len(data) - 5} 筆資料") return '\n'.join(summary_lines) if summary_lines else "資料格式無法解析" def _fallback_business_response(self, query_result: dict) -> str: """備用的業務回應生成""" if query_result.get('success'): data_count = len(query_result.get('data', [])) if data_count > 0: return f"✅ 查詢成功！找到 {data_count} 筆相關資料。" else: return "✅ 查詢完成，但沒有找到符合條件的資料。" else: error_msg = query_result.get('error', '未知錯誤') return f"❌ 查詢失敗：{error_msg}"