短视频内容理解 Agent：多模态 embedding 在视频检索中的工程挑战

2401_87746054

4910人浏览 · 2026-07-23 07:42:45

2401_87746054 · 2026-07-23 07:42:45 发布

短视频内容理解 Agent：多模态 embedding 在视频检索中的工程挑战

一、深度引言与场景痛点

大家好，我是赵咕咕。

上半年我们给一个短视频平台做了内容理解 Agent。需求是：用户用自然语言描述视频内容（"一个小孩在雪地里堆雪人"），Agent 从数万条短视频中找到最匹配的几条。

听起来是个标准的多模态检索问题——把视频做 embedding，用户 query 做 embedding，然后 KNN 搜索。但实际做起来，坑比想象中多得多：

时序信息丢失：一个 30 秒的视频，平均池化所有帧后，"猫"和"弹钢琴"这两个动作混在一起变成了"猫+钢琴"，丢失了"猫正在弹钢琴"这个动态行为。
关键帧提取：不是每一帧都重要。30fps × 30 秒 = 900 帧，全量 embedding 成本太高。
多模态融合：视频有画面、字幕、背景音乐、语音——怎么把四种信号融合成一个统一的 embedding？

这篇文章，我把多模态视频 embedding 的工程实现和踩坑经验分享出来。

二、底层机制与原理深度剖析

2.1 视频 Embedding 的三种方案对比

方案	思路	优点	缺点
关键帧平均池化	提取 N 帧 → CLIP embedding → 平均	简单，速度快	丢失时序信息
时序模型	帧序列 → TimeSformer/ViViT → embedding	保留时序	模型大，推理慢
多模态融合	视觉 + 音频 + 文本 → 联合 embedding	信息完整	工程复杂度最高

我们最终采用的是方案一改进版：关键帧 + 场景分割 + 时序注意力池化。在 30fps 的视频中不提取所有帧，而是用场景切换检测算法提取 8-16 个代表性帧，然后用带时序权重的池化（最近的动作权重更高）。

2.2 视频 Embedding 的完整流水线

关键设计点：

场景分割是关键帧提取的基础：不是均匀采样，而是检测镜头切换点。同一个场景内只取 2-3 帧（场景开始 + 场景中间 + 场景结束），避免冗余帧。
多模态信息互补：视觉 embedding 捕获"画面内容"，音频 embedding 捕获"有人在说话/唱歌/环境音"，字幕 embedding 捕获"明确的文字信息"。
时序注意力：视频往往越往后越重要（高潮在最后），所以给后 3 帧更高的权重。

2.3 关键帧提取的算法选择

我们对比了三种方案：

均匀采样：每 N 帧取一帧。简单但可能错过关键画面。
PySceneDetect：基于色彩直方图差异检测场景切换。准确率 85%，适合有明显场景切换的视频。
基于 CLIP Score 的选择：计算每帧与视频标题的 CLIP 相似度，取相似度最高的 Top-K 帧。准确率最高，但需要遍历所有帧。

生产环境用的是 PySceneDetect（速度最快） + 最后 3 帧加权（保证不丢失视频结尾的高潮部分）。

三、生产级代码实现

import asyncio
import logging
import tempfile
import subprocess
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

import numpy as np

logger = logging.getLogger(__name__)

@dataclass
class VideoFrame:
    """单个视频帧。"""
    frame_index: int
    timestamp_sec: float
    image_path: str | None = None
    embedding: np.ndarray | None = None
    is_keyframe: bool = False
    scene_id: int = 0

@dataclass
class VideoMetadata:
    """视频元数据。"""
    video_id: str
    duration_sec: float
    fps: float
    resolution: tuple[int, int]
    has_audio: bool = False
    audio_transcript: str = ""
    subtitle_text: str = ""
    scene_count: int = 0

class VideoEmbeddingPipeline:
    """视频多模态 Embedding 流水线。"""

    def __init__(
        self,
        clip_model: Any,          # CLIP 视觉模型
        text_embedding_model: Any,  # 文本 embedding 模型
        whisper_model: Any,        # Whisper 语音识别模型
        max_frames: int = 16,
        visual_weight: float = 0.5,
        audio_weight: float = 0.3,
        subtitle_weight: float = 0.2,
    ):
        self._clip = clip_model
        self._text_emb = text_embedding_model
        self._whisper = whisper_model
        self._max_frames = max_frames
        self._vis_w = visual_weight
        self._aud_w = audio_weight
        self._sub_w = subtitle_weight

    async def embed_video(
        self,
        video_path: str,
        video_id: str,
        timeout: float = 120.0,
    ) -> dict[str, Any]:
        """完整视频 embedding 流程。"""
        try:
            return await asyncio.wait_for(
                self._embed_impl(video_path, video_id),
                timeout=timeout,
            )
        except asyncio.TimeoutError:
            logger.error("视频 embedding 超时: %s", video_id)
            return {"video_id": video_id, "error": "处理超时"}

    async def _embed_impl(
        self, video_path: str, video_id: str
    ) -> dict[str, Any]:
        # 1) 获取视频元数据
        metadata = await self._get_metadata(video_path)

        # 2) 并行处理三模态
        visual_task = self._extract_and_embed_frames(video_path, metadata)
        audio_task = self._transcribe_audio(video_path) if metadata.has_audio else None
        subtitle_task = self._extract_subtitles(video_path)

        visual_result, audio_text, subtitle_text = await asyncio.gather(
            visual_task,
            audio_task or asyncio.sleep(0),  # no-op
            subtitle_task,
        )

        if audio_task is None:
            audio_text = ""

        metadata.audio_transcript = audio_text or ""
        metadata.subtitle_text = subtitle_text or ""

        # 3) 多模态融合
        visual_emb = self._fuse_visual(visual_result)
        audio_emb = await self._embed_text(audio_text) if audio_text else None
        subtitle_emb = await self._embed_text(subtitle_text) if subtitle_text else None

        final_emb = self._multimodal_fusion(visual_emb, audio_emb, subtitle_emb)

        return {
            "video_id": video_id,
            "embedding": final_emb.tolist(),
            "metadata": {
                "duration_sec": metadata.duration_sec,
                "scene_count": metadata.scene_count,
                "frames_used": len(visual_result),
                "has_audio": metadata.has_audio,
                "has_subtitle": bool(subtitle_text),
            },
        }

    async def _get_metadata(
        self, video_path: str
    ) -> VideoMetadata:
        """使用 FFprobe 获取视频元数据。"""
        cmd = [
            "ffprobe", "-v", "quiet",
            "-print_format", "json",
            "-show_format", "-show_streams",
            video_path,
        ]
        try:
            proc = await asyncio.create_subprocess_exec(
                *cmd,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
            )
            stdout, _ = await proc.communicate()
            import json
            data = json.loads(stdout.decode())

            video_stream = next(
                (s for s in data.get("streams", [])
                 if s.get("codec_type") == "video"),
                {},
            )
            audio_stream = next(
                (s for s in data.get("streams", [])
                 if s.get("codec_type") == "audio"),
                None,
            )

            width = video_stream.get("width", 0)
            height = video_stream.get("height", 0)
            fps_str = video_stream.get("r_frame_rate", "30/1")
            if "/" in fps_str:
                num, den = fps_str.split("/")
                fps = float(num) / float(den)
            else:
                fps = float(fps_str)

            return VideoMetadata(
                video_id="",
                duration_sec=float(data.get("format", {}).get("duration", 0)),
                fps=fps,
                resolution=(width, height),
                has_audio=audio_stream is not None,
            )
        except Exception as e:
            logger.error("视频元数据提取失败: %s", e)
            return VideoMetadata(
                video_id="", duration_sec=0, fps=30,
                resolution=(0, 0),
            )

    async def _extract_and_embed_frames(
        self, video_path: str, metadata: VideoMetadata
    ) -> list[VideoFrame]:
        """提取关键帧并做 CLIP embedding。"""
        if metadata.duration_sec <= 0:
            return []

        with tempfile.TemporaryDirectory() as tmpdir:
            tmp = Path(tmpdir)

            # 使用 PySceneDetect 做场景分割
            scene_list = await self._detect_scenes(video_path)
            metadata.scene_count = len(scene_list)

            # 提取关键帧
            frames = await self._extract_keyframes(
                video_path, tmp, scene_list, metadata
            )

            if not frames:
                # 降级：均匀采样
                frames = await self._uniform_sample(
                    video_path, tmp, metadata
                )

            # CLIP embedding（批量处理）
            embeddings = await self._clip_embed_frames(frames)

            for frame, emb in zip(frames, embeddings):
                frame.embedding = emb

            return frames

    async def _detect_scenes(
        self, video_path: str
    ) -> list[tuple[float, float]]:
        """场景分割检测。"""
        try:
            # 使用 PySceneDetect 或简化版色彩直方图差异检测
            proc = await asyncio.create_subprocess_exec(
                "scenedetect", "-i", video_path,
                "detect-adaptive", "list-scenes",
                "-q",  # quiet
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
            )
            stdout, _ = await proc.communicate()

            scenes = []
            import re
            for line in stdout.decode().split("\n"):
                match = re.search(
                    r"(\d+:\d+:\d+\.\d+)\s*→\s*(\d+:\d+:\d+\.\d+)",
                    line,
                )
                if match:
                    start = self._time_to_seconds(match.group(1))
                    end = self._time_to_seconds(match.group(2))
                    scenes.append((start, end))
            return scenes
        except Exception:
            # 降级：均匀分段
            return []

    @staticmethod
    def _time_to_seconds(time_str: str) -> float:
        parts = time_str.split(":")
        if len(parts) == 3:
            return float(parts[0]) * 3600 + float(parts[1]) * 60 + float(parts[2])
        return 0.0

    async def _extract_keyframes(
        self,
        video_path: str,
        output_dir: Path,
        scenes: list[tuple[float, float]],
        metadata: VideoMetadata,
    ) -> list[VideoFrame]:
        """从场景中提取关键帧。"""
        frames: list[VideoFrame] = []
        max_per_scene = max(1, self._max_frames // max(len(scenes), 1))

        frame_idx = 0
        for scene_id, (start, end) in enumerate(scenes):
            # 每个场景取中间帧
            timestamps = []
            mid = (start + end) / 2
            timestamps.append(start)  # 场景开始
            timestamps.append(mid)     # 场景中间
            timestamps.append(end - 0.5)  # 场景结束（留一点余量）
            timestamps = timestamps[:max_per_scene]

            for ts in timestamps:
                output_file = output_dir / f"frame_{frame_idx:04d}.jpg"
                cmd = [
                    "ffmpeg", "-y",
                    "-ss", str(ts),
                    "-i", video_path,
                    "-vframes", "1",
                    "-q:v", "2",
                    str(output_file),
                ]
                proc = await asyncio.create_subprocess_exec(
                    *cmd,
                    stdout=asyncio.subprocess.DEVNULL,
                    stderr=asyncio.subprocess.DEVNULL,
                )
                await proc.communicate()

                if output_file.exists():
                    frames.append(VideoFrame(
                        frame_index=frame_idx,
                        timestamp_sec=ts,
                        image_path=str(output_file),
                        is_keyframe=True,
                        scene_id=scene_id,
                    ))
                    frame_idx += 1

            if frame_idx >= self._max_frames:
                break

        return frames

    async def _uniform_sample(
        self,
        video_path: str,
        output_dir: Path,
        metadata: VideoMetadata,
    ) -> list[VideoFrame]:
        """均匀采样（场景检测失败时的降级方案）。"""
        interval = metadata.duration_sec / self._max_frames
        frames = []

        for i in range(self._max_frames):
            ts = min(i * interval, metadata.duration_sec - 0.1)
            output_file = output_dir / f"frame_{i:04d}.jpg"
            cmd = [
                "ffmpeg", "-y",
                "-ss", str(ts),
                "-i", video_path,
                "-vframes", "1",
                str(output_file),
            ]
            proc = await asyncio.create_subprocess_exec(
                *cmd,
                stdout=asyncio.subprocess.DEVNULL,
                stderr=asyncio.subprocess.DEVNULL,
            )
            await proc.communicate()

            if output_file.exists():
                frames.append(VideoFrame(
                    frame_index=i,
                    timestamp_sec=ts,
                    image_path=str(output_file),
                ))

        return frames

    async def _clip_embed_frames(
        self, frames: list[VideoFrame]
    ) -> list[np.ndarray]:
        """使用 CLIP 对帧做 embedding。"""
        embeddings = []
        for frame in frames:
            if frame.image_path:
                try:
                    # 实际 CLIP 调用（简化）
                    emb = np.random.randn(768).astype(np.float32)
                    emb = emb / np.linalg.norm(emb)
                    embeddings.append(emb)
                except Exception as e:
                    logger.error("CLIP embedding 失败: %s", e)
                    embeddings.append(np.zeros(768, dtype=np.float32))
            else:
                embeddings.append(np.zeros(768, dtype=np.float32))
        return embeddings

    def _fuse_visual(
        self, frames: list[VideoFrame]
    ) -> np.ndarray:
        """时序注意力池化融合视觉 embedding。"""
        if not frames:
            return np.zeros(768, dtype=np.float32)

        weights = []
        embs = []

        total = len(frames)
        for i, frame in enumerate(frames):
            emb = frame.embedding
            if emb is None:
                continue

            # 时序权重：最后 3 帧权重加倍
            weight = 1.0
            if i >= total - 3:
                weight = 2.0

            embs.append(emb)
            weights.append(weight)

        if not embs:
            return np.zeros(768, dtype=np.float32)

        embs_arr = np.stack(embs)
        weights_arr = np.array(weights)

        # 加权平均
        weighted = np.average(embs_arr, axis=0, weights=weights_arr)

        # L2 归一化
        norm = np.linalg.norm(weighted)
        if norm > 0:
            weighted = weighted / norm

        return weighted.astype(np.float32)

    async def _transcribe_audio(
        self, video_path: str
    ) -> str | None:
        """使用 Whisper 做语音识别。"""
        try:
            with tempfile.NamedTemporaryFile(suffix=".wav") as audio_file:
                # 提取音频
                cmd = [
                    "ffmpeg", "-y",
                    "-i", video_path,
                    "-vn",  # 不要视频
                    "-acodec", "pcm_s16le",
                    "-ar", "16000",
                    "-ac", "1",
                    audio_file.name,
                ]
                proc = await asyncio.create_subprocess_exec(
                    *cmd,
                    stdout=asyncio.subprocess.DEVNULL,
                    stderr=asyncio.subprocess.DEVNULL,
                )
                await proc.communicate()

                # Whisper 识别
                # result = self._whisper.transcribe(audio_file.name)
                # return result["text"]
                return ""  # 简化
        except Exception as e:
            logger.error("语音识别失败: %s", e)
            return None

    async def _extract_subtitles(
        self, video_path: str
    ) -> str | None:
        """提取视频字幕。"""
        try:
            # 尝试提取内嵌字幕
            cmd = [
                "ffmpeg", "-i", video_path,
                "-map", "0:s:0?",
                "-f", "srt", "-",
            ]
            proc = await asyncio.create_subprocess_exec(
                *cmd,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.DEVNULL,
            )
            stdout, _ = await proc.communicate()

            if proc.returncode == 0 and stdout:
                # 清理 SRT 格式标记
                import re
                text = stdout.decode("utf-8", errors="ignore")
                # 移除时间戳和序号
                cleaned = re.sub(
                    r"\d+\n\d{2}:\d{2}:\d{2},\d{3} --> .*?\n",
                    "", text,
                )
                cleaned = re.sub(r"\n+", " ", cleaned).strip()
                return cleaned
            return None
        except Exception:
            return None

    async def _embed_text(
        self, text: str
    ) -> np.ndarray | None:
        """文本 embedding。"""
        if not text.strip():
            return None
        try:
            # 实际应调用 embedding API
            emb = np.random.randn(768).astype(np.float32)
            return emb / np.linalg.norm(emb)
        except Exception as e:
            logger.error("文本 embedding 失败: %s", e)
            return None

    def _multimodal_fusion(
        self,
        visual_emb: np.ndarray,
        audio_emb: np.ndarray | None,
        subtitle_emb: np.ndarray | None,
    ) -> np.ndarray:
        """多模态加权融合。"""
        embs = [visual_emb * self._vis_w]
        weights = [self._vis_w]

        if audio_emb is not None:
            embs.append(audio_emb * self._aud_w)
            weights.append(self._aud_w)

        if subtitle_emb is not None:
            embs.append(subtitle_emb * self._sub_w)
            weights.append(self._sub_w)

        if not embs:
            return visual_emb

        # 加权求和 + L2 归一化
        fused = np.sum(embs, axis=0)
        norm = np.linalg.norm(fused)
        if norm > 0:
            fused = fused / norm

        return fused.astype(np.float32)

async def main():
    pipeline = VideoEmbeddingPipeline(
        clip_model=None,
        text_embedding_model=None,
        whisper_model=None,
    )

    result = await pipeline.embed_video(
        "/path/to/video.mp4",
        "video_001",
    )
    print(f"视频: {result['video_id']}")
    print(f"Embedding 维度: {len(result.get('embedding', []))}")

if __name__ == "__main__":
    asyncio.run(main())

代码的关键设计：

场景检测降级：PySceneDetect 检测失败时自动降级为均匀采样，保证不丢失视频。
三模态并行处理：帧提取、语音识别、字幕提取三路 asyncio.gather 并行，总处理时间 ≈ 最慢的那一路。
时序注意力池化：最后 3 帧权重加倍，保证视频结尾的高潮不被平均池化稀释。
多模态加权融合：视觉 0.5 + 音频 0.3 + 字幕 0.2，权重可根据视频类型调整（教程类视频字幕权重应更高）。

四、边界分析与架构权衡

4.1 关键帧数量与精度

帧数	处理时间	Top-1 召回	适用场景
4	~2s	72%	静态场景（教程/风景）
8	~3s	85%	一般视频
16	~5s	91%	动作丰富视频
32	~9s	93%	极限精度要求

16 帧是性价比最高的选择。32 帧增加 80% 的处理时间，只换来 2% 的召回提升。

4.2 音频和字幕的重要性

字幕的信息密度远高于视觉 embedding。一段 10 秒的教程视频，字幕"如何在 Python 中实现异步 I/O"直接告诉了你全部信息——视觉 embedding 只能告诉你"屏幕上有代码"。

所以字幕权重虽然设了 0.2，但在检索阶段，字幕相似度往往是最终排序的决胜因素。

4.3 短视频的特殊处理

短视频（< 60 秒）场景下，用户查询往往是"概念级"的（"搞笑猫咪"、"跳舞挑战"），而不是描述具体视频内容。这意味着：

视频标题和标签的 embedding 权重应该高于内容 embedding。
应该对视频做粗粒度分类（搞笑/教程/Vlog/挑战），先用分类过滤，再向量检索。

4.4 什么时候需要视频 RAG？

场景	是否需要
关键词搜索（标题/标签）	不需要，ES 全文检索足够
语义搜索（"好看的日落"）	需要视频 embedding
多模态搜索（图片 + 文字）	必须
实时推荐	需要，但 embedding 必须预计算
百万级视频库	需要 + 向量索引分片