diff --git a/fam-edge/src/fam_edge/model_adapters/adapter_factory.py b/fam-edge/src/fam_edge/model_adapters/adapter_factory.py index fb9b4b4..7a71c93 100644 --- a/fam-edge/src/fam_edge/model_adapters/adapter_factory.py +++ b/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, } diff --git a/fam-edge/src/fam_edge/model_adapters/base_adapter.py b/fam-edge/src/fam_edge/model_adapters/base_adapter.py index 8c3c956..b3c3aa7 100644 --- a/fam-edge/src/fam_edge/model_adapters/base_adapter.py +++ b/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: diff --git a/fam-edge/src/fam_edge/model_adapters/gemini_adapter.py b/fam-edge/src/fam_edge/model_adapters/gemini_adapter.py index 2de8ae5..83dea1d 100644 --- a/fam-edge/src/fam_edge/model_adapters/gemini_adapter.py +++ b/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。""" diff --git a/fam-edge/src/fam_edge/model_adapters/nvidia_adapter.py b/fam-edge/src/fam_edge/model_adapters/nvidia_adapter.py new file mode 100644 index 0000000..a094dbf --- /dev/null +++ b/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。"""