Wyszukiwanie wideo i audio według znaczenia: budowa systemu multimodalnego RAG z użyciem embeddingów Gemini 2
Multimodalny RAG z Gemini Embedding 2: jeden przestrzeń wektorowa dla tekstu, audio, wideo i plików PDF, więc brak kroku transkrypcji, plus dzielenie na fragmenty, ChromaDB i ponowne sortowanie.
Źródła: *My RAG System Missed 80% of My Data. This One Change Fixed It.*
Rozważmy bazę wiedzy stworzoną dla klienta: dziesiątki tysięcy dokumentów firmy, skatalogowanych i dostępnych do wyszukiwania, obsługiwanych przez sprawną procedurę odzyskiwania informacji, która wyróżniała się podczas prezentacji. Wszyscy byli zadowoleni, dopóki ktoś nie zapytał, dlaczego system nie potrafi odpowiadać na pytania z cotygodniowych nagranych spotkań z całym zespołem. Szczera odpowiedź była kłopotliwa – system w ogóle nie mógł ich dostrzec.
Znaczna część najcenniejszej wiedzy tej firmy znajdowała się w filmach wideo. Podczas każdego etapu pracy rejestrowano prezentacje produktu. Inżynierowie wyjaśniali schemat bazy danych, wskazując na niego na ekranie. Menedżerowie produktu przeglądali interfejs, rozmawiając o problemach klientów. Nic z tego nie było dostępne, ponieważ wszystkie dostępne modele embeddingowe rozumiały jedynie tekst.
Zwykłym rozwiązaniem jest transkrypcja: przepuszczenie dźwięku przez Whisper, zużycie wielu godzin obliczeniowych oraz konieczność zapłacenia niemałego rachunku za usługi chmurowe. To jednak ujawnia głębszą wadę. Transkrypcja pozbawia nagrania kontekstu, który nadaje im wartość. Fraza „Jak widać tutaj” staje się pustym zdaniem, szkic na tablicy toczystej zamienia się w ciszę, a ton, którym inżynier mówi „Na razie to w porządku”, znika. Słowa są indeksowane, ale ich znaczenie zostaje odrzucone.
Dlaczego ważny jest wspólny przestrzeń embeddingów
Zanim przejdziemy do kodu, warto zrozumieć, dlaczego Gemini Embedding 2 różni się charakterem i nie jest po prostu lepszą wersją tego samego rozwiązania.
Systemy wyszukiwania multimodalnego borykały się z jednym podstawowym problemem: każda z modalności funkcjonowała we własnym przestrzeni wektorowej. Embedding tekstowy i embedding audio to oba listy liczb, jednak liczby te nie mają żadnego wspólnego znaczenia, a podobieństwo kosinowe wektorów z różnych modeli, szkolenych na różnych celach, nic nie mówi. Rozwiązaniem było użycie oddzielnych procesów z dodaną na górze heurystyką oceny – wyszukiwania tekstowego, audio oraz ewentualnie obrazowego, a następnie próba sklasyfikowania połączonych wyników. Taki projekt zawodził z dwóch poważnych powodów.
Jednym z nich była utrata relacji międzymodalnych. Gdy wideo pokazuje to, co opisuje dokument, oba elementy nie mają żadnego matematycznego powiązania, gdy znajdują się w oddzielnych przestrzeniach, więc zapytanie może wyświetlić dokument, pomijając wideo, lub odwrotnie.
Drugim problemem były trudności operacyjne: trzy indeksy do utrzymywania, trzy pipeline do przetwarzania danych do monitorowania oraz trzy zestawy awarii do naprawiania. Większość zespołów zrezygnowała z wyszukiwania audio i wideo, decydując się na wyszukiwanie wyłącznie tekstowe.
Dzięki Gemini Embedding 2 tekst, obrazy, wideo, audio i dokumenty są mapowane do jednego przestrzeni. Zapytanie tekstowe i klip wideo na ten sam temat znajdują się blisko siebie, więc można użyć jednego, aby znaleźć drugi. To jest naprawdę nowa cecha.
Co akceptuje model
Każda z modalności ma określone ograniczenia, które wpływają na jego projekt:
- Tekst: 8 192 tokenów wejściowych na jedną próbę
- Obrazy: maksymalnie sześć na próbę, w formacie PNG lub JPEG
- Wideo: fragmenty nie dłuższe niż 128 sekund, w formacie MP4 lub MOV
- Audio: fragmenty nie dłuższe niż 80 sekund, w formacie MP3 lub WAV
- PDF-y: sześć stron na jedną próbę
Wybór spośród ośmiu typów zadań nie jest kwestią estetyczną. Są to SEMANTIC_SIMILARITY, CLASSIFICATION i CLUSTERING; RETRIEVAL_DOCUMENT, RETRIEVAL_QUERY i CODE_RETRIEVAL_QUERY; a także QUESTION_ANSWERING i FACT_VERIFICATION. Typ zadania określony w momencie tworzenia wektora wskazuje modelowi, w jakim kierunku powinien go ukształtować. Wektor dostosowany do wyszukiwania nie jest taki sam jak ten dostosowany do klasyfikacji, nawet przy identycznych danych wejściowych.
Cena wynosi 0,20 dolara za milion tokenów tekstowych, przy 50% zniżce przy użyciu API typu batch dla zadań, które nie wymagają odpowiedzi w czasie rzeczywistym. Przy masowym importowaniu dużej bazy mediów, metoda batch jest niemal zawsze lepszym rozwiązaniem.
Jedno ostrzeżenie dotyczące migracji: wektory z gemini-embedding-001 (model GA zawierający tylko tekst) nie mogą być łączone z wektorami z gemini-embedding-2-preview, ponieważ te dwa rozwiązania są niekompatybilne, a żaden mechanizm nie przekształca starych wektorów. Aktualizacja istniejącego systemu oznacza konieczność przewidzenia kosztów pełnej ponownej indeksacji.
Architektura, komponent po komponencie
Większość poradników pokazuje jedynie diagram przepływu i na tym się kończy. Bardziej przydatne jest wyjaśnienie każdej części, dlaczego istnieje oraz co się psuje bez niej. System składa się z dwóch niezależnych procesów: procesu pobierania danych, który działa offline i przetwarza media, oraz procesu wyszukiwania, który działa w czasie rzeczywistym i dostarcza odpowiedzi.
Proces pobierania danych
Chunker działa lokalnie w ffmpeg i dzieli materiały multimedialne na segmenty pasujące do ograniczeń modelu: poniżej 128 sekund dla wideo, poniżej 80 sekund dla audio. Ponadto nakłada się na sąsiednie segmenty, dzięki czemu rozmowa znajdująca się na granicy zachowuje swoje znaczenie po obu stronach.
Embedder wywołuje API Gemini i przekształca każdy fragment w wektor. Wideo i audio muszą najpierw zostać przesłane do API Files Google i oczekiwać na przetworzenie, zanim będzie można wywołać punkt końcowy do embedowania. Tekst trafia bezpośrednio do API, bez konieczności jego przesyłania.
Store to baza danych wektorowych. Przechowuje każdy wektor wraz z metadanymi: ścieżką do pliku źródłowego, czasami zapisu, typem danych oraz krótkim tekstem prezentacji. To właśnie te metadane umożliwiają odpowiedziom cytowanie źródeł i odnoszenie się do oryginalnej nagrania.
Wyszukiwanie
Część związana z zapytaniem jest celowo prosta, ponieważ złożoność leży w procesie pobierania danych. W momencie wysyłania zapytania jest ono zawarte w typie zadania RETRIEVAL_QUERY, przeprowadzana jest wyszukiwanka na podstawie podobieństwa kosinowego, zbierane są najlepsze wyniki niezależnie od ich formy, które następnie są formatowane w blok kontekstowy, a wszystko trafia do modelu generatywnego.
Odpowiadanie za to odpowiada Gemini 2.5 Pro – model, który potrafi pracować z opisami tekstowymi oraz odniesieniami do fragmentów wideo w tym samym kontekście. Rozumie on, że fragment oznaczony jako [VIDEO at 02:14] oznacza coś innego niż ten oznaczony jako [TEXT p.3], i odpowiednio je cytuje.
Rozdzielenie tych dwóch elementów
Wykonywanie procesu pobierania danych i wysyłania zapytań w jednym procesie to powszechny błąd. Proces pobierania danych jest powolny: przekształcenie godziny nagrania wideo na 40 fragmentów, przesłanie każdego z nich przez API Files oraz oczekiwanie na wyniki embeddingów zajmuje od 15 do 30 minut, co jest znacznie dłużej, niż może poczekać punkt końcowy obsługujący zapytania. Proces pobierania danych może być również ponawiany – jeśli jeden fragment się nie uda, należy spróbować ponownie tylko ten fragment, bez zakłócania procesu wysyłania zapytań, a łączenie tych dwóch czynności utrudnia to zadanie. Dlatego proces pobierania danych powinien odbywać się w tle w formie zadań partialnych, czy to za pomocą skryptu, wpisu do cron lub etapu w pipeline’u, natomiast wysyłanie zapytań – w API bez stanu. Baza danych wektorowych stanowi umowę pomiędzy nimi.
Budowa strony odpowiedzialnej za pobieranie danych
Środowisko
Najpierw zainstaluj wymagane zależności:
pip install google-genai chromadb google-generativeai python-dotenv ffmpeg-python
Na poziomie systemu również wymagany jest ffmpeg – można go zainstalować za pomocą polecenia brew install ffmpeg na Macu lub apt install ffmpeg na Ubuntu. Następnie mały moduł konfiguracyjny konfiguruje klienta:
# 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)
Proces przyjmowania treści musi obsługiwać wszystkie trzy formy danych, zarządzać przesyłaniem dużych plików multimedialnych, przechowywać istotne metadane oraz zachowywać się poprawnie w przypadku błędów. Krok po kroku:
Klient do wstawiania treści
Klient stanowi prostą warstwę otaczającą API Gemini. Ważny jest jeden szczegół: nagrania wideo i audio muszą zostać przesłane przed wstawieniem, w odróżnieniu od obrazów, które można wysłać bezpośrednio.
# 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
Pomocnik _wait_for_file łatwo pominąć, ale jego brak ma poważne konsekwencje. Pliki są przetwarzane asynchronicznie, a umieszczanie ich przed osiągnięciem stanu ACTIVE powoduje błąd. Interwał sprawdzania trwający pięć sekund jest ostrożnym wyborem. Małe pliki są przetwarzane szybciej, natomiast duże filmy mogą wymagać aż minuty.
Podział na fragmenty
Wszystko tutaj odbywa się lokalnie. ffmpeg dzieli pliki na segmenty, przy czym na każdej granicy istnieje obszar nakładania się. Powód tego nakładania jest łatwy do zrozumienia. Wyobraźmy sobie 30-sekundową dyskusję rozpoczynającą się w 85. sekundzie filmu. Przy segmentach o długości 90 sekund bez nakładania się pierwsze 5 sekund trafia do segmentu 0, a pozostałe 25 do segmentu 1, więc zapytanie dotyczące tej tematyki może zwrócić tylko fragment informacji. Przy 10 sekundach nakładania się oba segmenty zawierają ten fragment, więc zapytanie najprawdopodobniej zwróci ten z większą ilością treści dyskusji.
# 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
Magazyn wektorowy
ChromaDB nadaje się do tego przykładu, ponieważ nie trzeba niczego wokół niego konfigurować. Nie ma serwera do uruchomienia, żadnego kontenera ani konta w chmurze, ponieważ działa wewnątrz procesu Pythona i zapisuje dane do folderu na dysku. W systemie produkcyjnym z milionami fragmentów danych użyto by zamiast niego Qdrant, Pinecone lub Weaviate, które oferują interfejs niemal identyczny.
# 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()
Biegacz
Skrypt łączy wszystko ze sobą i przetwarza katalog z różnymi plikami multimedialnymi. System identyfikacji oparty na haszach sprawia, że proces pobierania jest idempotentny: uruchomienie pliku dwa razy albo kończy się błędem z powodu duplikatu ID, który można wykryć i pominąć, albo jest zabezpieczone sprawdzeniem przed dokonaniem pobrania.
# 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")
Budowa strony z zapytaniami
Przyjmowanie danych to trudna część, natomiast po utworzeniu magazynu zapytania są proste. Proces polega na umieszczeniu pytania, wyszukiwaniu w magazynie, formatowaniu wyników i generowaniu odpowiedzi.
# 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)
}
Parametr similarity_threshold wymaga szczególnej uwagi. Jeśli zostanie ustawiony zbyt nisko, nieistotny treść dostanie się do kontekstu i wprowadzi zamieszanie w generatorze. Jeśli zostanie ustawiony zbyt wysoko, prawidłowe wyniki zostaną odrzucone. Rozsądnym początkowym zakresem jest wartość od 0,6 do 0,7, którą należy dostosować poprzez przeprowadzanie rzeczywistych zapytań na oznaczonym zbiorze testowym.
Zwiększanie dokładności wyszukiwania
Podstawowy system, który funkcjonuje, to jedno, natomiast przydatny system produkcyjny wymaga czegoś więcej. Poprawy osiąga się na trzech etapach: jakości pobierania danych, formułowania zapytań oraz filtrowania po wyszukiwaniu.
Zawsze dopasuj do typu zadania
To jest najważniejszy czynnik wpływający na wyniki. Użycie RETRIEVAL_QUERY w momencie wyszukiwania oraz RETRIEVAL_DOCUMENT podczas przetwarzania danych nie jest opcjonalne. Te dwa mechanizmy działają w przeciwnych kierunkach, dostosowane do asymetrii procesu wyszukiwania, przy której krótkie zapytania odnoszą się do długich dokumentów. Podczas testów stosowanie RETRIEVAL_DOCUMENT w obu przypadkach, zgodnie z domyślnym ustawieniem, spowodowało spadek dokładności wyszukiwania o 13 punktów procentowych. Błąd ten nie wywołuje żadnego błędu i łatwo go przeoczyć.
Najpierw rozszerz zapytanie
Krótkie, niejednoznaczne zapytania szkodzą. Zapytanie „API limits engineering review” jest kiepskie. Znacznie lepiej sprawdza się bardziej szczegółowe pytanie, np. „Które decyzje dotyczące limitów szybkości API zostały podjęte na spotkaniu zespołu inżynierów i dlaczego?”. Jedno szybkie wezwanie modelu LLM przed krokiem tworzenia embeddingów może automatycznie przepisać takie zapytanie.
# 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")
Na podstawie pytań z nagrania wideo sama ekspansja zwiększyła stopę rozpoznania elementów z pierwszej piątki z 74% do 81%.
Wykonuj kilka wariantów równolegle
Zamiast jednego rozszerzonego zapytania, utwórz trzy warianty i połącz wyniki. Różne sformułowania ujawniają różne fragmenty, a ich połączenie pozwala na lepsze rozpoznanie.
# 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)
Wyszukiwanie z wieloma zapytaniami wymaga około dwóch do trzech wywołań API na pytanie, co powoduje dodatkową opóźnioność i koszty. Zwykle jest to opłacalne w sytuacjach, gdy brak odpowiedniego fragmentu ma duże konsekwencje finansowe. W przypadku mniej ważnych pytań wystarczy jedno rozszerzone zapytanie.
Ponowna klasyfikacja kandydatów
Wyszukiwanie oparte na embeddingach dobrze radzi sobie z znajdowaniem semantycznie powiązanych fragmentów, ale słabiej odróżnia to, co „bardzo istotne dla tego pytania”, od tego, co „tylko nieco związane z tym tematem”. Narzędzie do ponownej klasyfikacji eliminuje tę luki.
# 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
Zwykły schemat pozyskuje od 15 do 20 kandydatów poprzez wyszukiwanie wewnątrzsystemowe, przekazuje je wszystkie do ponownego rankowania i zachowuje 5 najlepszych. Ponowny ranker to lokalny kodownik krzyżowy, który może być uruchamiany bez ograniczeń i kończy przetwarzanie w mniej niż 100 ms na procesorze, więc opóźnienie prawie się nie zmienia.
Użyj progu podobieństwa
Gdy baza wiedzy nie zawiera nic istotnego, mechanizm wyszukiwania nadal zwraca wyniki, ponieważ zawsze pokazuje najbliższe dopasowania, nawet jeśli nie są one powiązane. Próg w procesie zapytania rozwiązuje ten problem. Fragmenty o podobieństwie kosinowym poniżej około 0,6 prawie nigdy nie są istotne, a przekazanie ich do generatora skutkuje fałszywymi odpowiedziami. Znaczenia tych progów różnią się w zależności od dziedziny. Treści techniczne z precyzyjną terminologią uzyskują wyższe oceny przy prawdziwych dopasowaniach, natomiast nagrania rozmówowe – niższe, nawet jeśli są istotne, dlatego progi należy kalibrować na oznaczonym zestawie testowym z rzeczywistych danych.
Wyniki w porównaniu ze starym ustawieniem
W porównaniu z poprzednim systemem, w wersji Whisper z transkrypcją oraz systemem RAG opartym wyłącznie na tekście wyróżniły się trzy aspekty. Największy postęp w dokładności odnotowano w przypadku pytań wymagających wizualnego kontekstu z wideo.
- Dokładność wyszukiwania w przypadku pytań pochodzących z wideo. Stary system osiągał około 61% skuteczności w identyfikacji odpowiedzi z pierwszej piątki, natomiast nowy – 84%. Różnica ta wynika niemal wyłącznie z pytań, na które odpowiadano wizualnie: pokazów ekranu, diagramów na tablicy i interakcji z interfejsem. System oparty na transkrypcji generował frazy takie jak „jak można tu zobaczyć” bez żadnego kontekstu wokół nich, podczas gdy modele multimodalne bezpośrednio przechwytywały elementy wizualne.
- Czas przetwarzania. W przypadku filmu trwającego 60 minut stary system potrzebował czterech do pięciu minut na transkrypcję, zanim mogło rozpocząć się indeksowanie. Nowy system tworzy składniki indeksowane już po załadowaniu i włączeniu pierwszego segmentu trwającego 90 sekund, mniej więcej po dwóch minutach. Całkowity czas przetwarzania jest nieco dłuższy ze względu na dodatkowe obciążenia związane z przesyłaniem danych, ale indeks staje się dostępny szybciej.
- Koszt za godzinę nagrania wideo. Transkrypcja za pomocą Whisper kosztuje około 0,36 dolara za godzinę nagrania dźwiękowego. Włączenie 40 fragmentów wideo przy obecnych cenach kosztuje mniej więcej od 0,09 do 0,14 dolara, w zależności od długości. Metoda multimodalna jest zarówno tańsza, jak i dokładniejsza.
Rozmiary i infrastruktura Matryoszki
Matryoshka Representation Learning (MRL) to jedna z bardziej interesujących cech. Model jest trenowany tak, aby najważniejsze informacje semantyczne znajdowały się w pierwszych wymiarach wektora, co oznacza, że pełny wektor o 3072 wymiarach może zostać skrócony z bardzo małą stratą jakości przy wyszukiwaniu. Rozmiar wyniku jest wybierany raz podczas wprowadzania danych i pozostaje niezmieniony, ponieważ mieszanie różnych wymiarowości w jednej kolekcji nie jest obsługiwane.
Dla większości systemów odpowiednią wartością domyślną jest 1536: połowa przestrzeni przechowywania, szybsze zapytania oraz jedynie około 2% straty jakości. 3072 ma sens tylko w systemach o wysokiej precyzji, gdzie kluczowa jest dokładność wyników, a koszt nie stanowi ograniczenia. Obniżenie wartości poniżej 1536 jest uzasadnione tylko w przypadku dziesiątek milionów fragmentów danych, gdy koszt przechowywania staje się istotny. Nawet przy 256 wymiarach jakość wynosi około 88%, co jest doskonale akceptowalne w wielu zastosowaniach.
Co to umożliwia
- Bazy wiedzy instytucjonalnej, w których nagrania z procesu wdrożenia, filmy szkoleniowe oraz instrukcje krok po kroku są umieszczone obok dokumentacji pisemnej. Nowy pracownik, który pyta, jak działa pipeline wdrożeniowy, otrzymuje zarówno podręcznik postępowania, jak i ośmiominutowe nagranie z instrukcjami.
- Systemy zgodności i audytu, które muszą umożliwiać porównywanie dowodów z różnych źródeł. Zespół prawniczy, który chce dowiedzieć się, gdzie omawiano z klientami politykę zwrotów, może znaleźć dokumenty dotyczące tej polityki oraz nagrania rozmów w ramach jednego wyszukiwania.
- Asystenci wsparcia, które pokazują dokładny 40-sekundowy fragment z demonstracji produktu obok artykułu pomocniczego. Obserwowanie, jak ktoś wykona zadanie, często jest skuteczniejsze niż czytanie o tym.
Zasady praktyczne
- Zacznij od najcenniejszych, zabezpieczonych treści: nagrań zawierających wiedzę, której nikt obecnie nie może znaleźć.
- Użyj 1536 wymiarów i wracaj do nich tylko wtedy, gdy koszt przechowywania stanie się rzeczywistym ograniczeniem.
- Od samego początku stosuj próg podobieństwa. Bez niego system tworzy błędne informacje, gdy baza wiedzy nie zawiera odpowiednich treści. 0,65 to rozsądny punkt wyjścia, dostosowany na podstawie rzeczywistych zapytań.
- Rozszerz zapytania przed ich embedowaniem. Jedna wywołanie LLM kosztuje ułamek centa i znacząco poprawia wyniki wyszukiwania w przypadku wideo, gdzie terminy wyszukiwania rzadko dokładnie odpowiadają wypowiedzianym słowom.
- Zachowaj całkowitą separację między zadaniami pobierania danych a serwerem zapytań, dzieląc się jedynie bazą danych wektorową. Wspólne stanowanie między nimi powoduje problemy z niezawodnością.
- Zadbaj o bogate metadane kawałków danych. Embeddingi zajmują się wyszukiwaniem semantycznym, natomiast metadane odpowiadają za resztę: filtrowanie według daty, mówcy, typu nagrania lub poziomu pewności.
Uwagi dotyczące wersji wstępnej
Model jest dostępny w wersji wstępnej publicznie, więc obowiązują pewne ograniczenia:
- Limity czasowe 80 sekund są niewygodne przy podcastach lub wykładach trwających kilka godzin, co skutkuje powstaniem wielu małych kawałków danych i licznych wywołań API. Prawdopodobnie zostaną one zwiększone po pełnej dostępności produktu.
- Klasa GoogleGenerativeAIEmbeddings z LangChain obsługuje tylko tekst. Do embedowania dźwięku i wideo potrzebny jest bezpośrednio SDK google-genai, które następnie jest łączone z pipeline’em LangChain – to rozwiązanie działa, ale nie jest eleganckie.
- W trybie wstępnym nie ma darmowej wersji, więc poszczególni programiści muszą płacić za każde wywołanie embedowania. 50% zniżka na API typu batch pomaga, ale wymaga przetwarzania danych jako zadań w tle, a nie w czasie rzeczywistym.
- Nie udostępnia się wag uwagi ani innych wskaźników interpretowalności. Niemożliwe jest sprawdzenie, które klatki najbardziej wpłynęły na wynik embedowania, więc debugowanie problemów z wyszukiwaniem jest trudniejsze niż w przypadku tekstu, ponieważ pracuje się na podstawie ogólnych wyników podobieństwa bez możliwości zrozumienia ich przyczyn.