""" GeminiAdapter - Google Gemini 云端 VLM 适配器 provider_name = "gemini" 模型: gemini-flash-latest 角色: vision (整视频直出结构化 JSON) + 智能问答 健康检查: GET /v1beta/models?key=... 熔断器: 启用 整视频分析: 用 Files API 上传完整视频 -> generateContent 直出结构化 JSON （本地不切片、不抽帧；Gemini 原生支持长视频） """ import os import time import json import requests from typing import Dict, List, Optional from .base_adapter import BaseModelAdapter from .circuit_breaker import CircuitBreaker from ..logger import setup_logger from ..ai_orchestrator.json_parser import parse_vlm_json, VLMOutputInvalidError logger = setup_logger('fam-edge.gemini_adapter') class GeminiAdapter(BaseModelAdapter): """Gemini 云端 VLM 适配器 (整视频直出结构化 JSON + 文本问答)""" def __init__(self, config: dict): super().__init__("gemini", config) self.model_name = config.get('model_name', 'gemini-flash-latest') self.model_chain = [self.model_name] + [ m for m in config.get('fallback_models', []) if m and m != self.model_name] self.api_key = self._resolve_key(config.get('api_key', '')) self.timeout = config.get('timeout', 600) 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._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: 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) if resp.status_code == 200: models = resp.json().get('models', []) names = [m.get('name', '') for m in models] missing = [m for m in self.model_chain if not any(m in n for n in names)] if not missing: logger.info(f"Gemini 健康检查通过: {' -> '.join(self.model_chain)}") return True logger.warning(f"Gemini 模型未找到: {missing}; 可用: {names[:5]}") return len(missing) < len(self.model_chain) logger.warning(f"Gemini 健康检查 HTTP {resp.status_code}") return False except Exception as e: logger.error(f"Gemini 健康检查异常: {e}") return False # ------------------------------------------------------------------ # 整视频分析：Files API 上传 -> generateContent # ------------------------------------------------------------------ def analyze_video(self, video_path: str, known_members_context: str, event_start_time: str = '') -> Optional[Dict]: if self._cb.is_open(): logger.warning("Gemini 熔断器 OPEN，跳过视频分析") return None if not self.api_key: logger.warning("Gemini API Key 未配置，跳过视频分析") return None if not os.path.isfile(video_path): logger.warning(f"Gemini 视频文件不存在: {video_path}") return None file_uri = self._upload_file(video_path) if not file_uri: self._cb.record_failure() return None prompt = self._build_video_prompt(known_members_context, event_start_time) try: text = self._generate_video(file_uri, prompt, max_tokens=4096, temperature=0.2) if text is None: self._cb.record_failure() return None try: result = parse_vlm_json(text) result = self._normalize(result) if not result or 'events' not in result: logger.error(f"Gemini 视频输出缺少 events: {text[:150]}") self._cb.record_failure() return None result['compute_provider'] = 'gemini' self._cb.record_success() logger.info(f"Gemini 整视频分析完成，events={len(result.get('events', []))}") return result except VLMOutputInvalidError as e: logger.error(f"Gemini 视频输出无法解析为 JSON: {e}") self._cb.record_failure() return None except requests.Timeout: logger.warning(f"Gemini 视频分析超时 ({self.timeout}s)") self._cb.record_failure() return None except Exception as e: logger.error(f"Gemini 视频分析异常: {e}") self._cb.record_failure() return None finally: self._delete_file(file_uri) def _upload_file(self, video_path: str) -> Optional[str]: """用 Files API resumable 可续传协议上传完整视频，返回可引用 URI。""" name = os.path.basename(video_path) size = os.path.getsize(video_path) upload_timeout = max(self.timeout, 900) # 上传端点必须是 /upload/v1beta/files（/v1beta/files 只是元数据端点，不接受上传协议） base = f"https://generativelanguage.googleapis.com/upload/v1beta/files?key={self.api_key}" # 1) 创建可续传上传会话 try: r0 = requests.post( base, headers={ "X-Goog-Upload-Protocol": "resumable", "X-Goog-Upload-Command": "start", "X-Goog-Upload-Header-Content-Length": str(size), "X-Goog-Upload-Header-Content-Type": "video/mp4", "X-Goog-Upload-File-Name": name, "Content-Type": "application/json", }, json={"file": {"display_name": name}}, timeout=60, ) except Exception as e: logger.error(f"Gemini 创建上传会话异常: {e}") return None if r0.status_code not in (200, 201): logger.warning(f"Gemini 创建上传会话失败 HTTP {r0.status_code}: {r0.text[:200]}") return None session_url = r0.headers.get('X-Goog-Upload-URL') if not session_url: logger.warning("Gemini 上传响应缺少 X-Goog-Upload-URL") return None # 2) 上传文件体（流式） try: with open(video_path, 'rb') as f: resp = requests.post( session_url, headers={ "Content-Length": str(size), "X-Goog-Upload-Command": "upload, finalize", "X-Goog-Upload-Offset": "0", "Content-Type": "video/mp4", }, data=f, timeout=upload_timeout, ) except requests.Timeout: logger.warning(f"Gemini 文件上传超时 ({upload_timeout}s)") return None except Exception as e: logger.error(f"Gemini 文件上传异常: {e}") return None if resp.status_code not in (200, 201): logger.warning(f"Gemini 文件上传失败 HTTP {resp.status_code}: {resp.text[:200]}") return None try: info = resp.json().get('file', {}) uri = info.get('uri') file_name = info.get('name') state = info.get('state') except (ValueError, KeyError): logger.warning("Gemini 文件上传响应解析失败") return None if not uri: return None # 等待 ACTIVE（大文件可能还在处理） if state != 'ACTIVE' and file_name: uri = self._wait_active(file_name) return uri def _wait_active(self, file_name: str, max_wait: int = 120) -> Optional[str]: url = f"{self._base_url}/{file_name}?key={self.api_key}" deadline = time.time() + max_wait while time.time() < deadline: try: r = requests.get(url, timeout=15) if r.status_code == 200: j = r.json() if j.get('state') == 'ACTIVE': return j.get('uri') except Exception: pass time.sleep(5) logger.warning(f"Gemini 文件 {file_name} 未在 {max_wait}s 内 ACTIVE") return None def _delete_file(self, file_uri: str): if not file_uri or 'files/' not in file_uri: return name = file_uri.split('files/', 1)[-1] try: requests.delete(f"{self._base_url}/files/{name}?key={self.api_key}", timeout=15) except Exception: pass def _generate_video(self, file_uri: str, prompt: str, max_tokens: int, temperature: float) -> Optional[str]: """带模型 fallback 链的 generateContent（视频文件引用）调用。""" parts = [ {"file_data": {"mime_type": "video/mp4", "file_uri": file_uri}}, {"text": prompt}, ] for model in self.model_chain: for attempt in range(2): if attempt == 0: logger.info(f"Gemini [{model}] 本轮请求超时 {self.timeout}s") try: resp = requests.post( f"{self._base_url}/models/{model}:generateContent?key={self.api_key}", json={"contents": [{"parts": parts}], "generationConfig": { "temperature": temperature, "maxOutputTokens": max_tokens}}, timeout=self.timeout ) except requests.Timeout: logger.warning(f"Gemini [{model}] 视频请求超时 ({self.timeout}s)") break except Exception as e: logger.error(f"Gemini [{model}] 视频请求异常: {e}") break if resp.status_code == 200: cands = resp.json().get('candidates', []) text = ''.join( p.get('text', '') for p in (cands[0].get('content', {}) if cands else {}).get('parts', []) ).strip() if cands else '' if text: if model != self.model_name: logger.info(f"Gemini 主模型不可用，由 fallback [{model}] 出结果") return text logger.warning(f"Gemini [{model}] 返回空文本") continue detail = resp.text[:150].replace('\n', ' ') if resp.status_code == 429: logger.warning(f"Gemini [{model}] 429 配额耗尽，切换下一模型") break if resp.status_code == 503: if attempt == 0: logger.warning(f"Gemini [{model}] 503 过载，3s 后重试") time.sleep(3) continue break logger.warning(f"Gemini [{model}] HTTP {resp.status_code}: {detail}") break return None @staticmethod def _normalize(result: dict) -> dict: """统一字段名：frame_details -> events（兼容旧结构）。""" events = result.get('events') if events is None and 'frame_details' in result: events = [] for f in result['frame_details']: events.append({ "timestamp": f.get('frame_timestamp', ''), "description": f.get('action', ''), "people": [f.get('person', '')] if f.get('person') else [], "is_attention_event": bool(f.get('is_attention_event', False)), }) if events is None: events = [] people = result.get('people_mentioned') or result.get('entities_json') or [] if isinstance(people, list) and people and isinstance(people[0], dict): people = [p.get('person', '') for p in people] people = [p for p in people if p] return { "global_summary": result.get('global_summary', ''), "events": events, "people_mentioned": people, } def _build_video_prompt(self, known_members: str, event_start_time: str) -> str: start_hint = "" if event_start_time: start_hint = f"\n视频开始时间（北京时间）约为：{event_start_time}。请据此推算每个事件的绝对时间戳。" return f"""你是家庭监控视频分析助手。下面是一段完整监控录像（已整段上传）。 请观看整段视频，提取其中有用的信息，只输出合法 JSON（不要 markdown、不要任何解释文字），结构如下： {{ "global_summary": "整个时段的整体摘要，简体中文，2-4 句，客观描述人物与主要活动", "events": [ {{ "timestamp": "事件发生时的绝对北京时间(格式 YYYY-MM-DD HH:MM:SS)", "description": "该时间点的画面/动作信息摘要（谁、在做什么、位置）", "people": ["出现在该时刻的人物，用已知成员真名或'人物A'/'人物B'"], "is_attention_event": false }} ], "people_mentioned": ["本视频出现过的所有人物标识/真名"] }}{start_hint} 规则： 1. 只描述客观画面，不要猜测或想象。 2. events 提取视频中"有意义的时间点"（人物出现/动作变化/异常），不要逐秒罗列；timestamp 用绝对北京时间。 3. 已知家庭成员（按特征匹配，匹配到用 real_name，否则用"人物X"）： {known_members or '（暂无已知成员）'} 4. is_attention_event：是否为跌倒、危险、异常哭闹等需关注事件（没有则为 false）。 5. 没有人物出现的时段不要单独成 event；people 留空数组。""" # ------------------------------------------------------------------ # 智能问答：纯文本 # ------------------------------------------------------------------ def chat(self, prompt: str, max_tokens: int = 512) -> Optional[str]: if not self.api_key: logger.warning("Gemini API Key 未配置，跳过问答") return None try: return self._generate_text(prompt, max_tokens=max_tokens, temperature=0.3) except Exception as e: logger.error(f"Gemini 问答异常: {e}") return None def _generate_text(self, text: str, max_tokens: int, temperature: float) -> Optional[str]: """纯文本 generateContent（复用模型 fallback 链）。""" for model in self.model_chain: try: resp = requests.post( f"{self._base_url}/models/{model}:generateContent?key={self.api_key}", json={"contents": [{"parts": [{"text": text}]}], "generationConfig": { "temperature": temperature, "maxOutputTokens": max_tokens}}, timeout=self.timeout ) except requests.Timeout: logger.warning(f"Gemini [{model}] 问答超时") continue except Exception as e: logger.error(f"Gemini [{model}] 问答异常: {e}") continue if resp.status_code == 200: cands = resp.json().get('candidates', []) out = ''.join( p.get('text', '') for p in (cands[0].get('content', {}) if cands else {}).get('parts', []) ).strip() if cands else '' if out: return out elif resp.status_code == 429: logger.warning(f"Gemini [{model}] 429，切换模型") continue return None def get_timeout(self) -> int: return self.timeout def get_circuit_breaker(self) -> CircuitBreaker: return self._cb