Recherche de vidéos et d’audio par signification : une architecture multimodale RAG utilisant Gemini Embedding 2
RAG multimodal avec Gemini Embedding 2 : un espace vectoriel unique pour le texte, l’audio, la vidéo et les PDF, ce qui élimine l’étape de transcription, ainsi que du chunking, ChromaDB et un réclassement.
Sources: *My RAG System Missed 80% of My Data. This One Change Fixed It.*
Imaginez une base de connaissances conçue pour un client : des dizaines de milliers de documents d’entreprise, indexés et recherchables, accompagnés d’un processus de récupération efficace qui se distinguait lors des démonstrations. Tout le monde était satisfait jusqu’à ce que quelqu’un demande pourquoi le système ne pouvait pas répondre aux questions issues des enregistrements hebdomadaires des réunions de l’ensemble des employés. La réponse honnête était gênante : le système ne pouvait absolument pas les voir.
Une grande partie des connaissances les plus précieuses de cette entreprise se trouvait sous forme de vidéos. Des présentations des produits étaient enregistrées à chaque itération. Les ingénieurs expliquaient le schéma de la base de données en le montrant à l’écran. Les chefs de produit naviguaient dans l’interface tout en parlant des problèmes des clients. Rien de tout cela n’était accessible, car tous les modèles d’incorporation disponibles ne comprenaient que du texte.
La méthode habituelle consiste en la transcription : faire passer l’audio par Whisper, consommer des heures de calcul, et accepter une facture cloud importante. Cela révèle une faille plus profonde. Une transcription élimine le contexte qui donne de la valeur aux enregistrements. « Comme vous pouvez le voir ici » devient une phrase vide. Un schéma griffonné sur un tableau blanc se transforme en silence. Le ton de l’ingénieur qui dit « c’est acceptable pour l’instant » disparaît. Les mots sont indexés, mais leur sens est perdu.
Pourquoi un espace d’embedding partagé est important
Avant d’examiner le code, il est utile de comprendre pourquoi Gemini Embedding 2 est différent par nature, et non simplement une version améliorée de la même chose.
Les systèmes de récupération multimodaux rencontrent tous un problème fondamental : chaque modalité évolue dans son propre espace vectoriel. Une embedding de texte et une embedding audio sont toutes deux des listes de nombres, mais ces nombres ne possèdent aucun sens commun, et la similarité cosinus entre des vecteurs provenant de modèles différents entraînés avec des objectifs distincts ne renseigne en rien. La solution temporaire consistait à utiliser des pipelines séparés auxquels on ajoutait une heuristique de scoring, c’est-à-dire une recherche de texte, une recherche audio et éventuellement une recherche d’image, suivie d’une tentative de classer les résultats combinés. Cette approche a échoué de deux manières graves.
Le premier problème était la perte des relations intermodales. Lorsqu’une vidéo montre ce que décrit un document, ces deux éléments n’ont aucun lien mathématique s’ils se trouvent dans des espaces distincts ; par conséquent, une requête peut faire apparaître le document tout en ignorant la vidéo, ou l’inverse.
L’autre problème était lié aux difficultés opérationnelles : trois index à maintenir, trois pipelines d’ingestion à surveiller, trois ensembles de pannes à déboguer. La plupart des équipes ont abandonné la recherche d’audio et de vidéo pour se contenter d’une recherche uniquement sur du texte.
Avec Gemini Embedding 2, le texte, les images, la vidéo, l’audio et les documents sont tous cartographiés dans un même espace. Une requête de texte et une vidéo sur le même sujet se trouvent proches l’une de l’autre, ce qui permet d’utiliser l’une pour trouver l’autre. C’est là l’aspect véritablement nouveau.
Ce que le modèle accepte
Chaque modalité présente une limite qui influence sa conception :
- Texte : 8 192 tokens d’entrée par appel
- Images : au maximum six par appel, au format PNG ou JPEG
- Video : des segments d’une durée ne dépassant pas 128 secondes, au format MP4 ou MOV
- Audio : des segments d’une durée ne dépassant pas 80 secondes, au format MP3 ou WAV
- PDFs : six pages par appel
Le choix parmi les huit types de tâches n’est pas purement esthétique. Il s’agit de SEMANTIC_SIMILARITY, CLASSIFICATION et CLUSTERING ; RETRIEVAL_DOCUMENT, RETRIEVAL_QUERY et CODE_RETRIEVAL_QUERY ; ainsi que de QUESTION_ANSWERING et FACT_VERIFICATION. Le type de tâche indiqué au moment de l’incorporation indique au modèle dans quelle direction orienter le vecteur. Un vecteur ajusté pour la récupération n’est pas identique à un vecteur ajusté pour la classification, même pour des entrées identiques.
Le tarif est de 0,20 $ par million de tokens de texte, avec une remise de 50 % via l’API par lots pour les travaux ne nécessitant pas de réponses en temps réel. Pour l’ingestion en masse d’une grande bibliothèque multimédia, l’approche par lots est presque toujours la meilleure solution.
Avertissement concernant la migration : les vecteurs provenant de gemini-embedding-001 (le modèle GA uniquement texte) ne peuvent pas être mélangés avec ceux de gemini-embedding-2-preview, car ces deux systèmes sont incompatibles et aucun outil ne permet de convertir les anciens vecteurs. Mettre à jour un système existant implique de prévoir un réindexage complet.
Architecture, composant par composant
La plupart des tutoriels se contentent de présenter un diagramme de flux. Il est plus utile d’expliquer chaque partie, la raison de son existence et ce qui cesse de fonctionner en son absence. Le système dispose de deux pipelines indépendants : l’ingestion, qui s’exécute hors ligne et traite les médias, et la consultation, qui fonctionne en temps réel et fournit des réponses.
Ingestion
Le chunker s’exécute localement via ffmpeg et divise les médias en segments correspondant aux limites du modèle : moins de 128 secondes pour la vidéo, moins de 80 pour l’audio. Il superpose également les segments voisins afin qu’une conversation située à une frontière conserve son sens des deux côtés.
Le embedder appelle l’API Gemini et transforme chaque chunk en vecteur. La vidéo et l’audio doivent d’abord être téléchargés via l’API Files de Google et attendre le traitement avant que l’endpoint d’incorporation ne puisse être utilisé. Le texte, quant à lui, est envoyé directement à l’API sans téléchargement préalable.
Le store est la base de données vectorielle. Elle conserve chaque vecteur accompagné de métadonnées : le chemin du fichier source, les timestamps, la modalité ainsi qu’une courte chaîne de prévisualisation. Ce sont ces métadonnées qui permettent aux réponses de citer des sources et de faire référence à l’enregistrement original.
Requêtes
La partie de la requête est délibérément simple, car la complexité réside dans l’ingestion des données. Au moment de la requête, la question est intégrée avec le type de tâche RETRIEVAL_QUERY, une recherche par similarité cosinus est effectuée, les meilleurs résultats, quel que soit leur format, sont rassemblés, ils sont formatés en un bloc de contexte et tout est transmis à un modèle génératif.
La réponse est générée par Gemini 2.5 Pro, un modèle capable de travailler avec des descriptions textuelles ainsi qu’avec des références à des segments vidéo dans le même contexte. Il comprend qu’un fragment étiqueté [VIDEO à 02:14] signifie quelque chose de différent d’un fragment étiqueté [TEXT p.3], et cite chacun en conséquence.
Séparer les deux éléments
Exécuter à la fois l’ingestion et les requêtes dans un même processus est une erreur courante. L’ingestion est lente : transformer une heure de vidéo en 40 fragments, envoyer chacun d’eux via l’API Files et attendre la génération des embeddings prend entre 15 et 30 minutes, ce qui est bien plus long que ce qu’un point de terminaison de requête peut supporter. De plus, l’ingestion peut être reprise. Si un fragment échoue, seules cette partie doit être retentée, sans perturber le processus de requête, et mélanger les deux rend cela plus difficile. Par conséquent, l’ingestion doit avoir lieu dans un job batch hors ligne, qu’il s’agisse d’un script, d’une entrée cron ou d’une étape de pipeline, tandis que les requêtes doivent être traitées via une API sans état. La base de données vectorielle constitue le lien entre ces deux composants.
Construire la partie d’ingestion
Environnement
Installez d’abord les dépendances :
pip install google-genai chromadb google-generativeai python-dotenv ffmpeg-python
FFmpeg est également nécessaire au niveau du système, à installer via brew install ffmpeg sur Mac ou apt install ffmpeg sur Ubuntu. Ensuite, un petit module de configuration configure le 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)
La prise en charge de la production doit gérer les trois modalités, administrer les téléchargements de médias volumineux, stocker des métadonnées utiles et fonctionner en cas d’échec de manière propre. Étape par étape :
Le client d’incorporation
Le client n’est qu’un enveloppe légère autour de l’API Gemini. Un détail est important : les vidéos et les audios doivent être téléchargés avant l’incorporation, contrairement aux images, qui peuvent être envoyées directement.
# 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
Le outil _wait_for_file est facile à ignorer, mais son omission a des conséquences coûteuses. Les téléchargements sont traités de manière asynchrone, et l’incorporation du contenu avant que l’état du fichier ne devienne ACTIVE provoque une erreur. Un intervalle de sondage de cinq secondes est prudent. Les petits fichiers sont traités plus rapidement, tandis que de gros vidéos peuvent nécessiter une minute entière.
Le découpeur de fichiers
Tout se passe localement ici. ffmpeg divise les fichiers en segments avec une zone de chevauchement à chaque limite. La raison de ce chevauchement est évidente : imaginez une discussion de 30 secondes commençant à la 85e seconde d’une vidéo. Avec des segments de 90 secondes sans chevauchement, les 5 premières secondes se trouvent dans le segment 0 et les 25 suivantes dans le segment 1, de sorte qu’une recherche sur ce sujet pourrait ne récupérer qu’un fragment. Avec 10 secondes de chevauchement, les deux segments contiennent la partie concernée, et la récupération donnera probablement le segment contenant la majeure partie de la 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
Le stockage vectoriel
ChromaDB convient à cet exemple car rien n’a besoin d’être configuré autour de lui. Il n’y a pas de serveur à faire fonctionner, pas de conteneur et pas de compte cloud, puisqu’il fonctionne à l’intérieur du processus Python et enregistre les données dans un dossier sur le disque. Pour un système de production avec des millions de fragments, on utiliserait plutôt Qdrant, Pinecone ou Weaviate, avec une interface presque identique.
# 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()
Le lanceur
Un script rassemble tout cela et traite un répertoire contenant des médias variés. Un système d’identification basé sur les hachages rend l’ingestion idempotente : exécuter un fichier deux fois échoue soit en cas d’ID déjà existant, ce qui peut être détecté et ignoré, soit grâce à une vérification effectuée avant l’ingestion.
# 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")
Construction du côté requête
L’ingestion est la partie difficile ; une fois le stock créé, les requêtes deviennent simples. Le flux consiste à intégrer la question, à rechercher dans le stock, à formater les résultats et à générer une réponse.
# 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)
}
Le paramètre similarity_threshold mérite une attention particulière. S’il est trop bas, du contenu irrelevant s’infiltre dans le contexte et perturbe le générateur. S’il est trop élevé, des résultats valables sont éliminés. Une fourchette de départ de 0,6 à 0,7 est raisonnable, à ajuster en effectuant des requêtes réelles sur un ensemble de test étiqueté.
Améliorer la précision de la récupération
Avoir un système de base fonctionnel, c’est une chose ; un système de production utile en exige davantage. Les améliorations se réalisent en trois étapes : la qualité de l’ingestion, la formulation de la requête et le filtrage après récupération.
Adaptez toujours le type de tâche
C’est le levier le plus important. RETRIEVAL_QUERY au moment de la requête et RETRIEVAL_DOCUMENT lors de l’ingestion ne sont pas optionnels. Ces deux vecteurs agissent dans des directions opposées, adaptés à l’asymétrie de la recherche, où les requêtes courtes renvoient vers des documents longs. Lors des tests, l’utilisation de RETRIEVAL_DOCUMENT pour les deux cas, c’est-à-dire la valeur par défaut, a entraîné une perte de 13 points de pourcentage en termes de précision de récupération. Ce bug ne génère aucune erreur et est facile à manquer.
Élargir d’abord la requête
Les requêtes courtes et ambiguës sont préjudiciables. « API limits engineering review » est médiocre. Une question plus complète, telle que « quelles décisions concernant les limites de débit de l’API ont été prises lors de la réunion d’évaluation technique, et pourquoi ? », fonctionne bien mieux. Une simple appel à un LLM avant l’étape d’incorporation peut automatiser cette réécriture :
# 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")
Dans un ensemble de questions issues d’une vidéo, l’expansion seule a porté le taux de rappel parmi les 5 meilleures réponses de 74 % à 81 %.
Exécuter plusieurs formulations en parallèle
Plutôt qu’une seule requête étendue, générer trois formulations et fusionner les résultats. Des formulations différentes mettent en évidence des fragments différents, et leur union permet de rappeler davantage d’informations.
# 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)
La recherche multi-demandes ajoute environ deux à trois appels API par question, avec un temps de réponse et des coûts supplémentaires. Elle est généralement rentable lorsque le manque d’un extrait pertinent a de lourdes conséquences. Pour des questions moins importantes, une seule demande étendue suffit.
Réclasser les résultats
La recherche par embedding est efficace pour trouver des segments sémantiquement liés, mais moins performante pour distinguer ce qui est « hautement pertinent pour cette question » de ce qui est « vaguement lié au même sujet ». Un outil de réclassement comble cette lacune.
# 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
Le schéma habituel récupère de 15 à 20 candidats grâce à la recherche intégrée, les fait tous passer par le réclassificateur et conserve les 5 meilleurs. Le réclassificateur est un encodeur croisé local, facile à exécuter, qui termine en moins de 100 ms sur un CPU, de sorte que la latence reste pratiquement inchangée.
Utiliser un seuil de similarité
Lorsque la base de connaissances ne contient rien de pertinent, la récupération renvoie néanmoins des résultats, car elle affiche toujours les correspondances les plus proches, même si elles ne sont pas pertinentes. Le seuil intégré dans le pipeline de requête gère ce problème. Les valeurs de similarité cosinus inférieures à environ 0,6 ne sont presque jamais pertinentes, et leur transmission au générateur produit des réponses fantaisistes. Les plages de valeurs varient selon le domaine. Le contenu technique doté d’une terminologie précise obtient de meilleurs résultats pour des correspondances authentiques, tandis que l’audio conversationnel obtient de moins bons résultats même s’il est pertinent ; par conséquent, les seuils doivent être calibrés à l’aide d’un ensemble de test étiqueté provenant de données réelles.
Résultats par rapport à l’ancienne configuration
Par rapport au système précédent, la transcription Whisper associée à un RAG texte uniquement a montré trois améliorations notables. La précision a connu la plus forte hausse pour les questions nécessitant un contexte visuel provenant d’une vidéo.
- Précision de récupération pour les questions basées sur des vidéos. L’ancien système atteignait une précision de rappel dans le top 5 d’environ 61 %, tandis que le nouveau système en atteint 84 %. Cette différence provient presque entièrement des questions répondues de manière visuelle : démonstrations à l’écran, schémas sur tableau blanc et interactions avec l’interface. Le système basé sur la transcription produisait des phrases comme « comme vous pouvez le voir ici » sans aucun contexte supplémentaire, tandis que les embeddings multimodaux capturaient directement les éléments visuels.
- Temps d’ingestion. Pour une vidéo de 60 minutes, l’ancien processus prenait quatre à cinq minutes pour effectuer la transcription avant que toute indexation ne puisse commencer. Le nouveau processus génère des segments indexés dès que le premier segment de 90 secondes est téléchargé et intégré, soit en environ deux minutes. Le temps total d’ingestion est légèrement plus long en raison des coûts de téléchargement, mais l’index devient utilisable plus rapidement.
- Coût par heure de vidéo. La transcription via Whisper coûtait environ 0,36 dollar par heure d’audio. L’intégration de 40 segments vidéo, aux prix actuels, coûte approximativement de 0,09 à 0,14 dollar, en fonction de la durée. La méthode multimodale est à la fois moins chère et plus précise.
Dimensions et infrastructure Matryoshka
Apprentissage par représentation Matryoshka (MRL) est l’une des fonctionnalités les plus intéressantes. Le modèle est entraîné de manière à ce que les informations sémantiques les plus importantes se trouvent dans les premières dimensions du vecteur, ce qui permet de tronquer le vecteur complet de 3072 dimensions avec très peu de perte en termes de qualité de récupération. La taille de sortie est déterminée une fois au moment de l’importation et reste inchangée, car le mélange de dimensions au sein d’une même collection n’est pas pris en charge.
Pour la plupart des systèmes, 1536 est la valeur par défaut idéale : cela représente la moitié de l’espace de stockage, accélère les requêtes et entraîne seulement une perte de qualité d’environ 2 %. 3072 n’a de sens que pour des systèmes à haute précision où le taux de rappel est crucial et où le coût n’est pas un facteur limitant. Une taille inférieure à 1536 n’est justifiée que lorsqu’il s’agit de dizaines de millions de fragments, lorsque le coût de stockage devient significatif. Même avec 256 dimensions, la qualité reste d’environ 88 %, ce qui est parfaitement acceptable pour de nombreuses applications.
Ce que cela permet
- Bases de connaissances institutionnelles où les enregistrements d’intégration, les vidéos de formation et les guides de mise en œuvre se trouvent aux côtés de la documentation écrite. Un nouvel employé qui demande comment fonctionne le pipeline de déploiement reçoit à la fois le manuel opérationnel et le guide de huit minutes.
- Systèmes de conformité et d’audit qui doivent permettre de retrouver des preuves dans différents formats. Une équipe juridique qui cherche où la politique de retours a été abordée avec les clients peut trouver à la fois les documents de politique et les enregistrements des appels grâce à une seule recherche.
- Copilotes de support qui affichent la séquence exacte de 40 secondes d’une démonstration du produit à côté de l’article d’aide. Regarder quelqu’un effectuer une tâche est souvent plus efficace que de la lire.
Règles pratiques
- Commencez par le contenu le plus précieux et actuellement inaccessible : des enregistrements contenant des connaissances que personne ne peut trouver pour l’instant.
- Utilisez 1536 dimensions, et n’y revenez que si le coût de stockage devient un véritable obstacle.
- Appliquez le seuil de similarité dès le premier jour. Sans lui, le système produit des erreurs lorsque la base de connaissances manque de contenu pertinent. 0,65 constitue un bon point de départ, ajusté en fonction des requêtes réelles.
- Élargissez les requêtes avant d’effectuer l’incorporation. Une requête adressée à un LLM coûte une fraction de centime et améliore considérablement le taux de récupération pour les vidéos, où les termes de recherche correspondent rarement exactement aux mots prononcés.
- Gardez le job d’ingestion et le serveur de requêtes complètement séparés, en ne partageant que la base de données vectorielle. Un état partagé entre eux engendre des problèmes de fiabilité.
- Assurez-vous que les métadonnées des blocs soient riches. Les embeddings gèrent la récupération sémantique, tandis que les métadonnées s’occupent du reste : le filtrage par date, par interlocuteur, par type d’enregistrement ou par niveau de confiance.
Problèmes dans la version préliminaire
Le modèle est en version préliminaire publique, et plusieurs limitations s’appliquent :
- La limite de 80 secondes pour l’audio est gênante pour les podcasts ou les conférences de plusieurs heures, ce qui génère de nombreux petits blocs et de nombreuses appels API. Elle sera probablement augmentée lors de la disponibilité générale.
- La classe GoogleGenerativeAIEmbeddings de LangChain ne prend en charge que du texte. L’incorporation d’audio et de vidéo nécessite directement le SDK google-genai, relié ensuite à un pipeline LangChain, ce qui fonctionne mais n’est pas élégant.
- Pendant la phase de bêta, il n’y a pas de niveau gratuit, de sorte que les développeurs individuels doivent payer pour chaque appel d’incorporation. La remise de 50 % offerte par l’API par lots aide, mais exige que l’ingestion soit structurée comme une tâche en arrière-plan et non comme un processus en temps réel.
- Aucun poids d’attention ni aucun autre signal d’interprétabilité n’est exposé. Il est impossible de déterminer quels frames ont le plus influencé une incorporation, ce qui rend le débogage des mauvaises récupérations plus difficile que pour le texte, car on ne dispose que de scores de similarité globaux sans savoir ce qui les a générés.