""" Pydantic AI 服務 - 使用 Pydantic AI 框架整合 Groq 和商品查詢功能 解決商品查詢準確性問題，特別是像 "是否有推薦貓砂?" 這類查詢 """ import logging from typing import Dict, Any, List, Optional from dataclasses import dataclass from pydantic import BaseModel, Field from pydantic_ai import Agent, RunContext from backend.services.enhanced_product_service import EnhancedProductService from backend.services.database_service import DatabaseService from backend.config import settings logger = logging.getLogger(__name__) # 依賴注入類型 @dataclass class ProductQueryDependencies: """商品查詢依賴""" enhanced_product_service: EnhancedProductService database_service: DatabaseService user_id: Optional[str] = None # 輸出模型 class ProductQueryResult(BaseModel): """商品查詢結果""" intent: str = Field(description="查詢意圖") response_text: str = Field(description="回應文字") products_found: int = Field(description="找到的商品數量") has_recommendations: bool = Field(description="是否包含推薦") stock_info_included: bool = Field(description="是否包含庫存資訊") search_keywords: List[str] = Field(default_factory=list, description="搜尋關鍵字") class ProductQueryService: """Pydantic AI 商品查詢服務""" def __init__(self): self.enhanced_product_service = EnhancedProductService() self.database_service = DatabaseService() # 創建 Pydantic AI Agent self.product_agent = Agent( f'groq:{settings.GROQ_MODEL}', deps_type=ProductQueryDependencies, output_type=ProductQueryResult, system_prompt=self._get_system_prompt() ) # 註冊工具 self._register_tools() def _get_system_prompt(self) -> str: """系統提示詞""" return """你是一個專業的商品查詢助手，專門協助用戶查詢商品資訊。 你的主要任務： 1. 理解用戶的商品查詢意圖，包括推薦、搜尋、庫存查詢等 2. 使用適當的工具查詢商品資料庫 3. 提供準確、有用的商品資訊回應 特別注意： - 當用戶詢問"推薦"、"有沒有"、"是否有"時，要積極查詢相關商品 - 優先顯示有庫存的商品 - 提供具體的商品名稱、庫存狀況和分類資訊 - 如果沒有找到完全匹配的商品，嘗試提供相似或相關的商品 回應要求： - 使用繁體中文 - 語氣友善專業 - 資訊準確完整 - 如果有多個商品，按庫存量排序推薦""" def _register_tools(self): """註冊 AI Agent 工具""" @self.product_agent.tool async def search_products( ctx: RunContext[ProductQueryDependencies], query_text: str, include_recommendations: bool = False ) -> Dict[str, Any]: """ 搜尋商品 Args: query_text: 搜尋關鍵字 include_recommendations: 是否包含推薦功能 """ try: if include_recommendations: # 使用推薦功能 result = ctx.deps.enhanced_product_service.get_product_recommendations( query_text=query_text, limit=10 ) else: # 使用一般搜尋 result = ctx.deps.enhanced_product_service.search_products_advanced( query_text=query_text, include_stock_info=True, limit=10 ) return { "success": result.success, "products": result.data, "count": result.count, "error": result.error } except Exception as e: logger.error(f"商品搜尋工具錯誤: {str(e)}") return { "success": False, "products": [], "count": 0, "error": str(e) } @self.product_agent.tool async def get_products_by_category( ctx: RunContext[ProductQueryDependencies], category_name: str ) -> Dict[str, Any]: """ 根據分類獲取商品 Args: category_name: 分類名稱 """ try: result = ctx.deps.enhanced_product_service.get_products_by_category( category_name=category_name, limit=10 ) return { "success": result.success, "products": result.data, "count": result.count, "error": result.error } except Exception as e: logger.error(f"分類查詢工具錯誤: {str(e)}") return { "success": False, "products": [], "count": 0, "error": str(e) } @self.product_agent.tool async def check_low_stock_products( ctx: RunContext[ProductQueryDependencies], threshold: int = 10 ) -> Dict[str, Any]: """ 檢查低庫存商品 Args: threshold: 庫存閾值 """ try: result = ctx.deps.enhanced_product_service.get_low_stock_products( threshold=threshold ) return { "success": result.success, "products": result.data, "count": result.count, "error": result.error } except Exception as e: logger.error(f"低庫存查詢工具錯誤: {str(e)}") return { "success": False, "products": [], "count": 0, "error": str(e) } def process_product_query_sync(self, user_message: str, user_id: str = None) -> Dict[str, Any]: """ 同步處理商品查詢 - 為了避免異步複雜性 Args: user_message: 用戶訊息 user_id: 用戶ID Returns: 查詢結果字典 """ try: logger.info(f"🔍 Pydantic AI 開始處理商品查詢: '{user_message}'") # 分析查詢意圖 intent_analysis = self.analyze_query_intent(user_message) logger.info(f"📊 意圖分析結果: {intent_analysis}") if not intent_analysis["is_product_query"]: logger.warning(f"❌ 非商品查詢意圖，拒絕處理") return { "success": False, "text": "這似乎不是商品查詢，請嘗試其他功能。", "mode": "product_query", "error": "非商品查詢意圖" } # 根據意圖類型選擇查詢方法 if intent_analysis["is_recommendation"]: logger.info(f"🛍️ 執行推薦查詢") # 推薦查詢 result = self.enhanced_product_service.get_product_recommendations( query_text=user_message, limit=5 ) else: logger.info(f"🔍 執行一般搜尋") # 一般搜尋 result = self.enhanced_product_service.search_products_advanced( query_text=user_message, include_stock_info=True, limit=10 ) logger.info(f"📋 查詢結果: 成功={result.success}, 數量={result.count}, 錯誤={result.error}") # 格式化回應 if result.success and result.data: response_text = self._format_product_response(result.data, intent_analysis) return { "success": True, "intent": "product_query", "text": response_text, "products_found": result.count, "has_recommendations": intent_analysis["is_recommendation"], "stock_info_included": True, "search_keywords": self._extract_keywords_from_message(user_message), "mode": "product_query", "user_id": user_id } else: return { "success": False, "text": f"抱歉，沒有找到相關商品。{result.error if result.error else ''}", "mode": "product_query", "error": result.error, "user_id": user_id } except Exception as e: logger.error(f"❌ 同步商品查詢錯誤: {str(e)}") import traceback logger.error(f"📋 錯誤詳情: {traceback.format_exc()}") return { "success": False, "text": f"抱歉，商品查詢時發生錯誤：{str(e)}", "mode": "product_query", "error": str(e), "user_id": user_id } async def process_product_query(self, user_message: str, user_id: str = None) -> Dict[str, Any]: """ 處理商品查詢 Args: user_message: 用戶訊息 user_id: 用戶ID Returns: 查詢結果字典 """ try: # 準備依賴 deps = ProductQueryDependencies( enhanced_product_service=self.enhanced_product_service, database_service=self.database_service, user_id=user_id ) # 執行 AI Agent result = await self.product_agent.run(user_message, deps=deps) # 記錄查詢 if user_id: try: self.database_service.save_message(user_id, user_message, "product_query") except Exception as e: logger.warning(f"記錄查詢失敗: {str(e)}") return { "success": True, "intent": result.output.intent, "text": result.output.response_text, "products_found": result.output.products_found, "has_recommendations": result.output.has_recommendations, "stock_info_included": result.output.stock_info_included, "search_keywords": result.output.search_keywords, "mode": "product_query", "user_id": user_id } except Exception as e: logger.error(f"Pydantic AI 商品查詢錯誤: {str(e)}") return { "success": False, "text": f"抱歉，商品查詢時發生錯誤：{str(e)}", "mode": "product_query", "error": str(e), "user_id": user_id } def is_available(self) -> bool: """檢查服務是否可用""" return bool(settings.GROQ_API_KEY) def analyze_query_intent(self, message: str) -> Dict[str, Any]: """ 分析查詢意圖 - 判斷是否為商品查詢 Args: message: 用戶訊息 Returns: 意圖分析結果 """ message_lower = message.lower() # 商品查詢關鍵字（擴展版） product_keywords = [ '推薦', '有沒有', '是否有', '請問有', '商品', '產品', '貨品', '查詢', '搜尋', '找', '庫存', '存貨', '價格', '多少錢', '貓砂', '狗糧', '寵物', '食品', '用品', '貓', '狗', '寵物用品', 'cat', 'dog', 'pet', 'litter', 'food' # 英文關鍵字 ] # 推薦查詢關鍵字 recommendation_keywords = ['推薦', '建議', '介紹', '有什麼', '哪些', '什麼好', '推薦一些'] # 庫存查詢關鍵字 inventory_keywords = ['庫存', '存貨', '剩餘', '還有', '現貨', '有多少', '剩多少'] is_product_query = any(keyword in message_lower for keyword in product_keywords) is_recommendation = any(keyword in message_lower for keyword in recommendation_keywords) is_inventory_check = any(keyword in message_lower for keyword in inventory_keywords) confidence = 0.5 if is_product_query: confidence += 0.3 if is_recommendation: confidence += 0.2 if is_inventory_check: confidence += 0.2 return { "is_product_query": is_product_query, "is_recommendation": is_recommendation, "is_inventory_check": is_inventory_check, "confidence": min(confidence, 1.0), "intent": "product_query" if is_product_query else "unknown" } def _format_product_response(self, products: List[Dict[str, Any]], intent_analysis: Dict[str, Any]) -> str: """格式化商品查詢回應""" if not products: return "沒有找到相關商品。" # 根據意圖類型調整回應風格 if intent_analysis["is_recommendation"]: header = f"為您推薦 {len(products)} 個商品：" elif intent_analysis["is_inventory_check"]: header = f"庫存查詢結果，找到 {len(products)} 個商品：" else: header = f"商品搜尋結果，找到 {len(products)} 個商品：" response_lines = [header, ""] for i, product in enumerate(products[:5], 1): # 最多顯示5個 name = product.get('product_name', 'N/A') stock = product.get('current_stock', 0) category = product.get('category_name', '') warehouse = product.get('warehouse', '') # 庫存狀態 stock_status = product.get('stock_status', self._get_stock_status_text(stock)) line = f"{i}. {name}" if category: line += f" ({category})" line += f"\n 庫存: {stock} - {stock_status}" if warehouse: line += f"\n 倉庫: {warehouse}" if product.get('recommendation_reason'): line += f"\n 推薦原因: {product['recommendation_reason']}" response_lines.append(line) if len(products) > 5: response_lines.append(f"\n... 還有 {len(products) - 5} 個商品") # 添加庫存提醒 low_stock_count = sum(1 for p in products if p.get('current_stock', 0) <= 10) if low_stock_count > 0: response_lines.append(f"\n⚠️ 其中 {low_stock_count} 個商品庫存偏低，建議盡快補貨。") return "\n".join(response_lines) def _get_stock_status_text(self, stock: int) -> str: """獲取庫存狀態文字""" if stock <= 0: return "缺貨 ❌" elif stock <= 5: return "庫存極低 🔴" elif stock <= 10: return "庫存偏低 🟡" elif stock <= 50: return "庫存正常 🟢" else: return "庫存充足 ✅" def _extract_keywords_from_message(self, message: str) -> List[str]: """從訊息中提取關鍵字""" # 移除常見的查詢詞彙 stop_words = ['推薦', '有沒有', '是否有', '請問', '想要', '需要', '找', '查詢', '搜尋', '？', '?'] # 分割並清理關鍵字 words = message.replace('?', '').replace('？', '').split() keywords = [word for word in words if word not in stop_words and len(word) > 1] return keywords if keywords else [message.strip()]