1b4f3a7bbe
Some checks failed
CD Pipeline / deploy (push) Failing after 59s
- 建立 Gitea Actions CD pipeline (.gitea/workflows/cd.yaml) - 部署模式: rsync Python 檔案至 188 → docker restart (volume mount) - Dockerfile/requirements 變動時自動重建 Docker image - 部署通知: Telegram (開始/成功/失敗) - 健康檢查: https://mo.wooo.work/health (最多 5 次重試) - 同步最新 CLAUDE.md / ADR-008 / memory (2026-04-19) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
432 lines
14 KiB
Python
432 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
AI 提供者抽象層
|
||
統一 Ollama 和 Gemini 的介面，支援動態切換 AI 提供者
|
||
"""
|
||
|
||
import os
|
||
import logging
|
||
from typing import Optional, Dict, Any, List, Union
|
||
from dataclasses import dataclass
|
||
from datetime import date, datetime
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# AI 提供者設定
|
||
AI_PROVIDER = os.getenv('AI_PROVIDER', 'ollama') # 預設使用 Ollama
|
||
|
||
# 引入服務
|
||
from .ollama_service import OllamaService, OllamaResponse
|
||
from .gemini_service import GeminiService, GeminiResponse, AVAILABLE_GEMINI_MODELS
|
||
|
||
|
||
@dataclass
|
||
class AIResponse:
|
||
"""統一的 AI 回應結構"""
|
||
success: bool
|
||
content: str
|
||
model: str
|
||
provider: str # 'ollama' 或 'gemini'
|
||
error: Optional[str] = None
|
||
total_duration: Optional[float] = None
|
||
input_tokens: int = 0
|
||
output_tokens: int = 0
|
||
total_tokens: int = 0
|
||
input_cost: float = 0.0
|
||
output_cost: float = 0.0
|
||
total_cost: float = 0.0
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
"""轉換為字典格式"""
|
||
return {
|
||
'success': self.success,
|
||
'content': self.content,
|
||
'model': self.model,
|
||
'provider': self.provider,
|
||
'error': self.error,
|
||
'total_duration': self.total_duration,
|
||
'input_tokens': self.input_tokens,
|
||
'output_tokens': self.output_tokens,
|
||
'total_tokens': self.total_tokens,
|
||
'input_cost': self.input_cost,
|
||
'output_cost': self.output_cost,
|
||
'total_cost': self.total_cost,
|
||
}
|
||
|
||
|
||
class AIProviderService:
|
||
"""
|
||
AI 提供者服務 - 統一的 AI 介面
|
||
|
||
支援動態切換 Ollama 和 Gemini，並提供統一的生成介面
|
||
"""
|
||
|
||
def __init__(self, default_provider: str = None):
|
||
"""
|
||
初始化 AI 提供者服務
|
||
|
||
Args:
|
||
default_provider: 預設提供者 ('ollama' 或 'gemini')
|
||
"""
|
||
self._default_provider = default_provider or AI_PROVIDER
|
||
self._ollama = OllamaService()
|
||
self._gemini = GeminiService()
|
||
|
||
# 狀態快取
|
||
self._status_cache = {'timestamp': 0, 'data': None}
|
||
self._CACHE_TTL = 60 # 60 秒
|
||
|
||
@property
|
||
def default_provider(self) -> str:
|
||
"""取得預設提供者"""
|
||
return self._default_provider
|
||
|
||
@default_provider.setter
|
||
def default_provider(self, value: str):
|
||
"""設定預設提供者"""
|
||
if value not in ('ollama', 'gemini'):
|
||
raise ValueError("Provider must be 'ollama' or 'gemini'")
|
||
self._default_provider = value
|
||
logger.info(f"AI 預設提供者已切換至: {value}")
|
||
|
||
def get_status(self, force_refresh: bool = False) -> Dict[str, Any]:
|
||
"""
|
||
取得所有 AI 服務的狀態
|
||
|
||
Args:
|
||
force_refresh: 是否強制刷新快取
|
||
|
||
Returns:
|
||
dict: 包含 Ollama 和 Gemini 狀態的字典
|
||
"""
|
||
import time
|
||
now = time.time()
|
||
|
||
# 使用快取
|
||
if not force_refresh and self._status_cache['data'] is not None:
|
||
if now - self._status_cache['timestamp'] < self._CACHE_TTL:
|
||
return self._status_cache['data']
|
||
|
||
# 檢查各服務狀態
|
||
ollama_connected = self._ollama.check_connection()
|
||
gemini_connected = self._gemini.check_connection()
|
||
|
||
status = {
|
||
'default_provider': self._default_provider,
|
||
'ollama': {
|
||
'connected': ollama_connected,
|
||
'model': self._ollama.model if ollama_connected else None,
|
||
'available_models': self._ollama.available_models if ollama_connected else [],
|
||
'type': 'local',
|
||
'cost': 'free'
|
||
},
|
||
'gemini': {
|
||
'connected': gemini_connected,
|
||
'model': self._gemini.model if gemini_connected else None,
|
||
'available_models': AVAILABLE_GEMINI_MODELS,
|
||
'type': 'cloud',
|
||
'cost': 'paid'
|
||
},
|
||
'recommended_provider': self._get_recommended_provider(ollama_connected, gemini_connected),
|
||
'timestamp': datetime.now().isoformat()
|
||
}
|
||
|
||
# 更新快取
|
||
self._status_cache = {'timestamp': now, 'data': status}
|
||
return status
|
||
|
||
def _get_recommended_provider(self, ollama_ok: bool, gemini_ok: bool) -> str:
|
||
"""根據可用性推薦提供者"""
|
||
if ollama_ok and gemini_ok:
|
||
return self._default_provider
|
||
elif ollama_ok:
|
||
return 'ollama'
|
||
elif gemini_ok:
|
||
return 'gemini'
|
||
else:
|
||
return 'none'
|
||
|
||
def _convert_response(self, response: Union[OllamaResponse, GeminiResponse],
|
||
provider: str) -> AIResponse:
|
||
"""將服務回應轉換為統一格式"""
|
||
if isinstance(response, OllamaResponse):
|
||
return AIResponse(
|
||
success=response.success,
|
||
content=response.content,
|
||
model=response.model,
|
||
provider='ollama',
|
||
error=response.error,
|
||
total_duration=response.total_duration,
|
||
input_tokens=0,
|
||
output_tokens=0,
|
||
total_tokens=0,
|
||
input_cost=0.0,
|
||
output_cost=0.0,
|
||
total_cost=0.0 # Ollama 是免費的
|
||
)
|
||
elif isinstance(response, GeminiResponse):
|
||
return AIResponse(
|
||
success=response.success,
|
||
content=response.content,
|
||
model=response.model,
|
||
provider='gemini',
|
||
error=response.error,
|
||
total_duration=response.total_duration,
|
||
input_tokens=response.input_tokens,
|
||
output_tokens=response.output_tokens,
|
||
total_tokens=response.total_tokens,
|
||
input_cost=response.input_cost,
|
||
output_cost=response.output_cost,
|
||
total_cost=response.total_cost
|
||
)
|
||
else:
|
||
return AIResponse(
|
||
success=False,
|
||
content='',
|
||
model='unknown',
|
||
provider=provider,
|
||
error='Unknown response type'
|
||
)
|
||
|
||
def generate(self, prompt: str, provider: str = None, model: str = None,
|
||
system_prompt: str = None, temperature: float = 0.7,
|
||
timeout: int = None) -> AIResponse:
|
||
"""
|
||
生成文字（統一介面）
|
||
|
||
Args:
|
||
prompt: 使用者提示
|
||
provider: 指定提供者 ('ollama' 或 'gemini')
|
||
model: 指定模型
|
||
system_prompt: 系統提示
|
||
temperature: 創意度
|
||
timeout: 超時時間（秒）
|
||
|
||
Returns:
|
||
AIResponse
|
||
"""
|
||
provider = provider or self._default_provider
|
||
|
||
if provider == 'gemini':
|
||
response = self._gemini.generate(
|
||
prompt=prompt,
|
||
model=model,
|
||
system_prompt=system_prompt,
|
||
temperature=temperature,
|
||
timeout=timeout
|
||
)
|
||
else: # ollama
|
||
response = self._ollama.generate(
|
||
prompt=prompt,
|
||
model=model,
|
||
system_prompt=system_prompt,
|
||
temperature=temperature,
|
||
timeout=timeout
|
||
)
|
||
|
||
return self._convert_response(response, provider)
|
||
|
||
def generate_sales_copy(self, product_name: str, provider: str = None,
|
||
model: str = None, trend_keywords: List[str] = None,
|
||
style: str = "吸睛", upcoming_holidays: List[Dict] = None,
|
||
bestseller_products: List[Dict] = None) -> AIResponse:
|
||
"""
|
||
生成銷售文案（統一介面）
|
||
|
||
Args:
|
||
product_name: 商品名稱
|
||
provider: 指定提供者
|
||
model: 指定模型
|
||
trend_keywords: 趨勢關鍵字
|
||
style: 文案風格
|
||
upcoming_holidays: 即將到來的假期
|
||
bestseller_products: 競品熱銷商品
|
||
|
||
Returns:
|
||
AIResponse
|
||
"""
|
||
provider = provider or self._default_provider
|
||
|
||
if provider == 'gemini':
|
||
response = self._gemini.generate_sales_copy(
|
||
product_name=product_name,
|
||
trend_keywords=trend_keywords,
|
||
style=style,
|
||
upcoming_holidays=upcoming_holidays,
|
||
bestseller_products=bestseller_products,
|
||
model=model
|
||
)
|
||
else: # ollama
|
||
response = self._ollama.generate_sales_copy(
|
||
product_name=product_name,
|
||
trend_keywords=trend_keywords,
|
||
style=style,
|
||
upcoming_holidays=upcoming_holidays,
|
||
bestseller_products=bestseller_products
|
||
)
|
||
|
||
return self._convert_response(response, provider)
|
||
|
||
def extract_keywords(self, text: str, provider: str = None,
|
||
model: str = None, max_keywords: int = 10) -> AIResponse:
|
||
"""
|
||
提取關鍵字（統一介面）
|
||
|
||
Args:
|
||
text: 要分析的文字
|
||
provider: 指定提供者
|
||
model: 指定模型
|
||
max_keywords: 最大關鍵字數量
|
||
|
||
Returns:
|
||
AIResponse
|
||
"""
|
||
provider = provider or self._default_provider
|
||
|
||
if provider == 'gemini':
|
||
response = self._gemini.extract_keywords(text, max_keywords, model)
|
||
else: # ollama
|
||
response = self._ollama.extract_keywords(text, max_keywords)
|
||
|
||
return self._convert_response(response, provider)
|
||
|
||
def search_product_insights(self, product_name: str, provider: str = None,
|
||
model: str = None, include_competitors: bool = True,
|
||
include_trends: bool = True,
|
||
web_context: str = "") -> AIResponse:
|
||
"""
|
||
搜尋商品市場洞察（統一介面）
|
||
|
||
Args:
|
||
product_name: 商品名稱
|
||
provider: 指定提供者
|
||
model: 指定模型
|
||
include_competitors: 是否包含競品分析
|
||
include_trends: 是否包含趨勢分析
|
||
web_context: 網路搜尋結果（用於 Ollama）
|
||
|
||
Returns:
|
||
AIResponse
|
||
"""
|
||
provider = provider or self._default_provider
|
||
|
||
if provider == 'gemini':
|
||
response = self._gemini.search_product_insights(
|
||
product_name=product_name,
|
||
include_competitors=include_competitors,
|
||
include_trends=include_trends,
|
||
model=model
|
||
)
|
||
else: # ollama
|
||
response = self._ollama.search_product_insights(
|
||
product_name=product_name,
|
||
include_competitors=include_competitors,
|
||
include_trends=include_trends,
|
||
web_context=web_context
|
||
)
|
||
|
||
return self._convert_response(response, provider)
|
||
|
||
def web_search(self, query: str, provider: str = None, model: str = None,
|
||
num_results: int = 5, search_type: str = "general") -> AIResponse:
|
||
"""
|
||
網路搜尋（統一介面）- 僅 Ollama 支援
|
||
|
||
Args:
|
||
query: 搜尋關鍵字
|
||
provider: 指定提供者（會忽略 gemini）
|
||
model: 指定模型
|
||
num_results: 返回結果數量
|
||
search_type: 搜尋類型
|
||
|
||
Returns:
|
||
AIResponse
|
||
"""
|
||
# 網路搜尋目前僅 Ollama 支援
|
||
response = self._ollama.web_search(query, num_results, search_type)
|
||
return self._convert_response(response, 'ollama')
|
||
|
||
def search_trend_keywords(self, category: str, provider: str = None,
|
||
model: str = None, time_range: str = "week") -> AIResponse:
|
||
"""
|
||
搜尋趨勢關鍵字（統一介面）- 僅 Ollama 支援
|
||
|
||
Args:
|
||
category: 商品分類
|
||
provider: 指定提供者（會忽略 gemini）
|
||
model: 指定模型
|
||
time_range: 時間範圍
|
||
|
||
Returns:
|
||
AIResponse
|
||
"""
|
||
# 趨勢關鍵字目前僅 Ollama 支援
|
||
response = self._ollama.search_trend_keywords(category, time_range)
|
||
return self._convert_response(response, 'ollama')
|
||
|
||
|
||
# 建立全域服務實例
|
||
ai_provider_service = AIProviderService()
|
||
|
||
|
||
# 便捷函數
|
||
def get_ai_status(force_refresh: bool = False) -> Dict[str, Any]:
|
||
"""取得 AI 服務狀態"""
|
||
return ai_provider_service.get_status(force_refresh)
|
||
|
||
|
||
def set_ai_provider(provider: str) -> bool:
|
||
"""設定預設 AI 提供者"""
|
||
try:
|
||
ai_provider_service.default_provider = provider
|
||
return True
|
||
except ValueError:
|
||
return False
|
||
|
||
|
||
def generate_copy(product_name: str, provider: str = None, model: str = None,
|
||
**kwargs) -> AIResponse:
|
||
"""生成文案（便捷函數）"""
|
||
return ai_provider_service.generate_sales_copy(
|
||
product_name=product_name,
|
||
provider=provider,
|
||
model=model,
|
||
**kwargs
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# 測試程式碼
|
||
logging.basicConfig(level=logging.INFO)
|
||
|
||
service = AIProviderService()
|
||
|
||
# 檢查狀態
|
||
print("檢查 AI 服務狀態...")
|
||
status = service.get_status()
|
||
print(f"Ollama: {'✅ 已連線' if status['ollama']['connected'] else '❌ 未連線'}")
|
||
print(f"Gemini: {'✅ 已連線' if status['gemini']['connected'] else '❌ 未連線'}")
|
||
print(f"預設提供者: {status['default_provider']}")
|
||
print(f"推薦提供者: {status['recommended_provider']}")
|
||
|
||
# 測試文案生成
|
||
if status['recommended_provider'] != 'none':
|
||
print(f"\n使用 {status['recommended_provider']} 測試文案生成...")
|
||
result = service.generate_sales_copy(
|
||
product_name="玻尿酸保濕面膜",
|
||
provider=status['recommended_provider'],
|
||
trend_keywords=["換季保養", "敏感肌"],
|
||
style="吸睛"
|
||
)
|
||
|
||
if result.success:
|
||
print(f"\n生成結果:\n{result.content}")
|
||
print(f"\n提供者: {result.provider}")
|
||
print(f"模型: {result.model}")
|
||
print(f"耗時: {result.total_duration:.2f} 秒")
|
||
if result.total_cost > 0:
|
||
print(f"費用: ${result.total_cost:.6f} USD")
|
||
else:
|
||
print(f"生成失敗: {result.error}")
|