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
-    model_name: "gemini-1.5-flash"
-    api_key: ""
-    timeout: 8
+    role: "vision"
+    enabled: true
+    model_name: "gemini-flash-latest"   # v1beta 下 gemini-1.5-flash 会 404
+    api_key: "${GEMINI_API_KEY}"
+    timeout: 15
     circuit_breaker:
       enabled: true
-      threshold: 5
-      cooldown: 900
+      threshold: 3
+      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]:
-        """并行调用所有健康模型进行视觉分析"""
-        model_outputs = {}
-        max_timeout = max((a.get_timeout() for a in adapters), default=240)
+        """视觉分析阶段：仅 role=vision 的适配器参与
 
-        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 纯文本模式
-        ollama_cfg = next(
-            (cfg for cfg in self.config.get('models', []) if cfg.get('provider') == 'ollama'),
-            None
+        # 调用文本角色模型（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 ollama_cfg:
-            raise VLMOutputInvalidError("没有 Ollama 配置，无法执行文本融合")
+        if not text_cfg:
+            raise VLMOutputInvalidError("没有文本角色模型配置，无法执行文本融合")
 
-        base_url = ollama_cfg.get('base_url', 'http://localhost:11434')
-        model_name = ollama_cfg.get('model_name', 'llava-phi3')
-        fusion_timeout = self.timeout_cfg.get('vlm_fusion', 120)
+        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)
 
         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
         )

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 扩展:
-    # "openai": OpenAIAdapter,
-    # "nvidia": NvidiaAdapter,
+    "nvidia": NvidiaVisionAdapter,
 }
 
 

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
-健康检查: GET models API
-熔断器: 启用，连续 5 次失败 -> OPEN 15 分钟
+模型: gemini-flash-latest (v1beta 下 gemini-1.5-flash 会 404，用 flash-latest 别名)
+角色: vision (视觉分析)
+健康检查: 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.api_key = config.get('api_key', '')
-        self.timeout = config.get('timeout', 8)
+        self.model_name = config.get('model_name', 'gemini-flash-latest')
+        self.api_key = self._resolve_key(config.get('api_key', ''))
+        self.timeout = config.get('timeout', 15)
         cb_cfg = config.get('circuit_breaker', {})
         self._cb = CircuitBreaker(
-            threshold=cb_cfg.get('threshold', 5),
-            cooldown=cb_cfg.get('cooldown', 900),
-            enabled=cb_cfg.get('enabled', True)  # 云端默认启用
+            threshold=cb_cfg.get('threshold', 3),
+            cooldown=cb_cfg.get('cooldown', 600),
+            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}",
-                timeout=10
-            )
+                f"{self._base_url}/models?key={self.api_key}", timeout=10)
             if resp.status_code == 200:
                 models = resp.json().get('models', [])
-                model_names = [m.get('name', '') for m in models]
-                has_model = any(self.model_name in name for name in model_names)
-                if has_model:
-                    logger.info(f"Gemini 健康检查通过: 模型 {self.model_name} 可用")
+                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}")
                     return True
-                else:
-                    logger.warning(f"Gemini 健康检查失败: 模型 {self.model_name} 未找到")
-                    return False
+                logger.warning(f"Gemini 模型未找到: {self.model_name}; 可用: {names[:5]}")
+                return False
+            logger.warning(f"Gemini 健康检查 HTTP {resp.status_code}")
             return False
         except Exception as e:
             logger.error(f"Gemini 健康检查异常: {e}")
@@ -62,69 +65,59 @@ 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
-        n = len(frame_paths)
-        prompt = self._build_visual_prompt(n, frame_timestamps, known_members_context)
+        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}")
 
-        # 构建 inline_data
-        parts = [{"text": prompt}]
-        for path in frame_paths:
-            try:
-                with open(path, 'rb') as f:
-                    img_data = 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}")
+        if not results:
+            self._cb.record_failure()
+            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 = 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 = requests.post(
                 f"{self._base_url}/models/{self.model_name}:generateContent?key={self.api_key}",
-                json={
-                    "contents": [{"parts": parts}],
-                    "generationConfig": {"temperature": 0.2, "topP": 0.8}
-                },
+                json={"contents": [{"parts": [
+                    {"text": prompt},
+                    {"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()
-                candidates = data.get('candidates', [])
-                if candidates:
-                    output = candidates[0].get('content', {}).get('parts', [{}])[0].get('text', '')
-                    self._cb.record_success()
-                    logger.info(f"Gemini 视觉分析完成，输出长度={len(output)}")
-                    return output
-                else:
-                    logger.warning("Gemini 返回空 candidates")
-                    self._cb.record_failure()
-                    return None
+                cands = resp.json().get('candidates', [])
+                if cands:
+                    parts = cands[0].get('content', {}).get('parts', [])
+                    text = ''.join(p.get('text', '') for p in parts).strip()
+                    return text or None
+                logger.warning("Gemini 返回空 candidates")
             else:
-                logger.error(f"Gemini 调用失败: {resp.status_code} {resp.text[:200]}")
-                self._cb.record_failure()
-                return None
-
+                logger.warning(f"Gemini 单帧失败 HTTP {resp.status_code}: {resp.text[:150]}")
         except requests.Timeout:
-            logger.warning(f"Gemini 调用超时 ({self.timeout}s)，降级跳过")
-            self._cb.record_failure()
-            return None
+            logger.warning(f"Gemini 单帧超时 ({self.timeout}s)")
         except Exception as e:
-            logger.error(f"Gemini 调用异常: {e}")
-            self._cb.record_failure()
-            return None
+            logger.error(f"Gemini 单帧异常: {e}")
+        return None
 
     def get_timeout(self) -> int:
         return self.timeout
@@ -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:
-        ts_lines = '\n'.join(
-            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.
+    def _build_prompt(self, ts: str, known_members: str) -> str:
+        return f"""你是家庭监控视频分析助手。请看这张监控截图（拍摄时间 {ts}），客观描述画面内容，不要猜测。
 
-Timestamps:
-{ts_lines}
+需报告：
+1. 人物：数量、衣着（颜色+类型）、可见动作
+2. 物品：玩具、奶瓶、家具等显眼物体
+3. 互动：人与人或人与物体的互动
 
-For each image, report:
-1. People: count, clothing (color + type), visible actions
-2. Objects: toys, bottles, furniture, etc.
-3. Interactions: between people or people and objects
+已知家庭成员（按特征匹配，匹配到用真名，否则用"人物X"）：
+{known_members or '（暂无）'}
 
-Known family members (match by features, use real_name if matched, otherwise "PersonX"):
-{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."""
+要求简洁客观，不要输出 JSON 或 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。"""