按语义搜索视频和音频：基于Gemini Embedding 2的多模态RAG构建方法
基于 Gemini Embedding 2 的多模态 RAG：为文本、音频、视频和 PDF 提供统一的向量空间，无需转录步骤，同时还具备分块处理、ChromaDB 存储以及重新排序功能。
来源: *My RAG System Missed 80% of My Data. This One Change Fixed It.*
试想为某客户构建的知识库：数以万计的公司文档被建立索引并可供搜索，还配有在演示中表现出色的高效检索流程。所有人都对此感到满意，直到有人问为何它无法回答从每周全员会议录像中提取的问题。诚实的答案令人尴尬——该系统根本无法识别这些录像内容。
这家公司许多最宝贵的知识都存储在视频中。每次迭代周期都会录制产品演示视频，工程师们会在屏幕上指着数据库结构进行讲解，产品经理则一边浏览界面一边谈论客户面临的痛点。但这些内容都无法被系统获取，因为现有的所有嵌入模型都只能处理文本信息。
常见的解决方案是转录：将音频输入Whisper，耗费数小时的计算资源，还要支付不低的云服务费用。但这其实暴露了更深层的问题——转录会去除让录音具有价值的上下文。“如您所见”会变成一句空话，白板上的草图也会化为沉默，工程师说“目前这样还行”的语气也会消失不见。文字虽然被索引了，但其含义却被丢弃了。
为何共享嵌入空间如此重要
在了解代码之前，先弄清楚Gemini Embedding 2为何在本质上有不同，而不仅仅是同一事物的改进版，会更有帮助。
多模态检索系统面临一个共同的核心问题：每种模态都存在于独立的向量空间中。文本嵌入和音频嵌入虽然都是数字列表，但这些数字之间并无共同含义，而且针对不同目标训练的不同模型所产生的向量之间的余弦相似度也毫无参考价值。为了解决这个问题，人们采用了独立的处理流程，并在之上添加评分机制，即结合文本搜索、音频搜索，可能还有图像搜索，然后再尝试对综合结果进行排序。但这种设计存在两个严重缺陷。
其一是跨模态关系的缺失。当视频展示了文档所描述的内容时，由于这两者处于不同的向量空间中，因此不存在数学上的关联，这就可能导致查询结果出现文档却找不到对应视频，或者反之的情况。
另一个问题是运营上的难题：需要维护三个索引，监控三条数据摄取管道，还要排查三组故障。大多数团队放弃了音视频检索功能，转而只使用文本搜索。
借助 Gemini Embedding 2，文本、图片、视频、音频和文档都能被映射到同一个空间中。关于同一主题的文本查询和视频片段会彼此靠近，因此任一内容都可以用来查找另一内容。这才是真正的新颖之处。
模型可处理的输入类型
每种输入类型都有其限制，这些限制决定了设计方式：
- 文本：每次调用最多可输入8,192个标记
- 图片：每次调用最多六张，格式为PNG或JPEG
- 视频：片段时长不超过128秒，格式为MP4或MOV
- 音频：片段时长不超过80秒，格式为MP3或WAV
从八种任务类型中选择并非出于表面考虑。这些类型包括语义相似度分析、分类与聚类；文档检索、查询检索与代码检索查询；以及问答与事实验证。在嵌入阶段指定的任务类型会告诉模型应如何塑造向量。即使输入相同，为检索而优化的向量也与为分类而优化的向量也有所不同。
费用为每百万文本标记0.20美元，对于不需要实时回复的任务，通过批量API可享受50%的折扣。对于大量媒体库的批量导入，批量处理几乎总是更佳选择。
一个迁移警告：gemini-embedding-001（纯文本生成模型）生成的向量无法与gemini-embedding-2-preview生成的向量混合使用，因为这两个系统不兼容，且没有工具可以转换旧向量。升级现有系统意味着需要准备资金进行完整的重新索引。
逐组件解析架构
大多数教程只展示流程图便止步不前。实际上，详细解释每个组件的作用、存在原因以及缺少该组件时会出现的问题会更有帮助。该系统包含两个独立的处理流程：离线运行的数据摄取流程，用于处理媒体文件；以及实时运行的查询流程，用于提供答案。
数据摄取流程
分块器在ffmpeg上本地运行，将媒体内容分割成符合模型处理限制的片段：视频时长不超过128秒，音频时长不超过80秒。它还会让相邻片段有所重叠，这样位于边界处的对话内容在两侧都能保持完整含义。
嵌入器会调用Gemini API，将每个分块转换为向量。视频和音频需先上传到Google的Files API并等待处理完成，之后才能调用嵌入端点；而文本则可直接发送给API，无需上传。
存储模块即向量数据库，它保存每个向量以及相关元数据：源文件路径、时间戳、内容类型和一段简短的预览文本。正是这些元数据使得答案能够引用来源并回溯到原始录音。
查询
查询端被刻意设计得简单，因为复杂性体现在数据摄取阶段。在查询时，问题会以RETRIEVAL_QUERY任务类型的形式被嵌入，随后进行余弦相似度搜索，无论结果形式如何都会收集最相关的结果，将其格式化为上下文块，最后全部传递给生成模型。
回答工作由Gemini 2.5 Pro负责，该模型能够处理文本描述以及同一上下文中的视频片段引用。它明白标记为[VIDEO at 02:14]的内容与标记为[TEXT p.3]的内容含义不同，并会分别对应地引用它们。
将两者区分开
在同一个进程中同时处理数据导入和查询是一个常见的错误。数据导入速度很慢：将一小时的视频分割成40个数据块，通过Files API逐个上传并等待嵌入结果，这一过程需要15到30分钟，远远超过查询端点所能等待的时间。此外，数据导入操作是可以重试的。如果某个数据块处理失败，只需重新尝试该数据块即可，而不应影响查询流程，而将两者混在一起会使得处理更加复杂。因此，数据导入应放在离线批处理任务中，无论是脚本、cron任务还是流水线阶段，而查询则应在无状态API中完成。向量数据库则是连接两者的桥梁。
构建数据导入部分
环境准备
首先安装所需的依赖项：
pip install google-genai chromadb google-generativeai python-dotenv ffmpeg-python
系统层面也需要 ffmpeg，可在 Mac 上使用 brew install ffmpeg，或在 Ubuntu 上使用 apt install ffmpeg。之后通过一个小的配置模块来设置客户端：
# config.py
import os
from google import genai
from google.genai import types
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")
EMBEDDING_MODEL = "gemini-embedding-2-preview"
GENERATION_MODEL = "gemini-2.5-pro"
# Output dimensionality options: 128, 256, 512, 768, 1024, 1536, 3072
# 1536 is the recommended default
EMBEDDING_DIMENSIONS = 1536
client = genai.Client(api_key=GEMINI_API_KEY)
生产环境的数据处理系统需要支持这三种数据格式，能够管理大容量媒体的上传，存储有意义的元数据，并在出错时实现优雅降级。具体实现步骤如下：
嵌入客户端
该客户端只是 Gemini API 的简单封装层。需要注意的一点是：与可直接上传的图像不同，视频和音频必须在嵌入处理之前上传。
# embedder.py
import time
from pathlib import Path
from google import genai
from google.genai import types
from config import client, EMBEDDING_MODEL, EMBEDDING_DIMENSIONS
def embed_text(text: str, task_type: str = "RETRIEVAL_DOCUMENT") -> list[float]:
"""Embed a plain text chunk."""
result = client.models.embed_content(
model=EMBEDDING_MODEL,
contents=text,
config=types.EmbedContentConfig(
task_type=task_type,
output_dimensionality=EMBEDDING_DIMENSIONS
)
)
return result.embeddings[0].values
def _wait_for_file(uploaded, max_wait: int = 300):
"""Poll until a File API upload is done processing."""
waited = 0
poll_interval = 5
while uploaded.state.name == "PROCESSING" and waited time.sleep(poll_interval)
waited += poll_interval
uploaded = client.files.get(name=uploaded.name)
if uploaded.state.name != "ACTIVE":
raise RuntimeError(
f"File never became ACTIVE. Final state: {uploaded.state.name}"
)
return uploaded
def embed_audio(audio_path: str) -> list[float]:
"""
Embed an audio file natively. No transcription step.
The model processes the audio signal directly and returns a
semantic embedding that captures speech content, tone, and
acoustic features. Max input: 80 seconds per file.
"""
uploaded = client.files.upload(path=str(audio_path))
uploaded = _wait_for_file(uploaded, max_wait=120)
result = client.models.embed_content(
model=EMBEDDING_MODEL,
contents=uploaded,
config=types.EmbedContentConfig(
task_type="RETRIEVAL_DOCUMENT",
output_dimensionality=EMBEDDING_DIMENSIONS
)
)
# Clean up: uploaded files count against your quota
client.files.delete(name=uploaded.name)
return result.embeddings[0].values
def embed_video(video_path: str) -> list[float]:
"""
Embed a video chunk natively. Gemini processes both the
audio track and visual frames together in one pass.
This is the key capability: visual demonstrations get captured
in the embedding alongside what is being said. Max input: 128 seconds.
"""
uploaded = client.files.upload(path=str(video_path))
uploaded = _wait_for_file(uploaded, max_wait=300)
result = client.models.embed_content(
model=EMBEDDING_MODEL,
contents=uploaded,
config=types.EmbedContentConfig(
task_type="RETRIEVAL_DOCUMENT",
output_dimensionality=EMBEDDING_DIMENSIONS
)
)
client.files.delete(name=uploaded.name)
return result.embeddings[0].values
def embed_with_context(text: str, image_bytes: bytes = None) -> list[float]:
"""
Embed text and an optional image together in a single call.
When both are passed, the model returns one vector that
represents the joint meaning. A query asking about a database
schema can retrieve a screenshot of that schema.
"""
contents = [text]
if image_bytes:
contents.append(
types.Part.from_bytes(data=image_bytes, mime_type="image/jpeg")
)
result = client.models.embed_content(
model=EMBEDDING_MODEL,
contents=contents,
config=types.EmbedContentConfig(
task_type="RETRIEVAL_DOCUMENT",
output_dimensionality=EMBEDDING_DIMENSIONS
)
)
return result.embeddings[0].values
_wait_for_file辅助函数虽可省略，但省略后会带来严重后果。上传操作是异步处理的，若在文件状态为ACTIVE之前就尝试嵌入内容，将会引发错误。五秒的轮询间隔已属保守设定——小文件处理速度更快，而大型视频则可能需要整整一分钟。
分块器
此处的一切操作均在本地进行。ffmpeg会将文件分割成多个片段，每个片段边界处都设有重叠区域。设置重叠区域的理由显而易见：假设视频从85秒处开始有30秒的讨论内容，若采用90秒无重叠的分块方式，前5秒会属于第0个片段，其余25秒属于第1个片段，这样查询该主题时可能只能获取到仅包含部分内容的片段；而若设置10秒的重叠区域，两个片段都会包含该讨论片段，查询时更有可能得到包含大部分讨论内容的片段。
# chunker.py
import subprocess
import json
from pathlib import Path
from dataclasses import dataclass
from typing import List
@dataclass
class MediaChunk:
file_path: str
start_time: float
end_time: float
source_file: str
modality: str
chunk_index: int
total_chunks: int # Useful for progress reporting
def get_media_duration(file_path: str) -> float:
"""Get exact duration via ffprobe. Works for both audio and video."""
cmd = [
"ffprobe", "-v", "quiet",
"-print_format", "json",
"-show_streams", str(file_path)
]
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
data = json.loads(result.stdout)
# Find the first stream with a duration value
for stream in data.get("streams", []):
if "duration" in stream:
return float(stream["duration"])
raise ValueError(f"Could not determine duration for: {file_path}")
def _run_ffmpeg_split(input_path: str, output_path: str,
start: float, duration: float):
"""Execute a single ffmpeg split operation."""
cmd = [
"ffmpeg", "-y",
"-ss", str(start),
"-i", str(input_path),
"-t", str(duration),
"-c", "copy", # No re-encoding: much faster, no quality loss
"-avoid_negative_ts", "make_zero",
str(output_path)
]
result = subprocess.run(cmd, capture_output=True)
if result.returncode != 0:
raise RuntimeError(
f"ffmpeg failed: {result.stderr.decode()}"
)
def chunk_video(
video_path: str,
chunk_duration: int = 90,
overlap: int = 10,
output_dir: str = "./chunks/video"
) -> List[MediaChunk]:
"""
Split video into overlapping chunks within the 128-second limit.
Default: 90-second chunks with 10-second overlap.
Overlap ensures topic transitions are captured in at least one chunk.
"""
Path(output_dir).mkdir(parents=True, exist_ok=True)
total_duration = get_media_duration(video_path)
source_name = Path(video_path).stem
# Pre-calculate chunk boundaries
boundaries = []
start = 0.0
while start end = min(start + chunk_duration, total_duration)
boundaries.append((start, end))
start += (chunk_duration - overlap)
chunks = []
for idx, (start, end) in enumerate(boundaries):
output_path = f"{output_dir}/{source_name}_{idx:04d}.mp4"
_run_ffmpeg_split(video_path, output_path, start, end - start)
chunks.append(MediaChunk(
file_path=output_path,
start_time=start,
end_time=end,
source_file=str(video_path),
modality="video",
chunk_index=idx,
total_chunks=len(boundaries)
))
return chunks
def chunk_audio(
audio_path: str,
chunk_duration: int = 60,
overlap: int = 5,
output_dir: str = "./chunks/audio"
) -> List[MediaChunk]:
"""
Split audio into overlapping chunks within the 80-second limit.
60 seconds per chunk gives a comfortable buffer under the 80-second cap.
"""
Path(output_dir).mkdir(parents=True, exist_ok=True)
total_duration = get_media_duration(audio_path)
source_name = Path(audio_path).stem
boundaries = []
start = 0.0
while start end = min(start + chunk_duration, total_duration)
boundaries.append((start, end))
start += (chunk_duration - overlap)
chunks = []
for idx, (start, end) in enumerate(boundaries):
output_path = f"{output_dir}/{source_name}_{idx:04d}.mp3"
_run_ffmpeg_split(audio_path, output_path, start, end - start)
chunks.append(MediaChunk(
file_path=output_path,
start_time=start,
end_time=end,
source_file=str(audio_path),
modality="audio",
chunk_index=idx,
total_chunks=len(boundaries)
))
return chunks
向量存储
ChromaDB非常适合这个示例，因为无需为其进行任何额外配置。它不需要运行服务器，也不需要容器或云账户，只需在Python进程内部运行并将数据保存到磁盘上的文件夹中。而对于需要处理数百万个数据块的生产环境，则会选用Qdrant、Pinecone或Weaviate，这些系统的界面几乎相同。
# vector_store.py
import chromadb
from chromadb.config import Settings
from pathlib import Path
class MultimodalVectorStore:
"""
Vector store wrapping ChromaDB for multimodal RAG.
Stores embeddings + metadata for text, audio, and video chunks.
"""
def __init__(self, persist_dir: str = "./chroma_db"):
self.client = chromadb.PersistentClient(
path=persist_dir,
settings=Settings(anonymized_telemetry=False)
)
self.collection = self.client.get_or_create_collection(
name="multimodal_rag",
# cosine distance is standard for semantic similarity
metadata={"hnsw:space": "cosine"}
)
def add_text_chunk(
self,
chunk_id: str,
text: str,
embedding: list[float],
source_file: str,
chunk_index: int,
page: int = None
):
self.collection.add(
ids=[chunk_id],
embeddings=[embedding],
documents=[text],
metadatas=[{
"modality": "text",
"source_file": source_file,
"chunk_index": chunk_index,
"page": page or 0,
"preview": text[:250]
}]
)
def add_media_chunk(
self,
chunk_id: str,
embedding: list[float],
source_file: str,
start_time: float,
end_time: float,
modality: str,
chunk_index: int
):
"""
Store a video or audio chunk.
Note: we store a formatted timestamp string in `documents`
so ChromaDB has something to display. The actual retrieval
quality comes entirely from the embedding, not this text.
"""
ts_start = f"{int(start_time // 60):02d}:{int(start_time % 60):02d}"
ts_end = f"{int(end_time // 60):02d}:{int(end_time % 60):02d}"
display = (
f"[{modality.upper()}] {Path(source_file).name} "
f"from {ts_start} to {ts_end}"
)
self.collection.add(
ids=[chunk_id],
embeddings=[embedding],
documents=[display],
metadatas=[{
"modality": modality,
"source_file": source_file,
"start_time": start_time,
"end_time": end_time,
"timestamp_start": ts_start,
"timestamp_end": ts_end,
"chunk_index": chunk_index,
"preview": display
}]
)
def search(
self,
query_embedding: list[float],
n_results: int = 5,
modality_filter: str = None
) -> list[dict]:
"""
Retrieve top-k most similar chunks across all modalities.
Optionally filter to a single modality for targeted search.
"""
where_clause = {"modality": modality_filter} if modality_filter else None
results = self.collection.query(
query_embeddings=[query_embedding],
n_results=n_results,
where=where_clause,
include=["documents", "metadatas", "distances"]
)
chunks = []
for doc, meta, dist in zip(
results["documents"][0],
results["metadatas"][0],
results["distances"][0]
):
chunks.append({
"content": doc,
"metadata": meta,
"modality": meta["modality"],
# ChromaDB returns cosine distance; convert to similarity score
"similarity": round(1.0 - dist, 4)
})
return sorted(chunks, key=lambda x: x["similarity"], reverse=True)
def count(self) -> int:
return self.collection.count()
运行程序
一个脚本负责将所有内容整合起来，并处理包含多种媒体类型的目录。基于哈希值的标识方案使得数据导入操作具有幂等性：若重复导入同一文件，要么会因标识重复而失败，从而被检测并跳过，要么会在导入前通过检查加以防止。
# ingest.py
import os
import hashlib
from pathlib import Path
from chunker import chunk_video, chunk_audio
from embedder import embed_text, embed_audio, embed_video
from vector_store import MultimodalVectorStore
store = MultimodalVectorStore(persist_dir="./chroma_db")
def make_chunk_id(source_path: str, chunk_index: int) -> str:
"""Stable, unique ID for any chunk. Same input always = same ID."""
raw = f"{os.path.abspath(source_path)}:{chunk_index}"
return hashlib.sha256(raw.encode()).hexdigest()[:20]
def ingest_text_file(file_path: str):
with open(file_path, "r", encoding="utf-8") as f:
text = f.read()
# Sliding window chunking: 800 chars with 100-char overlap
chunk_size, overlap = 800, 100
raw_chunks = []
start = 0
while start end = min(start + chunk_size, len(text))
raw_chunks.append(text[start:end])
start += chunk_size - overlap
for i, chunk_text in enumerate(raw_chunks):
embedding = embed_text(chunk_text, task_type="RETRIEVAL_DOCUMENT")
store.add_text_chunk(
chunk_id=make_chunk_id(file_path, i),
text=chunk_text,
embedding=embedding,
source_file=file_path,
chunk_index=i
)
print(f" Stored {len(raw_chunks)} text chunks from {Path(file_path).name}")
def ingest_video_file(file_path: str):
print(f" Chunking: {Path(file_path).name}")
chunks = chunk_video(file_path, chunk_duration=90, overlap=10)
for chunk in chunks:
print(
f" Embedding chunk {chunk.chunk_index + 1}/{chunk.total_chunks} "
f"({chunk.start_time:.0f}s to {chunk.end_time:.0f}s)"
)
try:
embedding = embed_video(chunk.file_path)
store.add_media_chunk(
chunk_id=make_chunk_id(file_path, chunk.chunk_index),
embedding=embedding,
source_file=file_path,
start_time=chunk.start_time,
end_time=chunk.end_time,
modality="video",
chunk_index=chunk.chunk_index
)
except Exception as e:
print(f" WARNING: Failed to embed chunk {chunk.chunk_index}: {e}")
finally:
# Always clean up temp files, even on failure
if os.path.exists(chunk.file_path):
os.remove(chunk.file_path)
print(f" Done. {len(chunks)} video chunks stored.")
def ingest_audio_file(file_path: str):
print(f" Chunking: {Path(file_path).name}")
chunks = chunk_audio(file_path, chunk_duration=60, overlap=5)
for chunk in chunks:
try:
embedding = embed_audio(chunk.file_path)
store.add_media_chunk(
chunk_id=make_chunk_id(file_path, chunk.chunk_index),
embedding=embedding,
source_file=file_path,
start_time=chunk.start_time,
end_time=chunk.end_time,
modality="audio",
chunk_index=chunk.chunk_index
)
except Exception as e:
print(f" WARNING: Failed to embed chunk {chunk.chunk_index}: {e}")
finally:
if os.path.exists(chunk.file_path):
os.remove(chunk.file_path)
print(f" Done. {len(chunks)} audio chunks stored.")
def ingest_directory(directory: str):
handlers = {
".txt": ingest_text_file,
".md": ingest_text_file,
".mp4": ingest_video_file,
".mov": ingest_video_file,
".mp3": ingest_audio_file,
".wav": ingest_audio_file,
}
all_files = list(Path(directory).rglob("*"))
media_files = [f for f in all_files if f.suffix.lower() in handlers]
print(f"Found {len(media_files)} files to ingest\n")
for file_path in media_files:
print(f"Processing: {file_path.name}")
handler = handlers[file_path.suffix.lower()]
handler(str(file_path))
print()
print(f"Ingestion complete. Total chunks indexed: {store.count()}")
if __name__ == "__main__":
ingest_directory("./knowledge_base")
构建查询模块
数据摄取是最困难的部分，一旦数据存储完成，查询操作就会变得简单。其流程为：嵌入查询语句、在存储中搜索、格式化结果并生成答案。
# query.py
import os
from pathlib import Path
import google.generativeai as genai
from embedder import embed_text
from vector_store import MultimodalVectorStore
from config import GENERATION_MODEL
store = MultimodalVectorStore(persist_dir="./chroma_db")
def format_context_for_llm(chunks: list[dict]) -> str:
"""
Format retrieved chunks into a context block for the generative model.
We include modality, source, and similarity score so the model
can calibrate its confidence and cite sources accurately.
"""
parts = []
for rank, chunk in enumerate(chunks, start=1):
meta = chunk["metadata"]
modality = chunk["modality"]
score = chunk["similarity"]
if modality == "text":
parts.append(
f"[SOURCE {rank} | TEXT | {Path(meta['source_file']).name} "
f"| chunk {meta['chunk_index']} | similarity {score}]\n"
f"{chunk['content']}"
)
elif modality == "video":
parts.append(
f"[SOURCE {rank} | VIDEO | {Path(meta['source_file']).name} "
f"| {meta['timestamp_start']} to {meta['timestamp_end']} "
f"| similarity {score}]\n"
f"Video segment covering this time range."
)
elif modality == "audio":
parts.append(
f"[SOURCE {rank} | AUDIO | {Path(meta['source_file']).name} "
f"| {meta['timestamp_start']} to {meta['timestamp_end']} "
f"| similarity {score}]\n"
f"Audio segment covering this time range."
)
return "\n\n---\n\n".join(parts)
def answer_query(
query: str,
n_results: int = 5,
modality_filter: str = None,
similarity_threshold: float = 0.6
) -> dict:
"""
Full RAG pipeline: embed the query, retrieve chunks, generate answer.
similarity_threshold: chunks below this score are dropped before generation.
Prevents low-quality matches from polluting the context.
"""
query_embedding = embed_text(query, task_type="RETRIEVAL_QUERY")
retrieved = store.search(
query_embedding=query_embedding,
n_results=n_results,
modality_filter=modality_filter
)
# Filter out weak matches
filtered = [c for c in retrieved if c["similarity"] >= similarity_threshold]
if not filtered:
return {
"answer": (
"No sufficiently relevant content was found in the knowledge base. "
"The most similar content had a similarity score below the threshold."
),
"sources": retrieved,
"query": query
}
context = format_context_for_llm(filtered)
system_prompt = """You are a helpful assistant with access to a multimodal
knowledge base that contains text documents, video recordings, and audio files.
When citing a source, reference it by its label (e.g., SOURCE 1, SOURCE 2).
For video and audio sources, always include the timestamp so the user can
navigate to the exact moment in the recording.
If the retrieved context does not contain enough information to answer
confidently, say so clearly rather than guessing."""
user_message = (
f"Using only the sources below, answer this question:\n\n"
f"Question: {query}\n\n"
f"Sources:\n{context}"
)
genai.configure(api_key=os.getenv("GEMINI_API_KEY"))
model = genai.GenerativeModel(GENERATION_MODEL)
response = model.generate_content(
user_message,
generation_config={"temperature": 0.1}
)
return {
"answer": response.text,
"sources": filtered,
"query": query,
"chunks_retrieved": len(retrieved),
"chunks_used": len(filtered)
}
similarity_threshold参数值得重视。若设置过低，无关内容会混入上下文，使生成器陷入混乱；若设置过高，则有效结果会被遗漏。0.6到0.7作为初始范围较为合理，可通过在带标签的测试集上执行实际查询来进一步调整。
提升检索精度
仅有基本功能正常的系统还不够，实用的实用系统还需要更多优化。提升精度可分为三个阶段：数据摄入质量、查询构建，以及检索后的过滤。
始终匹配任务类型
这是最关键的单一因素。在查询时必须使用RETRIEVAL_QUERY，在数据导入时则必须使用RETRIEVAL_DOCUMENT，二者均不可省略。这两个向量方向相反，旨在弥补搜索过程中的不对称性——即简短的查询往往需要对应较长的文档。在测试中，若对两者都采用默认的懒惰处理方式，会导致检索准确率下降13个百分点。该缺陷不会引发错误，因此很容易被忽视。
先扩展查询内容
简短且模糊的查询会带来负面影响。“API限制工程评审”这样的表述就很糟糕。而像“在工程评审会议上就API速率限制做出了哪些决定，原因是什么？”这样更完整的提问效果要好得多。在嵌入步骤之前调用一次大型语言模型即可自动完成内容的重写：
# query_expander.py
import google.generativeai as genai
import os
genai.configure(api_key=os.getenv("GEMINI_API_KEY"))
def expand_query(raw_query: str) -> str:
"""
Rewrite a short user query into a more detailed retrieval query.
Returns the expanded version. Falls back to original on failure.
"""
model = genai.GenerativeModel("gemini-2.0-flash")
prompt = (
"Rewrite the following search query to be more detailed and specific. "
"Add relevant context, related terminology, and clarify the intent. "
"Keep it as a single question. Do not add facts not implied by the original.\n\n"
f"Original query: {raw_query}\n\n"
"Expanded query:"
)
try:
response = model.generate_content(
prompt,
generation_config={"temperature": 0.2, "max_output_tokens": 200}
)
return response.text.strip()
except Exception:
return raw_query # Graceful fallback
# Usage in query pipeline:
# expanded = expand_query("API limits engineering review")
# query_embedding = embed_text(expanded, task_type="RETRIEVAL_QUERY")
在基于视频的测试题中，仅通过扩展查询就将前五名的召回率从74%提升到了81%。
同时运行多种表述方式
不要只使用一种扩展后的查询语句，而应生成三种不同的表述并合并结果。不同的表述会提取出不同的信息片段，合并后能获得更高的召回率。
# multi_query.py
import concurrent.futures
from query_expander import expand_query
from embedder import embed_text
from vector_store import MultimodalVectorStore
store = MultimodalVectorStore(persist_dir="./chroma_db")
def generate_query_variants(query: str) -> list[str]:
"""Generate multiple phrasings for a single question."""
import google.generativeai as genai
import os
genai.configure(api_key=os.getenv("GEMINI_API_KEY"))
model = genai.GenerativeModel("gemini-2.0-flash")
prompt = (
f"Generate 3 different ways to search for information about: {query}\n\n"
"Return exactly 3 queries, one per line, no numbering or bullets."
)
response = model.generate_content(prompt)
variants = [line.strip() for line in response.text.strip().split("\n") if line.strip()]
return ([query] + variants)[:4] # Always include original, cap at 4 total
def multi_query_search(query: str, n_per_query: int = 4) -> list[dict]:
"""
Search with multiple query variants and deduplicate results.
Returns unique chunks ranked by their best similarity score.
"""
variants = generate_query_variants(query)
def search_one(variant: str) -> list[dict]:
embedding = embed_text(variant, task_type="RETRIEVAL_QUERY")
return store.search(embedding, n_results=n_per_query)
# Run all variants in parallel
all_results = []
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
futures = {executor.submit(search_one, v): v for v in variants}
for future in concurrent.futures.as_completed(futures):
all_results.extend(future.result())
# Deduplicate by source file + chunk index, keeping best similarity score
seen = {}
for chunk in all_results:
meta = chunk["metadata"]
key = f"{meta['source_file']}:{meta['chunk_index']}"
if key not in seen or chunk["similarity"] > seen[key]["similarity"]:
seen[key] = chunk
return sorted(seen.values(), key=lambda x: x["similarity"], reverse=True)
多查询搜索每条问题大约需要两到三次API调用，还会带来额外的延迟和成本。在错过相关内容会带来高昂代价的情况下，这种做法通常是值得的。对于重要性较低的查询，一个扩展后的查询就足够了。
重新排序候选结果
嵌入搜索擅长找出语义上相关的片段，但在区分“与当前问题高度相关”和“仅大致属于同一主题”的内容方面能力较弱。重新排序机制可以弥补这一缺陷。
# reranker.py
from sentence_transformers import CrossEncoder
# This model runs locally, no API cost, fast inference
_reranker = None
def get_reranker():
global _reranker
if _reranker is None:
_reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")
return _reranker
def rerank_chunks(query: str, chunks: list[dict], top_k: int = 5) -> list[dict]:
"""
Re-rank retrieved chunks using a cross-encoder model.
Cross-encoders read both query and chunk together, giving much
more precise relevance scores than embedding cosine similarity.
Only practical on a small candidate set (10-20 chunks).
"""
reranker = get_reranker()
# For video/audio, we use the metadata preview as the text input.
# For text chunks, we use the actual content.
pairs = []
for chunk in chunks:
if chunk["modality"] == "text":
doc_text = chunk["content"]
else:
meta = chunk["metadata"]
doc_text = (
f"{chunk['modality']} recording: {meta['source_file']} "
f"at {meta.get('timestamp_start', '')} to {meta.get('timestamp_end', '')}"
)
pairs.append([query, doc_text])
scores = reranker.predict(pairs)
for chunk, score in zip(chunks, scores):
chunk["rerank_score"] = float(score)
return sorted(chunks, key=lambda x: x["rerank_score"], reverse=True)[:top_k]
# Usage:
# candidates = store.search(query_embedding, n_results=20) # Retrieve wide
# final = rerank_chunks(query, candidates, top_k=5) # Re-rank narro
常规模式通过嵌入搜索获取15到20个候选项，将它们全部送入重排序器后保留排名前5的。重排序器是一种本地交叉编码器，可自由运行，在CPU上处理时间不到100毫秒，因此延迟几乎不会发生变化。
使用相似度阈值
即使知识库中没有任何相关内容，检索功能仍会返回结果，因为它总会返回最接近的匹配项，即便这些项并无关联。查询流程中的阈值可解决这一问题。余弦相似度低于0.6的候选项几乎从不具备相关性，将其传递给生成器会导致虚假答案的产生。不同领域的阈值范围有所差异：包含精确术语的技术内容在真正匹配时的得分较高，而对话音频即便相关得分也较低，因此阈值应基于真实数据的标记测试集进行校准。
与旧配置的对比结果
与之前的系统相比，Whisper转录结合纯文本RAG在三个方面表现突出。在需要视频视觉上下文的问题上，准确率提升最为显著。
- 基于视频问题的检索准确率。旧系统的前5名召回率约为61%，而新系统则达到84%。这一差距几乎完全来自那些需要通过视觉方式解答的问题，比如屏幕演示、白板图示以及界面交互。基于转录文本的系统会生成诸如“如您所见”这样的短语，且周围没有任何其他信息，而多模态嵌入技术则能直接捕捉视觉内容。
- 数据导入时间。对于60分钟的视频，旧流程在开始索引之前需要花费4到5分钟进行转录。新流程在上传并嵌入前90秒的内容后即可生成索引片段，大约只需2分钟。由于上传带来的额外开销，整体导入时间略有延长，但索引的可用时间却更早。
- 每小时视频的成本。Whisper的转录费用约为每小时音频0.36美元。按照当前价格，嵌入40个视频片段的费用大约在0.09美元到0.14美元之间，具体取决于视频长度。多模态处理方式既更便宜又更准确。
Matryoshka系统的规模与基础设施
套娃表示学习（MRL）是其中较为有趣的功能之一。该模型经过训练，使得最重要的语义信息集中在向量的前几个维度中，因此即使将3072维的向量进行截断，检索质量也会仅有极小的损失。输出维度在数据导入时确定后便不会改变，因为同一个集合内不允许混合不同的维度。
对于大多数系统而言，1536是合适的默认值：能节省一半的存储空间，查询速度更快，且质量损失仅约2%。只有在对召回率要求极高且成本不受限制的高精度系统中，才需要使用3072维。只有当数据块数量达到数千万级、存储成本变得显著时，才有必要将维度降至1536以下。即便使用256维，质量也能保持在88%左右，对于许多应用来说已经足够理想。
由此可实现的功能
- 机构知识库，其中包含入职培训录像、教学视频和工程流程演示，与书面文档并存。新员工若想了解部署流程，既能看到操作手册，也能观看8分钟的流程演示。
- 合规与审计系统，要求能够跨不同形式查找相关证据。法律团队若想查询与客户讨论退货政策的记录，可通过一次搜索即可找到政策文件和通话录像。
- 辅助支持工具，会在帮助文章旁显示产品演示中恰好40秒的片段。观看他人操作往往比阅读说明更有效。
实际操作规则
- 从价值最高的已锁定内容开始：那些包含目前无人能找到的知识的录音。
- 使用1536个维度，只有当存储成本成为真正限制时才重新考虑调整。
- 从一开始就应用相似度阈值。如果没有该阈值，当知识库中缺乏相关内容时系统就会产生错误信息。0.65是一个合理的起始值，可通过实际查询进行优化。
- 在嵌入之前先扩展查询条件。每次调用大型语言模型只需几美分，且能显著提升视频内容的检索效果，因为搜索词很少能与实际说话内容完全匹配。
- 应将数据摄取任务与查询服务器完全分开，仅共享向量数据库。两者之间的状态共享会引发可靠性问题。
- 要确保分块元数据足够丰富。嵌入模型负责语义检索，而元数据则用于处理其他功能：按日期、说话人、录音类型或置信度进行过滤。
预览版的缺陷
该模型目前处于公开预览阶段，存在一些限制：
- 80秒的音频时长限制对播客或数小时的讲座来说很不方便，会导致生成大量小分块并触发大量API调用。在正式发布时这一限制很可能会被提高。
- LangChain的GoogleGenerativeAIEmbeddings类仅支持文本处理。音频和视频的嵌入需要直接使用google-genai SDK，再通过接口接入LangChain流程，虽然可行但不够简洁。
- 在预览阶段没有免费套餐，因此每位开发者都需要为每次嵌入调用付费。批量 API 虽然有 50% 的折扣，但要求将数据摄取处理为后台任务而非实时流程。
- 不提供注意力权重或其他可解释性信号，因此无法查看是哪些帧对嵌入结果影响最大，这使得调试较差的检索效果比处理文本更为困难，因为只能依据总体相似度分数，而无法了解其背后的原因。