手把手教你用 RAG 技术实现长视频智能问答系统

前言

最近在做一个长视频分析项目，需要让用户能够通过自然语言查询视频内容。调研后发现 RAG（Retrieval-Augmented Generation）是非常适合的技术方案。经过一段时间的开发，终于完成了这个系统，今天来分享一下整个过程。

**
**

项目地址：https://github.com/DOGOLD/rag-video-qa

一、项目概述

1.1 解决什么问题？

想象一下：

  • 你有一个 2 小时的教学视频，想快速找到某个知识点的位置
  • 你想了解视频中提到的某个概念的具体解释
  • 你想对视频内容进行深度问答，但不想看完整视频

传统的解决方案是人工标注或者文本搜索，但效率很低。而我的系统可以：

  • ✅ 自动转录视频音频为文本
  • ✅ 智能分割视频场景
  • ✅ 构建向量索引实现语义搜索
  • ✅ 基于视频内容进行智能问答

1.2 核心特性

  • 🎥 多源视频输入：支持本地视频和 YouTube 链接
  • 🔊 智能语音转录：使用 OpenAI Whisper 模型
  • 👁️ 帧级特征提取：使用 CLIP 模型
  • 📖 自动场景分割：基于内容相似度
  • 🔍 向量语义检索：使用 ChromaDB
  • 💬 RAG 问答：结合 GPT 模型生成答案
  • 🐳 Docker 部署：一行命令启动服务

二、技术架构

2.1 整体流程

视频输入 → 音频转录 → 帧提取 → 场景分割 → 向量索引 → 用户问答
   ↓         ↓          ↓         ↓          ↓         ↓
 YouTube   Whisper    CLIP     聚类算法   ChromaDB   GPT-4
  /本地                    算法                API

**我用的是中转站
**
这里我使用的是自己的中转站api

2.2 核心技术栈

模块 技术选型 说明
音视频处理 FFmpeg 视频帧提取、音频分离
语音转录 Whisper OpenAI 开源的语音识别模型
图像嵌入 CLIP OpenAI 的图文对比学习模型
向量数据库 ChromaDB 轻量级向量检索数据库
LLM OpenAI API GPT-3.5/4 生成答案
前端框架 Streamlit 快速构建数据应用
视频下载 yt-dlp 支持 YouTube 等多平台

2.3 目录结构

rag-video-qa/
├── main.py              # 主应用入口
├── audio_transcriber.py # 音频转录模块
├── video_downloader.py  # 视频下载模块
├── frame_processor.py   # 帧处理模块
├── scene_segmenter.py   # 场景分割模块
├── vector_index.py      # 向量索引模块
├── qa_engine.py         # 问答引擎模块
├── config.py            # 配置文件
├── requirements.txt     # 依赖列表
├── Dockerfile           # Docker 配置
└── docker-compose.yml   # 容器编排配置

三、核心功能实现

3.1 音频转录

使用 Whisper 模型进行语音识别。关键点是需要将长音频分段处理，因为 Whisper 单次处理有限制。

# audio_transcriber.py
def transcribe_audio(self, audio_path, language="zh"):
    # 将音频分段，每段30秒
    segments = self._split_audio(audio_path, segment_duration=30)

    all_results = []
    for i, segment in enumerate(segments):
        # 使用 Whisper 转录
        result = self.model.transcribe(
            segment,
            language=language,
            task="transcribe"
        )
        all_results.extend(result["segments"])

    return self._merge_segments(all_results)

**
**

3.2 帧特征提取

使用 CLIP 模型提取视频帧的视觉特征：

# frame_processor.py
def extract_frame_embeddings(self, video_path):
    frames = self._extract_frames(video_path)
    embeddings = []

    for frame in frames:
        # CLIP 图像编码
        img = Image.fromarray(frame)
        img_input = self.preprocess(img).unsqueeze(0)

        with torch.no_grad():
            features = self.clip_model.encode_image(img_input)

        embeddings.append(features.numpy().tolist())

    return embeddings

3.3 场景分割

基于转录文本和帧嵌入，使用聚类算法分割场景：

# scene_segmenter.py
from sklearn.cluster import AgglomerativeClustering

def segment_scenes(self, transcripts, frame_embeddings):
    # 构建相似度矩阵
    similarity_matrix = self._build_similarity_matrix(
        transcripts,
        frame_embeddings
    )

    # 层次聚类
    n_scenes = max(1, len(transcripts) // 10)  # 根据内容长度确定场景数
    clustering = AgglomerativeClustering(
        n_clusters=n_scenes,
        metric='precomputed',
        linkage='average'
    )
    labels = clustering.fit_predict(1 - similarity_matrix)

    # 生成场景列表
    scenes = self._create_scenes(transcripts, frame_embeddings, labels)
    return scenes

3.4 向量索引构建

使用 ChromaDB 存储和检索向量：

# vector_index.py
import chromadb

class VectorIndex:
    def __init__(self, index_dir):
        self.client = chromadb.PersistentClient(path=index_dir)
        self.collection = self.client.get_or_create_collection(
            name="video_scenes"
        )

    def add_scenes(self, scenes):
        embeddings = [s["frame_embedding"] for s in scenes]
        metadatas = [{
            "scene_id": int(s["scene_id"]),
            "start": float(s["start"]),
            "end": float(s["end"]),
            "text": str(s["text"])
        } for s in scenes]

        self.collection.add(
            embeddings=embeddings,
            metadatas=metadatas,
            ids=[f"scene_{i}" for i in range(len(scenes))]
        )

    def query(self, query_embedding, top_k=3):
        return self.collection.query(
            query_embeddings=[query_embedding],
            n_results=top_k
        )

四、踩坑与解决方案

这部分是本文最有价值的部分，记录了我在开发过程中遇到的各种问题及解决方案。

4.1 Whisper 长音频处理失败

问题描述：

ValueError: audio features have variable timesteps

原因分析：
Whisper 模型默认只能处理约 30 秒的音频，如果直接传入长音频会报错。

解决方案：
将音频分段处理：

def _split_audio(self, audio_path, segment_duration=30):
    """将长音频分割为小段"""
    audio = AudioSegment.from_file(audio_path)
    segments = []

    for i in range(0, len(audio), segment_duration * 1000):
        segment = audio[i:i + segment_duration * 1000]
        temp_path = f"/tmp/segment_{i}.wav"
        segment.export(temp_path, format="wav")
        segments.append(temp_path)

    return segments

4.2 帧嵌入与文本嵌入维度不匹配

问题描述：

ValueError: operands could not be broadcast together

原因分析：
TF-IDF 生成的文本嵌入是低维的（如 14 维），而 CLIP 帧嵌入是高维的（768 维），无法直接合并。

解决方案：
简化设计，仅使用帧嵌入进行检索：

def add_scenes(self, scenes, text_embedding_model=None):
    for scene in scenes:
        if scene["frame_embedding"] is not None:
            emb = scene["frame_embedding"]
        else:
            continue

        self.collection.add(
            embeddings=[emb.tolist()],
            metadatas=[{
                "scene_id": int(scene["scene_id"]),
                "start": float(scene["start"]),
                "end": float(scene["end"]),
                "text": str(scene["text"])
            }],
            ids=[f"scene_{scene['scene_id']}"]
        )

4.3 ChromaDB 数据类型错误

问题描述：

ValueError: Expected metadata value to be a str, int, float, bool...
got 7 which is a int64

原因分析：
NumPy 的 int64 类型不被 ChromaDB 支持，需要转换为 Python 原生类型。

解决方案：
显式类型转换：

metadatas.append({
    "scene_id": int(scene["scene_id"]),    # 转换 int64 → int
    "start": float(scene["start"]),        # 转换 float64 → float
    "end": float(scene["end"]),
    "text": str(scene["text"])
})

4.4 YouTube 下载被识别为机器人

问题描述：

ERROR: Sign in to confirm you're not a bot

原因分析：
YouTube 的反爬机制越来越严格，yt-dlp 会被识别为机器人。

解决方案：
支持上传 cookies 文件绕过验证：

def download_youtube_video(url, output_dir, cookies_file=None):
    ydl_opts = {
        'format': 'bestvideo[ext=mp4]+bestaudio[ext=m4a]/best',
        'outtmpl': os.path.join(output_dir, '%(title)s.%(ext)s'),
    }

    if cookies_file and os.path.exists(cookies_file):
        ydl_opts['cookiefile'] = cookies_file

    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
        info = ydl.extract_info(url, download=True)
        return ydl.prepare_filename(info)

4.5 sklearn API 兼容性

问题描述：

TypeError: AgglomerativeClustering.__init__() got an
unexpected keyword argument 'affinity'

原因分析：
新版本 scikit-learn 将 affinity 参数重命名为 metric。

解决方案：

# 旧版本
clustering = AgglomerativeClustering(
    n_clusters=n_scenes,
    affinity='precomputed',
    linkage='average'
)

# 新版本
clustering = AgglomerativeClustering(
    n_clusters=n_scenes,
    metric='precomputed',  # 注意这里
    linkage='average'
)

五、Docker 部署

5.1 为什么用 Docker？

部署 AI 应用的环境配置非常复杂：

  • 需要 Python 3.10+
  • 需要 FFmpeg
  • 需要下载 Whisper、CLIP 等大模型
  • 模型文件可能达到数 GB

使用 Docker 可以一键安装所有依赖，而且环境隔离，不影响主机。

5.2 Dockerfile

FROM python:3.10-slim

WORKDIR /app

# 安装系统依赖
RUN apt-get update && apt-get install -y \
    ffmpeg git gcc g++ python3-dev && \
    rm -rf /var/lib/apt/lists/*

# 安装 Python 依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 下载预训练模型
RUN python -c "import whisper; whisper.load_model('base')" && \
    python -c "import clip; clip.load('ViT-L/14')"

# 复制应用代码
COPY . .

EXPOSE 8501

CMD ["streamlit", "run", "main.py", "--server.port=8501"]

5.3 docker-compose.yml

version: '3.8'

services:
  rag-video-qa:
    build: .
    container_name: rag-video-qa
    ports:
      - "8501:8501"
    volumes:
      - ./data:/app/data
      - ./models:/app/models
    environment:
      - TRANSFORMERS_CACHE=/app/models
    restart: unless-stopped

5.4 一键启动

# 构建并启动
docker-compose up -d

# 查看日志
docker-compose logs -f

# 停止服务
docker-compose down

六、使用指南

6.1 环境准备

方式一：Docker 部署（推荐）

git clone https://github.com/DOGOLD/rag-video-qa.git
cd rag-video-qa
docker-compose up -d

方式二：本地运行

pip install -r requirements.txt
streamlit run main.py

6.2 配置 API Key

在左侧边栏输入 OpenAI API Key：

  • API Key：你的 OpenAI 密钥
  • Base URL：API 地址（使用代理服务时需要修改）
  • 模型选择：系统会自动获取可用模型列表

**在这里插入图片描述

6.3 上传视频并处理

  1. 选择输入方式（本地文件 / YouTube 链接）
  2. 上传或输入视频
  3. 点击「开始处理」
  4. 等待处理完成（根据视频长度可能需要几分钟）

6.4 开始问答

处理完成后，在问答区域输入问题：

**在这里插入图片描述
**

七、性能优化建议

7.1 GPU 加速

如果有 NVIDIA 显卡，开启 GPU 加速可以大幅提升性能：

# config.py
USE_GPU = True  # 开启 GPU

# Whisper 使用 GPU
model = whisper.load_model("base", device="cuda")

# CLIP 使用 GPU
clip_model, _ = clip.load("ViT-L/14", device="cuda")

7.2 模型选择权衡

模型 参数量 速度 准确性 内存需求
Tiny 39M ⚡⚡⚡ ⭐ ~1GB
Base 74M ⚡⚡ ⭐⭐ ~1GB
Small 244M ⚡ ⭐⭐⭐ ~2GB
Medium 769M 🔄 ⭐⭐⭐⭐ ~5GB
Large 1550M 🐢 ⭐⭐⭐⭐⭐ ~10GB

建议：中文视频使用 base 或 small，英文视频可以用 medium。

7.3 处理速度参考

视频时长 Whisper 转录 CLIP 提取 场景分割 总计
5 分钟 ~30s ~2min ~10s ~3min
30 分钟 ~3min ~12min ~30s ~16min
1 小时 ~6min ~25min ~1min ~32min

八、未来展望

目前系统还有一些可以改进的地方：

8.1 短期优化

  • 添加批处理支持，同时处理多个视频
  • 支持更多视频平台（Bilibili、抖音等）
  • 添加视频摘要自动生成功能

8.2 长期规划

  • WebRTC 实时问答
  • 多语言翻译支持
  • 视频内容知识图谱构建
  • 对话式交互界面

九、总结

本文详细介绍了基于 RAG 技术的长视频问答系统的开发过程，包括：

  1. 技术架构：使用 Whisper + CLIP + ChromaDB + GPT 的组合
  2. 核心实现：音频转录、帧提取、场景分割、向量检索
  3. 踩坑记录：详细记录了开发过程中遇到的问题和解决方案
  4. Docker 部署：一键启动服务的完整方案

项目已经开源到 GitHub，欢迎大家使用和交流。如果有问题或建议，可以在 GitHub 提 Issue。

参考资料

【全文完】

如果觉得文章有帮助，欢迎：

有任何问题可以在评论区留言，我会尽量解答！

# 音视频
Logo
DeepSeek技术社区

欢迎加入DeepSeek 技术社区。在这里，你可以找到志同道合的朋友，共同探索AI技术的奥秘。

更多推荐

Cursor Free VIP破解工具：终极免费Pro功能一键解锁指南

还在为Cursor的"试用请求已达上限"而烦恼吗？你是否经常遇到"本机使用过多免费试用账户"的提示？别担心，今天我要介绍的这个开源工具能帮你轻松解决这些问题！Cursor Free VIP是一款功能强大的破解工具，通过智能重置机器ID和修改授权状态，让你免费享受Cursor Pro的全部高级功能。这个工具支持Windows、macOS和Linux三大操作系统，提供多语言界面，操作简单快捷，让AI编

avatar DeepSeek技术社区

CC-Switch 2026最新完整版 下载+安装+全功能配置教程

供应商名称：自定义备注（如XX中转）Base URL：平台提供地址（结尾带/v1）API Key：中转平台密钥模型名称：填写平台支持代码模型名勾选「适配Claude协议路由」，保存启用。

avatar DeepSeek技术社区
cover

AI第一性原理：从一个神经元到ChatGPT的硬核拆解

avatar DeepSeek技术社区