diff --git a/routes/ai_routes.py b/routes/ai_routes.py index 89ab092..7ed185f 100644 --- a/routes/ai_routes.py +++ b/routes/ai_routes.py @@ -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) # 額外的上下文資訊 diff --git a/routes/bot_api_routes.py b/routes/bot_api_routes.py index f03b72e..19da946 100644 --- a/routes/bot_api_routes.py +++ b/routes/bot_api_routes.py @@ -483,7 +483,7 @@ def bot_generate_copy(): - product_name: 商品名稱 (必填) - trend_keywords: 趨勢關鍵字 (選填, 陣列) - style: 文案風格 (選填, 預設 '吸睛') - - provider: AI 提供者 (選填, 'ollama' 或 'gemini') + - provider: AI 提供者 (選填;一律 Ollama-first,Gemini 僅作失敗備援) Returns: - 生成的銷售文案 diff --git a/services/ai_provider.py b/services/ai_provider.py index fb47805..2a767f2 100644 --- a/services/ai_provider.py +++ b/services/ai_provider.py @@ -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: diff --git a/tests/test_ai_provider_ollama_first.py b/tests/test_ai_provider_ollama_first.py new file mode 100644 index 0000000..5777ca4 --- /dev/null +++ b/tests/test_ai_provider_ollama_first.py @@ -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"