Suchen in Video und Audio nach Bedeutung: Aufbau eines multimodalen RAG-Systems mit Gemini Embedding 2
Multimodales RAG mit Gemini Embedding 2: ein Vektorraum für Text, Audio, Video und PDFs, sodass kein Transkriptionsschritt nötig ist, zusätzlich Chunking, ChromaDB und Neuranking.
Quellen: *My RAG System Missed 80% of My Data. This One Change Fixed It.*
Betrachten wir eine für einen Kunden erstellte Wissensdatenbank: Zehntausende von Unternehmensdokumenten, die indiziert und suchbar sind, unterstützt durch einen übersichtlichen Abrufmechanismus, der in den Demonstrationen hervorstach. Alle waren zufrieden – bis jemand fragte, warum das System keine Fragen beantworten konnte, die aus den wöchentlichen Gesprächsaufnahmen aller Mitarbeiter stammten. Die ehrliche Antwort war peinlich: Das System konnte diese Aufnahmen überhaupt nicht erkennen.
Ein Großteil des wertvollsten Wissens dieses Unternehmens befand sich in Videos. Jeder Sprint wurde mit Produktvorführungen aufgezeichnet. Ingenieure erklärten das Datenbankschema, während sie darauf auf dem Bildschirm zeigten. Produktmanager navigierten durch die Benutzeroberfläche und sprachen dabei über die Probleme der Kunden. All das war nicht zugänglich, denn alle verfügbaren Embedding-Modelle verstanden nur Text.
Die übliche Lösung ist die Transkription: Das Audio wird durch Whisper laufen gelassen, es werden Stunden Rechenzeit verbraucht, und am Ende muss man eine beträchtliche Rechnung im Cloud-Dienst begleichen. Dadurch wird ein tieferliegender Mangel offensichtlich. Eine Transkription entzieht den Aufnahmen den Kontext, der ihren Wert ausmacht. „Wie Sie hier sehen können“ wird zu einem leeren Satz. Ein Skizzenaufzeichnung auf einer Tafel verwandelt sich in Stille. Der Tonfall eines Ingenieurs, der sagt „Das ist vorerst in Ordnung“, verschwindet. Die Wörter werden indiziert, während die Bedeutung verloren geht.
Warum ein gemeinsamer Embedding-Raum wichtig ist
Bevor es um den Code geht, hilft es, zu verstehen, warum Gemini Embedding 2 in seiner Art anders ist und nicht nur eine verbesserte Version desselben Konzepts darstellt.
Multimodale Suchsysteme haben ein gemeinsames Kernproblem: Jede Modalmethode existiert in ihrem eigenen Vektorraum. Sowohl Texte als auch Audiodaten werden als Zahlenlisten dargestellt, doch diese Zahlen tragen keine gemeinsame Bedeutung, und die Kosinusähnlichkeit zwischen Vektoren unterschiedlicher Modelle, die mit unterschiedlichen Zielen trainiert wurden, sagt nichts aus. Die Lösung bestand darin, separate Verarbeitungspfade zu verwenden, zu denen eine Bewertungsmethode hinzugefügt wurde – eine Textsuche, eine Audiosuche und eventuell eine Bildsuche –, um anschließend versuchen, die kombinierten Ergebnisse zu ranken. Dieses Design scheiterte auf zwei ernsthafte Weisen.
Ein Problem war der Verlust der cross-modalen Beziehungen. Wenn ein Video zeigt, was in einem Dokument beschrieben wird, gibt es zwischen den beiden Elementen, da sie in separaten Räumen liegen, keine mathematische Verbindung; daher kann eine Suche das Dokument finden, das Video jedoch übersehen, oder umgekehrt.
Der andere Aspekt waren die betrieblichen Herausforderungen: Drei Indizes, die gepflegt werden mussten, drei Einpflegepipelines, die überwacht werden mussten, sowie drei verschiedene Fehlerarten, die behoben werden mussten. Die meisten Teams gaben die Suche nach Audio- und Videodaten auf und begnügten sich mit einer Suchfunktion nur für Texte.
Mit Gemini Embedding 2 werden Texte, Bilder, Videos, Audiodaten und Dokumente alle in einen gemeinsamen Raum abgebildet. Eine Textanfrage und ein Videoclip zum selben Thema landen in der Nähe voneinander, sodass entweder die eine oder die andere verwendet werden kann, um das andere zu finden. Das ist der wirklich neue Aspekt.
Was das Modell akzeptiert
Jede Datentypklasse hat eine Obergrenze, die die Gestaltung beeinflusst:
- Text: 8.192 Tokens Eingabe pro Aufruf
- Bilder: höchstens sechs pro Aufruf, im Format PNG oder JPEG
- Video: Abschnitte mit einer Länge von maximal 128 Sekunden, im Format MP4 oder MOV
- Audiodaten: Abschnitte mit einer Länge von maximal 80 Sekunden, im Format MP3 oder WAV
Die Auswahl aus den acht Aufgabentypen ist nicht nur kosmetischer Natur. Es handelt sich um SEMANTIC_SIMILARITY, CLASSIFICATION und CLUSTERING; RETRIEVAL_DOCUMENT, RETRIEVAL_QUERY und CODE_RETRIEVAL_QUERY; sowie QUESTION_ANSWERING und FACT_VERIFICATION. Der bei der Einbettung festgelegte Aufgabentyp gibt dem Modell an, in welche Richtung der Vektor gestaltet werden soll. Ein für die Suche optimierter Vektor unterscheidet sich von einem für die Klassifizierung optimierten Vektor – selbst bei identischen Eingaben.
Der Preis beträgt 0,20 US-Dollar pro Million Text-Token, wobei durch die Batch-API ein Rabatt von 50 Prozent für Arbeiten gewährt wird, die keine Echtzeitantworten benötigen. Für die Masseneingabe einer großen Mediathek ist die Batch-Methode fast immer der bessere Weg.
Eine Migrationswarnung: Vektoren von gemini-embedding-001 (das rein textbasierte GA-Modell) können nicht mit denen von gemini-embedding-2-preview vermischt werden, da die beiden Systeme inkompatibel sind und es keine Möglichkeit gibt, alte Vektoren umzuwandeln. Ein Upgrade eines bestehenden Systems erfordert daher die Planung einer vollständigen Neindeckung.
Architektur, Komponente für Komponente
Die meisten Anleitungen zeigen lediglich ein Flussdiagramm und hören auf. Es ist nützlicher, jeden Bestandteil zu erklären – warum er vorhanden ist und was ohne ihn schiefgeht. Das System verfügt über zwei unabhängige Pipelines: den Eingabepipeline, der offline läuft und Medien verarbeitet, sowie den Abfrapipeline, der in Echtzeit arbeitet und Antworten liefert.
Eingabepipeline
Der Chunker läuft lokal unter ffmpeg und teilt die Medien in Segmente auf, die in die Grenzen des Modells passen: unter 128 Sekunden für Video, unter 80 für Audio. Zudem überschneiden sich die benachbarten Segmente, sodass ein Gespräch, das an einer Grenze endet, auf beiden Seiten seine Bedeutung beibehält.
Der Embedder ruft die Gemini-API auf und wandelt jeden Chunk in einen Vektor um. Video- und Audiodateien müssen zunächst in die Google Files-API hochgeladen und der Verarbeitung zugeführt werden, bevor der Embedding-Endpunkt aufgerufen werden kann. Texte werden hingegen ohne Hochladen direkt an die API gesendet.
Die Speicherstelle ist die Vektordatenbank. Sie speichert jeden Vektor zusammen mit Metadaten: dem Pfad zur Quelldatei, Zeitstempeln, der Modalität sowie einem kurzen Vorschau-Text. Genau diese Metadaten ermöglichen es, Antworten mit Quellen zu versehen und auf die ursprüngliche Aufnahme zu verweisen.
Abfragen
Die Abfrageseite ist absichtlich einfach, denn die Komplexität liegt in der Erfassung der Daten. Zum Zeitpunkt der Abfrage wird die Frage mit dem Task-Typ RETRIEVAL_QUERY eingebettet, es wird eine Suche nach kosinusähnlicher Ähnlichkeit durchgeführt, die besten Ergebnisse unabhängig von ihrer Art werden gesammelt, in einen Kontextblock formatiert und an ein generatives Modell übergeben.
Die Beantwortung erfolgt durch Gemini 2.5 Pro, ein Modell, das mit textbasierten Beschreibungen sowie Verweisen auf Videoabschnitte im selben Kontext arbeiten kann. Es versteht, dass ein Abschnitt mit der Kennzeichnung [VIDEO at 02:14] etwas anderes bedeutet als einer mit der Kennzeichnung [TEXT p.3], und zitiert jeweils entsprechend.
Trennung der beiden Komponenten
Das Ausführen von Einpflegvorgängen und Abfragen im selben Prozess ist ein häufiger Fehler. Die Einpflege ist langsam: Um eine Stunde Videomaterial in 40 Blöcke zu zerlegen, jeden Block über die Files API zu senden und auf die Erstellung der Embeddings zu warten, vergehen 15 bis 30 Minuten – deutlich länger, als ein Abfrageendpunkt warten kann. Zudem lässt sich die Einpflege wiederholen. Fällt ein bestimmter Block fehl, sollte nur dieser Block erneut versucht werden, ohne den Abfrageablauf zu stören; das Mischen beider Vorgänge erschwert dies. Daher gehört die Einpflege zu einem offline Batch-Job – sei es ein Skript, eine Cron-Einträge oder eine Pipeline-Phase – während die Abfragen über eine stateless API abgewickelt werden. Die Vektordatenbank dient als Verbindung zwischen beiden Komponenten.
Erstellung der Einpflegkomponente
Umgebung
Installieren Sie zunächst die Abhängigkeiten:
pip install google-genai chromadb google-generativeai python-dotenv ffmpeg-python
Auch auf Systemebene ist ffmpeg erforderlich, wobei man auf einem Mac mit „brew install ffmpeg“ oder auf Ubuntu mit „apt install ffmpeg“ vorgehen kann. Anschließend stellt ein kleines Konfigurationsmodul den Client ein:
# 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)
Die Verarbeitung der Erzeugnisse muss alle drei Formate handhaben, das Hochladen großer Mediendateien verwalten, sinnvolle Metadaten speichern und fehlerfrei funktionieren. Schritt für Schritt:
Der Embedding-Client
Der Client ist eine einfache Abdeckung für die Gemini-API. Ein wichtiger Punkt: Video- und Audiodateien müssen vor dem Einbetten hochgeladen werden, im Gegensatz zu Bildern, die direkt gesendet werden können.
# 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
Der Hilfsfunktion _wait_for_file kann leicht übersprungen werden, doch ihr Fehlen hat erhebliche Folgen. Die Hochläader werden asynchron verarbeitet, und das Einbetten vor dem Erreichen des Zustands „ACTIVE“ führt zu einem Fehler. Ein Abfristintervall von fünf Sekunden ist vorsichtig gewählt. Kleine Dateien werden schneller verarbeitet, während große Videos bis zu einer ganzen Minute dauern können.
Der Chunker
Alles hier findet lokal statt. ffmpeg teilt Dateien in Segmente auf, wobei an jeder Grenze ein Überschneidungsbereich vorhanden ist. Der Grund für diese Überschneidung ist leicht nachvollziehbar. Stellen Sie sich eine 30-sekündige Diskussion vor, die 85 Sekunden in ein Video hinein beginnt. Bei 90-sekündigen Segmenten ohne Überschneidung landen die ersten 5 Sekunden im Segment 0 und die restlichen 25 Sekunden im Segment 1 – daher könnte eine Abfrage zu diesem Thema nur ein Segment liefern, das nur einen Teil der Diskussion enthält. Bei 10 Sekunden Überschneidung enthalten beide Segmente den Übergangsbereich, sodass die Abfrage wahrscheinlich das Segment zurückgibt, das den größten Teil der Diskussion enthält.
# 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
Der Vektor-Speicher
ChromaDB eignet sich für dieses Beispiel, weil nichts extra eingerichtet werden muss. Es gibt keinen Server, keinen Container und kein Cloud-Konto, da es innerhalb des Python-Prozesses arbeitet und die Daten in einem Ordner auf der Festplatte speichert. Für ein Produktivsystem mit Millionen von Datenblöcken würden man Qdrant, Pinecone oder Weaviate verwenden, wobei die Benutzeroberfläche fast identisch ist.
# 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()
Der Ausführende
Ein Skript verbindet alles miteinander und verarbeitet ein Verzeichnis mit gemischten Medien. Ein auf Hash basierendes ID-System sorgt dafür, dass die Einpflegung idempotent ist: Wenn ein Datei zweimal ausgeführt wird, scheitert dies entweder an einer doppelten ID, was erkannt und übersprungen werden kann, oder es wird vor der Einpflegung durch eine Überprüfung geschützt.
# 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")
Aufbau der Abfrageseite
Die Eingabe der Daten ist der schwierige Teil, doch sobald der Speicher vorhanden ist, ist das Abfragen einfach. Der Ablauf besteht darin, die Frage einzubetten, im Speicher zu suchen, die Ergebnisse zu formatieren und eine Antwort zu erzeugen.
# 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)
}
Der Parameter similarity_threshold verdient Aufmerksamkeit. Wenn er zu niedrig eingestellt wird, dringen irrelevante Inhalte in den Kontext ein und verwirren den Generator. Wenn er zu hoch eingestellt wird, werden gültige Ergebnisse ausgeschlossen. Ein anfänglicher Wertebereich von 0,6 bis 0,7 ist angemessen und sollte durch das Ausführen eigentlicher Abfragen an einem gelabelten Testset angepasst werden.
Erhöhung der Suchgenauigkeit
Ein grundlegend funktionsfähiges System ist eine Sache – ein nützliches Produktivsystem erfordert jedoch mehr. Verbesserungen erzielt man in drei Schritten: Qualität der Dateneingabe, Formulierung der Abfrage sowie Filtern nach der Suche.
Passen Sie den Aufgabentyp immer an
Das ist der wichtigste Hebel überhaupt. RETRIEVAL_QUERY zum Zeitpunkt der Abfrage sowie RETRIEVAL_DOCUMENT bei der Aufnahme sind keine Optionen. Die beiden Vektoren wirken in entgegengesetzte Richtungen und sind auf die Asymmetrie der Suche abgestimmt, wobei kurze Abfragen auf lange Dokumente verweisen. Bei Tests, bei denen für beide Fälle der standardmäßige, nachlässige Ansatz mit RETRIEVAL_DOCUMENT verwendet wurde, kostete dies 13 Prozentpunkte an Genauigkeit bei der Dokumentenerfassung. Der Fehler verursacht keine Fehlermeldung und ist leicht übersehbar.
Erweitern Sie die Abfrage zunächst
Kurze, unklare Abfragen schaden. „API limits engineering review“ ist unzureichend. Eine ausführlichere Frage wie „Welche Entscheidungen bezüglich der API-Grenzwerte wurden in der Engineering-Review-Sitzung getroffen und warum?“ funktioniert deutlich besser. Ein kurzer Aufruf eines LLM vor dem Embedding-Schritt kann die Umformulierung automatisieren:
# 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")
Bei einer Reihe von Fragen, die aus Videos stammen, erhöhte allein die Erweiterung die Erinnerungsrate der Top-5 von 74 % auf 81 %.
Mehrere Formulierungen parallel ausführen
Anstelle einer einzigen erweiterten Abfrage sollten drei Formulierungen erstellt und die Ergebnisse zusammengeführt werden. Verschiedene Wortstellungen bringen unterschiedliche Teile zum Vorschein, wodurch mehr Informationen erinnert werden können.
# 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)
Die Mehrfachabfragesuche erfordert pro Frage etwa zwei bis drei API-Aufrufe, was zu zusätzlicher Verzögerung und Kosten führt. Sie lohnt sich in der Regel dann, wenn das Fehlen eines relevanten Clips teuer ist. Für weniger wichtige Fragen reicht eine erweiterte Abfrage aus.
Die Kandidaten neu ranken
Die Embedding-Suche ist gut darin, semantisch verwandte Abschnitte zu finden, weniger jedoch darin, zwischen „sehr relevant für diese Frage“ und „nur vage zum gleichen Thema“ zu unterscheiden. Ein Re-Ranker schließt diese Lücke.
# 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
Das übliche Muster holt 15 bis 20 Kandidaten durch eingebettete Suche ab, führt sie alle durch den Re-Ranker und behält die besten 5. Der Re-Ranker ist ein lokaler Cross-Encoder, der kostenlos ausgeführt werden kann und in weniger als 100 ms auf einer CPU fertig ist, sodass sich die Latenz kaum verändert.
Verwenden Sie einen Ähnlichkeitsschwellenwert
Auch wenn die Wissensdatenbank nichts Relevantes enthält, liefert die Suche weiterhin Ergebnisse, da sie stets die nächsten Treffer zurückgibt, selbst wenn diese nicht relevant sind. Der Schwellenwert im Abfragesystem kümmert sich darum. Blöcke mit einer Kosinus-Ähnlichkeit unter etwa 0,6 sind fast nie relevant, und ihre Übermittlung an den Generator führt zu erfundenen Antworten. Die Werte variieren je nach Domain. Technischer Inhalt mit präziser Terminologie erhält höhere Punktzahlen für echte Treffer, während konversationeller Audio-Inhalt selbst bei Relevanz niedrigere Punktzahlen erhält – daher sollten die Schwellenwerte anhand eines mit Etiketten versehenen Testsets aus echten Daten kalibriert werden.
Ergebnisse im Vergleich zur alten Konfiguration
Im Vergleich zum vorherigen System zeichneten sich bei Whisper-Transkription in Kombination mit textbasiertem RAG drei Aspekte besonders aus. Die Genauigkeit verbesserte sich am stärksten bei Fragen, die einen visuellen Kontext aus Videos benötigten.
- Genauigkeit bei der Abrufung von Informationen zu fragen mit Videoquellen. Das alte System erreichte eine Recall-Rate von etwa 61 % für die Top-5-Ergebnisse, während das neue System 84 % erreichte. Der Unterschied entstand fast ausschließlich bei Fragen, die visuell beantwortet werden mussten: Schritt-für-Schritt-Anleitungen auf dem Bildschirm, Diagramme an der Tafel sowie Interaktionen mit der Benutzeroberfläche. Das auf Transkripten basierende System erzeugte Ausdrücke wie „wie Sie hier sehen können“ ohne jeglichen Kontext, während die multimodalen Embeddings die visuellen Elemente direkt erfassen konnten.
- Einpflegzeit. Bei einem 60-minütigen Video benötigte das alte Verfahren vier bis fünf Minuten für die Transkription, bevor mit der Indexierung begonnen werden konnte. Das neue Verfahren erzeugt bereits nach dem Hochladen und Einbetten des ersten 90-sekündigen Abschnitts indexierte Teile, also nach etwa zwei Minuten. Die Gesamtzeit zur Einpflegung ist aufgrund der Hochladeaufwände etwas länger, doch der Index wird früher nutzbar.
- Kosten pro Stunde Video. Die Transkription mit Whisper kostete etwa 0,36 US-Dollar pro Stunde Audio. Das Einbetten von 40 Video-Teilen kostet bei den aktuellen Preisen je nach Länge etwa 0,09 bis 0,14 US-Dollar. Der multimodale Ansatz ist sowohl günstiger als auch genauer.
Matryoshka-Dimensionen und Infrastruktur
Matryoshka Representation Learning (MRL) ist eine der interessanteren Eigenschaften. Das Modell wird so trainiert, dass die wichtigsten semantischen Informationen in den ersten Dimensionen des Vektors liegen, was bedeutet, dass der vollständige 3072-dimensionalen Vektor mit nur geringem Qualitätsverlust gekürzt werden kann. Die Ausgabegröße wird einmal bei der Eingabe festgelegt und beibehalten, da eine Mischung von Dimensionen innerhalb derselben Sammlung nicht unterstützt wird.
Für die meisten Systeme ist 1536 die richtige Standardgröße: die Hälfte der Speicherkapazität, schnellere Abfragen und nur ein Qualitätsverlust von etwa 2 %. 3072 macht nur bei Systemen mit hoher Präzision Sinn, bei denen die Vollständigkeit der Ergebnisse entscheidend ist und Kosten kein Limit darstellen. Ein Wert unter 1536 ist nur dann gerechtfertigt, wenn es um Dutzende Millionen von Datenblöcken geht, bei denen die Speicherkosten eine Rolle spielen. Selbst bei 256 Dimensionen bleibt die Qualität bei etwa 88 %, was für viele Anwendungen vollkommen akzeptabel ist.
Was dies ermöglicht
- Institutionelle Wissensbasen, in denen Einarbeitungsaufnahmen, Schulungsvideos und technische Anleitungen neben schriftlicher Dokumentation vorhanden sind. Ein Neuzugang, der fragt, wie der Bereitstellungsprozess funktioniert, erhält sowohl das Handbuch als auch die achtminütige Anleitung.
- Konformitäts- und Prüfungssysteme, die Beweismittel aus verschiedenen Quellen abgleichen müssen. Ein Rechtsteam, das nachdenkt, wo die Rückgabepolitik mit Kunden besprochen wurde, kann mithilfe einer Suche sowohl die Policy-Dokumente als auch die Anrufaufnahmen finden.
- Hilfskopiloten, die neben dem Hilfsteil genau den 40-sekündigen Clip aus einer Produktvorführung anzeigen. Jemandem zuzusehen, wie eine Aufgabe ausgeführt wird, ist oft effektiver als das Lesen darüber.
Praktische Regeln
- Beginnen Sie mit dem wertvollsten gesicherten Inhalt: Aufnahmen, die Wissen enthalten, das derzeit niemand finden kann.
- Verwenden Sie 1536 Dimensionen und kehren Sie erst dann darauf zurück, wenn die Speicherkosten zu einem echten Engpass werden.
- Wenden Sie bereits ab dem ersten Tag die Ähnlichkeitsschwelle an. Ohne sie erzeugt das System Fehlinformationen, wenn die Wissensdatenbank an relevantem Inhalt mangelt. 0,65 ist ein angemessener Ausgangspunkt, der anhand tatsächlicher Abfragen angepasst wird.
- Erweitern Sie die Abfragen vor dem Einbetten. Ein Aufruf eines LLM kostet nur einen Bruchteil eines Cent und verbessert deutlich die Trefferquote bei Videos, bei denen Suchbegriffe selten exakt den gesprochenen Worten entsprechen.
- Halten Sie die Eingabearbeitung und den Abfrageserver vollständig getrennt und teilen Sie nur die Vektordatenbank. Ein gemeinsamer Zustand zwischen ihnen führt zu Zuverlässigkeitsproblemen.
- Sorgen Sie dafür, dass die Metadaten der Blöcke umfangreich sind. Embeddings kümmern sich um die semantische Suche, während die Metadaten alles Weitere bewältigen: Filtern nach Datum, Sprecher, Aufnahmetyp oder Zuverlässigkeit.
Unvollkommenheiten in der Vorschau
Das Modell ist in einer öffentlichen Vorschau verfügbar, weshalb einige Einschränkungen gelten:
- Die 80-Sekunden-Limit für Audio ist für Podcasts oder mehrstündige Vorträge unpraktisch, da dadurch viele kleine Blöcke und viele API-Aufrufe entstehen. Dieses Limit wird voraussichtlich bei der allgemeinen Verfügbarkeit erhöht.
- Die Klasse GoogleGenerativeAIEmbeddings von LangChain unterstützt nur Text. Für Audio- und Video-Embeddings ist das google-genai SDK direkt erforderlich, welches anschließend in einen LangChain-Pipeline eingebunden wird. Dies funktioniert zwar, ist aber nicht elegant gelöst.
- Während der Vorabversion gibt es keine kostenlose Stufe, sodass einzelne Entwickler für jeden Embedding-Aufruf bezahlen müssen. Der 50%-Rabatt für die Batch-API hilft zwar, erfordert aber, dass die Datenverarbeitung als Hintergrundaufgabe und nicht als Echtzeitprozess strukturiert wird.
- Es werden weder Aufmerksamkeitsgewichte noch andere Signalien zur Interpretierbarkeit bereitgestellt. Es ist unmöglich zu überprüfen, welche Frames den Embedding am stärksten beeinflusst haben, wodurch das Beheben von Problemen bei der Suche schwieriger ist als bei Texten, da nur aggregierte Ähnlichkeitswerte vorliegen, ohne Kenntnis der zugrunde liegenden Faktoren.