feat(orchestrator): 视觉 fallback 降级 + 文本融合角色化

- run_visual_analysis 仅 vision 角色参与, fallback 顺序降级(Gemini→NVIDIA NIM)首个成功即采用
- run_text_fusion 固定用 role=text 的 Ollama(qwen2.5:7b) 融合, 支持 num_predict
- config 改为多模型池(gemini/nvidia vision + ollama text)

fam-edge/config/config.yaml

@@ -1,4 +1,4 @@
-# FAM-Edge 配置文件 (Oracle 端) - 实际部署配置
+# FAM-Edge 配置文件 (Oracle 端) - 多模型池配置
 # Tailscale: Oracle=100.74.137.126, NAS=100.70.234.39
 
 # NAS 端回调地址
@@ -13,6 +13,11 @@ server:
   port: 5000
   max_concurrent_tasks: 1
 
+# 编排调度模式: fallback(顺序降级, 默认) | ensemble(并行交叉验证)
+orchestrator:
+  mode: "fallback"
+  overall_timeout: 600
+
 # 关键帧筛选参数（自适应：帧数随视频时长动态计算）
 video:
   candidate_per_minute: 2      # 每分钟粗抽候选帧数
@@ -27,8 +32,6 @@ video:
   max_long_edge: 1024
 
 # 超时（秒）
-# vlm_visual 实测: 1024px 帧视觉编码 ~36s/帧 + 生成 ~12s/60token (Oracle ARM CPU)
-# 30min 视频 12 帧 × ~50s ≈ 600s，超时需覆盖最坏情况
 timeout:
   download: 60
   vlm_visual: 600
@@ -36,26 +39,39 @@ timeout:
   callback: 30
   overall: 1800
 
-# 模型清单
+# 多模型池配置
+# 视觉分析: Gemini(主) -> NVIDIA NIM(备) 顺序降级; 全失败 -> 任务 FAILED 走重试
+# 文本融合/对话: 本地 Ollama qwen2.5:7b 专职 (不参与视觉)
 models:
-  - provider: "ollama"
-    enabled: true
-    model_name: "llava-phi3"
-    base_url: "http://localhost:11434"
-    timeout: 600
-    # num_predict 必须小: ARM CPU ~5 tok/s，500 会单帧跑数分钟触发超时
-    num_predict: 60
-    circuit_breaker:
-      enabled: false
-      threshold: 5
-      cooldown: 900
-
   - provider: "gemini"
-    enabled: false
+    role: "vision"
-    model_name: "gemini-1.5-flash"
+    enabled: true
-    api_key: ""
+    model_name: "gemini-flash-latest"   # v1beta 下 gemini-1.5-flash 会 404
-    timeout: 8
+    api_key: "${GEMINI_API_KEY}"
+    timeout: 15
     circuit_breaker:
       enabled: true
-      threshold: 5
+      threshold: 3
-      cooldown: 900
+      cooldown: 600
+
+  - provider: "nvidia"
+    role: "vision"
+    enabled: true
+    model_name: "meta/llama-3.2-11b-vision-instruct"
+    base_url: "https://integrate.api.nvidia.com/v1"
+    api_key: "${NVIDIA_API_KEY}"
+    timeout: 20
+    circuit_breaker:
+      enabled: true
+      threshold: 3
+      cooldown: 600
+
+  - provider: "ollama"
+    role: "text"
+    enabled: true
+    model_name: "qwen2.5:7b"
+    base_url: "http://localhost:11434"
+    timeout: 300
+    num_predict: 1024
+    circuit_breaker:
+      enabled: false

fam-edge/src/fam_edge/ai_orchestrator/orchestrator.py

@@ -97,45 +97,87 @@ class AIOrchestrator:
                            frame_paths: List[str],
                            frame_timestamps: List[str],
                            known_members_context: str) -> Dict[str, str]:
-        """并行调用所有健康模型进行视觉分析"""
+        """视觉分析阶段：仅 role=vision 的适配器参与
-        model_outputs = {}
-        max_timeout = max((a.get_timeout() for a in adapters), default=240)
 
-        with ThreadPoolExecutor(max_workers=len(adapters)) as pool:
+        orchestrator.mode:
+        - fallback (默认): 按 config 顺序依次尝试，首个成功即采用（单元素 dict）
+        - ensemble: 并行所有健康 vision 模型，全部成功结果都保留（交叉验证）
+        """
+        vision_adapters = [a for a in adapters if getattr(a, 'role', 'vision') == 'vision']
+        if not vision_adapters:
+            logger.error("没有 vision 角色的可用适配器")
+            return {}
+
+        mode = self.config.get('orchestrator', {}).get('mode', 'fallback')
+
+        if mode == 'ensemble':
+            return self._run_visual_ensemble(
+                vision_adapters, frame_paths, frame_timestamps, known_members_context)
+
+        # fallback: 顺序降级，首个成功即采用
+        model_outputs = {}
+        for adapter in vision_adapters:
+            if adapter.get_circuit_breaker().is_open():
+                logger.warning(f"[{adapter.provider_name}] 熔断器 OPEN，跳过")
+                continue
+            start = time.time()
+            try:
+                output = adapter.analyze_frames(
+                    frame_paths, frame_timestamps, known_members_context)
+                duration_ms = int((time.time() - start) * 1000)
+                if output:
+                    adapter.get_circuit_breaker().record_success()
+                    log_task(logger, 0, f'model_{adapter.provider_name}',
+                             f'视觉分析成功', duration_ms=duration_ms)
+                    model_outputs[adapter.provider_name] = output
+                    logger.info(f"fallback 采用 [{adapter.provider_name}]，停止降级")
+                    break
+                else:
+                    adapter.get_circuit_breaker().record_failure()
+                    logger.warning(f"[{adapter.provider_name}] 视觉分析返回空，降级下一模型")
+            except Exception as e:
+                logger.error(f"[{adapter.provider_name}] 视觉分析异常: {e}")
+                adapter.get_circuit_breaker().record_failure()
+        return model_outputs
+
+    def _run_visual_ensemble(self, vision_adapters, frame_paths,
+                             frame_timestamps, known_members_context) -> Dict[str, str]:
+        """并行调用所有健康 vision 模型，保留全部成功结果（交叉验证）"""
+        model_outputs = {}
+        max_timeout = max((a.get_timeout() for a in vision_adapters), default=240)
+        with ThreadPoolExecutor(max_workers=len(vision_adapters)) as pool:
             futures = {}
-            for adapter in adapters:
+            for adapter in vision_adapters:
                 if adapter.get_circuit_breaker().is_open():
                     logger.warning(f"[{adapter.provider_name}] 熔断器 OPEN，跳过")
                     continue
                 future = pool.submit(
                     adapter.analyze_frames,
-                    frame_paths, frame_timestamps, known_members_context
+                    frame_paths, frame_timestamps, known_members_context)
-                )
                 futures[future] = adapter.provider_name
-
             for future in as_completed(futures, timeout=max_timeout + 10):
                 provider = futures[future]
                 start = time.time()
                 try:
-                    adapter = next(a for a in adapters if a.provider_name == provider)
+                    adapter = next(a for a in vision_adapters if a.provider_name == provider)
                     output = future.result(timeout=adapter.get_timeout())
                     duration_ms = int((time.time() - start) * 1000)
                     if output:
                         model_outputs[provider] = output
                         adapter.get_circuit_breaker().record_success()
-                        log_task(logger, 0, f'model_{provider}', f'视觉分析成功，输出长度={len(output)}', duration_ms=duration_ms)
+                        log_task(logger, 0, f'model_{provider}',
+                                 f'视觉分析成功，输出长度={len(output)}', duration_ms=duration_ms)
                     else:
                         adapter.get_circuit_breaker().record_failure()
                         logger.warning(f"[{provider}] 视觉分析返回空")
                 except FuturesTimeout:
                     logger.warning(f"[{provider}] 视觉分析超时")
-                    adapter = next(a for a in adapters if a.provider_name == provider)
+                    adapter = next(a for a in vision_adapters if a.provider_name == provider)
                     adapter.get_circuit_breaker().record_failure()
                 except Exception as e:
                     logger.error(f"[{provider}] 视觉分析异常: {e}")
-                    adapter = next(a for a in adapters if a.provider_name == provider)
+                    adapter = next(a for a in vision_adapters if a.provider_name == provider)
                     adapter.get_circuit_breaker().record_failure()
-
         return model_outputs
 
     def run_text_fusion(self, model_outputs: Dict[str, str],
@@ -152,17 +194,19 @@ class AIOrchestrator:
             known_members=known_members_context or '（暂无已知成员）'
         )
 
-        # 调用 Ollama 纯文本模式
+        # 调用文本角色模型（role=text，默认 ollama / qwen2.5:7b）做融合
-        ollama_cfg = next(
+        text_cfg = next(
-            (cfg for cfg in self.config.get('models', []) if cfg.get('provider') == 'ollama'),
+            (cfg for cfg in self.config.get('models', []) if cfg.get('role') == 'text'), None
-            None
+        ) or next(
+            (cfg for cfg in self.config.get('models', []) if cfg.get('provider') == 'ollama'), None
         )
-        if not ollama_cfg:
+        if not text_cfg:
-            raise VLMOutputInvalidError("没有 Ollama 配置，无法执行文本融合")
+            raise VLMOutputInvalidError("没有文本角色模型配置，无法执行文本融合")
 
-        base_url = ollama_cfg.get('base_url', 'http://localhost:11434')
+        base_url = text_cfg.get('base_url', 'http://localhost:11434')
-        model_name = ollama_cfg.get('model_name', 'llava-phi3')
+        model_name = text_cfg.get('model_name', 'qwen2.5:7b')
-        fusion_timeout = self.timeout_cfg.get('vlm_fusion', 120)
+        fusion_timeout = self.timeout_cfg.get('vlm_fusion', 300)
+        num_predict = text_cfg.get('num_predict', 1024)
 
         start = time.time()
         resp = requests.post(
@@ -172,7 +216,7 @@ class AIOrchestrator:
                 "prompt": prompt,
                 "stream": False,
                 "format": "json",
-                "options": {"temperature": 0.0}
+                "options": {"temperature": 0.0, "num_predict": num_predict}
             },
             timeout=fusion_timeout
         )