wooo/ewoooc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

守住 AI provider Ollama-first
All checks were successful
CD Pipeline / deploy (push) Successful in 56s
Details

Browse Source
This commit is contained in:
OoO
2026-05-13 12:07:33 +08:00
parent 0c9f9278f1
commit 2130c4f54b
4 changed files with 157 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
routes/ai_routes.py
View File

@@ -3,7 +3,7 @@
"""
AI 推薦路由模組
提供時事熱點商品推薦與文案生成功能
支援 Ollama 和 Gemini 雙 AI 提供者
支援 Ollama-first AI 生成，Gemini 僅作 Ollama 失敗備援
"""
from flask import Blueprint, render_template, request, jsonify, session
@@ -123,8 +123,13 @@ def api_set_provider():
data = request.get_json()
provider = data.get('provider', 'ollama')
if provider not in ('ollama', 'gemini'):
return jsonify({'success': False, 'error': '無效的提供者，請使用 ollama 或 gemini'}), 400
if provider == 'gemini':
return jsonify({
'success': False,
'error': 'Gemini 僅可作為 Ollama 失敗備援，不可設為預設提供者'
}), 400
if provider not in ('ollama',):
return jsonify({'success': False, 'error': '無效的提供者，請使用 ollama'}), 400
success = set_ai_provider(provider)
if success:
@@ -361,14 +366,14 @@ def api_get_weather():
@ai_bp.route('/api/ai/generate_copy', methods=['POST'])
@login_required
def api_generate_copy():
"""生成銷售文案（支援 Ollama/Gemini 雙提供者）"""
"""生成銷售文案（Ollama-first；Gemini 僅作失敗備援）"""
try:
data = request.get_json()
product_name = data.get('product_name', '')
trend_keywords = data.get('trend_keywords', [])
style = data.get('style', '吸睛')
model = data.get('model', None)
provider = data.get('provider', None) # 'ollama' 或 'gemini'
provider = data.get('provider', None) # gemini 會被視為 fallback-only，不會直接主呼叫
save_to_history = data.get('save_to_history', True)
# 額外的上下文資訊

2
routes/bot_api_routes.py
View File

@@ -483,7 +483,7 @@ def bot_generate_copy():
- product_name: 商品名稱 (必填)
- trend_keywords: 趨勢關鍵字 (選填, 陣列)
- style: 文案風格 (選填, 預設 '吸睛')
- provider: AI 提供者 (選填, 'ollama' 或 'gemini')
- provider: AI 提供者 (選填；一律 Ollama-first，Gemini 僅作失敗備援)
Returns:
- 生成的銷售文案

161
services/ai_provider.py
View File

@@ -2,7 +2,7 @@
# -*- coding: utf-8 -*-
"""
AI 提供者抽象層
統一 Ollama 和 Gemini 的介面，支援動態切換 AI 提供者
統一 Ollama 和 Gemini 的介面；通用生成一律 Ollama-first，Gemini 僅作備援
"""
import os
@@ -13,8 +13,8 @@ from datetime import date, datetime
logger = logging.getLogger(__name__)
# AI 提供者設定
AI_PROVIDER = os.getenv('AI_PROVIDER', 'ollama') # 預設使用 Ollama
# AI 提供者設定：Gemini 不可作為預設，只能在 Ollama 失敗後備援
AI_PROVIDER = os.getenv('AI_PROVIDER', 'ollama')
# 引入服務
from .ollama_service import OllamaService, OllamaResponse
@@ -60,7 +60,8 @@ class AIProviderService:
"""
AI 提供者服務 - 統一的 AI 介面
支援動態切換 Ollama 和 Gemini，並提供統一的生成介面
通用文案 / 關鍵字 / 洞察生成一律先走 Ollama 三主機級聯。
Gemini 只在 Ollama 主路徑失敗時作為 fallback，不允許被設為預設主路徑。
"""
def __init__(self, default_provider: str = None):
@@ -68,9 +69,9 @@ class AIProviderService:
初始化 AI 提供者服務
Args:
default_provider: 預設提供者 ('ollama' 或 'gemini')
default_provider: 預設提供者（Gemini 會被降回 Ollama）
"""
self._default_provider = default_provider or AI_PROVIDER
self._default_provider = self._sanitize_default_provider(default_provider or AI_PROVIDER)
self._ollama = OllamaService()
self._gemini = GeminiService()
self._elephant = ElephantService()
@@ -87,11 +88,20 @@ class AIProviderService:
@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
self._default_provider = self._sanitize_default_provider(value)
logger.info(f"AI 預設提供者已切換至: {value}")
@staticmethod
def _sanitize_default_provider(value: str) -> str:
"""Gemini 是 fallback-only；任何預設 Gemini 設定都強制回 Ollama。"""
normalized = (value or 'ollama').strip().lower()
if normalized == 'gemini':
logger.warning("AI_PROVIDER=gemini 已被拒絕；Gemini 僅作 Ollama 失敗備援")
return 'ollama'
if normalized not in ('ollama', 'elephant'):
raise ValueError("Provider must be 'ollama' or 'elephant'; gemini is fallback-only")
return normalized
def get_status(self, force_refresh: bool = False) -> Dict[str, Any]:
"""
取得所有 AI 服務的狀態
@@ -148,18 +158,15 @@ class AIProviderService:
def _get_recommended_provider(self, ollama_ok: bool, gemini_ok: bool, elephant_ok: bool) -> str:
"""根據可用性推薦提供者"""
if ollama_ok:
return 'ollama'
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:
if gemini_ok:
return 'gemini'
else:
return 'none'
if elephant_ok:
return 'elephant'
return 'none'
def _convert_response(self, response: Union[OllamaResponse, GeminiResponse],
provider: str) -> AIResponse:
@@ -218,6 +225,21 @@ class AIProviderService:
error='Unknown response type'
)
def _gemini_fallback(self, ollama_result: AIResponse, fallback_call) -> AIResponse:
"""Ollama 失敗時才呼叫 Gemini，讓通用 AI 入口符合 Ollama-first。"""
if ollama_result.success:
return ollama_result
logger.warning("Ollama 主路徑失敗，啟用 Gemini 備援：%s", ollama_result.error)
try:
gemini_response = fallback_call()
gemini_result = self._convert_response(gemini_response, 'gemini')
if not gemini_result.success and ollama_result.error:
gemini_result.error = f"Ollama 失敗：{ollama_result.error}；Gemini 備援失敗：{gemini_result.error}"
return gemini_result
except Exception as exc:
ollama_result.error = f"{ollama_result.error}; Gemini 備援例外：{exc}"
return ollama_result
def generate(self, prompt: str, provider: str = None, model: str = None,
system_prompt: str = None, temperature: float = 0.7,
timeout: int = None) -> AIResponse:
@@ -226,7 +248,7 @@ class AIProviderService:
Args:
prompt: 使用者提示
provider: 指定提供者 ('ollama' 或 'gemini')
provider: 指定提供者；Gemini 只會在 Ollama 失敗後備援
model: 指定模型
system_prompt: 系統提示
temperature: 創意度
@@ -235,17 +257,9 @@ class AIProviderService:
Returns:
AIResponse
"""
provider = provider or self._default_provider
provider = (provider or self._default_provider or 'ollama').strip().lower()
if provider == 'gemini':
response = self._gemini.generate(
prompt=prompt,
model=model,
system_prompt=system_prompt,
temperature=temperature,
timeout=timeout
)
elif provider == 'elephant':
if provider == 'elephant':
response = self._elephant.generate(
prompt=prompt,
model=model,
@@ -253,16 +267,27 @@ class AIProviderService:
temperature=temperature,
timeout=timeout
)
else: # ollama
response = self._ollama.generate(
return self._convert_response(response, provider)
ollama_response = self._ollama.generate(
prompt=prompt,
model=model,
system_prompt=system_prompt,
temperature=temperature,
timeout=timeout
)
ollama_result = self._convert_response(ollama_response, 'ollama')
return self._gemini_fallback(
ollama_result,
lambda: self._gemini.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,
@@ -283,27 +308,27 @@ class AIProviderService:
Returns:
AIResponse
"""
provider = provider or self._default_provider
provider = (provider or self._default_provider or 'ollama').strip().lower()
if provider == 'gemini':
response = self._gemini.generate_sales_copy(
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
)
ollama_result = self._convert_response(ollama_response, 'ollama')
return self._gemini_fallback(
ollama_result,
lambda: 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:
@@ -319,14 +344,12 @@ class AIProviderService:
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)
ollama_response = self._ollama.extract_keywords(text, max_keywords)
ollama_result = self._convert_response(ollama_response, 'ollama')
return self._gemini_fallback(
ollama_result,
lambda: self._gemini.extract_keywords(text, max_keywords, model),
)
def search_product_insights(self, product_name: str, provider: str = None,
model: str = None, include_competitors: bool = True,
@@ -346,24 +369,22 @@ class AIProviderService:
Returns:
AIResponse
"""
provider = provider or self._default_provider
if provider == 'gemini':
response = self._gemini.search_product_insights(
ollama_response = self._ollama.search_product_insights(
product_name=product_name,
include_competitors=include_competitors,
include_trends=include_trends,
web_context=web_context
)
ollama_result = self._convert_response(ollama_response, 'ollama')
return self._gemini_fallback(
ollama_result,
lambda: 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:

55
tests/test_ai_provider_ollama_first.py Normal file
View File

@@ -0,0 +1,55 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""AIProviderService Ollama-first / Gemini fallback contract."""
from services.ai_provider import AIProviderService
from services.gemini_service import GeminiResponse
from services.ollama_service import OllamaResponse
def test_default_provider_gemini_is_forced_to_ollama():
service = AIProviderService(default_provider="gemini")
assert service.default_provider == "ollama"
def test_requested_gemini_still_uses_ollama_first(monkeypatch):
service = AIProviderService(default_provider="ollama")
monkeypatch.setattr(
service._ollama,
"generate_sales_copy",
lambda **_kw: OllamaResponse(True, "ollama copy", "llama3.1:8b"),
)
monkeypatch.setattr(
service._gemini,
"generate_sales_copy",
lambda **_kw: (_ for _ in ()).throw(AssertionError("Gemini 不應在 Ollama 成功時被呼叫")),
)
result = service.generate_sales_copy("測試商品", provider="gemini")
assert result.success is True
assert result.provider == "ollama"
assert result.content == "ollama copy"
def test_gemini_is_called_only_after_ollama_failure(monkeypatch):
service = AIProviderService(default_provider="ollama")
monkeypatch.setattr(
service._ollama,
"generate_sales_copy",
lambda **_kw: OllamaResponse(False, "", "llama3.1:8b", error="ollama down"),
)
monkeypatch.setattr(
service._gemini,
"generate_sales_copy",
lambda **_kw: GeminiResponse(True, "gemini fallback", "gemini-2.5-flash"),
)
result = service.generate_sales_copy("測試商品")
assert result.success is True
assert result.provider == "gemini"
assert result.content == "gemini fallback"