d5c0feab5e
All checks were successful
CD Pipeline / deploy (push) Successful in 1m35s
## Telegram Bot 功能修復 - 補全 16 個 await: 按鈕的 handler（日期選擇/目標設定/促銷追蹤等）， 新增 _handle_await_callback + _process_await_input 完整狀態機 - cmd: 按鈕加入 ⏳ 即時回饋 + try/except 防 BadRequest - handle_callback 加頂層 try/except 錯誤兜底 - 補 momo:cmd:suggestion + momo:menu:main callback handler - 修復 _enhanced_keyword_matching context NameError ## AI 模型遷移（hermes3@111 → qwen2.5@188） - hermes_analyst_service: URL 192.168.0.111→188, hermes3→qwen2.5:7b-instruct - code_review_pipeline: 改用 HERMES_URL/HERMES_MODEL 常數 - elephant_alpha_orchestrator / nemoton_dispatcher: registry/footprint 同步 - aider_heal_executor: OLLAMA_API_BASE fallback 改 188 - ai_routes: footprint display 字串改 qwen2.5:7b-instruct ## ElephantAlpha 404 修復 - elephant_service: openrouter→NVIDIA NIM, nvidia/llama-3.1-nemotron-ultra-253b-v1 - ai_provider: 模型 ID 同步更新 ## TELEGRAM_CHAT_ID 環境變數修正 - cicd_routes + aider_heal_executor: 優先讀 TELEGRAM_CHAT_IDS[0]， fallback TELEGRAM_CHAT_ID，修復通知靜默失敗 ## AI 對話 logging 改善 - telegram_ai_integration: Hermes 降級改 WARNING，OpenClaw 失敗加 exc_info - hermes_analyst_service: 連線失敗 log 加 host/model context ## DB Schema 修復 - migrations/019: action_plans 補齊全欄位，DROP NOT NULL action_type - autoheal_models: ActionPlan ORM 同步為超集 schema Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
469 lines
16 KiB
Python
469 lines
16 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
|
||
from .elephant_service import ElephantService, ElephantResponse
|
||
|
||
|
||
@dataclass
|
||
class AIResponse:
|
||
"""統一的 AI 回應結構"""
|
||
success: bool
|
||
content: str
|
||
model: str
|
||
provider: str # 'ollama', 'gemini' 或 'elephant'
|
||
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._elephant = ElephantService()
|
||
|
||
# 狀態快取
|
||
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', 'elephant'):
|
||
raise ValueError("Provider must be 'ollama', 'gemini' or 'elephant'")
|
||
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()
|
||
elephant_connected = self._elephant.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'
|
||
},
|
||
'elephant': {
|
||
'connected': elephant_connected,
|
||
'model': self._elephant.model if elephant_connected else None,
|
||
'available_models': [{'id': 'nvidia/llama-3.1-nemotron-ultra-253b-v1', 'name': 'Nemotron Ultra 253B'}],
|
||
'type': 'cloud',
|
||
'cost': 'efficient'
|
||
},
|
||
'recommended_provider': self._get_recommended_provider(ollama_connected, gemini_connected, elephant_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, elephant_ok: bool) -> str:
|
||
"""根據可用性推薦提供者"""
|
||
if self._default_provider == 'elephant' and elephant_ok:
|
||
return 'elephant'
|
||
if ollama_ok and gemini_ok:
|
||
return self._default_provider
|
||
elif elephant_ok:
|
||
return 'elephant'
|
||
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
|
||
)
|
||
elif isinstance(response, ElephantResponse):
|
||
return AIResponse(
|
||
success=response.success,
|
||
content=response.content,
|
||
model=response.model,
|
||
provider='elephant',
|
||
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
|
||
)
|
||
elif provider == 'elephant':
|
||
response = self._elephant.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}")
|