Пошук па сэнсе ў відэа і аўдыё: мультымадальны RAG на Gemini Embedding 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 за мільйон тэкстовых токенав, пры чым за дапамогою API для пакетных задаń дапускаецца 50% зніжка для задаń, якім не патрэбны рэальнага часу адпаведзі. Для масовага запоўнення велікай бібліятэкі медыя-матэрыялав пакетны падход практычна завжды ёсць лепшым варыянтам.
Адзін паведамленне пра міграцыю: вектары з 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()
Бегун
Скрытый скрыпт абсалявае ўсё і обрабоцвае каталог з міксованымі медыя-файламі. Схема ID, адаснованая на хэшах, робіць працэс абсалявання ідэмпатным: абсаляванне файла два разы альбо збіваецца з-за дуплікатнага ID, што можна зафіксаваць і працягнуць далей, альбо контролюецца праз перакананне пры абсаляванні.
# 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 пад момент прыема дакументаў не ўзначальваюцца як неабавязковыя. Шырыя вектары дзейнасці дзейнуюць у працупаказваных напрамках, прыстосаваныя да асіметрыі адзыскання інфармацыі, калі короткія запіты спрамоўваюцца да дугіх дакументаў. У тэстах, калі для обох параметраў выкарыстоўваліся значэнні за замовчаннем, адносна недбалыя, точнасць адзыскання знізілася на 13 працэнтных пунктав. Шырокі баг не вызвал ніякіх адхылэнняў і лёгка працягваецца за гледком.
Спачатку расшырыце запіт
Короткія, нечыстыя запіты наносяць шкоду. Фраза „API limits engineering review“ ўжо неякская. Болей чытальны запіт, такі як „Калія рашэнні ў чысце лімітаў API былі прыйняты на заседанні інжынерскага адзору, і чаму?“ працюе набліжна лепш. Адно шырока запытанне да LLM пры перадзеянні практыкі embedding можа автаматызаваць перапісву запіту:
# 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")
У апартаментах з пытаннямі, взятымі з відэа, сама толькі розгортка падняла рэгрэт кантэнту з 74% да 81% сярод топ-5.
Адрабатнае выкарыстоўванне калькюляцый
У замяну адной розгорнутай запыткі створыце тры варыянты формулювання і з’едыніце рэзультаты. Разныя формулювання адкрываюць разныя фрагменты, і ў з’едыненні адзначаецца вышэйшы рэгрэт.
# 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% падтрымкі сярод пяці найболей прыдатных рэсультатаў, а ў новай — 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 коштуе частку цэнты і значна павышае рэзультаты пошуку ў відэа, дзе запіты рэдка калі точна падходзяць да сказаных слоў.
- Зберагачыце задачу прыема дадзейнаў і сервер запытання абсалютна аддзельна, дзеячы толькі за схемай вектарнай базы дадзэйнаў. Аднаўленне стану між імі стварае проблемы з надзеяйнасцю.
- Зберагачыце метаданы частак максымальна паўнамасштабныя. Функцыі embeddings адпавядаюць за семантычны пошук, а метаданы — за рэшэнне всіх іншых задач: фільтрацыю па даце, спікеры, тыпе запісу чы ўпэўненасці.
Недагэнутыя аспекты прыватнай працэсы
Модэль належыць да прыватнай працэсы, і існуюць калькі лімітацыяў:
- Ліміт у 80 секунд для аудыя ёсць некамфортным для подкастаў чы лекцыяў трываласцю у калькі гадзін, ўнаследкі чаго ствараецца множнацтва маленькіх частак і веліка калькі запытанняў да API. Оерагчыма гэты ліміт будзе падняты пасля загальнага выклікання.
- Клас GoogleGenerativeAIEmbeddings з LangChain падтрымлівае толькі тэкст. Для аудыя і відэа патрэбныя embeddings выкалічваюцца безпасульна за дапамогою google-genai SDK, пасля чаго яны перадаюцца ў пайплайн LangChain; гэта працюе, але не ўсё так адмыслена.
- У прыёмнай версіі няма бесплатнага тарыфу, таму адзінныя распрацоўшчыкі плацяюць за кожны вызов прабраматавання. 50% зніжка на API для абрабатавання пакетаў дапамагае, але вымагае структуравання працэса абрабатавання як фонавага задання, а не рэальнага часу.
- Не адкрываюцца ні вагі ўвагі, ні іншыя сігналы для інтэрпрэтавання. Немагчыма выявіць, якія кадры найбольш паўлінілі прабраматавання, таму дыбаггін проблем з аднаходжэнням інфармацыі ёсць сложнейшы, чым у віпадку з тэкстам, калі працюеш з сумарнымі рэйтынгамі схожасці без можлівасці адразу пазнакаць, што іх вызвало.