Поиск видео и аудио по смыслу: создание мультимодальной системы 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
- PDF-файлы: по шесть страниц за один запрос
Выбор из восьми типов задач имеет важное значение. Это SEMANTIC_SIMILARITY, CLASSIFICATION и CLUSTERING; RETRIEVAL_DOCUMENT, RETRIEVAL_QUERY и CODE_RETRIEVAL_QUERY; а также QUESTION_ANSWERING и FACT_VERIFICATION. Тип задачи, указанный при генерации вектора, определяет, в каком направлении должен быть сформирован вектор. Вектор, настроенный для поиска, отличается от вектора, настроенного для классификации, даже при идентичных входных данных.
Цена составляет 0,20 доллара за миллион токенов текста; при использовании API для обработки больших объемов данных предусмотрена скидка 50%, так как такая обработка не требует мгновенных ответов. Для массовой загрузки крупной библиотеки медиафайлов API для обработки больших объемов данных практически всегда является более подходящим решением.
Одно предупреждение по миграции: векторы из gemini-embedding-001 (модель GA только с текстом) нельзя смешивать с векторами из gemini-embedding-2-preview, поскольку эти две среды несовместимы, и ничто не позволяет преобразовать старые векторы. Для обновления существующей системы необходимо выделить средства на полную переиндексацию.
Архитектура, компонент за компонентом
В большинстве учебников показана лишь схема потока данных. Более полезно объяснить каждую часть, зачем она нужна и что сломается без неё. Система состоит из двух независимых потоков обработки: потока загрузки данных, который работает офлайн и обрабатывает медиафайлы, и потока запросов, который работает в реальном времени и предоставляет ответы.
Загрузка данных
Чанкер работает локально в ffmpeg и разделяет медиаконтент на фрагменты, соответствующие ограничениям модели: менее 128 секунд для видео и менее 80 секунд для аудио. Он также создает перекрытие между соседними фрагментами, чтобы диалог, происходящий на границе, сохранял свой смысл с обеих сторон.
Эмбеддер обращается к API Gemini и преобразует каждый фрагмент в вектор. Для видео и аудио сначала необходимо загрузить их в API Files Google и дождаться обработки, прежде чем можно будет вызвать конечную точку эмбеддинга. Текст передается непосредственно в API без предварительной загрузки.
Хранилище — это база данных векторов. Оно хранит каждый эмбеддинг вместе с метаданными: путем исходного файла, временными метками, типом данных и кратким превью-текстом. Именно эти метаданные позволяют ответам цитировать источники и возвращаться к оригинальной записи.
Запросы
Сторона запросов преднамеренно проста, поскольку сложность заключается в процессе получения данных. В момент запроса вопрос включается в задачу типа 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: обзор инженерами» слишком неопределённа. Гораздо лучше использовать более полные вопросы, такие как «Какие решения относительно ограничений скорости API были приняты на собрании инженеров и почему?». Один быстрый вызов большой языковой модели перед этапом генерации эмбеддингов может автоматически переписать запрос в более подходящий вид:
# 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 мс на процессоре, поэтому задержка практически не меняется.
Использование порога сходства
Даже если в базе знаний нет ничего релевантного, алгоритм поиска всё равно возвращает результаты, поскольку он всегда показывает наиболее близкие кандидаты, даже если они не связаны с запросом. Порог в конвейере обработки запросов решает эту проблему. Фрагменты с коэффициентом косинусного сходства ниже примерно 0,6 почти никогда не являются релевантными, и их передача генератору приводит к выдуманным ответам. Значения порога варьируются в зависимости от области. Технический контент с точными терминами получает более высокие оценки за настоящие совпадения, а аудиозаписи диалогов — более низкие, даже если они релевантны, поэтому пороги необходимо настраивать на маркированном тестовом наборе из реальных данных.
Результаты по сравнению со старой конфигурацией
По сравнению с предыдущей системой у системы Whisper с транскрипцией плюс RAG с текстовым форматом проявились три значимых преимущества. Наиболее заметно улучшилась точность при обработке вопросов, требующих визуального контекста из видео.
- Точность поиска ответов на вопросы, основанные на видео. В старой системе уровень восстановления ответов из числа пяти лучших составлял примерно 61%, тогда как в новой — 84%. Разница в основном обусловлена вопросами, на которые даются ответы с использованием визуальных элементов: пошаговые демонстрации на экране, схемы на доске и взаимодействие с интерфейсом. Система, основанная на транскрипции, генерировала фразы вроде «как вы можете видеть здесь» без какого-либо контекста, тогда как мультимодальные эмбеддинги непосредственно улавливали визуальную информацию.
- Время обработки. Для 60-минутного видео старая система тратила от четырех до пяти минут на транскрипцию, прежде чем можно было начать индексацию. Новая система генерирует индексируемые фрагменты сразу после загрузки и встраивания первого 90-секундного отрезка, примерно через две минуты. Общее время обработки увеличивается незначительно из-за дополнительных операций загрузки, но индекс становится готов к использованию раньше.
- Стоимость за час воспроизведения видео. Стоимость транскрипции с помощью Whisper составляла примерно 0,36 доллара за час аудио. Встраивание 40 фрагментов видео по текущим ценам обходится примерно от 0,09 до 0,14 доллара в зависимости от длительности. Мультимодальный подход является более дешевым и точным.
Размеры и инфраструктура Matryoshka
Матрёшкинский метод обучения представлениям (MRL) является одной из наиболее интересных особенностей. Модель обучается таким образом, что наиболее важная семантическая информация находится в первых измерениях вектора, что позволяет сокращать полный 3072-мерный вектор с минимальной потерей качества при поиске. Размер выходных данных определяется один раз при загрузке и остаётся неизменным, поскольку смешивание различных размерностей в одной коллекции не поддерживается.
Для большинства систем подходящим стандартным значением является 1536: это позволяет сократить объём хранилища, ускорить запросы и потерять всего около 2% качества. Значение 3072 имеет смысл только для систем высокой точности, где критически важна полнота восстановления данных, а стоимость не является ограничением. Снижение размерности ниже 1536 оправдано только при наличии десятков миллионов фрагментов данных, когда стоимость хранения становится значимой. Даже при размерности 256 качество остаётся на уровне примерно 88%, что считается вполне приемлемым для многих задач.
Что это позволяет делать
- Базы знаний организации, в которых записи процедур адаптации сотрудников, учебные видео и пошаговые инструкции по работе с технологиями находятся рядом с письменной документацией. Новый сотрудник, спрашивающий, как работает процесс развертывания, может получить как руководство по эксплуатации, так и восьмиминутное видео с пошаговым объяснением.
- Системы соблюдения норм и аудита, которые должны обеспечивать совпадение доказательств из разных источников. Юридическая команда, запрашивающая информацию о том, где обсуждалась политика возврата с клиентами, может найти документы с политикой и записи звонков в ходе одного поиска.
- Помощники-копилоты, которые показывают точный 40-секундный фрагмент демонстрации продукта рядом с статьей с инструкциями. Просмотр того, как кто-то выполняет задачу, часто эффективнее, чем просто чтение о ней.
Практические правила
- Начните с контента наивысшей ценности, который находится в защищенном доступе: записей, содержащих знания, которые в настоящее время никто не может найти.
- Используйте 1536 измерений и возвращайтесь к их использованию только тогда, когда стоимость хранения станет реальным ограничением.
- С самого начала применяйте порог сходства. Без него система может выдавать ложную информацию, когда в базе знаний отсутствует соответствующий контент. 0,65 — это разумная отправная точка, отрегулированная на основе реальных запросов.
- Расширяйте запросы перед их вставкой в систему. Один вызов большой языковой модели стоит совсем немного и значительно улучшает точность поиска в видео, где поисковые запросы редко полностью совпадают с произнесенными словами.
- Сохраняйте полную разделенность между задачей обработки данных и сервером запросов, делая общим лишь векторную базу данных. Общее состояние между ними приводит к проблемам с надежностью.
- Сделайте метаданные кусков максимально подробными. Эмбеддинги отвечают за семантическое поиск, а метаданные — за остальное: фильтрацию по дате, исполнителю, типу записи или уровню уверенности.
Недостатки превью-версии
Модель находится в режиме публичного превью, и существуют некоторые ограничения:
- Лимит в 80 секунд для аудио кажется неудобным для подкастов или лекций продолжительностью в несколько часов, что приводит к образованию множества мелких кусков и большому количеству запросов к API. Вероятно, этот лимит будет увеличен при полном выпуске продукта.
- Класс GoogleGenerativeAIEmbeddings из LangChain поддерживает только текст. Для эмбеддингов аудио и видео требуется прямое использование SDK google-genai, которое затем интегрируется в пайплайн LangChain; это работает, но выглядит неэлегантно.
- Во время превью нет бесплатного тарифа, поэтому отдельные разработчики оплачивают каждый вызов для встраивания контента. 50%-ная скидка на API для обработки пакетов помогает, но требует организации процесса обработки в качестве фоновой задачи, а не в реальном времени.
- Не предоставляются веса внимания или другие показатели интерпретируемости. Невозможно определить, какие кадры наиболее сильно повлияли на формирование эмбеддинга, поэтому отладка проблем с поиском сложнее, чем при работе с текстом, так как используются только суммарные показатели сходства без возможности увидеть причины их формирования.