feat(fam-edge): 云端模型调用统计 - 新增 model_calls 表(provider/model/时间/耗时/成功/失败原因),gemini/nvidia 每次请求经 model_call_hook 记录并随 sync delta 下发;gemini 支持模型级 model_timeouts(lite 实测22.6s×4≈90s)

This commit is contained in:
ericwyuan
2026-08-21 12:21:49 +08:00
parent 1b0f62e5e9
commit b87b38b140
6 changed files with 90 additions and 3 deletions

View File

@@ -55,6 +55,10 @@ models:
- "gemini-flash-lite-latest" - "gemini-flash-lite-latest"
api_key: "${GEMINI_API_KEY}" api_key: "${GEMINI_API_KEY}"
timeout: 600 timeout: 600
# 模型级独立超时(最终值,不参与编排层 ×2 放大)
# gemini-flash-lite 实测耗时 ~22.6s(真实 370MB 监控视频),按用户要求 ×4 ≈ 90s
model_timeouts:
"gemini-flash-lite-latest": 90
circuit_breaker: circuit_breaker:
enabled: true enabled: true
threshold: 5 threshold: 5

View File

@@ -33,6 +33,21 @@ class BaseModelAdapter(ABC):
self.config = config self.config = config
# 角色: vision=视觉分析, text=智能问答兜底(本地模型); 默认 vision # 角色: vision=视觉分析, text=智能问答兜底(本地模型); 默认 vision
self.role = config.get('role', '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: def get_role(self) -> str:
"""返回适配器角色: 'vision''text'""" """返回适配器角色: 'vision''text'"""

View File

@@ -13,6 +13,7 @@ import os
import time import time
import json import json
import requests import requests
from datetime import datetime
from typing import Dict, List, Optional from typing import Dict, List, Optional
from .base_adapter import BaseModelAdapter 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] 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.api_key = self._resolve_key(config.get('api_key', ''))
self.timeout = config.get('timeout', 600) 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', {}) cb_cfg = config.get('circuit_breaker', {})
self._cb = CircuitBreaker( self._cb = CircuitBreaker(
threshold=cb_cfg.get('threshold', 3), threshold=cb_cfg.get('threshold', 3),
@@ -224,9 +229,13 @@ class GeminiAdapter(BaseModelAdapter):
{"text": prompt}, {"text": prompt},
] ]
for model in self.model_chain: for model in self.model_chain:
# 模型级独立超时优先;未单独配置的用适配器默认(可能已被编排层 ×N 放大)
model_timeout = self.model_timeouts.get(model, self.timeout)
for attempt in range(2): for attempt in range(2):
if attempt == 0: 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: try:
resp = requests.post( resp = requests.post(
f"{self._base_url}/models/{model}:generateContent?key={self.api_key}", f"{self._base_url}/models/{model}:generateContent?key={self.api_key}",
@@ -234,14 +243,20 @@ class GeminiAdapter(BaseModelAdapter):
"generationConfig": { "generationConfig": {
"temperature": temperature, "temperature": temperature,
"maxOutputTokens": max_tokens}}, "maxOutputTokens": max_tokens}},
timeout=self.timeout timeout=model_timeout
) )
except requests.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 break
except Exception as e: except Exception as e:
duration = time.time() - t0
self._emit_model_call(model, started, duration, False, str(e))
logger.error(f"Gemini [{model}] 视频请求异常: {e}") logger.error(f"Gemini [{model}] 视频请求异常: {e}")
break break
duration = time.time() - t0
if resp.status_code == 200: if resp.status_code == 200:
cands = resp.json().get('candidates', []) cands = resp.json().get('candidates', [])
@@ -250,21 +265,28 @@ class GeminiAdapter(BaseModelAdapter):
for p in (cands[0].get('content', {}) if cands else {}).get('parts', []) for p in (cands[0].get('content', {}) if cands else {}).get('parts', [])
).strip() if cands else '' ).strip() if cands else ''
if text: if text:
self._emit_model_call(model, started, duration, True)
if model != self.model_name: if model != self.model_name:
logger.info(f"Gemini 主模型不可用,由 fallback [{model}] 出结果") logger.info(f"Gemini 主模型不可用,由 fallback [{model}] 出结果")
return text return text
self._emit_model_call(model, started, duration, False, "empty_text")
logger.warning(f"Gemini [{model}] 返回空文本") logger.warning(f"Gemini [{model}] 返回空文本")
continue continue
detail = resp.text[:150].replace('\n', ' ') detail = resp.text[:150].replace('\n', ' ')
if resp.status_code == 429: if resp.status_code == 429:
self._emit_model_call(model, started, duration, False, "429_quota")
logger.warning(f"Gemini [{model}] 429 配额耗尽,切换下一模型") logger.warning(f"Gemini [{model}] 429 配额耗尽,切换下一模型")
break break
if resp.status_code == 503: if resp.status_code == 503:
if attempt == 0: if attempt == 0:
self._emit_model_call(model, started, duration, False, "503_overload_retry")
logger.warning(f"Gemini [{model}] 503 过载3s 后重试") logger.warning(f"Gemini [{model}] 503 过载3s 后重试")
time.sleep(3) time.sleep(3)
continue continue
self._emit_model_call(model, started, duration, False, "503_overload")
break break
self._emit_model_call(model, started, duration, False,
f"http_{resp.status_code}")
logger.warning(f"Gemini [{model}] HTTP {resp.status_code}: {detail}") logger.warning(f"Gemini [{model}] HTTP {resp.status_code}: {detail}")
break break
return None return None

View File

@@ -11,7 +11,9 @@ SDK: openai (NIM 兼容 OpenAI API 规范)
import os import os
import json import json
import re import re
import time
import requests import requests
from datetime import datetime
from typing import Dict, List, Optional from typing import Dict, List, Optional
from .base_adapter import BaseModelAdapter from .base_adapter import BaseModelAdapter
@@ -144,6 +146,8 @@ class NvidiaVisionAdapter(BaseModelAdapter):
return None return None
prompt = self._build_video_prompt(known_members_context, event_start_time) 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: try:
resp = self._client.chat.completions.create( resp = self._client.chat.completions.create(
model=self.model_name, model=self.model_name,
@@ -158,16 +162,20 @@ class NvidiaVisionAdapter(BaseModelAdapter):
extra_body={"media_io_kwargs": {"video": {"num_frames": 128}}}, extra_body={"media_io_kwargs": {"video": {"num_frames": 128}}},
timeout=self.timeout timeout=self.timeout
) )
duration = time.time() - t0
content = resp.choices[0].message.content content = resp.choices[0].message.content
if not content: if not content:
self._emit_model_call(self.model_name, started, duration, False, "empty_content")
logger.warning("NVIDIA 视频分析返回空 content") logger.warning("NVIDIA 视频分析返回空 content")
self._cb.record_failure() self._cb.record_failure()
return None return None
data = self._parse_json(content) data = self._parse_json(content)
if not data or 'events' not in data: 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]}") logger.warning(f"NVIDIA 视频 JSON 解析失败: {content[:150]}")
self._cb.record_failure() self._cb.record_failure()
return None return None
self._emit_model_call(self.model_name, started, duration, True)
self._cb.record_success() self._cb.record_success()
logger.info(f"NVIDIA 整视频分析完成events={len(data.get('events', []))}") logger.info(f"NVIDIA 整视频分析完成events={len(data.get('events', []))}")
return { return {
@@ -177,6 +185,8 @@ class NvidiaVisionAdapter(BaseModelAdapter):
"compute_provider": "nvidia", "compute_provider": "nvidia",
} }
except Exception as e: except Exception as e:
duration = time.time() - t0
self._emit_model_call(self.model_name, started, duration, False, str(e))
self._cb.record_failure() self._cb.record_failure()
logger.warning(f"NVIDIA 视频分析异常: {e}") logger.warning(f"NVIDIA 视频分析异常: {e}")
return None return None

View File

@@ -81,8 +81,21 @@ class OracleDB:
key TEXT PRIMARY KEY, key TEXT PRIMARY KEY,
value TEXT 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_videos_updated ON videos(updated_at);
CREATE INDEX IF NOT EXISTS idx_events_video ON events(video_id); 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 列(生产-消费队列重试上限用) # 兼容旧库:补 retry_count 列(生产-消费队列重试上限用)
cols = [r[1] for r in c.execute("PRAGMA table_info(videos)").fetchall()] 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,)) cur = self._conn.execute("SELECT * FROM videos WHERE id=?", (video_id,))
return cur.fetchone() 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, def ensure_video(self, filename: str, local_path: str,
camera_name: str = '', event_start_time: str = '', camera_name: str = '', event_start_time: str = '',
duration_sec: float = 0.0, drive_file_id: str = '') -> int: duration_sec: float = 0.0, drive_file_id: str = '') -> int:
@@ -220,6 +247,10 @@ class OracleDB:
people = self._conn.execute( people = self._conn.execute(
"SELECT * FROM people WHERE updated_at > ? ORDER BY id ASC", (since_iso,) "SELECT * FROM people WHERE updated_at > ? ORDER BY id ASC", (since_iso,)
).fetchall() ).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): def _ser(row):
d = dict(row) d = dict(row)
@@ -229,6 +260,7 @@ class OracleDB:
"videos": [_ser(v) for v in videos], "videos": [_ser(v) for v in videos],
"events": [_ser(e) for e in events], "events": [_ser(e) for e in events],
"people": [_ser(p) for p in people], "people": [_ser(p) for p in people],
"model_calls": [_ser(m) for m in model_calls],
"server_time": _now_iso(), "server_time": _now_iso(),
} }

View File

@@ -93,6 +93,10 @@ class VideoProcessor:
last_err = "no_vision_adapter" last_err = "no_vision_adapter"
for adapter in self._ordered_vision_adapters(): 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 # 按模型原配置超时 × multiplier默认 1x队列消费默认 2x
orig_timeout = adapter.get_timeout() orig_timeout = adapter.get_timeout()
if timeout_multiplier != 1.0: if timeout_multiplier != 1.0: