From b87b38b14013d3644233293d847439e4832113ca Mon Sep 17 00:00:00 2001 From: ericwyuan Date: Fri, 21 Aug 2026 12:21:49 +0800 Subject: [PATCH] =?UTF-8?q?feat(fam-edge):=20=E4=BA=91=E7=AB=AF=E6=A8=A1?= =?UTF-8?q?=E5=9E=8B=E8=B0=83=E7=94=A8=E7=BB=9F=E8=AE=A1=20-=20=E6=96=B0?= =?UTF-8?q?=E5=A2=9E=20model=5Fcalls=20=E8=A1=A8(provider/model/=E6=97=B6?= =?UTF-8?q?=E9=97=B4/=E8=80=97=E6=97=B6/=E6=88=90=E5=8A=9F/=E5=A4=B1?= =?UTF-8?q?=E8=B4=A5=E5=8E=9F=E5=9B=A0)=EF=BC=8Cgemini/nvidia=20=E6=AF=8F?= =?UTF-8?q?=E6=AC=A1=E8=AF=B7=E6=B1=82=E7=BB=8F=20model=5Fcall=5Fhook=20?= =?UTF-8?q?=E8=AE=B0=E5=BD=95=E5=B9=B6=E9=9A=8F=20sync=20delta=20=E4=B8=8B?= =?UTF-8?q?=E5=8F=91=EF=BC=9Bgemini=20=E6=94=AF=E6=8C=81=E6=A8=A1=E5=9E=8B?= =?UTF-8?q?=E7=BA=A7=20model=5Ftimeouts=EF=BC=88lite=20=E5=AE=9E=E6=B5=8B2?= =?UTF-8?q?2.6s=C3=974=E2=89=8890s=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- fam-edge/config/config.yaml | 4 +++ .../fam_edge/model_adapters/base_adapter.py | 15 +++++++++ .../fam_edge/model_adapters/gemini_adapter.py | 28 ++++++++++++++-- .../fam_edge/model_adapters/nvidia_adapter.py | 10 ++++++ fam-edge/src/fam_edge/oracle_db.py | 32 +++++++++++++++++++ fam-edge/src/fam_edge/video_processor.py | 4 +++ 6 files changed, 90 insertions(+), 3 deletions(-) diff --git a/fam-edge/config/config.yaml b/fam-edge/config/config.yaml index 09f0e17..bceed39 100644 --- a/fam-edge/config/config.yaml +++ b/fam-edge/config/config.yaml @@ -55,6 +55,10 @@ models: - "gemini-flash-lite-latest" api_key: "${GEMINI_API_KEY}" timeout: 600 + # 模型级独立超时（最终值，不参与编排层 ×2 放大） + # gemini-flash-lite 实测耗时 ~22.6s（真实 370MB 监控视频），按用户要求 ×4 ≈ 90s + model_timeouts: + "gemini-flash-lite-latest": 90 circuit_breaker: enabled: true threshold: 5 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 7518cb7..8a60d72 100644 --- a/fam-edge/src/fam_edge/model_adapters/base_adapter.py +++ b/fam-edge/src/fam_edge/model_adapters/base_adapter.py @@ -33,6 +33,21 @@ class BaseModelAdapter(ABC): self.config = config # 角色: vision=视觉分析, text=智能问答兜底(本地模型); 默认 vision self.role = config.get('role', 'vision') + # 模型调用统计回调（由编排层注入）: + # hook(provider, model, started_at, duration_sec, success, error) + self.model_call_hook = None + + def _emit_model_call(self, model: str, started_at: str, + duration_sec: float, success: bool, + error: str = ''): + """上报一次模型调用统计（供前端展示成功/失败/耗时/失败原因）""" + hook = self.model_call_hook + if hook is None: + return + try: + hook(self.provider_name, model, started_at, duration_sec, success, error) + except Exception: + pass # 统计失败不影响主流程 def get_role(self) -> str: """返回适配器角色: 'vision' 或 'text'""" 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 d42347b..59d61f4 100644 --- a/fam-edge/src/fam_edge/model_adapters/gemini_adapter.py +++ b/fam-edge/src/fam_edge/model_adapters/gemini_adapter.py @@ -13,6 +13,7 @@ import os import time import json import requests +from datetime import datetime from typing import Dict, List, Optional from .base_adapter import BaseModelAdapter @@ -33,6 +34,10 @@ class GeminiAdapter(BaseModelAdapter): 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) + # 模型级独立超时（最终值，不参与编排层 ×N 放大）: {model_name: seconds} + # 例: {"gemini-flash-lite-latest": 90}（按实测耗时 ×4 配置） + self.model_timeouts = { + str(k): int(v) for k, v in (config.get('model_timeouts') or {}).items()} cb_cfg = config.get('circuit_breaker', {}) self._cb = CircuitBreaker( threshold=cb_cfg.get('threshold', 3), @@ -224,9 +229,13 @@ class GeminiAdapter(BaseModelAdapter): {"text": prompt}, ] for model in self.model_chain: + # 模型级独立超时优先；未单独配置的用适配器默认（可能已被编排层 ×N 放大） + model_timeout = self.model_timeouts.get(model, self.timeout) for attempt in range(2): if attempt == 0: - logger.info(f"Gemini [{model}] 本轮请求超时 {self.timeout}s") + logger.info(f"Gemini [{model}] 本轮请求超时 {model_timeout}s") + started = datetime.now().strftime('%Y-%m-%d %H:%M:%S') + t0 = time.time() try: resp = requests.post( f"{self._base_url}/models/{model}:generateContent?key={self.api_key}", @@ -234,14 +243,20 @@ class GeminiAdapter(BaseModelAdapter): "generationConfig": { "temperature": temperature, "maxOutputTokens": max_tokens}}, - timeout=self.timeout + timeout=model_timeout ) except requests.Timeout: - logger.warning(f"Gemini [{model}] 视频请求超时 ({self.timeout}s)") + duration = time.time() - t0 + self._emit_model_call(model, started, duration, False, + f"timeout({model_timeout}s)") + logger.warning(f"Gemini [{model}] 视频请求超时 ({model_timeout}s)") break except Exception as e: + duration = time.time() - t0 + self._emit_model_call(model, started, duration, False, str(e)) logger.error(f"Gemini [{model}] 视频请求异常: {e}") break + duration = time.time() - t0 if resp.status_code == 200: cands = resp.json().get('candidates', []) @@ -250,21 +265,28 @@ class GeminiAdapter(BaseModelAdapter): for p in (cands[0].get('content', {}) if cands else {}).get('parts', []) ).strip() if cands else '' if text: + self._emit_model_call(model, started, duration, True) if model != self.model_name: logger.info(f"Gemini 主模型不可用，由 fallback [{model}] 出结果") return text + self._emit_model_call(model, started, duration, False, "empty_text") logger.warning(f"Gemini [{model}] 返回空文本") continue detail = resp.text[:150].replace('\n', ' ') if resp.status_code == 429: + self._emit_model_call(model, started, duration, False, "429_quota") logger.warning(f"Gemini [{model}] 429 配额耗尽，切换下一模型") break if resp.status_code == 503: if attempt == 0: + self._emit_model_call(model, started, duration, False, "503_overload_retry") logger.warning(f"Gemini [{model}] 503 过载，3s 后重试") time.sleep(3) continue + self._emit_model_call(model, started, duration, False, "503_overload") break + self._emit_model_call(model, started, duration, False, + f"http_{resp.status_code}") logger.warning(f"Gemini [{model}] HTTP {resp.status_code}: {detail}") break return None diff --git a/fam-edge/src/fam_edge/model_adapters/nvidia_adapter.py b/fam-edge/src/fam_edge/model_adapters/nvidia_adapter.py index 3a70746..a505739 100644 --- a/fam-edge/src/fam_edge/model_adapters/nvidia_adapter.py +++ b/fam-edge/src/fam_edge/model_adapters/nvidia_adapter.py @@ -11,7 +11,9 @@ SDK: openai (NIM 兼容 OpenAI API 规范) import os import json import re +import time import requests +from datetime import datetime from typing import Dict, List, Optional from .base_adapter import BaseModelAdapter @@ -144,6 +146,8 @@ class NvidiaVisionAdapter(BaseModelAdapter): return None prompt = self._build_video_prompt(known_members_context, event_start_time) + started = datetime.now().strftime('%Y-%m-%d %H:%M:%S') + t0 = time.time() try: resp = self._client.chat.completions.create( model=self.model_name, @@ -158,16 +162,20 @@ class NvidiaVisionAdapter(BaseModelAdapter): extra_body={"media_io_kwargs": {"video": {"num_frames": 128}}}, timeout=self.timeout ) + duration = time.time() - t0 content = resp.choices[0].message.content if not content: + self._emit_model_call(self.model_name, started, duration, False, "empty_content") logger.warning("NVIDIA 视频分析返回空 content") self._cb.record_failure() return None data = self._parse_json(content) if not data or 'events' not in data: + self._emit_model_call(self.model_name, started, duration, False, "json_parse_failed") logger.warning(f"NVIDIA 视频 JSON 解析失败: {content[:150]}") self._cb.record_failure() return None + self._emit_model_call(self.model_name, started, duration, True) self._cb.record_success() logger.info(f"NVIDIA 整视频分析完成，events={len(data.get('events', []))}") return { @@ -177,6 +185,8 @@ class NvidiaVisionAdapter(BaseModelAdapter): "compute_provider": "nvidia", } except Exception as e: + duration = time.time() - t0 + self._emit_model_call(self.model_name, started, duration, False, str(e)) self._cb.record_failure() logger.warning(f"NVIDIA 视频分析异常: {e}") return None diff --git a/fam-edge/src/fam_edge/oracle_db.py b/fam-edge/src/fam_edge/oracle_db.py index 2be2dba..a6d178a 100644 --- a/fam-edge/src/fam_edge/oracle_db.py +++ b/fam-edge/src/fam_edge/oracle_db.py @@ -81,8 +81,21 @@ class OracleDB: key TEXT PRIMARY KEY, value TEXT ); + CREATE TABLE IF NOT EXISTS model_calls ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + provider TEXT, + model TEXT, + video_id INTEGER, + filename TEXT, + started_at TEXT, + duration_sec REAL, + success INTEGER DEFAULT 0, + error TEXT, + created_at TEXT + ); CREATE INDEX IF NOT EXISTS idx_videos_updated ON videos(updated_at); CREATE INDEX IF NOT EXISTS idx_events_video ON events(video_id); + CREATE INDEX IF NOT EXISTS idx_model_calls_created ON model_calls(created_at); """) # 兼容旧库：补 retry_count 列（生产-消费队列重试上限用） cols = [r[1] for r in c.execute("PRAGMA table_info(videos)").fetchall()] @@ -101,6 +114,20 @@ class OracleDB: cur = self._conn.execute("SELECT * FROM videos WHERE id=?", (video_id,)) return cur.fetchone() + def record_model_call(self, provider: str, model: str, + video_id, filename, + started_at: str, duration_sec: float, + success: bool, error: str = ''): + """记录一次云端模型调用（前端统计成功/失败/耗时/失败原因）""" + now = _now_iso() + self._conn.execute( + "INSERT INTO model_calls (provider, model, video_id, filename, " + "started_at, duration_sec, success, error, created_at) " + "VALUES (?,?,?,?,?,?,?,?,?)", + (provider, model, video_id, filename, started_at, + duration_sec, 1 if success else 0, error or '', now)) + self._conn.commit() + def ensure_video(self, filename: str, local_path: str, camera_name: str = '', event_start_time: str = '', duration_sec: float = 0.0, drive_file_id: str = '') -> int: @@ -220,6 +247,10 @@ class OracleDB: people = self._conn.execute( "SELECT * FROM people WHERE updated_at > ? ORDER BY id ASC", (since_iso,) ).fetchall() + # 模型调用统计（created_at >= since 配合 NAS 端幂等 upsert 防漏同秒记录） + model_calls = self._conn.execute( + "SELECT * FROM model_calls WHERE created_at >= ? ORDER BY id ASC", + (since_iso,)).fetchall() def _ser(row): d = dict(row) @@ -229,6 +260,7 @@ class OracleDB: "videos": [_ser(v) for v in videos], "events": [_ser(e) for e in events], "people": [_ser(p) for p in people], + "model_calls": [_ser(m) for m in model_calls], "server_time": _now_iso(), } diff --git a/fam-edge/src/fam_edge/video_processor.py b/fam-edge/src/fam_edge/video_processor.py index 6c3dc58..916707d 100644 --- a/fam-edge/src/fam_edge/video_processor.py +++ b/fam-edge/src/fam_edge/video_processor.py @@ -93,6 +93,10 @@ class VideoProcessor: last_err = "no_vision_adapter" for adapter in self._ordered_vision_adapters(): + # 模型调用统计 hook（带当前 video_id/filename，前端展示用） + adapter.model_call_hook = ( + lambda p, m, s, d, ok, e, _vid=video_id, _fn=filename: + self.db.record_model_call(p, m, _vid, _fn, s, d, ok, e)) # 按模型原配置超时 × multiplier（默认 1x；队列消费默认 2x） orig_timeout = adapter.get_timeout() if timeout_multiplier != 1.0: