fix: Gemini 模型 fallback 链 — 修复全量降级 NVIDIA

根因: 免费层配额 GenerateRequestsPerDayPerProjectPerModel 每天每模型仅 20 请求，日均 30+ 任务耗尽后持续 429，全部降级 NVIDIA (最近40任务 35 nvidia / 4 gemini)。 - gemini_adapter 抽 _generate() 模型链调用: 429 立即切换下一模型 (flash -> flash-lite，各自独立 20/天配额)，503 退避 3s 重试一次再切换；视觉分析与问答统一走该链 - config: gemini 增 fallback_models，nvidia 恢复 nemotron-omni (原生视频输入；此前 Oracle 线上手动改过未回传仓库，被旧配置覆盖回 llama-3.2 逐帧模式) + timeout 30->120 - 健康检查校验整条模型链实测 task 73: flash 429 -> lite 1s 内接管，12 帧直出，event_52 落库 provider=gemini。
2026-08-21 00:07:34 +08:00
parent 7efbee89fb
commit eaf4ef3bd3
2 changed files with 88 additions and 61 deletions
--- a/fam-edge/config/config.yaml
+++ b/fam-edge/config/config.yaml
@@ -68,7 +68,9 @@ models:
  - provider: "gemini"
    role: "vision"
    enabled: true
-    model_name: "gemini-flash-latest"   # v1beta 下 gemini-1.5-flash 会 404
+    model_name: "gemini-flash-latest"   # 主模型（每日免费配额 20 请求，按模型独立）
+    fallback_models:                     # 429 配额耗尽/503 过载时依次切换
+      - "gemini-flash-lite-latest"
    api_key: "${GEMINI_API_KEY}"
    timeout: 90
    circuit_breaker:
@@ -79,10 +81,10 @@ models:
  - provider: "nvidia"
    role: "vision"
    enabled: true
-    model_name: "meta/llama-3.2-11b-vision-instruct"
+    model_name: "nvidia/nemotron-3-nano-omni-30b-a3b-reasoning"   # Omni 原生视频输入；llama-3.2-11b-vision 仅逐帧
    base_url: "https://integrate.api.nvidia.com/v1"
    api_key: "${NVIDIA_API_KEY}"
-    timeout: 30
+    timeout: 120
    circuit_breaker:
      enabled: true
      threshold: 5
--- a/fam-edge/src/fam_edge/model_adapters/gemini_adapter.py
+++ b/fam-edge/src/fam_edge/model_adapters/gemini_adapter.py
@@ -9,6 +9,7 @@ provider_name = "gemini"
 视觉分析: 多图单请求直出结构化 JSON（global_summary/entities_json/frame_details）
 """
 import os
+import time
 import base64
 import requests
 from typing import Dict, List, Optional
@@ -27,6 +28,9 @@ class GeminiAdapter(BaseModelAdapter):
    def __init__(self, config: dict):
        super().__init__("gemini", config)
        self.model_name = config.get('model_name', 'gemini-flash-latest')
+        # 免费层配额按模型独立（20 请求/天/模型），fallback 链用于跨模型借用配额
+        self.model_chain = [self.model_name] + [
+            m for m in config.get('fallback_models', []) if m and m != self.model_name]
        self.api_key = self._resolve_key(config.get('api_key', ''))
        self.timeout = config.get('timeout', 30)
        cb_cfg = config.get('circuit_breaker', {})
@@ -52,11 +56,12 @@ class GeminiAdapter(BaseModelAdapter):
            if resp.status_code == 200:
                models = resp.json().get('models', [])
                names = [m.get('name', '') for m in models]
-                if any(self.model_name in n for n in names):
-                    logger.info(f"Gemini 健康检查通过: {self.model_name}")
+                missing = [m for m in self.model_chain if not any(m in n for n in names)]
+                if not missing:
+                    logger.info(f"Gemini 健康检查通过: {' -> '.join(self.model_chain)}")
                    return True
-                logger.warning(f"Gemini 模型未找到: {self.model_name}; 可用: {names[:5]}")
-                return False
+                logger.warning(f"Gemini 模型未找到: {missing}; 可用: {names[:5]}")
+                return len(missing) < len(self.model_chain)
            logger.warning(f"Gemini 健康检查 HTTP {resp.status_code}")
            return False
        except Exception as e:
@@ -98,21 +103,8 @@ class GeminiAdapter(BaseModelAdapter):
        parts.insert(0, {"text": self._build_structured_prompt(known_members_context)})

        try:
-            resp = requests.post(
-                f"{self._base_url}/models/{self.model_name}:generateContent?key={self.api_key}",
-                json={"contents": [{"parts": parts}],
-                      "generationConfig": {"temperature": 0.2, "maxOutputTokens": 2048}},
-                timeout=self.timeout
-            )
-            if resp.status_code == 200:
-                cands = resp.json().get('candidates', [])
-                if cands:
-                    text = ''.join(
-                        p.get('text', '')
-                        for p in cands[0].get('content', {}).get('parts', [])
-                    ).strip()
-                    if not text:
-                        logger.warning("Gemini 返回空文本")
+            text = self._generate(parts, max_tokens=2048, temperature=0.2)
+            if text is None:
                self._cb.record_failure()
                return None
            try:
@@ -132,9 +124,6 @@ class GeminiAdapter(BaseModelAdapter):
                logger.error(f"Gemini 输出无法解析为 JSON: {e}")
                self._cb.record_failure()
                return None
-            else:
-                logger.warning(f"Gemini 视觉分析 HTTP {resp.status_code}: {resp.text[:150]}")
-                self._cb.record_failure()
        except requests.Timeout:
            logger.warning(f"Gemini 视觉分析超时 ({self.timeout}s)")
            self._cb.record_failure()
@@ -143,6 +132,58 @@ class GeminiAdapter(BaseModelAdapter):
            self._cb.record_failure()
        return None

+    def _generate(self, parts: List[dict], max_tokens: int,
+                  temperature: float) -> Optional[str]:
+        """带模型 fallback 链的 generateContent 调用
+
+        - 429（每日免费配额耗尽，按模型独立）→ 立即换下一个模型，不重试
+        - 503（模型过载，临时性）→ 同模型退避 3s 重试一次，仍失败换下一个
+        """
+        for model in self.model_chain:
+            for attempt in range(2):
+                try:
+                    resp = requests.post(
+                        f"{self._base_url}/models/{model}:generateContent?key={self.api_key}",
+                        json={"contents": [{"parts": parts}],
+                              "generationConfig": {
+                                  "temperature": temperature, "maxOutputTokens": max_tokens}},
+                        timeout=self.timeout
+                    )
+                except requests.Timeout:
+                    logger.warning(f"Gemini [{model}] 请求超时 ({self.timeout}s)")
+                    break
+                except Exception as e:
+                    logger.error(f"Gemini [{model}] 请求异常: {e}")
+                    break
+
+                if resp.status_code == 200:
+                    cands = resp.json().get('candidates', [])
+                    text = ''.join(
+                        p.get('text', '')
+                        for p in (cands[0].get('content', {}) if cands else {}).get('parts', [])
+                    ).strip() if cands else ''
+                    if text:
+                        if model != self.model_name:
+                            logger.info(f"Gemini 主模型不可用，由 fallback 模型 [{model}] 出结果")
+                        return text
+                    logger.warning(f"Gemini [{model}] 返回空文本")
+                    continue
+
+                detail = resp.text[:150].replace('\n', ' ')
+                if resp.status_code == 429:
+                    logger.warning(f"Gemini [{model}] 429 每日免费配额耗尽，切换下一模型")
+                    break
+                if resp.status_code == 503:
+                    if attempt == 0:
+                        logger.warning(f"Gemini [{model}] 503 过载，3s 后重试")
+                        time.sleep(3)
+                        continue
+                    logger.warning(f"Gemini [{model}] 503 重试仍失败，切换下一模型")
+                    break
+                logger.warning(f"Gemini [{model}] HTTP {resp.status_code}: {detail}")
+                break
+        return None
+
    def _build_structured_prompt(self, known_members: str) -> str:
        return f"""你是家庭监控视频分析助手。下面按时间顺序排列了多张监控截图。
 请分析整个时段，只输出合法 JSON（不要 markdown、不要任何解释文字），结构如下：
@@ -181,23 +222,7 @@ class GeminiAdapter(BaseModelAdapter):
            logger.warning("Gemini API Key 未配置，跳过问答")
            return None
        try:
-            resp = requests.post(
-                f"{self._base_url}/models/{self.model_name}:generateContent?key={self.api_key}",
-                json={"contents": [{"parts": [{"text": prompt}]}],
-                      "generationConfig": {"temperature": 0.3, "maxOutputTokens": max_tokens}},
-                timeout=self.timeout
-            )
-            if resp.status_code == 200:
-                cands = resp.json().get('candidates', [])
-                if cands:
-                    text = ''.join(
-                        p.get('text', '')
-                        for p in cands[0].get('content', {}).get('parts', [])
-                    ).strip()
-                    return text or None
-            logger.warning(f"Gemini 问答 HTTP {resp.status_code}")
-        except requests.Timeout:
-            logger.warning(f"Gemini 问答超时 ({self.timeout}s)")
+            return self._generate([{"text": prompt}], max_tokens=max_tokens, temperature=0.3)
        except Exception as e:
            logger.error(f"Gemini 问答异常: {e}")
            return None