[架构重构] 移除本地Ollama融合，云端直出JSON直存DB，Q&A三模型降级

1. 视频摘要链路：云端VLM直出结构化JSON → Edge format_cloud_result格式化校验 → 直存NAS DB（移除run_text_fusion本地融合） 2. 智能问答链路：Gemini→NVIDIA→Ollama降级，新增chat()纯文本问答方法 3. 适配器重构：base/gemini/nvidia/ollama adapter新增chat()；gemini多图单请求结构化JSON；nvidia逐帧调用聚合 4. 端点变更：/api/edge/chat → /api/edge/chat/ask，调orchestrator.run_qa() 5. chat_handler改经Edge Q&A编排，不再直连Ollama 6. 配置更新：ollama_url → qa_url，Ollama role注释改为Q&A兜底 7. README同步更新架构描述、拓扑图、时序图、模块表
2026-08-20 10:21:09 +08:00
parent 486eee4feb
commit babf5b09a9
10 changed files with 503 additions and 267 deletions
--- a/fam-edge/src/fam_edge/ai_orchestrator/orchestrator.py
+++ b/fam-edge/src/fam_edge/ai_orchestrator/orchestrator.py
@@ -1,14 +1,16 @@
 """
-AI-Orchestrator - 多模型并行编排
+AI-Orchestrator - 多模型编排

-流程:
+视频分析链路（新框架）:
 1. 加载所有启用的模型适配器
 2. 健康检查
 3. 抽帧 + 关键帧筛选 + 压缩
-4. 并行调用所有健康模型（ThreadPoolExecutor）
-5. 文本融合（多模型输出平等交叉验证）
-6. 回调 NAS
-7. 清理临时文件
+4. 云端 VLM 视觉分析（Gemini 主 / NVIDIA 兜底），直出结构化 JSON
+5. format_cloud_result：对云端结果做**格式化/校验**（无本地模型调用，不汇总摘要）
+6. 同步返回 NAS → 落库
+
+智能问答链路（新框架）:
+- run_qa：Gemini → NVIDIA → 本地 Ollama（仅当两云端都失败才用本地兜底）
 """
 import time
 import json
@@ -16,63 +18,17 @@ import base64
 import requests
 from datetime import datetime, timedelta
 from concurrent.futures import ThreadPoolExecutor, as_completed, TimeoutError as FuturesTimeout
-from typing import Dict, List, Optional
+from typing import Dict, List, Optional, Tuple

 from ..logger import setup_logger, log_task
 from ..config_loader import load_config
 from ..model_adapters.adapter_factory import build_adapters
 from ..model_adapters.base_adapter import BaseModelAdapter
 from ..video_preprocessor.preprocessor import VideoPreprocessor
-from .json_parser import parse_vlm_json, VLMOutputInvalidError
+from .json_parser import VLMOutputInvalidError, validate_schema

 logger = setup_logger('fam-edge.orchestrator')

-FUSION_SYSTEM_PROMPT = """你是一个无情的数据提取器。不要输出任何思考过程，只输出合法 JSON。
-
-输入参数:
- 多模型视觉分析日志（每个模型独立输出，平等对待，交叉验证）:
-{model_outputs}
- 已知成员清单: {known_members}
-
-执行规则:
-1. 多个模型的输出平等对待，交叉验证：
-   - 多个模型一致描述的内容 → 可信度高，必须纳入 frame_details，source_providers 列出所有一致的模型
-   - 仅单一模型描述的内容 → 纳入 frame_details，source_providers 仅含该模型
-   - 多个模型冲突时（如人物动作描述不一致）→ 以多数模型一致为准，source_providers 列出多数派模型
-2. 画面人物按特征匹配已知成员清单：
-   - 匹配到已命名成员（real_name 非空）→ person 字段填 real_name
-   - 匹配到未命名成员（real_name 为空）→ person 字段填 abstract_label
-   - 都不匹配 → 按出现顺序赋予新标识"人物B"、"人物C"...
-3. 提取每张关键帧对应的时间点、人物、动作、衣着，输出到 frame_details 数组。
-4. frame_details 每条必须包含 source_providers 数组。
-5. compute_provider 字段填入本次实际成功调用的所有模型标识数组（去重）。
-6. 仅输出合法 JSON，不输出任何思考过程、markdown 标记或注释。
-
-输出 JSON 结构:
-{{
-  "global_summary": "字符串，整个时段的整体摘要，简体中文",
-  "entities_json": [
-    {{
-      "person": "字符串",
-      "action": "字符串",
-      "clothing": "字符串"
-    }}
-  ],
-  "frame_details": [
-    {{
-      "frame_index": "数字",
-      "frame_timestamp": "字符串，ISO 8601 格式时间戳",
-      "person": "字符串",
-      "action": "字符串",
-      "clothing": "字符串",
-      "is_attention_event": "布尔值",
-      "source_providers": "数组"
-    }}
-  ],
-  "compute_provider": "数组"
-}}
-"""
-

 class AIOrchestrator:
    """AI 编排器"""
@@ -96,7 +52,7 @@ class AIOrchestrator:
    def run_visual_analysis(self, adapters: List[BaseModelAdapter],
                           frame_paths: List[str],
                           frame_timestamps: List[str],
-                           known_members_context: str) -> Dict[str, str]:
+                           known_members_context: str) -> Dict[str, dict]:
        """视觉分析阶段：仅 role=vision 的适配器参与

        orchestrator.mode:
@@ -141,7 +97,7 @@ class AIOrchestrator:
        return model_outputs

    def _run_visual_ensemble(self, vision_adapters, frame_paths,
-                             frame_timestamps, known_members_context) -> Dict[str, str]:
+                             frame_timestamps, known_members_context) -> Dict[str, dict]:
        """并行调用所有健康 vision 模型，保留全部成功结果（交叉验证）"""
        model_outputs = {}
        max_timeout = max((a.get_timeout() for a in vision_adapters), default=240)
@@ -180,68 +136,115 @@ class AIOrchestrator:
                    adapter.get_circuit_breaker().record_failure()
        return model_outputs

-    def run_text_fusion(self, model_outputs: Dict[str, str],
-                         known_members_context: str,
-                         task_id: int) -> dict:
-        """文本融合阶段 - 多模型输出平等交叉验证"""
-        # 构建 model_outputs 文本
-        outputs_text = '\n'.join(
-            f"  - {provider} 输出: {output}" for provider, output in model_outputs.items()
-        )
+    def format_cloud_result(self, provider: str, raw_result: dict,
+                             known_members_context: str = '',
+                             task_id: int = 0) -> dict:
+        """格式化云端 VLM 直出的结构化结果（**无本地模型调用**）。

-        prompt = FUSION_SYSTEM_PROMPT.format(
-            model_outputs=outputs_text,
-            known_members=known_members_context or '（暂无已知成员）'
-        )
+        - 云端模型已产出结构化数据（frame_details / 可选 global_summary / entities_json）
+        - 本方法仅做：字段归一化、source_providers 与 compute_provider 填充、
+          entities 推导、global_summary 缺失时格式化生成
+        - 解析/校验失败抛 VLMOutputInvalidError
+        """
+        if not isinstance(raw_result, dict):
+            raise VLMOutputInvalidError("云端视觉模型未返回结构化数据(dict)")

-        # 调用文本角色模型（role=text，默认 ollama / qwen2.5:7b）做融合
-        text_cfg = next(
-            (cfg for cfg in self.config.get('models', []) if cfg.get('role') == 'text'), None
-        ) or next(
-            (cfg for cfg in self.config.get('models', []) if cfg.get('provider') == 'ollama'), None
-        )
-        if not text_cfg:
-            raise VLMOutputInvalidError("没有文本角色模型配置，无法执行文本融合")
+        data = dict(raw_result)
+        frame_details = data.get('frame_details')
+        if not isinstance(frame_details, list) or not frame_details:
+            raise VLMOutputInvalidError("云端结果缺少非空的 frame_details")

-        base_url = text_cfg.get('base_url', 'http://localhost:11434')
-        model_name = text_cfg.get('model_name', 'qwen2.5:7b')
-        fusion_timeout = self.timeout_cfg.get('vlm_fusion', 300)
-        num_predict = text_cfg.get('num_predict', 1024)
+        # 归一化每条 frame_detail
+        normalized = []
+        for f in frame_details:
+            if not isinstance(f, dict):
+                continue
+            sp = f.get('source_providers')
+            if not isinstance(sp, list) or not sp:
+                sp = [provider]
+            normalized.append({
+                "frame_index": int(f.get("frame_index", len(normalized) + 1)),
+                "frame_timestamp": str(f.get("frame_timestamp", "")),
+                "person": str(f.get("person", "无人")),
+                "action": str(f.get("action", "")),
+                "clothing": str(f.get("clothing", "")),
+                "is_attention_event": bool(f.get("is_attention_event", False)),
+                "source_providers": [str(p) for p in sp],
+            })
+        if not normalized:
+            raise VLMOutputInvalidError("frame_details 解析后为空")
+        data['frame_details'] = normalized

-        start = time.time()
-        resp = requests.post(
-            f"{base_url}/api/generate",
-            json={
-                "model": model_name,
-                "prompt": prompt,
-                "stream": False,
-                "format": "json",
-                "options": {"temperature": 0.0, "num_predict": num_predict}
-            },
-            timeout=fusion_timeout
-        )
+        # compute_provider：本次实际成功的云端模型
+        data['compute_provider'] = [provider]

-        if resp.status_code != 200:
-            raise VLMOutputInvalidError(f"融合阶段 Ollama 调用失败: {resp.status_code}")
+        # entities_json：缺失时由 frame_details 推导（按人物去重）
+        if not data.get('entities_json'):
+            seen = set()
+            ents = []
+            for f in normalized:
+                p = f['person']
+                if p and p != '无人' and p not in seen:
+                    seen.add(p)
+                    ents.append({
+                        "person": p,
+                        "action": f['action'],
+                        "clothing": f['clothing'],
+                    })
+            data['entities_json'] = ents

-        raw_output = resp.json().get('response', '')
-        duration_ms = int((time.time() - start) * 1000)
-        log_task(logger, task_id, 'vlm_fusion', f'融合完成，原始输出长度={len(raw_output)}', duration_ms=duration_ms)
+        # global_summary：云端未给则格式化生成（非 LLM 汇总，仅拼接事实）
+        if not data.get('global_summary'):
+            data['global_summary'] = self._build_summary_from_frames(normalized)

-        # 解析 JSON（三层容错）
-        result = parse_vlm_json(raw_output)
+        return validate_schema(data)

-        # 确保 compute_provider 与实际调用的模型一致
-        result['compute_provider'] = list(model_outputs.keys())
+    def _build_summary_from_frames(self, frame_details: List[dict]) -> str:
+        """当云端模型未提供 global_summary 时，由 frame_details 格式化生成摘要。
+        注意：这是确定性事实拼接，非 LLM 二次汇总。"""
+        persons = {}
+        has_attention = False
+        for f in frame_details:
+            p = f['person']
+            if p and p != '无人':
+                persons.setdefault(p, set()).add(f['action'])
+            if f.get('is_attention_event'):
+                has_attention = True
+        if not persons:
+            summary = "整个时段内画面中未检测到人物出现，主要为环境静态画面。"
+        else:
+            parts = []
+            for p, acts in persons.items():
+                acts_desc = "、".join(sorted(a for a in acts if a)) or "无明显动作"
+                parts.append(f"{p}（{acts_desc}）")
+            summary = f"时段内检测到：{'；'.join(parts)}。"
+        if has_attention:
+            summary += " ⚠️ 存在需关注的异常事件。"
+        return summary

-        # 确保 frame_details 的 source_providers 只包含实际成功的模型
-        valid_providers = set(model_outputs.keys())
-        for frame in result.get('frame_details', []):
-            frame['source_providers'] = [
-                p for p in frame.get('source_providers', []) if p in valid_providers
-            ] or list(valid_providers)
+    def run_qa(self, prompt: str, max_tokens: int = 512) -> Tuple[Optional[str], Optional[str]]:
+        """智能问答编排：Gemini → NVIDIA → 本地 Ollama（仅当两云端都失败才用本地兜底）。

-        return result
+        返回 (answer, provider)；全部失败返回 (None, None)。
+        """
+        qa_order = ['gemini', 'nvidia', 'ollama']
+        for name in qa_order:
+            adapter = next((a for a in self.adapters if a.provider_name == name), None)
+            if adapter is None:
+                logger.warning(f"[qa] 未配置模型 {name}，跳过")
+                continue
+            try:
+                if adapter.get_circuit_breaker().is_open():
+                    logger.warning(f"[qa] {name} 熔断器 OPEN，跳过")
+                    continue
+                answer = adapter.chat(prompt, max_tokens=max_tokens)
+                if answer:
+                    logger.info(f"[qa] 由 {name} 回答（长度={len(answer)}）")
+                    return answer, name
+                logger.warning(f"[qa] {name} 返回空")
+            except Exception as e:
+                logger.error(f"[qa] {name} 调用异常: {e}")
+        return None, None

    def send_callback(self, webhook_url: str, task_id: int,
                      result: dict, camera_name: str = '',
@@ -340,8 +343,10 @@ class AIOrchestrator:
            if not model_outputs:
                raise Exception('All models failed in visual analysis')

-            # 4. 文本融合
-            fusion_result = self.run_text_fusion(model_outputs, known_members, task_id)
+            # 4. 云端直出结果格式化（无本地融合）
+            provider = next(iter(model_outputs))
+            fusion_result = self.format_cloud_result(
+                provider, model_outputs[provider], known_members, task_id)

            # 5. 回调
            # 从视频文件名推断 camera_name
@@ -359,7 +364,7 @@ class AIOrchestrator:
            log_task(logger, task_id, 'overall', f'任务完成', duration_ms=total_ms)

        except VLMOutputInvalidError as e:
-            logger.error(f"[task_id={task_id}] VLM 输出解析失败: {e}")
+            logger.error(f"[task_id={task_id}] 云端结果格式化失败: {e}")
            self.send_failure_callback(webhook_url, task_id, 'vlm_fusion', str(e))
        except Exception as e:
            logger.error(f"[task_id={task_id}] 任务处理失败: {e}", exc_info=True)
@@ -429,8 +434,10 @@ class AIOrchestrator:
            if not model_outputs:
                raise Exception('All models failed in visual analysis')

-            # 4. 文本融合
-            fusion_result = self.run_text_fusion(model_outputs, known_members, task_id)
+            # 4. 云端直出结果格式化（无本地融合）
+            provider = next(iter(model_outputs))
+            fusion_result = self.format_cloud_result(
+                provider, model_outputs[provider], known_members, task_id)

            total_ms = int((time.time() - start_time) * 1000)
            log_task(logger, task_id, 'overall', '推送任务完成', duration_ms=total_ms)