File size: 5,766 Bytes
7e828dc |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 |
"""
測試 Pydantic AI 整合 - 驗證商品查詢功能
特別測試 "是否有推薦貓砂?" 這類查詢的準確性
"""
import asyncio
import sys
import os
# 添加專案根目錄到 Python 路徑
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from backend.services.message_router import MessageRouter
from backend.services.pydantic_ai_service import ProductQueryService
from backend.services.enhanced_product_service import EnhancedProductService
def test_message_router():
"""測試訊息路由器"""
print("🧪 測試訊息路由器...")
router = MessageRouter()
# 測試案例
test_cases = [
"是否有推薦貓砂？",
"有什麼狗糧推薦？",
"查詢寵物用品庫存",
"iPhone 還有庫存嗎？",
"推薦一些好用的商品",
"你好！今天天氣如何？", # 應該路由到聊天模式
"/help", # 應該顯示幫助
]
for i, message in enumerate(test_cases, 1):
print(f"\n--- 測試案例 {i}: {message} ---")
try:
result = router.route_message(message, "test_user")
print(f"模式: {result.get('mode', 'unknown')}")
print(f"成功: {result.get('success', False)}")
print(f"回應: {result.get('text', 'No response')[:100]}...")
if result.get('products_found'):
print(f"找到商品數量: {result['products_found']}")
except Exception as e:
print(f"❌ 錯誤: {str(e)}")
# 顯示統計
print(f"\n📊 路由統計:")
stats = router.get_route_statistics()
for mode, count in stats['stats'].items():
print(f" {mode}: {count}")
def test_product_query_service():
"""測試 Pydantic AI 商品查詢服務"""
print("\n🛍️ 測試 Pydantic AI 商品查詢服務...")
try:
service = ProductQueryService()
if not service.is_available():
print("❌ Pydantic AI 服務不可用 (請檢查 GROQ_API_KEY)")
return
# 測試意圖分析
test_messages = [
"是否有推薦貓砂？",
"有什麼狗糧推薦？",
"查詢iPhone庫存",
"今天天氣如何？", # 非商品查詢
]
print("\n🔍 意圖分析測試:")
for message in test_messages:
intent = service.analyze_query_intent(message)
print(f" '{message}' -> 商品查詢: {intent['is_product_query']}, 信心度: {intent['confidence']:.2f}")
# 測試實際查詢
print("\n🛍️ 實際查詢測試:")
query_result = service.process_product_query_sync("是否有推薦貓砂？", "test_user")
print(f"成功: {query_result['success']}")
print(f"意圖: {query_result.get('intent', 'unknown')}")
print(f"回應: {query_result.get('text', 'No response')}")
if query_result.get('products_found'):
print(f"找到商品: {query_result['products_found']}")
except Exception as e:
print(f"❌ 測試錯誤: {str(e)}")
def test_enhanced_product_service():
"""測試增強商品服務"""
print("\n🔧 測試增強商品服務...")
try:
service = EnhancedProductService()
# 測試商品推薦
print("🎯 測試商品推薦功能:")
result = service.get_product_recommendations("貓砂", limit=5)
print(f"成功: {result.success}")
print(f"找到商品數量: {result.count}")
if result.success and result.data:
print("推薦商品:")
for i, product in enumerate(result.data[:3], 1):
print(f" {i}. {product.get('product_name', 'N/A')} - 庫存: {product.get('current_stock', 0)}")
elif result.error:
print(f"錯誤: {result.error}")
# 測試進階搜尋
print("\n🔍 測試進階搜尋功能:")
result = service.search_products_advanced("寵物", include_stock_info=True, limit=3)
print(f"成功: {result.success}")
print(f"找到商品數量: {result.count}")
if result.success and result.data:
print("搜尋結果:")
for i, product in enumerate(result.data, 1):
print(f" {i}. {product.get('product_name', 'N/A')} - {product.get('stock_status', 'N/A')}")
except Exception as e:
print(f"❌ 測試錯誤: {str(e)}")
def test_database_connection():
"""測試資料庫連接"""
print("\n🗄️ 測試資料庫連接...")
try:
from backend.database.connection import test_database_connection
if test_database_connection():
print("✅ 資料庫連接正常")
else:
print("❌ 資料庫連接失敗")
except Exception as e:
print(f"❌ 資料庫測試錯誤: {str(e)}")
def main():
"""主測試函數"""
print("🚀 開始 Pydantic AI 整合測試\n")
# 1. 測試資料庫連接
test_database_connection()
# 2. 測試增強商品服務
test_enhanced_product_service()
# 3. 測試 Pydantic AI 服務
test_product_query_service()
# 4. 測試訊息路由器
test_message_router()
print("\n✅ 測試完成！")
if __name__ == "__main__":
# 檢查環境變數
from backend.config import settings
print("🔧 環境檢查:")
print(f" GROQ_API_KEY: {'✅ 已設定' if settings.GROQ_API_KEY else '❌ 未設定'}")
print(f" DATABASE_URL: {'✅ 已設定' if settings.DATABASE_URL else '❌ 未設定'}")
print()
# 執行測試
main()
|