告别识别率焦虑：视频 AI 工程化实战 —— 检测→判定→聚合→治理全链路拆解

背景

很多视频 AI 项目上线失败，不是识别率不够，而是工程能力缺失：无法批量跑、无法复盘、无法控成本。

vl_video(本人实现的一套方案)的价值是把识别问题做成了工程流水线。本文不列接口清单，直接拆架构与关键代码，给你一套可迁移的方法。

架构决策

主链路按“检测 -> 判定 -> 聚合 -> 治理”四段设计：

1. 检测：颜色 ROI + 候选会话。
2. 判定：多轮模型调用 + 投票 + 二次复核。
3. 聚合：片段事件映射回整视频时间轴。
4. 治理：评测指标、成本统计、调试资产。

trade-off

1. 单体大模型直看整段 vs 候选切片后识别
   前者开发快，后者成本与稳定性更优，适合持续运营。

2. 强依赖检测模型 vs 轻量前景差分
   前者准确，后者便宜；工程上应保留双模并支持切换和回退。

3. 追求吞吐 vs 追求可解释
   吞吐导向会牺牲复核链路；上线场景更应优先可解释和可审计。

失败复盘

1. 候选过密导致模型预算失控。
   修复：按候选分数分配调用次数，并设置单视频硬预算。

2. 模型返回格式漂移导致解析失败。
   修复：Prompt 强约束 + 三层 JSON 提取兜底。

3. 聚合后计数正确但事件顺序异常。
   修复：事件统一映射绝对时间再排序，禁止窗口内局部顺序直接拼接。

案例代码

# 节选自 vl_video/src/core/analyzer.py class VideoAnalyzer: CONFIDENCE_WEIGHT = {"high": 3.0, "medium": 2.0, "low": 1.0} def __init__(self, config, logger=None): self.config = config self.logger = logger self._multimodal_cls = None detector_config = CandidateDetectorConfig( sample_fps=config.candidate_sample_fps, sample_width=config.candidate_sample_width, absence_duration_sec=config.candidate_min_gap_sec, gap_fill_sec=config.candidate_gap_fill_sec, min_presence_sec=config.candidate_min_presence_sec, min_segment_sec=config.candidate_min_segment_sec, max_windows=config.max_candidate_windows_per_video, enable_light_normalization=config.enable_light_normalization, diff_threshold=config.candidate_peak_threshold, long_segment_threshold_sec=config.long_segment_threshold_sec, long_segment_search_window_sec=config.long_segment_search_window_sec, long_segment_min_side_sec=config.long_segment_min_side_sec, long_segment_quiet_sec=config.long_segment_quiet_sec, detection_method=config.detection_method, enable_yolo_person_detection=config.enable_yolo_person_detection, yolo_model_path=config.yolo_model_path, yolo_conf_threshold=config.yolo_conf_threshold, yolo_imgsz=config.yolo_imgsz, yolo_device=config.yolo_device, person_near_cabinet_padding=config.person_near_cabinet_padding, allow_person_detection_fallback=config.allow_person_detection_fallback, ) self.candidate_detector = CandidateDetector(detector_config) self.cabinet_locator = CabinetLocator( probe_frames=config.cabinet_probe_frames, max_regions=config.cabinet_max_regions, ) self.window_saver = WindowSaver( debug_root=config.debug_output_dir, save_windows=config.save_candidate_windows, ) self.event_aggregator = EventAggregator()

设计意图：配置前移，构建统一“可调参数面”。
风险点：配置膨胀会增加误配置概率，需配套校验与默认值策略。

# 节选自 vl_video/src/detection/candidate_detector.py def _detect_presence(self, sampled_rgb, sampled_gray, roi_payload): if self.config.detection_method == "foreground_diff": return self._detect_presence_with_heuristic(sampled_gray, roi_payload) if self.config.enable_yolo_person_detection: try: return self._detect_presence_with_yolo(sampled_rgb, sampled_gray, roi_payload) except Exception as exc: if not self.config.allow_person_detection_fallback: raise RuntimeError( "YOLO 人体检测依赖不可用，请先安装 `ultralytics`、`torch`，" "或将 `detection_method` 设为 'foreground_diff' 再运行。" ) from exc return self._detect_presence_with_heuristic(sampled_gray, roi_payload) return self._detect_presence_with_heuristic(sampled_gray, roi_payload) def detect(self, video_path: str, roi_payload=None) -> dict: metadata = probe_video(video_path) sampled_rgb = sample_rgb_frames( video_path=video_path, sample_fps=self.config.sample_fps, sample_width=self.config.sample_width, metadata=metadata, ) if sampled_rgb.shape[0] < 2: return { "metadata": metadata, "score_series": [], "windows": [], "presence_flags": [], "near_cabinet_flags": [], "motion_scores": [], "cabinet_activity_scores": [], }

设计意图：检测策略模块化，让精度模式和资源模式可热切换。
风险点：回退路径如果无日志，线上会出现“悄悄降级”难排障问题。

# 节选自 vl_video/src/core/event_aggregator.py def aggregate(self, analyzed_segments): final_events = [] uncertain_events = [] take_count = 0 put_count = 0 sorted_segments = sorted(analyzed_segments, key=lambda item: item.get("start_sec", 0)) for segment in sorted_segments: take_count += int(segment.get("take_count", 0)) put_count += int(segment.get("put_count", 0)) duration = max(segment.get("end_sec", 0) - segment.get("start_sec", 0), 0.1) fps = segment.get("fps", 0) for event in segment.get("events", []): center_sec = segment["start_sec"] + duration * float(event.get("relative_position", 0.5)) event_record = { "label": event["label"], "confidence": event["confidence"], "evidence": event["evidence"], "description": event["description"], "start_sec": center_sec, "end_sec": center_sec, "start_frame": int(round(center_sec * fps)), "end_frame": int(round(center_sec * fps)), "source_window_ids": [segment["window_id"]], "debug_paths": [segment.get("clip_path", "")], } event_record["event_id"] = f"event_{len(final_events) + 1:03d}" final_events.append(event_record) final_events.sort(key=lambda e: e["start_sec"]) logic_trace = " ".join([f"{e['label']}@{e['start_sec']:.2f}s" for e in final_events]) return { "logic_trace": logic_trace, "take_battery_num": take_count, "put_battery_num": put_count, "events": final_events, "uncertain_events": uncertain_events, }

设计意图：把模型输出转换成审计友好的时间线证据。
风险点：若 fps/时间基准不一致，会导致跨窗口时间错位。

总结

这套主链路可复用到多数视频 AI 场景，关键不是框架名，而是设计原则：

1. 先降输入熵，再做语义判定。
2. 把模型不确定性留在系统内部消化。
3. 输出必须可复核，才能长期运营。