Searching Video and Audio by Meaning: A Multimodal RAG Build With Gemini Embedding 2
Multimodal RAG with Gemini Embedding 2: one vector space for text, audio, video and PDFs, so no transcription step, plus chunking, ChromaDB and re-ranking.
This article is published in English.
Sources: *My RAG System Missed 80% of My Data. This One Change Fixed It.*
Consider a knowledge base built for a client: tens of thousands of company documents, indexed and searchable, fronted by a tidy retrieval pipeline that shone in demos. Everyone was pleased until somebody asked why it could not answer questions drawn from the weekly all-hands recordings. The honest answer was awkward. The system could not see them at all.
Much of that company's most valuable knowledge lived in video. Product walkthroughs were recorded every sprint. Engineers explained the database schema while pointing at it on screen. Product managers clicked through the interface while talking about customer pain. None of it was reachable, because every embedding model on hand understood only text.
The usual patch is transcription: run the audio through Whisper, spend hours of compute, accept a nontrivial cloud bill. It exposes a deeper flaw. A transcript strips away the context that gives recordings their value. "As you can see here" becomes an empty sentence. A whiteboard sketch turns into silence. The tone of an engineer saying "this is fine for now" disappears. The words get indexed and the meaning gets thrown out.
Why a shared embedding space matters
Before the code, it helps to see why Gemini Embedding 2 is different in kind and not just a better version of the same thing.
Multimodal retrieval systems used to share one core problem: each modality lived in its own vector space. A text embedding and an audio embedding are both lists of numbers, yet the numbers carry no common meaning, and cosine similarity between vectors from different models trained on different objectives tells nothing. The workaround was separate pipelines with a scoring heuristic stitched on top, a text search plus an audio search plus maybe an image search, then an attempt to rank the combined output. That design failed in two serious ways.
One was the loss of cross-modal relationships. When a video shows what a document describes, the two items have no mathematical link when they live in separate spaces, so a query could surface the document and miss the video, or the reverse.
The other was operational pain: three indexes to maintain, three ingestion pipelines to monitor, three sets of failures to debug. Most teams gave up on audio and video retrieval and settled for text-only search.
With Gemini Embedding 2, text, images, video, audio and documents all get mapped into one space. A text query and a video clip about the same topic land close together, so either can be used to find the other. That is the genuinely new part.
What the model accepts
Each modality has a ceiling that shapes the design:
- Text: 8,192 tokens of input per call
- Images: at most six per call, PNG or JPEG
- Video: chunks no longer than 128 seconds, MP4 or MOV
- Audio: chunks no longer than 80 seconds, MP3 or WAV
- PDFs: six pages per call
Choosing among the eight task types is not cosmetic. They are SEMANTIC_SIMILARITY, CLASSIFICATION and CLUSTERING; RETRIEVAL_DOCUMENT, RETRIEVAL_QUERY and CODE_RETRIEVAL_QUERY; plus QUESTION_ANSWERING and FACT_VERIFICATION. The task type given at embedding time tells the model which direction to shape the vector. A vector tuned for retrieval is not the same as one tuned for classification, even for identical input.
Pricing sits at $0.20 per million text tokens, with a 50% discount through the batch API for work that does not need real-time answers. For bulk ingestion of a large media library, batch is nearly always the better route.
One migration warning: vectors from gemini-embedding-001 (the text-only GA model) cannot be mixed with those from gemini-embedding-2-preview, since the two spaces are incompatible and nothing converts old vectors. Upgrading an existing system means budgeting for a complete re-index.
Architecture, component by component
Most tutorials show a flow diagram and stop. It is more useful to explain each part, why it exists and what breaks without it. The system has two independent pipelines: ingestion, which runs offline and processes media, and querying, which runs in real time and serves answers.
Ingestion
The chunker runs locally on ffmpeg and splits media into segments that fit the model's limits: under 128 seconds for video, under 80 for audio. It also overlaps neighbouring segments so that a conversation sitting on a boundary keeps its meaning on both sides.
The embedder calls the Gemini API and turns each chunk into a vector. Video and audio first need uploading to Google's Files API and a wait for processing before the embedding endpoint can be called. Text goes straight to the API with no upload.
The store is the vector database. It keeps each embedding together with metadata: source file path, timestamps, modality and a short preview string. That metadata is what allows answers to cite sources and link back to the original recording.
Querying
The query side is deliberately simple, because the complexity lives in ingestion. At query time the question is embedded with the RETRIEVAL_QUERY task type, a cosine similarity search runs, the top results are gathered whatever their modality, they are formatted into a context block and everything goes to a generative model.
Answering falls to Gemini 2.5 Pro, a model able to work with text descriptions and with references to video segments in the same context. It understands that a chunk labelled [VIDEO at 02:14] means something different from one labelled [TEXT p.3], and cites each accordingly.
Keeping the two apart
Running ingestion and querying in one process is a common mistake. Ingestion is slow: turning an hour of video into 40 chunks, pushing each through the Files API and waiting on embeddings consumes 15 to 30 minutes, far longer than a query endpoint can wait. Ingestion is also retriable. If one chunk fails, only that chunk should be retried, without disturbing the query path, and mixing the two makes that harder. So ingestion belongs in an offline batch job, whether a script, a cron entry or a pipeline stage, and querying in a stateless API. The vector database is the contract between them.
Building the ingestion side
Environment
Install the dependencies first:
pip install google-genai chromadb google-generativeai python-dotenv ffmpeg-python
ffmpeg is also required at system level, with brew install ffmpeg on a Mac or apt install ffmpeg on Ubuntu. Then a small configuration module sets up the client:
# 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)
Production ingestion has to handle all three modalities, manage uploads for large media, store meaningful metadata and fail gracefully. Layer by layer:
The embedding client
The client is a thin wrapper over the Gemini API. One detail matters: video and audio must be uploaded before embedding, unlike images, which can be sent directly.
# 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
The _wait_for_file helper is easy to skip and costly to omit. Uploads are processed asynchronously, and embedding before the file state is ACTIVE raises an error. A five-second polling interval is conservative. Small files finish faster, while large videos can take a full minute.
The chunker
Everything here is local. ffmpeg splits files into segments with an overlap region at each boundary. The reason for overlap is easy to see. Picture a 30-second discussion starting at 85 seconds into a video. With 90-second chunks and no overlap, its first 5 seconds land in chunk 0 and the other 25 in chunk 1, so a query on that topic may retrieve a chunk holding only a fragment. With 10 seconds of overlap, both chunks contain the junction, and retrieval will likely return the chunk with most of the discussion.
# 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
The vector store
ChromaDB suits the example because nothing has to be set up around it. There is no server to run, no container and no cloud account, since it works inside the Python process and saves to a folder on disk. A production system with millions of chunks would swap in Qdrant, Pinecone or Weaviate, with an interface that looks almost the same.
# 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()
The runner
A script ties it together and processes a directory of mixed media. A hash-based ID scheme makes ingestion idempotent: running a file twice either fails on a duplicate ID, which can be caught and skipped, or is guarded by a check before ingesting.
# 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")
Building the query side
Ingestion is the hard part, and once the store exists querying is clean. The flow is to embed the question, search the store, format the results and generate an answer.
# 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)
}
The similarity_threshold parameter deserves attention. Set too low, irrelevant content leaks into the context and confuses the generator. Set too high, valid results get dropped. A starting range of 0.6 to 0.7 is reasonable, to be tuned by running real queries against a labelled test set.
Raising retrieval accuracy
A basic system working is one thing, and a useful production system needs more. Gains arrive at three stages: ingestion quality, query formulation and filtering after retrieval.
Match the task type, always
This is the biggest single lever. RETRIEVAL_QUERY at query time and RETRIEVAL_DOCUMENT at ingestion are not optional. The two push vectors in opposite directions, tuned to the asymmetry of search, where short queries point into long documents. In testing, using RETRIEVAL_DOCUMENT for both, the lazy default, cost 13 percentage points of retrieval accuracy. The bug raises no error and is easy to miss.
Expand the query first
Short, ambiguous queries hurt. "API limits engineering review" is poor. A fuller question such as "which decisions on API rate limits came out of the engineering review meeting, and why?" works much better. One quick LLM call before the embedding step can automate the rewrite:
# 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")
On a set of questions sourced from video, expansion alone lifted top-5 recall from 74% to 81%.
Run several phrasings in parallel
Instead of a single expanded query, generate three phrasings and merge the results. Different wordings surface different chunks, and the union recalls more.
# 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)
Multi-query search adds roughly two to three API calls per question, with extra latency and cost. It is usually worth it where missing a relevant clip is expensive. For lower-stakes questions, one expanded query is enough.
Re-rank the candidates
Embedding search is good at finding semantically related chunks and weaker at telling "highly relevant to this question" from "vaguely on the same topic". A re-ranker closes that gap.
# 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
The usual pattern retrieves 15 to 20 candidates by embedding search, passes all of them through the re-ranker and keeps the top 5. The re-ranker is a local cross-encoder, free to run and finishing in under 100 ms on a CPU, so latency barely changes.
Use a similarity threshold
When the knowledge base holds nothing relevant, retrieval still returns results, since it always returns the closest matches even if they are unrelated. The threshold in the query pipeline handles that. Chunks below about 0.6 cosine similarity are almost never relevant, and passing them to the generator produces hallucinated answers. Ranges vary by domain. Technical content with precise terminology scores higher for true matches, conversational audio scores lower even when relevant, so thresholds should be calibrated on a labelled test set from real data.
Results against the old setup
Compared with the previous system, Whisper transcription plus text-only RAG, three measures stood out. Accuracy improved most sharply on questions needing visual context from video.
- Retrieval accuracy on video-sourced questions. The old pipeline reached about 61% top-5 recall and the new one 84%. The gap comes almost entirely from questions answered visually: screen walkthroughs, whiteboard diagrams and interface interactions. The transcript-based system produced phrases like "as you can see here" with nothing around them, while the multimodal embeddings captured the visuals directly.
- Ingestion time. For a 60-minute video, the old pipeline spent four to five minutes transcribing before any indexing could begin. The new one yields indexed chunks once the first 90-second segment is uploaded and embedded, roughly two minutes in. Total ingestion runs slightly longer because of upload overhead, but the index becomes usable sooner.
- Cost per hour of video. Whisper transcription cost about $0.36 per hour of audio. Embedding 40 chunks of video at current pricing runs roughly $0.09 to $0.14, depending on length. The multimodal route is both cheaper and more accurate.
Matryoshka dimensions and infrastructure
Matryoshka Representation Learning (MRL) is one of the more interesting features. The model is trained so the most important semantic information sits in the first dimensions of the vector, which means the full 3072-dimensional vector can be truncated with very little loss in retrieval quality. The output size is chosen once at ingestion time and kept, because mixing dimensionalities within one collection is unsupported.
For most systems 1536 is the right default: half the storage, faster queries and only about a 2% quality loss. 3072 makes sense only for high-precision systems where recall is critical and cost is no constraint. Going below 1536 is justified only with tens of millions of chunks, when storage cost becomes meaningful. Even at 256 dimensions quality holds around 88%, perfectly acceptable for many uses.
What this makes possible
- Institutional knowledge bases where onboarding recordings, training videos and engineering walkthroughs sit beside written documentation. A new hire asking how the deployment pipeline works gets both the runbook and the eight-minute walkthrough.
- Compliance and audit systems that must match evidence across modalities. A legal team asking where the returns policy was discussed with customers can surface policy documents and call recordings in one search.
- Support copilots that show the exact 40-second clip from a product demo next to the help article. Watching someone do a task often beats reading about it.
- Research and interview analysis over audio from user sessions. Semantic search finds every instance of users expressing confusion about a feature, with no need to transcribe and hand-code each session.
Practical rules
- Start with the highest-value locked content: recordings that hold knowledge nobody can currently find.
- Use 1536 dimensions, and revisit only if storage cost becomes a real constraint.
- Apply the similarity threshold from day one. Without it the system hallucinates when the knowledge base lacks relevant content. 0.65 is a fair starting point, tuned on real queries.
- Expand queries before embedding. One LLM call costs a fraction of a cent and noticeably improves recall on video, where search terms rarely match the spoken words exactly.
- Keep the ingestion job and query server fully separate, sharing only the vector database. Shared state between them breeds reliability problems.
- Keep chunk metadata rich. Embeddings handle semantic retrieval, and metadata handles the rest: filtering by date, speaker, recording type or confidence.
Rough edges in the preview
The model is a public preview, and a few limitations apply:
- The 80-second audio limit is awkward for podcasts or multi-hour lectures, producing many small chunks and many API calls. It will probably be raised at general availability.
- LangChain's GoogleGenerativeAIEmbeddings class supports only text. Audio and video embedding need the google-genai SDK directly, bridged back into a LangChain pipeline, which works but is inelegant.
- There is no free tier during the preview, so individual developers pay for every embedding call. The batch API's 50% discount helps but requires structuring ingestion as a background job, not a real-time process.
- No attention weights or other interpretability signal are exposed. It is impossible to inspect which frames most influenced an embedding, so debugging poor retrieval is harder than with text, working from aggregate similarity scores with no view of what drove them.