LlamaIndex VideoDB Retriever：基于 VideoDBRetriever 的视频片段 RAG 检索器 API 参考与实现解析-平芜编程栈

LlamaIndex VideoDB Retriever：基于 VideoDBRetriever 的视频片段 RAG 检索器 API 参考与实现解析

【免费下载链接】llama_indexLlamaIndex is the leading document agent and OCR platform项目地址: https://gitcode.com/GitHub_Trending/ll/llama_index

本文以 LlamaIndex 仓库中 VideoDB Retriever 的 API 参考文档为核心，系统讲解VideoDBRetriever的完整构造参数、检索执行流程与底层实现细节，并结合仓库中的源码、示例 Notebook 与测试用例，给出从单视频检索、场景（视觉）索引检索到多视频合集检索的可复制实战方案。读完后你可以直接基于 VideoDB 托管的视频索引，用 LlamaIndex 构建面向视频的 RAG 管线，并将检索结果转化为文本回答与可播放的视频片段流。

1. 文档定位：API 参考页与`VideoDBRetriever`模块的对应关系

仓库中的 API 参考页 videodb.md 使用 mkdocs-autorefs 风格声明了对llama_index.retrievers.videodb模块的文档化范围，指定成员为VideoDBRetriever：

::: llama_index.retrievers.videodb options: members: - VideoDBRetriever

这意味着该参考页描述的对象只有一个类：VideoDBRetriever，其真实定义位于独立集成包中：

类实现：base.py
包导出：__init__.py，其中__all__ = ["VideoDBRetriever"]
集成包说明：README.md
配套示例：videodb_retriever.ipynb

VideoDBRetriever是 LlamaIndex 核心检索器基类的子类：

class VideoDBRetriever(BaseRetriever): def __init__(self, api_key=None, collection="default", video=None, ...)

基类BaseRetriever定义在 base_retriever.py（第 34 行），它负责统一的retrieve()入口、回调管理器（CallbackManager）与对象映射。集成包的测试用例也验证了这一继承关系：

# tests/test_retrievers_videdb.py from llama_index.core.base.base_retriever import BaseRetriever from llama_index.retrievers.videodb import VideoDBRetriever def test_class(): names_of_base_classes = [b.__name__ for b in VideoDBRetriever.__mro__] assert BaseRetriever.__name__ in names_of_base_classes

测试文件路径：test_retrievers_videdb.py。

2. 安装与依赖前提

根据集成包 pyproject.toml 的实际声明（当前版本 0.5.0）：

[project] name = "llama-index-retrievers-videodb" version = "0.5.0" requires-python = ">=3.10,<4.0" dependencies = [ "videodb>=0.2.0", "llama-index-core>=0.13.0,<0.15", ]

安装命令（来自集成包 README）：

pip install llama-index llama-index-retrievers-videodb videodb

运行前提：

需要一个 VideoDB API Key。VideoDB 是一个面向视频内容的数据库，提供视频存储、索引（语义索引/场景索引）、搜索与流式播放能力；API Key 可从 VideoDB 控制台获取；
若要走 LlamaIndex 的回答合成环节（get_response_synthesizer默认使用 OpenAI），还需配置OPENAI_API_KEY；
检索器每次检索都会通过网络连接 VideoDB 服务，因此运行环境需要可达其 API 端点（支持通过base_url参数自定义端点，见下文）。

3. 构造参数完整参考

以下参数表完整来自 base.py 中VideoDBRetriever.__init__的签名（第 16–46 行），并逐项说明其源码行为：

参数	类型	默认值	说明
`api_key`	`Optional[str]`	`None`	VideoDB API 密钥。传`None`时自动回退读取环境变量`VIDEO_DB_API_KEY`；两者都缺失时直接抛出`Exception`，提示必须二选一提供
`collection`	`Optional[str]`	`"default"`	VideoDB 集合（collection）名称或 ID，作为检索范围
`video`	`Optional[str]`	`None`	视频 ID。指定后只对单条视频执行`video.search()`；不指定则对整个集合执行`coll.search()`
`score_threshold`	`Optional[float]`	`0.2`	相似度分数阈值，低于该值的片段不会被返回
`result_threshold`	`Optional[int]`	`5`	最多返回的结果数量
`search_type`	`Optional[str]`	`SearchType.semantic`	检索方式，来自`videodb`SDK，可选语义检索或关键词检索
`index_type`	`Optional[str]`	`IndexType.spoken_word`	目标索引类型，默认“口播/语音内容”索引，也可设为`IndexType.scene`（场景/视觉索引）
`scene_index_id`	`Optional[str]`	`None`	场景索引 ID。仅当`index_type == IndexType.scene`且该值非空时，才会作为`index_id`加入检索参数
`base_url`	`Optional[str]`	`None`	自定义 VideoDB API 端点；非`None`时会传入`connect()`建立连接
`callback_manager`	`Optional[CallbackManager]`	`None`	LlamaIndex 回调管理器，透传给基类`BaseRetriever.__init__`

API Key 的解析逻辑（源码第 31–36 行）：

if api_key is None: api_key = os.environ.get("VIDEO_DB_API_KEY") if api_key is None: raise Exception( "No API key provided. Set an API key either as an environment variable " "(VIDEO_DB_API_KEY) or pass it as an argument." )

因此两种等价写法：

# 方式一：环境变量 os.environ["VIDEO_DB_API_KEY"] = "your_api_key" # 方式二：显式传参 retriever = VideoDBRetriever(api_key="your_api_key")

4. 检索执行流程：`_retrieve`的实现剖析

VideoDBRetriever的核心逻辑全部在_retrieve(query_bundle)方法中（base.py）。其执行顺序为：

建立连接：每次检索时以api_key（及可选base_url）调用videodb.connect()建立新连接。从源码结构看，连接是逐次请求创建的，而非常驻客户端；
两条检索路径（由是否传入video决定）：
- 单视频路径：conn.get_collection(collection)→coll.get_video(video)→video.search(...)。检索参数包含query（取自query_bundle.query_str）、search_type、index_type、score_threshold、result_threshold；
- 集合路径：conn.get_collection(collection)→coll.search(query, ...)，跨集合内所有视频检索；
场景索引的特殊分支：仅当index_type == IndexType.scene且scene_index_id有值时，才把index_id=scene_index_id加入单视频检索参数。这解释了为什么场景检索必须先通过 VideoDB 的场景索引接口拿到索引 ID 再传给检索器；
结果到 LlamaIndex 节点的映射：遍历search_res.get_shots()，每个 shot 转成一个TextNode并包裹进NodeWithScore：

textnode = TextNode( text=shot.text, metadata={ "collection_id": collection_id, "video_id": shot.video_id, "length": shot.video_length, "title": shot.video_title, "start": shot.start, "end": shot.end, "type": self.index_type, }, ) nodes.append(NodeWithScore(node=textnode, score=score))

这套 metadata 是整个视频 RAG 的关键：text是片段的转写/描述文本（供 LLM 合成回答），start/end是该片段在视频中的时间区间（供生成可播放的视频剪辑流），video_id标识片段来源视频（跨视频检索时用于拼接 Timeline）。

5. 实战一：单视频视频 RAG（语音索引 + 场景索引）

以下流程对应官方示例 videodb_retriever.ipynb，可直接按步骤复制执行。

5.1 连接 VideoDB 并上传视频

from videodb import connect conn = connect() coll = conn.create_collection( name="VideoDB Retrievers", description="VideoDB Retrievers" ) # 支持公有 URL、YouTube 链接或本地文件 video = coll.upload(url="https://www.youtube.com/watch?v=aRgP3n0XiMc") print(f"Video uploaded with ID: {video.id}")

5.2 口播（spoken content）检索

先对视频做语音索引，再用index_type=IndexType.spoken_word（默认值）检索：

video.index_spoken_words() from llama_index.retrievers.videodb import VideoDBRetriever from videodb import SearchType, IndexType spoken_retriever = VideoDBRetriever( collection=coll.id, video=video.id, search_type=SearchType.semantic, index_type=IndexType.spoken_word, score_threshold=0.1, ) spoken_query = "Nationwide exams" nodes_spoken_index = spoken_retriever.retrieve(spoken_query)

用 LlamaIndex 的回答合成器基于检索节点生成文本：

from llama_index.core import get_response_synthesizer response_synthesizer = get_response_synthesizer() response = response_synthesizer.synthesize(spoken_query, nodes=nodes_spoken_index) print(response)

5.3 视频片段流：利用 metadata 中的时间区间

每个检索节点 metadata 中的start/end表示该片段时间区间。利用 VideoDB 的可编程流式接口，可以即时拼接出与检索结果一一对应的视频剪辑：

from videodb import play_stream results = [ (node.metadata["start"], node.metadata["end"]) for node in nodes_spoken_index ] stream_link = video.generate_stream(results) play_stream(stream_link)

5.4 场景（visual content）检索

场景检索需要先用index_scenes建立场景索引并保存返回的index_id，再将该 ID 传给检索器的scene_index_id参数：

from videodb import SceneExtractionType index_id = video.index_scenes( extraction_type=SceneExtractionType.shot_based, extraction_config={"frame_count": 3}, prompt="Describe the scene in detail", ) scene_retriever = VideoDBRetriever( collection=coll.id, video=video.id, search_type=SearchType.semantic, index_type=IndexType.scene, scene_index_id=index_id, score_threshold=0.1, ) scene_query = "accident scenes" nodes_scene_index = scene_retriever.retrieve(scene_query)

响应合成与片段流生成方式与 5.2、5.3 相同，直接复用nodes_scene_index。

6. 实战二：简单多模态 RAG（语音 + 场景双通道）

示例 Notebook 的第四步演示了同时利用两种模态回答组合查询（如“Show me 1. Accident Scene 2. Discussion about nationwide exams”）的简单做法：

查询拆分：用 LLM 把原始查询拆成口语部分与视觉部分：

from llama_index.llms.openai import OpenAI def split_spoken_visual_query(query): transformation_prompt = """ Divide the following query into two distinct parts: one for spoken content and one for visual content. Format the response strictly as: Spoken: <spoken_query> Visual: <visual_query> Query: {query} """ response = OpenAI(model="gpt-4").complete(transformation_prompt.format(query=query)) divided_query = response.text.strip().split("\n") spoken_query = divided_query[0].replace("Spoken:", "").strip() scene_query = divided_query[1].replace("Visual:", "").strip() return spoken_query, scene_query

双检索器并行检索（各自对应一种index_type）：

spoken_retriever = VideoDBRetriever( collection=coll.id, video=video.id, search_type=SearchType.semantic, index_type=IndexType.spoken_word, score_threshold=0.1, ) scene_retriever = VideoDBRetriever( collection=coll.id, video=video.id, search_type=SearchType.semantic, index_type=IndexType.scene, scene_index_id=index_id, score_threshold=0.1, ) nodes_spoken_index = spoken_retriever.retrieve(spoken_query) nodes_scene_index = scene_retriever.retrieve(scene_query)

合并结果：文本侧直接把两路节点相加后交给合成器；视频侧对时间区间做并集（Union）合并后生成剪辑流。示例中给出的区间合并实现：

def merge_intervals(intervals): if not intervals: return [] intervals.sort(key=lambda x: x[0]) merged = [intervals[0]] for interval in intervals[1:]: if interval[0] <= merged[-1][1]: merged[-1][1] = max(merged[-1][1], interval[1]) else: merged.append(interval) return merged results = [ [node.metadata["start"], node.metadata["end"]] for node in nodes_spoken_index + nodes_scene_index ] merged_results = merge_intervals(results) stream_link = video.generate_stream(merged_results) play_stream(stream_link)

示例文档也说明，除并集（Union）外还可用交集（Intersection）策略：只保留在两种模态中都出现的时间段，结果更保守。

7. 实战三：集合级检索与跨视频 Timeline 拼接

不传video参数即切换为集合级检索，对应源码中的coll.search()分支：

# 假设 video_2 已上传并分别建立语音索引与场景索引 spoken_retriever = VideoDBRetriever( collection=coll.id, search_type=SearchType.semantic, index_type=IndexType.spoken_word, score_threshold=0.2, ) scene_retriever = VideoDBRetriever( collection=coll.id, search_type=SearchType.semantic, index_type=IndexType.scene, score_threshold=0.2, ) nodes_spoken_index = spoken_retriever.retrieve(spoken_query) nodes_scene_index = scene_retriever.retrieve(scene_query)

跨多个视频合成剪辑流时，需要基于每条节点的video_id、start、end构造VideoAsset，再挂载到Timeline上统一编译：

from videodb.timeline import Timeline from videodb.asset import VideoAsset timeline = Timeline(conn) for node_obj in nodes_scene_index + nodes_spoken_index: node = node_obj.node node_asset = VideoAsset( asset_id=node.metadata["video_id"], start=node.metadata["start"], end=node.metadata["end"], ) timeline.add_inline(node_asset) stream_url = timeline.generate_stream() play_stream(stream_url)

8. 检索器配置要点汇总

示例 Notebook 的“Configuring VideoDBRetriever”小节给出了全部配置场景，可对照本文第 3 节的参数表理解：

只检索单条视频：传入该视频的 ID

VideoDBRetriever(video="my_video_id")

只检索某个集合：传入集合 ID

VideoDBRetriever(collection="my_coll_id")

指定索引类型：语音索引与场景索引（场景索引必须同时提供scene_index_id）

from videodb import IndexType spoken_retriever = VideoDBRetriever(index_type=IndexType.spoken_word) scene_retriever = VideoDBRetriever( index_type=IndexType.scene, scene_index_id="my_index_id" )

指定检索方式：关键词检索或语义检索

from videodb import SearchType, IndexType keyword_spoken_search = VideoDBRetriever( search_type=SearchType.keyword, index_type=IndexType.spoken_word ) semantic_spoken_search = VideoDBRetriever( search_type=SearchType.semantic, index_type=IndexType.spoken_word )

调优阈值：result_threshold（返回条数上限，默认 5）与score_threshold（分数下限，默认 0.2）

custom_retriever = VideoDBRetriever(result_threshold=2, score_threshold=0.5)

9. 打包、发布与集成元信息

集成包的 pyproject.toml 中还包含 LlamaHub 集成元数据，标明了该包对外的类与导入路径：

[tool.llamahub] classes = ["VideoDBRetriever"] import_path = "llama_index.retrievers.videodb" [tool.llamahub.class_authors] VideoDBRetriever = "video-db"

构建配置（[tool.hatch.build.targets.wheel]）仅打包llama_index/目录，说明该包是一个纯集成插件，不修改llama-index-core。

10. 适用前提与使用限制小结

适用版本：本文基于仓库中当前集成包版本 0.5.0，依赖videodb>=0.2.0与llama-index-core>=0.13.0,<0.15，Python 要求>=3.10,<4.0；
外部服务依赖：VideoDBRetriever是服务型检索器，所有检索都经由 VideoDB API 完成，本地没有离线模式；缺少 API Key 时构造器会直接抛异常；
场景索引前置步骤：IndexType.scene检索依赖事先通过 VideoDB 场景索引接口生成的scene_index_id，否则该参数不会进入检索参数（源码中的条件分支）；
连接模型：从源码结构看，_retrieve每次调用都会重新connect()，高频调用场景下连接建立开销需要自行评估；
扩展方向：除了直接使用 VideoDB 托管索引，也可以加载其转写文本与场景数据后在 LlamaIndex 侧自行建立索引，仓库中另有配套的多模态示例 multi_modal_videorag_videodb.ipynb 可作进一步参考。

【免费下载链接】llama_indexLlamaIndex is the leading document agent and OCR platform项目地址: https://gitcode.com/GitHub_Trending/ll/llama_index

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

LlamaIndex VideoDB Retriever：基于 VideoDBRetriever 的视频片段 RAG 检索器 API 参考与实现解析