Пошук у відео та аудіо за змістом: створення багатомодального RAG з використанням ембеддингів Gemini 2
Мультимодальний RAG з Gemini Embedding 2: один простір векторів для тексту, аудіо, відео та PDF-файлів, тому немає потреби у кроці транскрипції, а також є функції чанкінгу, ChromaDB та переранжування.
Джерела: *My RAG System Missed 80% of My Data. This One Change Fixed It.*
Уявіть базу знань, створену для клієнта: десятки тисяч корпоративних документів, індексованих та здатних до пошуку, з чіткою системою пошуку, яка вражала під час демонстрацій. Усі були задоволені, поки хтось не запитав, чому система не може відповідати на запитання з щотижневих записів зустрічей усіх співробітників. Чесна відповідь була незручною – система взагалі не могла їх бачити.
Більша частина найцінніших знань цієї компанії зберігалась у відео. Кожного спрінту записувалися огляди продукту. Інженери пояснювали схему бази даних, показуючи її на екрані. Менеджери з продукту переглядали інтерфейс, розповідаючи про проблеми клієнтів. Усе це було недоступне, адже всі наявні моделі вбудовування розуміли лише текст.
Звичайним рішенням є транскрипція: передати аудіо через Whisper, витратити години обчислень та прийняти значну рахунок за послуги хмари. Однак це виявляє глибшу проблему. Транскрипція позбавляє записи контексту, який надає їм цінності. Фраза „Як ви можете бачити тут“ перетворюється на порожнє речення. Нотатки на дошці перетворюються на тишу. Тон, з яким інженер каже „Наразі все гаразд“, зникає. Слова індексуються, але їхній зміст втрачається.
Чому важливий спільний простір ембеддингів
Перш ніж розглядати код, корисно зрозуміти, чому Gemini Embedding 2 відрізняється за своєю суттю, а не є просто покращеною версією того ж самого.
Системи багатомодального пошуку стикалися з однією основною проблемою: кожен модус існував у власному векторному просторі. Ембеддинги тексту та аудіо — це обидва списки чисел, проте ці числа не мають спільного значення, а косинусна схожість векторів від різних моделей, навчених на різних завданнях, нічого не говорить. Рішенням було використання окремих потоків обробки з доданою методикою оцінки, тобто пошук тексту, пошук аудіо та, можливо, пошук зображень, після чого намагалися визначити рангінг об’єднаних результатів. Ця конструкція зазнала невдачі з двох серйозних причин.
Однією з них була втрата міжмодальних зв’язків. Коли відео демонструє те, що описує документ, ці два елементи не мають математичного зв’язку, оскільки знаходяться у різних просторах, тож запит може видаати документ, але пропустити відео, або навпаки.
Іншою проблемою були технічні труднощі: потрібно було підтримувати три індекси, стежити за трьома потоками отримання даних та виправляти три набори проблем. Більшість команд відмовилися від пошуку аудіо та відео та задовольнилися пошуком лише в текстовому форматі.
За допомогою Gemini Embedding 2 текст, зображення, відео, аудіо та документи все отримують відображення в одному просторі. Текстовий запит та відеокліп на одну тему знаходяться поруч, тож можна використовувати будь-який з них для пошуку іншого. Саме це є справді новим аспектом.
Що приймає модель
Кожен тип даних має обмеження, які впливають на його обробку:
- Текст: 8 192 токенів вхідних даних за один запит
- Зображення: не більше шести за один запит, формат PNG або JPEG
- Відео: фрагменти тривалістю не більше 128 секунд, формат MP4 або MOV
- Аудіо: фрагменти тривалістю не більше 80 секунд, формат MP3 або WAV
Вибір серед восьми типів завдань не є формальністю. Це SEMANTIC_SIMILARITY, CLASSIFICATION та CLUSTERING; RETRIEVAL_DOCUMENT, RETRIEVAL_QUERY та CODE_RETRIEVAL_QUERY; а також QUESTION_ANSWERING та FACT_VERIFICATION. Тип завдання, вказаний під час формування вектора, визначає, у якому напрямку має бути сформований вектор. Вектор, налаштований для пошуку, відрізняється від вектора, налаштованого для класифікації, навіть для ідентичних вхідних даних.
Ціна становить 0,20 долара за мільйон текстових токенів, зі знижкою 50% через API для пакетних операцій для роботи, якій не потрібні відповіді в реальному часі. Для масового завантаження великої бібліотеки медіа-файлів пакетний підхід майже завжди є кращим варіантом.
Одне попередження щодо міграції: вектори з gemini-embedding-001 (модель GA лише з текстом) не можна поєднувати з векторами з gemini-embedding-2-preview, оскільки ці два середовища несумісні, і немає способу конвертувати старі вектори. Оновлення існуючої системи означає необхідність виділення коштів на повну переіндексацію.
Архітектура, компонент за компонентом
Більшість навчальних матеріалів показують схему потоку та на цьому зупиняються. Корисніше пояснювати кожну частину, наводити причини її існування та те, що зламається без неї. Система має два незалежні канали обробки: канал прийому даних, який працює офлайн та обробляє медіафайли, та канал запитів, який працює в реальному часі та надає відповіді.
Прийом даних
Chunker працює локально в ffmpeg та розділяє медіаконтент на фрагменти, які відповідають обмеженням моделі: менше 128 секунд для відео та менше 80 секунд для аудіо. Він також створює перекриття між сусідніми фрагментами, щоб розмова, що проходить по межі, зберігала свій сенс з обох сторін.
Embedder викликає API Gemini та перетворює кожен фрагмент на вектор. Відео та аудіо спочатку потрібно завантажити в API Files Google та почекати на обробку, перш ніж можна буде викликати кінцеву точку вбудовування. Текст надсилається безпосередньо в API без попереднього завантаження.
Store — це база даних векторів. Вона зберігає кожен вектор разом із метаданими: шляхом до вихідного файлу, часовими позначками, типом контенту та коротким рядком попереднього перегляду. Саме ці метадані дозволяють відповідям посилатися на джерела та пов’язуватися з оригінальною записью.
Запити
Частина, що обробляє запит, навмисно проста, оскільки складність полягає у процесі отримання даних. Під час формування запиту запит вбудовується з типом завдання RETRIEVAL_QUERY, виконується пошук за косинусовою схожістю, збираються найкращі результати незалежно від їхньої форми, вони форматуються у блок контексту, і все це надсилається до генеративної моделі.
Відповіді формує модель Gemini 2.5 Pro, яка може працювати з текстовими описами та посиланнями на відеофрагменти в одному контексті. Вона розуміє, що фрагмент, позначений як [VIDEO at 02:14], означає щось інше, ніж фрагмент, позначений як [TEXT p.3], і відповідно посилається на кожен з них.
Розділення цих двох елементів
Виконання процесів імпорту даних та запитів у одному процесі — це поширена помилка. Процес імпорту є повільним: перетворення години відео на 40 частин, надсилання кожної з них через Files API та очікування на генерацію ембеддингів займає від 15 до 30 хвилин, що значно більше, ніж може чекати кінцева точка обробки запитів. Крім того, процес імпорту дозволяє повторювати спроби. Якщо одна частина даних не обробилася, слід спробувати її ще раз, не порушуючи процес обробки запитів, а поєднання цих двох процесів ускладнює ситуацію. Тому імпорт даних має відбуватися у офлайн-завданні пакетної обробки — чи то через скрипт, запис у cron-таймері чи етап конвеєра обробки, — а запити — через безстановий API. Векторна база даних є контрактом між ними.
Створення частини для імпорту даних
Середовище
Спочатку встановіть необхідні залежності:
pip install google-genai chromadb google-generativeai python-dotenv ffmpeg-python
На рівні системи також необхідний ffmpeg — його можна встановити за допомогою команди brew install ffmpeg на Mac або apt install ffmpeg на Ubuntu. Після цього невеликий модуль конфігурації налаштовує клієнта:
# 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)
Процес обробки контенту має підтримувати всі три формати, керувати завантаженням великих медіафайлів, зберігати корисні метадані та гідно реагувати на помилки. Шар за шаром:
Клієнт для вбудовування
Цей клієнт є простим обгортком навколо API Gemini. Важлива деталь: відео та аудіо мають бути завантажені перед вбудовуванням, на відміну від зображень, які можна надсилати безпосередньо.
# embedder.py
import time
from pathlib import Path
from google import genai
from google.genai import types
from config import client, EMBEDDING_MODEL, EMBEDDING_DIMENSIONS
def embed_text(text: str, task_type: str = "RETRIEVAL_DOCUMENT") -> list[float]:
"""Embed a plain text chunk."""
result = client.models.embed_content(
model=EMBEDDING_MODEL,
contents=text,
config=types.EmbedContentConfig(
task_type=task_type,
output_dimensionality=EMBEDDING_DIMENSIONS
)
)
return result.embeddings[0].values
def _wait_for_file(uploaded, max_wait: int = 300):
"""Poll until a File API upload is done processing."""
waited = 0
poll_interval = 5
while uploaded.state.name == "PROCESSING" and waited time.sleep(poll_interval)
waited += poll_interval
uploaded = client.files.get(name=uploaded.name)
if uploaded.state.name != "ACTIVE":
raise RuntimeError(
f"File never became ACTIVE. Final state: {uploaded.state.name}"
)
return uploaded
def embed_audio(audio_path: str) -> list[float]:
"""
Embed an audio file natively. No transcription step.
The model processes the audio signal directly and returns a
semantic embedding that captures speech content, tone, and
acoustic features. Max input: 80 seconds per file.
"""
uploaded = client.files.upload(path=str(audio_path))
uploaded = _wait_for_file(uploaded, max_wait=120)
result = client.models.embed_content(
model=EMBEDDING_MODEL,
contents=uploaded,
config=types.EmbedContentConfig(
task_type="RETRIEVAL_DOCUMENT",
output_dimensionality=EMBEDDING_DIMENSIONS
)
)
# Clean up: uploaded files count against your quota
client.files.delete(name=uploaded.name)
return result.embeddings[0].values
def embed_video(video_path: str) -> list[float]:
"""
Embed a video chunk natively. Gemini processes both the
audio track and visual frames together in one pass.
This is the key capability: visual demonstrations get captured
in the embedding alongside what is being said. Max input: 128 seconds.
"""
uploaded = client.files.upload(path=str(video_path))
uploaded = _wait_for_file(uploaded, max_wait=300)
result = client.models.embed_content(
model=EMBEDDING_MODEL,
contents=uploaded,
config=types.EmbedContentConfig(
task_type="RETRIEVAL_DOCUMENT",
output_dimensionality=EMBEDDING_DIMENSIONS
)
)
client.files.delete(name=uploaded.name)
return result.embeddings[0].values
def embed_with_context(text: str, image_bytes: bytes = None) -> list[float]:
"""
Embed text and an optional image together in a single call.
When both are passed, the model returns one vector that
represents the joint meaning. A query asking about a database
schema can retrieve a screenshot of that schema.
"""
contents = [text]
if image_bytes:
contents.append(
types.Part.from_bytes(data=image_bytes, mime_type="image/jpeg")
)
result = client.models.embed_content(
model=EMBEDDING_MODEL,
contents=contents,
config=types.EmbedContentConfig(
task_type="RETRIEVAL_DOCUMENT",
output_dimensionality=EMBEDDING_DIMENSIONS
)
)
return result.embeddings[0].values
Допоміжну функцію _wait_for_file легко проігнорувати, але її відсутність має серйозні наслідки. Завантаження обробляються асинхронно, і вбудовування контенту до того, як статус файлу стане ACTIVE, спричиняє помилку. Інтервал опитування у п’ять секунд є обережним. Малі файли обробляються швидше, тоді як великі відео можуть зайняти цілих хвилину.
Розділювач на частини
Усе тут відбувається локально. ffmpeg розділяє файли на сегменти з зоною перекриття на кожному кордоні. Причина наявності цієї зони є зрозумілою. Уявіть 30-секундну дискусію, яка починається на 85-й секунді відео. Якщо сегменти мають тривалість 90 секунд без перекриття, перші 5 секунд потраплять у сегмент 0, а решта 25 — у сегмент 1, тож запит щодо цієї теми може повернути сегмент, який містить лише частину інформації. З перекриттям у 10 секунд обидва сегменти міститимуть цю ділянку, і, ймовірно, запит поверне сегмент із більшою частиною дискусії.
# chunker.py
import subprocess
import json
from pathlib import Path
from dataclasses import dataclass
from typing import List
@dataclass
class MediaChunk:
file_path: str
start_time: float
end_time: float
source_file: str
modality: str
chunk_index: int
total_chunks: int # Useful for progress reporting
def get_media_duration(file_path: str) -> float:
"""Get exact duration via ffprobe. Works for both audio and video."""
cmd = [
"ffprobe", "-v", "quiet",
"-print_format", "json",
"-show_streams", str(file_path)
]
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
data = json.loads(result.stdout)
# Find the first stream with a duration value
for stream in data.get("streams", []):
if "duration" in stream:
return float(stream["duration"])
raise ValueError(f"Could not determine duration for: {file_path}")
def _run_ffmpeg_split(input_path: str, output_path: str,
start: float, duration: float):
"""Execute a single ffmpeg split operation."""
cmd = [
"ffmpeg", "-y",
"-ss", str(start),
"-i", str(input_path),
"-t", str(duration),
"-c", "copy", # No re-encoding: much faster, no quality loss
"-avoid_negative_ts", "make_zero",
str(output_path)
]
result = subprocess.run(cmd, capture_output=True)
if result.returncode != 0:
raise RuntimeError(
f"ffmpeg failed: {result.stderr.decode()}"
)
def chunk_video(
video_path: str,
chunk_duration: int = 90,
overlap: int = 10,
output_dir: str = "./chunks/video"
) -> List[MediaChunk]:
"""
Split video into overlapping chunks within the 128-second limit.
Default: 90-second chunks with 10-second overlap.
Overlap ensures topic transitions are captured in at least one chunk.
"""
Path(output_dir).mkdir(parents=True, exist_ok=True)
total_duration = get_media_duration(video_path)
source_name = Path(video_path).stem
# Pre-calculate chunk boundaries
boundaries = []
start = 0.0
while start end = min(start + chunk_duration, total_duration)
boundaries.append((start, end))
start += (chunk_duration - overlap)
chunks = []
for idx, (start, end) in enumerate(boundaries):
output_path = f"{output_dir}/{source_name}_{idx:04d}.mp4"
_run_ffmpeg_split(video_path, output_path, start, end - start)
chunks.append(MediaChunk(
file_path=output_path,
start_time=start,
end_time=end,
source_file=str(video_path),
modality="video",
chunk_index=idx,
total_chunks=len(boundaries)
))
return chunks
def chunk_audio(
audio_path: str,
chunk_duration: int = 60,
overlap: int = 5,
output_dir: str = "./chunks/audio"
) -> List[MediaChunk]:
"""
Split audio into overlapping chunks within the 80-second limit.
60 seconds per chunk gives a comfortable buffer under the 80-second cap.
"""
Path(output_dir).mkdir(parents=True, exist_ok=True)
total_duration = get_media_duration(audio_path)
source_name = Path(audio_path).stem
boundaries = []
start = 0.0
while start end = min(start + chunk_duration, total_duration)
boundaries.append((start, end))
start += (chunk_duration - overlap)
chunks = []
for idx, (start, end) in enumerate(boundaries):
output_path = f"{output_dir}/{source_name}_{idx:04d}.mp3"
_run_ffmpeg_split(audio_path, output_path, start, end - start)
chunks.append(MediaChunk(
file_path=output_path,
start_time=start,
end_time=end,
source_file=str(audio_path),
modality="audio",
chunk_index=idx,
total_chunks=len(boundaries)
))
return chunks
Сховище векторів
ChromaDB підходить для цього прикладу, оскільки не потрібно нічого налаштовувати навколо нього. Не потрібен сервер для роботи, жодних контейнерів чи облікових записів у хмарі, адже він працює всередині процесу Python та зберігає дані у папці на диску. Для продакшн-систем із мільйонами частинок використовують Qdrant, Pinecone чи Weaviate, які мають майже такий самий інтерфейс.
# vector_store.py
import chromadb
from chromadb.config import Settings
from pathlib import Path
class MultimodalVectorStore:
"""
Vector store wrapping ChromaDB for multimodal RAG.
Stores embeddings + metadata for text, audio, and video chunks.
"""
def __init__(self, persist_dir: str = "./chroma_db"):
self.client = chromadb.PersistentClient(
path=persist_dir,
settings=Settings(anonymized_telemetry=False)
)
self.collection = self.client.get_or_create_collection(
name="multimodal_rag",
# cosine distance is standard for semantic similarity
metadata={"hnsw:space": "cosine"}
)
def add_text_chunk(
self,
chunk_id: str,
text: str,
embedding: list[float],
source_file: str,
chunk_index: int,
page: int = None
):
self.collection.add(
ids=[chunk_id],
embeddings=[embedding],
documents=[text],
metadatas=[{
"modality": "text",
"source_file": source_file,
"chunk_index": chunk_index,
"page": page or 0,
"preview": text[:250]
}]
)
def add_media_chunk(
self,
chunk_id: str,
embedding: list[float],
source_file: str,
start_time: float,
end_time: float,
modality: str,
chunk_index: int
):
"""
Store a video or audio chunk.
Note: we store a formatted timestamp string in `documents`
so ChromaDB has something to display. The actual retrieval
quality comes entirely from the embedding, not this text.
"""
ts_start = f"{int(start_time // 60):02d}:{int(start_time % 60):02d}"
ts_end = f"{int(end_time // 60):02d}:{int(end_time % 60):02d}"
display = (
f"[{modality.upper()}] {Path(source_file).name} "
f"from {ts_start} to {ts_end}"
)
self.collection.add(
ids=[chunk_id],
embeddings=[embedding],
documents=[display],
metadatas=[{
"modality": modality,
"source_file": source_file,
"start_time": start_time,
"end_time": end_time,
"timestamp_start": ts_start,
"timestamp_end": ts_end,
"chunk_index": chunk_index,
"preview": display
}]
)
def search(
self,
query_embedding: list[float],
n_results: int = 5,
modality_filter: str = None
) -> list[dict]:
"""
Retrieve top-k most similar chunks across all modalities.
Optionally filter to a single modality for targeted search.
"""
where_clause = {"modality": modality_filter} if modality_filter else None
results = self.collection.query(
query_embeddings=[query_embedding],
n_results=n_results,
where=where_clause,
include=["documents", "metadatas", "distances"]
)
chunks = []
for doc, meta, dist in zip(
results["documents"][0],
results["metadatas"][0],
results["distances"][0]
):
chunks.append({
"content": doc,
"metadata": meta,
"modality": meta["modality"],
# ChromaDB returns cosine distance; convert to similarity score
"similarity": round(1.0 - dist, 4)
})
return sorted(chunks, key=lambda x: x["similarity"], reverse=True)
def count(self) -> int:
return self.collection.count()
Бігун
Скрипт об’єднує все це та обробляє каталог з різними медіафайлами. Схема ідентифікації, заснована на хешах, робить процес вставки ідемпотентним: якщо файл запускати двічі, це або призведе до помилки через дублікат ідентифікатора, що можна виявити та проігнорувати, або буде заблоковано перевіркою перед вставкою.
# ingest.py
import os
import hashlib
from pathlib import Path
from chunker import chunk_video, chunk_audio
from embedder import embed_text, embed_audio, embed_video
from vector_store import MultimodalVectorStore
store = MultimodalVectorStore(persist_dir="./chroma_db")
def make_chunk_id(source_path: str, chunk_index: int) -> str:
"""Stable, unique ID for any chunk. Same input always = same ID."""
raw = f"{os.path.abspath(source_path)}:{chunk_index}"
return hashlib.sha256(raw.encode()).hexdigest()[:20]
def ingest_text_file(file_path: str):
with open(file_path, "r", encoding="utf-8") as f:
text = f.read()
# Sliding window chunking: 800 chars with 100-char overlap
chunk_size, overlap = 800, 100
raw_chunks = []
start = 0
while start end = min(start + chunk_size, len(text))
raw_chunks.append(text[start:end])
start += chunk_size - overlap
for i, chunk_text in enumerate(raw_chunks):
embedding = embed_text(chunk_text, task_type="RETRIEVAL_DOCUMENT")
store.add_text_chunk(
chunk_id=make_chunk_id(file_path, i),
text=chunk_text,
embedding=embedding,
source_file=file_path,
chunk_index=i
)
print(f" Stored {len(raw_chunks)} text chunks from {Path(file_path).name}")
def ingest_video_file(file_path: str):
print(f" Chunking: {Path(file_path).name}")
chunks = chunk_video(file_path, chunk_duration=90, overlap=10)
for chunk in chunks:
print(
f" Embedding chunk {chunk.chunk_index + 1}/{chunk.total_chunks} "
f"({chunk.start_time:.0f}s to {chunk.end_time:.0f}s)"
)
try:
embedding = embed_video(chunk.file_path)
store.add_media_chunk(
chunk_id=make_chunk_id(file_path, chunk.chunk_index),
embedding=embedding,
source_file=file_path,
start_time=chunk.start_time,
end_time=chunk.end_time,
modality="video",
chunk_index=chunk.chunk_index
)
except Exception as e:
print(f" WARNING: Failed to embed chunk {chunk.chunk_index}: {e}")
finally:
# Always clean up temp files, even on failure
if os.path.exists(chunk.file_path):
os.remove(chunk.file_path)
print(f" Done. {len(chunks)} video chunks stored.")
def ingest_audio_file(file_path: str):
print(f" Chunking: {Path(file_path).name}")
chunks = chunk_audio(file_path, chunk_duration=60, overlap=5)
for chunk in chunks:
try:
embedding = embed_audio(chunk.file_path)
store.add_media_chunk(
chunk_id=make_chunk_id(file_path, chunk.chunk_index),
embedding=embedding,
source_file=file_path,
start_time=chunk.start_time,
end_time=chunk.end_time,
modality="audio",
chunk_index=chunk.chunk_index
)
except Exception as e:
print(f" WARNING: Failed to embed chunk {chunk.chunk_index}: {e}")
finally:
if os.path.exists(chunk.file_path):
os.remove(chunk.file_path)
print(f" Done. {len(chunks)} audio chunks stored.")
def ingest_directory(directory: str):
handlers = {
".txt": ingest_text_file,
".md": ingest_text_file,
".mp4": ingest_video_file,
".mov": ingest_video_file,
".mp3": ingest_audio_file,
".wav": ingest_audio_file,
}
all_files = list(Path(directory).rglob("*"))
media_files = [f for f in all_files if f.suffix.lower() in handlers]
print(f"Found {len(media_files)} files to ingest\n")
for file_path in media_files:
print(f"Processing: {file_path.name}")
handler = handlers[file_path.suffix.lower()]
handler(str(file_path))
print()
print(f"Ingestion complete. Total chunks indexed: {store.count()}")
if __name__ == "__main__":
ingest_directory("./knowledge_base")
Створення частини запитів
Збір даних — це складна частина, а коли база даних вже створена, запити стають простими. Послідовність дій полягає у тому, щоб сформулювати запит, провести пошук у базі даних, сформатувати результати та створити відповідь.
# query.py
import os
from pathlib import Path
import google.generativeai as genai
from embedder import embed_text
from vector_store import MultimodalVectorStore
from config import GENERATION_MODEL
store = MultimodalVectorStore(persist_dir="./chroma_db")
def format_context_for_llm(chunks: list[dict]) -> str:
"""
Format retrieved chunks into a context block for the generative model.
We include modality, source, and similarity score so the model
can calibrate its confidence and cite sources accurately.
"""
parts = []
for rank, chunk in enumerate(chunks, start=1):
meta = chunk["metadata"]
modality = chunk["modality"]
score = chunk["similarity"]
if modality == "text":
parts.append(
f"[SOURCE {rank} | TEXT | {Path(meta['source_file']).name} "
f"| chunk {meta['chunk_index']} | similarity {score}]\n"
f"{chunk['content']}"
)
elif modality == "video":
parts.append(
f"[SOURCE {rank} | VIDEO | {Path(meta['source_file']).name} "
f"| {meta['timestamp_start']} to {meta['timestamp_end']} "
f"| similarity {score}]\n"
f"Video segment covering this time range."
)
elif modality == "audio":
parts.append(
f"[SOURCE {rank} | AUDIO | {Path(meta['source_file']).name} "
f"| {meta['timestamp_start']} to {meta['timestamp_end']} "
f"| similarity {score}]\n"
f"Audio segment covering this time range."
)
return "\n\n---\n\n".join(parts)
def answer_query(
query: str,
n_results: int = 5,
modality_filter: str = None,
similarity_threshold: float = 0.6
) -> dict:
"""
Full RAG pipeline: embed the query, retrieve chunks, generate answer.
similarity_threshold: chunks below this score are dropped before generation.
Prevents low-quality matches from polluting the context.
"""
query_embedding = embed_text(query, task_type="RETRIEVAL_QUERY")
retrieved = store.search(
query_embedding=query_embedding,
n_results=n_results,
modality_filter=modality_filter
)
# Filter out weak matches
filtered = [c for c in retrieved if c["similarity"] >= similarity_threshold]
if not filtered:
return {
"answer": (
"No sufficiently relevant content was found in the knowledge base. "
"The most similar content had a similarity score below the threshold."
),
"sources": retrieved,
"query": query
}
context = format_context_for_llm(filtered)
system_prompt = """You are a helpful assistant with access to a multimodal
knowledge base that contains text documents, video recordings, and audio files.
When citing a source, reference it by its label (e.g., SOURCE 1, SOURCE 2).
For video and audio sources, always include the timestamp so the user can
navigate to the exact moment in the recording.
If the retrieved context does not contain enough information to answer
confidently, say so clearly rather than guessing."""
user_message = (
f"Using only the sources below, answer this question:\n\n"
f"Question: {query}\n\n"
f"Sources:\n{context}"
)
genai.configure(api_key=os.getenv("GEMINI_API_KEY"))
model = genai.GenerativeModel(GENERATION_MODEL)
response = model.generate_content(
user_message,
generation_config={"temperature": 0.1}
)
return {
"answer": response.text,
"sources": filtered,
"query": query,
"chunks_retrieved": len(retrieved),
"chunks_used": len(filtered)
}
Параметр similarity_threshold заслуговує на увагу. Якщо встановити його занадто низько, у контекст потраплятиме неактуальний контент, що спричинить плутанину у генераторі. Якщо занадто високо, будуть відкинуті коректні результати. Розумним початковим діапазоном є 0,6–0,7, який можна налаштувати шляхом виконання реальних запитів на позначеному тестовому наборі.
Підвищення точності пошуку
Проста працездатна система — це одне, а корисна система для продакшну вимагає більшого. Покращення досягаються на трьох етапах: якість отримання даних, сформулювання запиту та фільтрація після пошуку.
Завжди підбирайте параметри відповідно до типу завдання
Це найбільший окремий фактор впливу. Використання RETRIEVAL_QUERY під час запиту та RETRIEVAL_DOCUMENT під час обробки даних є обов’язковим. Ці два вектори дії спрямовані в протилежних напрямках та налаштовані з урахуванням асиметрії пошуку, коли короткі запити ведуть до довгих документів. Під час тестування, коли для обох випадків використовувався лінивий стандартний варіант RETRIEVAL_DOCUMENT, точність пошуку знизилась на 13 відсоткових пунктів. Ця проблема не викликає помилок та легко залишається непоміченою.
Спочатку розширте запит
Короткі, неоднозначні запити завдають шкоди. Фраза „API limits engineering review“ є поганою. Більш повне запитання, наприклад „Які рішення щодо лімітів API були прийняті на зустрічі з інженерами, і чому?“, працює набагато краще. Одне швидке звернення до LLM перед кроком створення ембеддингів може автоматизувати переписування запиту:
# 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")
На основі набору запитань з відео лише розширення дозволило підвищити рівень відтворення інформації серед топ-5 з 74% до 81%.
Виконуйте кілька варіантів одночасно
Замість одного розширеного запиту створіть три варіанти формулювання та об’єднайте результати. Різні формулювання виявляють різні фрагменти інформації, а їх об’єднання дозволяє відтворити більше даних.
# multi_query.py
import concurrent.futures
from query_expander import expand_query
from embedder import embed_text
from vector_store import MultimodalVectorStore
store = MultimodalVectorStore(persist_dir="./chroma_db")
def generate_query_variants(query: str) -> list[str]:
"""Generate multiple phrasings for a single question."""
import google.generativeai as genai
import os
genai.configure(api_key=os.getenv("GEMINI_API_KEY"))
model = genai.GenerativeModel("gemini-2.0-flash")
prompt = (
f"Generate 3 different ways to search for information about: {query}\n\n"
"Return exactly 3 queries, one per line, no numbering or bullets."
)
response = model.generate_content(prompt)
variants = [line.strip() for line in response.text.strip().split("\n") if line.strip()]
return ([query] + variants)[:4] # Always include original, cap at 4 total
def multi_query_search(query: str, n_per_query: int = 4) -> list[dict]:
"""
Search with multiple query variants and deduplicate results.
Returns unique chunks ranked by their best similarity score.
"""
variants = generate_query_variants(query)
def search_one(variant: str) -> list[dict]:
embedding = embed_text(variant, task_type="RETRIEVAL_QUERY")
return store.search(embedding, n_results=n_per_query)
# Run all variants in parallel
all_results = []
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
futures = {executor.submit(search_one, v): v for v in variants}
for future in concurrent.futures.as_completed(futures):
all_results.extend(future.result())
# Deduplicate by source file + chunk index, keeping best similarity score
seen = {}
for chunk in all_results:
meta = chunk["metadata"]
key = f"{meta['source_file']}:{meta['chunk_index']}"
if key not in seen or chunk["similarity"] > seen[key]["similarity"]:
seen[key] = chunk
return sorted(seen.values(), key=lambda x: x["similarity"], reverse=True)
Пошук з кількома запитами додає приблизно два-три виклики API на запит, що призводить до додаткової затримки та витрат. Це зазвичай виправдано, коли втрата релевантного фрагмента коштує дорого. Для менш важливих запитів достатньо одного розширеного запиту.
Переурядкування кандидатів
Пошук за допомогою ембеддингів добре справляється з пошуком семантично пов’язаних фрагментів, але гірше розрізняє те, що є «дуже релевантним для цього запиту», від того, що є «ледь пов’язаним з тією самою темою». Функція переурядкування заповнює цю прогалину.
# reranker.py
from sentence_transformers import CrossEncoder
# This model runs locally, no API cost, fast inference
_reranker = None
def get_reranker():
global _reranker
if _reranker is None:
_reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")
return _reranker
def rerank_chunks(query: str, chunks: list[dict], top_k: int = 5) -> list[dict]:
"""
Re-rank retrieved chunks using a cross-encoder model.
Cross-encoders read both query and chunk together, giving much
more precise relevance scores than embedding cosine similarity.
Only practical on a small candidate set (10-20 chunks).
"""
reranker = get_reranker()
# For video/audio, we use the metadata preview as the text input.
# For text chunks, we use the actual content.
pairs = []
for chunk in chunks:
if chunk["modality"] == "text":
doc_text = chunk["content"]
else:
meta = chunk["metadata"]
doc_text = (
f"{chunk['modality']} recording: {meta['source_file']} "
f"at {meta.get('timestamp_start', '')} to {meta.get('timestamp_end', '')}"
)
pairs.append([query, doc_text])
scores = reranker.predict(pairs)
for chunk, score in zip(chunks, scores):
chunk["rerank_score"] = float(score)
return sorted(chunks, key=lambda x: x["rerank_score"], reverse=True)[:top_k]
# Usage:
# candidates = store.search(query_embedding, n_results=20) # Retrieve wide
# final = rerank_chunks(query, candidates, top_k=5) # Re-rank narro
Звичайний підхід знаходить від 15 до 20 кандидатів за допомогою пошуку, пропускає їх усіх через алгоритм переоцінки та залишає топ-5. Алгоритм переоцінки — це локальний крос-кодер, який можна запустити без проблем та який завершує роботу менш ніж за 100 мс на CPU, тому затримка майже не змінюється.
Використання порогу схожості
Коли база знань не містить нічого релевантного, система все одно повертає результати, адже вона завжди повертає найбільш близькі варіанти, навіть якщо вони не пов’язані. Поріг у процесі обробки запиту вирішує цю проблему. Фрагменти з коефіцієнтом косинусової схожості нижче приблизно 0,6 майже ніколи не є релевантними, і їх передача генератору призводить до вигаданих відповідей. Діапазони змінюються залежно від галузі. Технічний контент із точною термінологією отримує вищі оцінки за справжні зіставлення, а аудіо для розмов має нижчі оцінки навіть у разі релевантності, тому пороги слід налаштовувати на позначеному тестовому наборі з реальних даних.
Результати порівняно зі старою конфігурацією
У порівнянні з попередньою системою, система Whisper з транскрипцією та моделлю RAG лише з текстом, три показники вирізнилися. Найбільше покращилась точність у запитаннях, які потребували візуального контексту з відео.
- Точність пошуку інформації у запитаннях з відео. Стара система досягала приблизно 61% рівня відтворення топ-5 відповідей, тоді як нова — 84%. Різниця зумовлена переважно запитаннями, на які відповідали за допомогою візуалізацій: показів екрана, діаграм на дошці та взаємодії з інтерфейсом. Система, заснована на транскрипції, створювала фрази на кшталт „Як ви можете бачити тут“ без будь-якого контексту навколо, тоді як багатомодальні ембеддинги безпосередньо фіксували візуальні елементи.
- Час обробки. Для 60-хвилинного відео стара система витрачала від чотирьох до п’яти хвилин на транскрипцію, перш ніж можна було розпочати індексування. Нова система створює індексовані фрагменти вже після завантаження та вбудовування першого 90-секундного сегмента, приблизно через дві хвилини. Загальний час обробки трохи більший через додаткові витрати на завантаження, але індекс стає придатним раніше.
- Витрати за годину відео. Транскрипція за допомогою Whisper коштувала приблизно 0,36 долара за годину аудіо. Вбудовування 40 фрагментів відео за поточними цінами коштує приблизно від 0,09 до 0,14 долара, залежно від довжини. Багатомодальний підхід є як дешевшим, так і точнішим.
Розміри та інфраструктура Matryoshka
Матрьошкине навчання представлень (MRL) є однією з найцікавіших особливостей. Модель навчається так, щоб найважливіша семантична інформація знаходилася у перших вимірах вектора, що означає, що повний 3072-вимірний вектор можна скоротити з мінімальними втратами у якості пошуку. Розмір вихідних даних вибирається один раз під час завантаження та залишається незмінним, оскільки змішування розмірностей у одній колекції не підтримується.
Для більшості систем оптимальним значенням за замовчуванням є 1536: це половина простору зберігання, швидші запити та лише близько 2% втрати якості. 3072 має сенс лише для систем високої точності, де критично важливим є рівень відтворення даних, а витрати не є обмеженням. Зниження розмірності нижче 1536 є обґрунтованим лише у випадку десятків мільйонів фрагментів, коли витрати на зберігання стають значущими. Навіть при 256 вимірах якість залишається на рівні близько 88%, що є цілком прийнятним для багатьох застосувань.
Що це дозволяє робити
- Бази інституційних знань, де записи процедур адаптації, навчальні відео та пошагові інструкції поєднуються з письмовою документацією. Новий співробітник, який запитує, як працює процес розгортання, отримує як посібник, так і вісімхвилинний огляд.
- Системи відповідності та аудиту, які мають збігатися за доказами у різних форматах. Юридична команда, яка запитує, де обговорювалась політика повернень з клієнтами, може знайти документи політики та записи дзвінків за одним пошуком.
- Помічники підтримки, які показують саме 40-секундний уривок з демонстрації продукту поруч із статтею допомоги. Перегляд того, як хтось виконує завдання, часто кращий, ніж читання про це.
Практичні правила
- Почніть з найціннішого заблокованого контенту: записів, які містять знання, які наразі ніхто не може знайти.
- Використовуйте 1536 вимірів та повертайтеся до них лише тоді, коли витрати на зберігання стануть справжньою перешкодою.
- З самого початку застосовуйте поріг схожості. Без нього система створює хибні дані, коли база знань не містить відповідного контенту. 0,65 — це розумна вихідна точка, налаштована на основі реальних запитів.
- Розширюйте запити перед вставкою даних. Одне викликання LLM коштує частку цента та значно покращує результати пошуку у відео, де запити рідко точно збігаються з вимовленими словами.
- Зберігайте процес отримання даних та сервер запитів повністю окремими, ділячись лише базою даних векторів. Спільний стан між ними створює проблеми з надійністю.
- Зробіть метадані частин багатими. Ембеддинги відповідають за семантичне пошукове функціонування, а метадані — за все інше: фільтрацію за датою, спікером, типом запису чи рівнем впевненості.
Недоліки у попередньому перегляді
Модель знаходиться у стадії публічного попереднього перегляду, тому існують певні обмеження:
- Ліміт у 80 секунд для аудіо є незручним для подкастів чи багатогодинних лекцій, оскільки це призводить до створення багатьох малих частин та великої кількості викликів API. Ймовірно, цей ліміт буде підвищений під час загальної доступності.
- Клас GoogleGenerativeAIEmbeddings від LangChain підтримує лише текст. Для ембеддингу аудіо та відео потрібен безпосередній використання SDK google-genai, який потім підключається до потоку LangChain; цей підхід працює, але є неефективним.
- Під час прев’ю немає безкоштовного тарифу, тому окремі розробники оплачують кожен виклик для вбудовування даних. 50% знижка на API для пакетної обробки допомагає, але вимагає організації процесу отримання даних як фонової задачі, а не у реальному часі.
- Не відображаються ваги уваги чи інші сигнали для інтерпретованості. Неможливо визначити, які кадри найбільше вплинули на створення ембеддингу, тому виправлення проблем із пошуком є складнішим, ніж у випадку з текстом, оскільки працюється лише з сумарними оцінками схожості без можливості побачити їхні причини.