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

手把手教你用 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 核心技术栈

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.pydeftranscribe_audio(self,audio_path,language="zh"):# 将音频分段，每段30秒segments=self._split_audio(audio_path,segment_duration=30)all_results=[]fori,segmentinenumerate(segments):# 使用 Whisper 转录result=self.model.transcribe(segment,language=language,task="transcribe")all_results.extend(result["segments"])returnself._merge_segments(all_results)

**
**

3.2 帧特征提取

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

# frame_processor.pydefextract_frame_embeddings(self,video_path):frames=self._extract_frames(video_path)embeddings=[]forframeinframes:# CLIP 图像编码img=Image.fromarray(frame)img_input=self.preprocess(img).unsqueeze(0)withtorch.no_grad():features=self.clip_model.encode_image(img_input)embeddings.append(features.numpy().tolist())returnembeddings

3.3 场景分割

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

# scene_segmenter.pyfromsklearn.clusterimportAgglomerativeClusteringdefsegment_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)returnscenes

3.4 向量索引构建

使用 ChromaDB 存储和检索向量：

# vector_index.pyimportchromadbclassVectorIndex:def__init__(self,index_dir):self.client=chromadb.PersistentClient(path=index_dir)self.collection=self.client.get_or_create_collection(name="video_scenes")defadd_scenes(self,scenes):embeddings=[s["frame_embedding"]forsinscenes]metadatas=[{"scene_id":int(s["scene_id"]),"start":float(s["start"]),"end":float(s["end"]),"text":str(s["text"])}forsinscenes]self.collection.add(embeddings=embeddings,metadatas=metadatas,ids=[f"scene_{i}"foriinrange(len(scenes))])defquery(self,query_embedding,top_k=3):returnself.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=[]foriinrange(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)returnsegments

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

问题描述：

ValueError: operands could not be broadcast together

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

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

defadd_scenes(self,scenes,text_embedding_model=None):forsceneinscenes:ifscene["frame_embedding"]isnotNone:emb=scene["frame_embedding"]else:continueself.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 文件绕过验证：

defdownload_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'),}ifcookies_fileandos.path.exists(cookies_file):ydl_opts['cookiefile']=cookies_filewithyt_dlp.YoutubeDL(ydl_opts)asydl:info=ydl.extract_info(url,download=True)returnydl.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-qaports:-"8501:8501"volumes:-./data:/app/data-./models:/app/modelsenvironment:-TRANSFORMERS_CACHE=/app/modelsrestart:unless-stopped

5.4 一键启动

# 构建并启动docker-composeup-d# 查看日志docker-composelogs-f# 停止服务docker-composedown

六、使用指南

6.1 环境准备

方式一：Docker 部署（推荐）

gitclone https://github.com/DOGOLD/rag-video-qa.gitcdrag-video-qadocker-composeup-d

方式二：本地运行

pipinstall-rrequirements.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.pyUSE_GPU=True# 开启 GPU# Whisper 使用 GPUmodel=whisper.load_model("base",device="cuda")# CLIP 使用 GPUclip_model,_=clip.load("ViT-L/14",device="cuda")

7.2 模型选择权衡

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

7.3 处理速度参考

八、未来展望

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

8.1 短期优化

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

8.2 长期规划

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

九、总结

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

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

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

参考资料

• OpenAI Whisper
• OpenAI CLIP
• ChromaDB
• Streamlit 文档
• yt-dlp

【全文完】

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

• ⭐ Star 项目：https://github.com/DOGOLD/rag-video-qa
• 💬 留言讨论
• 🔄 转发分享

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