feat(adapter): 云端视觉适配器 + role 角色区分

- base_adapter 增加 role 字段(vision/text)与 get_role()
- gemini_adapter 修复 v1beta 下模型名 404(gemini-1.5-flash→gemini-flash-latest), 改逐帧调用
- 新增 nvidia_adapter(openai SDK, 规避 NIM 单次限 1 图逐帧), 注册 adapter_factory
- 视觉分析仅 vision 角色参与, 文本融合交给 role=text 模型

fam-edge/src/fam_edge/model_adapters/adapter_factory.py

@@ -11,6 +11,7 @@ from typing import List
 from .base_adapter import BaseModelAdapter
 from .ollama_adapter import OllamaAdapter
 from .gemini_adapter import GeminiAdapter
+from .nvidia_adapter import NvidiaVisionAdapter
 from ..logger import setup_logger
 
 logger = setup_logger('fam-edge.adapter_factory')
@@ -18,9 +19,7 @@ logger = setup_logger('fam-edge.adapter_factory')
 _ADAPTER_REGISTRY = {
     "ollama": OllamaAdapter,
     "gemini": GeminiAdapter,
-    # v1.1 扩展:
+    "nvidia": NvidiaVisionAdapter,
-    # "openai": OpenAIAdapter,
-    # "nvidia": NvidiaAdapter,
 }
 
 

fam-edge/src/fam_edge/model_adapters/base_adapter.py

@@ -17,6 +17,12 @@ class BaseModelAdapter(ABC):
     def __init__(self, provider_name: str, config: dict):
         self.provider_name = provider_name  # 如 "ollama", "gemini"
         self.config = config
+        # 角色: vision=视觉分析, text=文本融合/对话; 默认 vision
+        self.role = config.get('role', 'vision')
+
+    def get_role(self) -> str:
+        """返回适配器角色: 'vision' 或 'text'"""
+        return self.role
 
     @abstractmethod
     def health_check(self) -> bool:

fam-edge/src/fam_edge/model_adapters/gemini_adapter.py

@@ -1,10 +1,12 @@
 """
-GeminiAdapter - Google Gemini 云端模型适配器
+GeminiAdapter - Google Gemini 云端 VLM 适配器
 
 provider_name = "gemini"
-模型: gemini-1.5-flash
+模型: gemini-flash-latest (v1beta 下 gemini-1.5-flash 会 404，用 flash-latest 别名)
-健康检查: GET models API
+角色: vision (视觉分析)
-熔断器: 启用，连续 5 次失败 -> OPEN 15 分钟
+健康检查: GET /v1beta/models?key=...
+熔断器: 启用
+逐帧分析: 与 NVIDIA 统一流程，逐帧调用（也规避多图返回不稳定）
 """
 import os
 import base64
@@ -19,41 +21,42 @@ logger = setup_logger('fam-edge.gemini_adapter')
 
 
 class GeminiAdapter(BaseModelAdapter):
-    """Gemini 云端 VLM 适配器"""
+    """Gemini 云端 VLM 适配器 (逐帧)"""
 
     def __init__(self, config: dict):
         super().__init__("gemini", config)
-        self.model_name = config.get('model_name', 'gemini-1.5-flash')
+        self.model_name = config.get('model_name', 'gemini-flash-latest')
-        self.api_key = config.get('api_key', '')
+        self.api_key = self._resolve_key(config.get('api_key', ''))
-        self.timeout = config.get('timeout', 8)
+        self.timeout = config.get('timeout', 15)
         cb_cfg = config.get('circuit_breaker', {})
         self._cb = CircuitBreaker(
-            threshold=cb_cfg.get('threshold', 5),
+            threshold=cb_cfg.get('threshold', 3),
-            cooldown=cb_cfg.get('cooldown', 900),
+            cooldown=cb_cfg.get('cooldown', 600),
-            enabled=cb_cfg.get('enabled', True)  # 云端默认启用
+            enabled=cb_cfg.get('enabled', True)
         )
         self._base_url = "https://generativelanguage.googleapis.com/v1beta"
 
+    def _resolve_key(self, raw: str) -> str:
+        if raw.startswith('${') and raw.endswith('}'):
+            return os.environ.get(raw[2:-1], '')
+        return raw
+
     def health_check(self) -> bool:
-        """GET models API，检查可用性"""
         if not self.api_key:
             logger.warning("Gemini API Key 未配置，健康检查失败")
             return False
         try:
             resp = requests.get(
-                f"{self._base_url}/models?key={self.api_key}",
+                f"{self._base_url}/models?key={self.api_key}", timeout=10)
-                timeout=10
-            )
             if resp.status_code == 200:
                 models = resp.json().get('models', [])
-                model_names = [m.get('name', '') for m in models]
+                names = [m.get('name', '') for m in models]
-                has_model = any(self.model_name in name for name in model_names)
+                if any(self.model_name in n for n in names):
-                if has_model:
+                    logger.info(f"Gemini 健康检查通过: {self.model_name}")
-                    logger.info(f"Gemini 健康检查通过: 模型 {self.model_name} 可用")
                     return True
-                else:
+                logger.warning(f"Gemini 模型未找到: {self.model_name}; 可用: {names[:5]}")
-                    logger.warning(f"Gemini 健康检查失败: 模型 {self.model_name} 未找到")
                 return False
+            logger.warning(f"Gemini 健康检查 HTTP {resp.status_code}")
             return False
         except Exception as e:
             logger.error(f"Gemini 健康检查异常: {e}")
@@ -62,68 +65,58 @@ class GeminiAdapter(BaseModelAdapter):
     def analyze_frames(self, frame_paths: List[str],
                        frame_timestamps: List[str],
                        known_members_context: str) -> Optional[str]:
-        """调用 Gemini 视觉分析"""
         if self._cb.is_open():
             logger.warning("Gemini 熔断器 OPEN，跳过调用")
             return None
-
         if not self.api_key:
             logger.warning("Gemini API Key 未配置，跳过调用")
             return None
 
-        # 构建 Prompt
+        results = []
-        n = len(frame_paths)
+        for path, ts in zip(frame_paths, frame_timestamps):
-        prompt = self._build_visual_prompt(n, frame_timestamps, known_members_context)
+            desc = self._analyze_one(path, ts, known_members_context)
+            if desc:
+                results.append(f"[帧] 时间: {ts}\n{desc}")
 
-        # 构建 inline_data
+        if not results:
-        parts = [{"text": prompt}]
+            self._cb.record_failure()
-        for path in frame_paths:
+            return None
+        self._cb.record_success()
+        logger.info(f"Gemini 视觉分析完成，{len(results)} 帧有描述")
+        return "\n".join(results)
+
+    def _analyze_one(self, path: str, ts: str,
+                     known_members: str) -> Optional[str]:
         try:
             with open(path, 'rb') as f:
-                    img_data = base64.b64encode(f.read()).decode('utf-8')
+                img = base64.b64encode(f.read()).decode('utf-8')
-                parts.append({
-                    "inline_data": {
-                        "mime_type": "image/jpeg",
-                        "data": img_data
-                    }
-                })
         except Exception as e:
             logger.error(f"读取图片失败 {path}: {e}")
+            return None
 
+        prompt = self._build_prompt(ts, known_members)
         try:
             resp = requests.post(
                 f"{self._base_url}/models/{self.model_name}:generateContent?key={self.api_key}",
-                json={
+                json={"contents": [{"parts": [
-                    "contents": [{"parts": parts}],
+                    {"text": prompt},
-                    "generationConfig": {"temperature": 0.2, "topP": 0.8}
+                    {"inline_data": {"mime_type": "image/jpeg", "data": img}}
-                },
+                ]}], "generationConfig": {"temperature": 0.2, "maxOutputTokens": 300}},
                 timeout=self.timeout
             )
-
             if resp.status_code == 200:
-                data = resp.json()
+                cands = resp.json().get('candidates', [])
-                candidates = data.get('candidates', [])
+                if cands:
-                if candidates:
+                    parts = cands[0].get('content', {}).get('parts', [])
-                    output = candidates[0].get('content', {}).get('parts', [{}])[0].get('text', '')
+                    text = ''.join(p.get('text', '') for p in parts).strip()
-                    self._cb.record_success()
+                    return text or None
-                    logger.info(f"Gemini 视觉分析完成，输出长度={len(output)}")
-                    return output
-                else:
                 logger.warning("Gemini 返回空 candidates")
-                    self._cb.record_failure()
-                    return None
             else:
-                logger.error(f"Gemini 调用失败: {resp.status_code} {resp.text[:200]}")
+                logger.warning(f"Gemini 单帧失败 HTTP {resp.status_code}: {resp.text[:150]}")
-                self._cb.record_failure()
-                return None
-
         except requests.Timeout:
-            logger.warning(f"Gemini 调用超时 ({self.timeout}s)，降级跳过")
+            logger.warning(f"Gemini 单帧超时 ({self.timeout}s)")
-            self._cb.record_failure()
-            return None
         except Exception as e:
-            logger.error(f"Gemini 调用异常: {e}")
+            logger.error(f"Gemini 单帧异常: {e}")
-            self._cb.record_failure()
         return None
 
     def get_timeout(self) -> int:
@@ -132,25 +125,15 @@ class GeminiAdapter(BaseModelAdapter):
     def get_circuit_breaker(self) -> CircuitBreaker:
         return self._cb
 
-    def _build_visual_prompt(self, n: int, timestamps: List[str], known_members: str) -> str:
+    def _build_prompt(self, ts: str, known_members: str) -> str:
-        ts_lines = '\n'.join(
+        return f"""你是家庭监控视频分析助手。请看这张监控截图（拍摄时间 {ts}），客观描述画面内容，不要猜测。
-            f"[Image {i+1}] Time: {ts}" for i, ts in enumerate(timestamps)
-        )
-        return f"""You are a home surveillance video analysis assistant. Describe what you see in the following {n} images chronologically. Be objective.
 
-Timestamps:
+需报告：
-{ts_lines}
+1. 人物：数量、衣着（颜色+类型）、可见动作
+2. 物品：玩具、奶瓶、家具等显眼物体
+3. 互动：人与人或人与物体的互动
 
-For each image, report:
+已知家庭成员（按特征匹配，匹配到用真名，否则用"人物X"）：
-1. People: count, clothing (color + type), visible actions
+{known_members or '（暂无）'}
-2. Objects: toys, bottles, furniture, etc.
-3. Interactions: between people or people and objects
 
-Known family members (match by features, use real_name if matched, otherwise "PersonX"):
+要求简洁客观，不要输出 JSON 或 markdown。"""
-{known_members or 'None'}
-
-Output format (plain text, one paragraph per image, keep timestamp markers):
-[Image 1] Time: {timestamps[0] if timestamps else ''}
-Description: ...
-
-Be concise and objective. Do not output JSON or markdown."""

fam-edge/src/fam_edge/model_adapters/nvidia_adapter.py

@@ -0,0 +1,134 @@
+"""
+NvidiaVisionAdapter - NVIDIA NIM 云端 VLM 适配器
+
+provider_name = "nvidia"
+模型: meta/llama-3.2-11b-vision-instruct
+角色: vision (视觉分析)
+SDK: openai (NIM 兼容 OpenAI API 规范)
+限制: NIM 单次请求最多 1 张图 -> 适配器内部逐帧调用
+熔断器: 启用
+"""
+import os
+import base64
+from typing import List, Optional
+
+from .base_adapter import BaseModelAdapter
+from .circuit_breaker import CircuitBreaker
+from ..logger import setup_logger
+
+logger = setup_logger('fam-edge.nvidia_adapter')
+
+try:
+    from openai import OpenAI
+except ImportError:
+    OpenAI = None
+
+
+class NvidiaVisionAdapter(BaseModelAdapter):
+    """NVIDIA NIM 云端 VLM 适配器 (逐帧)"""
+
+    def __init__(self, config: dict):
+        super().__init__("nvidia", config)
+        self.model_name = config.get('model_name', 'meta/llama-3.2-11b-vision-instruct')
+        self.api_key = self._resolve_key(config.get('api_key', ''))
+        self.base_url = config.get('base_url', 'https://integrate.api.nvidia.com/v1')
+        self.timeout = config.get('timeout', 20)
+        cb_cfg = config.get('circuit_breaker', {})
+        self._cb = CircuitBreaker(
+            threshold=cb_cfg.get('threshold', 3),
+            cooldown=cb_cfg.get('cooldown', 600),
+            enabled=cb_cfg.get('enabled', True)
+        )
+        self._client = None
+        if OpenAI is not None and self.api_key:
+            try:
+                self._client = OpenAI(base_url=self.base_url, api_key=self.api_key)
+            except Exception as e:
+                logger.error(f"NVIDIA OpenAI 客户端初始化失败: {e}")
+                self._client = None
+
+    def _resolve_key(self, raw: str) -> str:
+        if raw.startswith('${') and raw.endswith('}'):
+            return os.environ.get(raw[2:-1], '')
+        return raw
+
+    def health_check(self) -> bool:
+        if self._client is None:
+            logger.warning("NVIDIA OpenAI SDK 未就绪或 Key 未配置，健康检查失败")
+            return False
+        try:
+            self._client.models.list()
+            logger.info("NVIDIA 健康检查通过")
+            return True
+        except Exception as e:
+            logger.warning(f"NVIDIA 健康检查失败: {e}")
+            return False
+
+    def analyze_frames(self, frame_paths: List[str],
+                       frame_timestamps: List[str],
+                       known_members_context: str) -> Optional[str]:
+        if self._cb.is_open():
+            logger.warning("NVIDIA 熔断器 OPEN，跳过调用")
+            return None
+        if self._client is None:
+            logger.warning("NVIDIA 客户端未初始化，跳过调用")
+            return None
+
+        results = []
+        for path, ts in zip(frame_paths, frame_timestamps):
+            desc = self._analyze_one(path, ts, known_members_context)
+            if desc:
+                results.append(f"[帧] 时间: {ts}\n{desc}")
+
+        if not results:
+            self._cb.record_failure()
+            return None
+        self._cb.record_success()
+        logger.info(f"NVIDIA 视觉分析完成，{len(results)} 帧有描述")
+        return "\n".join(results)
+
+    def _analyze_one(self, path: str, ts: str,
+                     known_members: str) -> Optional[str]:
+        try:
+            with open(path, 'rb') as f:
+                b64 = base64.b64encode(f.read()).decode('utf-8')
+        except Exception as e:
+            logger.error(f"读取图片失败 {path}: {e}")
+            return None
+
+        prompt = self._build_prompt(ts, known_members)
+        try:
+            resp = self._client.chat.completions.create(
+                model=self.model_name,
+                messages=[{"role": "user", "content": [
+                    {"type": "text", "text": prompt},
+                    {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}}
+                ]}],
+                temperature=0.2,
+                max_tokens=512,
+                timeout=self.timeout
+            )
+            content = resp.choices[0].message.content
+            return content.strip() if content else None
+        except Exception as e:
+            logger.warning(f"NVIDIA 单帧异常: {e}")
+            return None
+
+    def get_timeout(self) -> int:
+        return self.timeout
+
+    def get_circuit_breaker(self) -> CircuitBreaker:
+        return self._cb
+
+    def _build_prompt(self, ts: str, known_members: str) -> str:
+        return f"""你是家庭监控视频分析助手。请看这张监控截图（拍摄时间 {ts}），客观描述画面内容，不要猜测。
+
+需报告：
+1. 人物：数量、衣着（颜色+类型）、可见动作
+2. 物品：玩具、奶瓶、家具等显眼物体
+3. 互动：人与人或人与物体的互动
+
+已知家庭成员（按特征匹配，匹配到用真名，否则用"人物X"）：
+{known_members or '（暂无）'}
+
+要求简洁客观，不要输出 JSON 或 markdown。"""