Практычныя прытамулкі: Мая система RAG працювала некоректна з 80% маёў дадзеных. Аднаго змены хапілася, каб усё было налажана.
Практычныя прыказкі: Мая система RAG працявала некоректна, перажыла 80% маёў дадзеных, але гэтае змены ўсё выправілі: контракты, перакрыцчы і месца для коду для команд, якія викорыстоўваюць гэты патэрн.
У гэтым карыце практычным напамінанні перакладзена схема працы, яка практычна ад сыр'ёў да готовай системы для: Мая система RAG працявала некоректна, перазрачыўшы 80% маёй дакументаціі. Адна змянка выправіла гэта. Увага сфокусавана на практычных кроках, чысткіх пераконтралях і коде, які можна проста дадаць у репазітарый, не прабуючы здогадвацца пра мету. У стадії агляду неабходна прадзефінавацыя вхідных дадзеных, адпаведальнага за крок і крэтарыяў завершэння працы перш чым змяніць код. Аператары должны магчымае перадзваніць крок з вядомай точкі контролю, не прабуючы здагадвацца пра схованы стан. Конфігурацыю трэба залічыць паза кодам прыкладнення. Файлы сераў, сховішчы секрэтных дадзеных і флагі функцый належаць у аднам месца, якое аператары можаць пераглядаць, не чытаяўшы весь код.
Што на самай працэ змянілася з Gemini Embedding 2
Калі працуеце над этапам «Што на самай справе змянілася», спачатку запісаце кантракт: неабяжлівыя даны, сигнал успеху і тое, што выходзіць пад частковыя адказы. Такі список пераканаецца дапамагчы залічваць будучыя змены ў кодзе чыстаючы. Документавайце як шлях успеху, так і шлях вярнення. Перапрыбуткі, людзкіе контралі і обработка некоректных паведамленняў ёсць частью продукту, а не пасляднім дапрацоўкам. Зявляйце логі з ідэнтыфікаторам запиту, ідэнтыфікаторам моделі і часам затрымкі праз кожны вызов. Без такога следу періодычныя памылкі прадастаўця выглядаюць як багі ў прыемніку.
Проблема адной прасторы для вбудоввання
Калі працюеце над стадзіяй «The Single Embedding Space», спачатку запісайце умовы викорыстоўвання: неабходныя даны, сігнал успеху і тое, што выходзіць пад частым неудачам. Такі список дапамагае залічыць змяны ў кодзе пазней. Валіце маленькія, тэставаныя елементы замест большых скрыптов. Калі якісь крок не выходзіць, неудача должна адносіцца да адной відпаведальнасці, а не да заплутанага ланцуга задач. Перад налаштаваннем запитоў пераканайцеся, як модель адпаведзае на фіксаваны набор запитанняў. Частае зміненне запитоў рэдка калі-небудзь выправляе слабкую спроможнасць моделі да пошуку інформацыі.
Што на самай працэ падтрымлівае модель
Калі працуеце над стадзіяй «Што на самай працоўна модэль», спачатку запісуйце кантракт: неабходныя вхідныя даны, сигнал успеху і тое, што выходзіць у разе частковага невыпання. Такі список контролю дапамагае заставіць пазнейшыя змены коду быць чыстымі. Спрэчвайце гэтую стадзію як кантракт межа вхіднымі данымі і перакананымі выходнымі рэзультатамі. Даўце назвы артыфактам, задаць правіла пераканання успеху і адмовіцеся ад тыхоўскага частковага завершэння. Зберагачвайце кеш стабільных інструкцыяў системы і схемаў інструментаў. Перадача ідэнтычных прамулі ёсць частым выклікам зношэння ресурсаў. Калі працуеце над стадзіяй «Што на самай працоўна модэль», спачатку запісуйце кантракт: неабходныя вхідныя даны, сигнал успеху і тое, што выходзіць у разе частковага невыпання. Такі список контролю дапамагае заставіць пазнейшыя змены коду быць чыстымі. Зберагачвайце настройкі параду ўнутры коду аплікацыі. Файлы сераўнавання, хранільнікі секрэтных дадзеных і флагі функцыйяў должны знаходзіцца ў аднам месцы, якое аператары можаць пераглядаць без неабяжнай чытанняў усіх элементаў структуры.
Архітектура, расказаная дакладна
Архітектуру, яка адаптуецца паэтапна, лепейша розглядаць як мерыябельную структуру. Запісайце адны ідеальны прыклад, адну ситуацыю неудачы і прыметкі па поверненні да пачатковага стану пры расшырэнні масштаба. Дакументавайце як успешны, так і варыянт вярнення да нормальнага стану. Практыкі павторных спроб, кантрольныя етапы і обработка некоректных паведамленняў ёсць часткай продукту, а не элементамі пазнейшага дапрацоўкі. Раздзеліце правілы частковай обработкі дадзеных ад правіл ўтрымання іх. Змена адных не павінна вымагаць перапісву іншых, калі зменяюцца паказатэлі якасці.
Практыка прыемкі дадзеных
Этап каналу прыема дадзейна працюе наякша, калі яго спрыяваць як меравальную плошчу. Зафіксавайце адны ідеальны прыем дадзейна, адзін прыклад неудачы і запіс пра вярнэнне да пачатковага стану, перш чым расширваць сферу дзеяння. Валіце маленькія, тэставаныя элементы замест большых скрыптов. Калі які-небудзь крок не выйшае, прычына неудачы павінна вказываць на адну конкрэтную адпаведальнасць, а не на заплутаны канал прыема дадзейна. Раздзеліце правілы часткавання дадзейна ад правіл наявнасці інформацыі. Змена адных правіл не павінна вымагаць перапісву іншых, калі зменяюцца паказнікі якосці.
Канал запыткаў
Этап «Канал запыткаў» працюе наякша, калі яго спрыяваць як мерыемую структуру. Зберагучы адну ідеальную версію выходных дадзеных, адны прыклад неудачы і запіс паўранення, перш чым расширваць сферу дзеяння. Спрыяйце гэтаму этапу як кантракту межаў вхідных дадзеных і перакананых выходных. Даўце назвы артыфактам, задаце критэрыя успеху і не падзеўляйцеся частым, непূরным выкананням задач. Раздзеліце правілы часткавага оброблення дадзеных ад правіл ўтрымання іх. Змена адных не павінна вымагаць перапісву іншых, калі зменяюцыся паказнікі якосці. Этап «Канал запыткаў» працюе наякша, калі яго спрыяваць як мерыемую структуру. Зберагучы адну ідеальную версію выходных дадзеных, адны прыклад неудачы і запіс паўранення, перш чым расширваць сферу дзеяння. Зберагачы настройкі парадульна ад коду прыемліцеля. Файлы сераўнавання сяродовішча, хранілішчы секрэтных дадзеных і флагі функций павінны знаходзіцца ў адном месцы, куды аператары можаць адбавляць контроль без неабходнасці чытання всей структуры.
Чаму гэтыя два каналы павінны застацца раздзельнымі
Для стадіі «Чаму гэтыя два каналы» неабходна прадзеўжэнне вводных дадзеных, адначыяе апаведчанню пра таго, хто адпавядае за даны крок, і крэтарыяў выходу пры змены коду. Аператары павінны магчымае перзапусканне крока з вядомай точкі контролю, не падозрываючы пра схованы стан. Неабходна аддзекаваць як шлях успеху, так і шлях вярнення да нормальнага стану. Перапрыбуткі, людзкія перакрыцця і обработка некоректных паведамленняў ёсць часткай продукту, а не чымось, што дадаецца пазней. Паказваць трэба тыя часткі тексту, якія фактычна служылі падставай для адпаведзення. Без цых цитатаў аператары не зможаць разлічыць галюцинацію ад прасоўкі ў індэксаванні.
Установка среды
У стадії налагоджэння среды неабяцкова практычна вказаць інпуты, адміністратара крока та критэрыя завершэння пры перадзеіснавленні коду. Аператары должны магчымаць перзапуск крока з вядомай точкі контролю, не падозрываючы схованы стан. Лепш выбіраць маленькія, тэставаныя елементы замест большых скрыптов. Калі крок не выйшаў, прычына неудачы павінна вказываць на адну конкрэтную адпаведальнасць, а не на заплутаны ланцюг задач. Прыкладзіце фрагменты, якія насправды ляглі в основу адпаведнай адказы. Без цых праменаванняў аператары не зможаць розразліць галюцинацыю ад працягу індэксавання.
pip install google-genai chromadb google-generativeai python-dotenv ffmpeg-python
# 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)
Стварэнне ланцюга прыемкі дадзеных
Для стадіі стварэння каналу прыема дадзейнаў неабходна прадзефінаваць вхідныя даны, адпаведальную особу за выкананне крока і критэрыя завершэння пры перамены коду. Аператары должны магчымае запускаваць крок з вядомай точкі контролю, не спрабоўваючы здагадвацца пра схованы стан. Спрыяйце тэй стадіі як даговору межа вхіднымі данымі і перакананымі выходнымі рэзультатамі. Даўце назвы артыфактам, прадзефінаваць пераканання ў успеху і адмовіцеся ад беззвучнага частковага завершэння. Цітуйце тыя часткі, якія фактычна лежалі в основе адпаведнай адказы. Без цітатаў аператары не можуць разлічыць галюцинацію ад прасоў у індэксаванні. Для стадіі стварэння каналу прыема дадзейнаў неабходна прадзефінаваць вхідныя даны, адпаведальную особу за выкананне крока і критэрыя завершэння пры перамены коду. Аператары должны магчымае запускаваць крок з вядомай точкі контролю, не спрабоўваючы здагадвацца пра схованы стан. Зберагачыце настройкі параду ад коду прыемніка. Файлы сяродавішча, хранільнікі секрэтных дадзейнаў і флагі функций должны знаходзіцца ў аднам месцы, якое аператары можуць пераглядаць без чытання.
весь граф.Шаг 1: Кліѐнт для вбудоввання
Працюючы над шагам 1, спачатку запісайце умовы викорыстоўвання: неабходныя даны, сігнал успеху і тое, што вядзецца пад частковым невяснам. Такі список дапамагае заліцьваты змяны ў кодзе. Запісуйце адночасна шлях успеху і шлях вяснавання проблемы. Перапрыбуткі, людзкі контроль і обработка некоректных паведамленняў є частью продукту, а не дадатковыми правкамі. Перад налаштаваннем запитоў пераканайцеся ў рівні вяснавання на фіксованай сэтцы запитаў. Частая зміна запитоў рэдка калі вярнуе слабкія результаты пошуку.
# 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
Шаг 2: Фрагментавальнік медыя-даных
Калі працюеце над стадзіяй «Шаг 2: Медыя», спачатку запісайце контракт: неабяжлівыя даны, сігнал успеху і тое, што выходзіць у разе частковага неяксамоства. Такі список пераканальвае ў тым, каб пазнейшыя змены коду былі чыстымі. Валіце маленькія, тэставаныя елементы замест вялікіх скрыптав. Калі якісь шаг неяксамоства, гэтае неяксамоства павінна адносіцца да адной адпаведальнасці, а не да заплутанага ланца. Перад налаштаваннем прапозыцый замерайце рэкалі на фіксаваным наборе запытанняў. Частая змена прапозыцый рэдка калі лечыць слабую систему аднаходжэння інфармацыі.
# 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
Шаг 3: Хранілішча вектараў
Калі працуеце над стэпам 3 «Вектар», спачатку запісайце контракт: неабяжлівыя данні, сігнал успеху і тое, што выходзіць у разе частковага нявыпання. Такі список пераконвае ў тым, што пазнейшыя змены коду будуць чыстымі. Спрэчвайце гэты стэп як контракт межа даннімі і перакананымі выходамі. Дайце назвы артыфактам, задаце правілы пераканання успеху і не падзеўляйцеся частковым завершэнням без паведамлення. Змяркуйце рэкалі на фіксаваным наборе запитаў прычыну налаштавання прапты. Частае змены прапты рэдка калі вярнуюць слабую якасць адзысквання інформаціі. Калі працуеце над стэпам 3 «Вектар», спачатку запісайце контракт: неабяжлівыя данні, сігнал успеху і тое, што выходзіць у разе частковага нявыпання. Такі список пераконвае ў тым, што пазнейшыя змены коду будуць чыстымі. Зберагаюце налаштаванні за межамі коду прыемліка. Файлы сяродавішча, хранільнікі секрэтных дадзеных і флагі функций должны знаходзіцца ў аднам месцы, якое аператары можаць пераглядаць без неабяжлівага чытання всей структуры.
# 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()
Стэп 4: Адпрацоўчык інгестіі
Этап 4 – стадія прыемкі даных – работае наўпершыя, калі яго расследжваць як вимерную плошчу. Зафіксавайце адна ідеальная транскрыпцыю, адзін прыклад неудачы і запіс пра вярнэнне да пачатковага стану, перш чым расширваць сферу дзеяння. Запісвайце адночасна шлях успеху і шлях вярнэння да нормальнага стану. Перапрыбуткі, людзкі контроль і обработка некоректных паведамленняў є часткай продукту, а не элементамі пазнейшага доўнелення. Раздзельвайце правілы частковай обработкі даных і правілы ўзяць даныя. Змена аднаго з іх не должна вымагаць перапісвання другога, калі зменяюцыся паказнікі якосці.
# 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)
}
Падвышэнне точнасці адзыскання дадзеных
Этап падчырання тачнасці адзысквання працуе наяўнейша, калі яго спрыяваць як меравальную плошчу. Зафіксавайце адны ідеальны прымер, адзін кейс неудачы і запіс пра вярнэнне да пачатковага стану перш чым расширваць масштаб. Спрыяйце гэтам этапу як кантракту межа вхіднымі дадзеннямі і паверыранымі выходнымі рэзультатамі. Дайце назвы артыфактам, задаце критэрыя успеху і адмовіцеся ад тыхняй частковай роботы без паведамлення. Раздзеліце правілы часткавання дадзеных ад правілаў адзысквання. Змена ў одных не павінна прыводзіць да перапісвы іншых, калі змянююцца паказнікі якасці. Этап падчырання тачнасці адзысквання працуе наяўнейша, калі яго спрыяваць як меравальную плошчу. Зафіксавайце адны ідеальны прымер, адзін кейс неудачы і запіс пра вярнэнне да пачатковага стану перш чым расширваць масштаб. Зберагаўце настройкі праза код аплікацыі. Файлы сяродавішча, хранільнікі секрэтных дадзеных і флагі функций павінны знаходзіцца ў адном месцы, якое аператары можаць пераглядаць без неабяжнага чытання всіх элементаў.
1. Іспользуйце правы тип задання, ў кожны раз
Для стадіі «1. Выкорыстоўваць правыя данні» неабходна прадзефінаваць вхідныя даны, адпаведальную особу за кожны крок і крэтарыя выходу пры зміне коду. Аператары должны магчыма было перзапускаць крок з вядомай точкі контролю, не падозрываючы прыхованы стан. Неабходна аддокументаваць як шлях успеху, так і шлях вярнення да нормальнага стану. Перапрыбуткі, людзкія перакрыцця та обробка некоректных паведамленняў є частью продукту, а не елементамі пазнейшай дапрацоўкі. Паказваць трэба тыя часткі тексту, якія фактычна лежаць у падставе адпаведнай адказы. Без ціх цитатаў аператары не зможуць розразліць галюцинацыю ад прасоў у індэксацыі.
2. Дадзіце розглед запыту пры ўбудоўванні
Для стадіі 2 «Расшырэнне запита» неабяжна практычна адзінавочка: перад змянайом кодам неабяжна задаць вхідныя даны, адпаведнага адпаведальніка за шаг і критэрыя завершэння. Аперацыям неабяжна маты магчымасць парадзельна запускаць шаг з вядомай точкі контролю, не падозрюючы прыхованых станоў. Лепш выбіраць маленькія, тэставаныя елементы замест большых скрыптов. Калі шаг не выйшаў, прычына неабяжна стосавацца адной конкрэтнай адпаведальнасці, а не заплутанага ланцацька. Неабяжна цітаваць тыя часткі тексту, якія фактычна лежаць у падставе адпаведнай адказы; без цітатаў аперацыям немагчыма розразліць галюцинацію ад прычыны, зв’язанай з недастаткам індэксавання.
# 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")
3. Адкліканне калькуляцый паралельна
Для стадіі «3. Выкананне калькаў разам» неабходна перад змянай коду задаць вхідныя даны, адпаведальнага за этап і крэтыры завершэння. Аператары должны магчымае перайсці на гэты этап з вядомага пункта контролю, не спрабоўваючы здагадвацца пра схованы стан. Спрыяйце таму, каб гэтая стадія была схожа на контракт межа вхіднымі данымі і перакананымі выходнымі рэзультатамі. Даць назвы артыфактам, задаць крэтыры успеху і не прабоўваць прыймаць часткова завершаныя рэзультаты без падтверджэння. Указваць тыя часткі тексту, якія фактычна лежаць у падставе адпаведнай адпаведзі. Без цых цытатаў аператары не зможаць розлічыць галюцинацію ад працягу індэксавання. Для стадіі «3. Выкананне калькаў разам» неабходна перад змянай коду задаць вхідныя даны, адпаведальнага за этап і крэтыры завершэння. Аператары должны магчымае перайсці на гэты этап з вядомага пункта контролю, не спрабоўваючы здагадвацца пра схованы стан. Конфігурацыю трэба заставляць праза код аплікацыі. Файлы сераўнавання, хранільнікі секрэтных данных і флагі функцияў должны знаходзіцца ў аднам месцы, якое аператары можаць пераглядаць, не чытаяўшы весь код аплікацыі.
# 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)
4. Пераранжаванне атрыманых фрагментаў
Кал працуеце над этапам пераранжавання атрыманых фрагментаў, спачатку запісайце умовы: неабходныя данні, сігнал успеху і тое, што выканаецца у разы ўзельнага неякшага рэзультата. Такі чарточка дапамагае заліцьваты пазнейшыя змены ў кодзе. Документавайце як шлях успеху, так і шлях вярнення да нормы. Перапрыбуткі, людзкія перакрыцчы і обробка некоректных паведамленняў є частью продукту, а не пазнейшым дапрацоўкам. Перад налаштаваннем запрасаў пераканайцеся ў рэкалі на адзначаныя касета пытанняў. Частая зміна запрасаў рэдка калі вялікі эфект на слабую систему атрымання інформацыі.
# 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
5. Задаць порог сэмабільнасці і прыдбацца да яго
Калі працюеце над 5-м этапам «Установка сэродненасці», спачатку запісайце умовы контракту: неабяжлівыя даны, сігнал успеху і тое, што выходзіць пад частыя неудачы. Такі список контроля дапамагае залічваць змяны ў кодзе чыстаю думкай. Валіце маленькія, тэставаныя елементы замест большых скрыптов. Калі якісь крок не выйшоў, неудача должна вказваць на адну адпаведальнасць, а не на заплутаны ланцужок задач. Перад налаштаваннем запитоў пераканайцеся, як працуе алгорытм на фіксаванай сэтцы запитаў. Частая зміна запитоў рэдка калі вярнайце слабкую эфектыўнасць алгорытму.
Рэзультаты
Калі працуеце на стадзіі «Рэзультаты», спачатку запісайце контракт: неабяжлівыя даннэ, сігнал успеху і тое, што выканаецца у разе частковага нявыпалення. Такі список пераканаецца, каб пазнейшыя змены коду былі чыстымі. Спрэцьвуйце да гэтай стадзіі як да контракту межа даннемі і перакананымі выходамі. Дайце назвы артыфактам, задацьте перакананні на успех і адмовіцеся ад тыхоўага частковага завершэння. Замерайце рівень запамятовання на фіксаваным наборе пытанняў прычым рэгулюванні запрошэнняў. Частае змены запрошэнняў рэдка калі выправляюць слабкую систему аднаходжэння інформацыі. Калі працуеце на стадзіі «Рэзультаты», спачатку запісайце контракт: неабяжлівыя даннэ, сігнал успеху і тое, што выканаецца у разе частковага нявыпалення. Такі список пераканаецца, каб пазнейшыя змены коду былі чыстымі. Зберагаеце настройкі за межамі коду прыемліка. Файлы сяродавішча, хранільнікі секрэтных дадзенняў і флагі функций должны знаходзіцца ў аднам месцы, якое аператары можаць пераглядаць без неабяжлівага чытання всей структуры.
Што значаюць выміры Матрёшкі для вашай інфраструктуры
Этап «Выклікі размераў матрыёшкі» працуе наяўней, калі яго спрыяваць як вимерную паверхню. Запісаўце адна «золатая» транскрыпцыю, адзін прыклад неудачы і запіс пра вярнэнне да пачатковага стану перш чым расширваць масштабы. Дакументаваць трэба як шлях успеху, так і шлях вярнення. Перапрыбуткі, людзкія контралі і обработка некоректных паведамленняў є часткай продукту, а не чымсь, што дадаецца пазней. Раздзеліце політыку часткавай обработкі і політыку выкарыстоўвання дадзеных. Змена адной з іх не должна прымусіваць перапісванне другой, калі зменяюцыся паказнікі якосці.
Што можна стварыць з гэтага
Этап «Што можна стварыць» работае наяўней, калі яго спрыяваць як мерыемую паверхню. Запісаце адна ідеальная версія, адзін прыклад неудачы і прыметкі па адвярненню роботы пры расшырэнні масштаба. Валіце маленькія, тэставаныя елементы замест большых скрыптов. Калі якісь крок не выйшае, прычына неудачы павінна вказываць на адную адпаведальнасць, а не на заплутаны ланцюг задач. Раздзеляйце правілы часткавання і правілы выявлення. Змена адных не павінна прыводзіць да перапісву іншых, калі змянююцца паказнікі якосці.
Практычныя выводы
Этап «Практычныя вывыкі» працюе наўсёх краща, калі яго спрыяваць як мерыемую паверхню. Зафіксавайце адны ідеальны транскрыпт, адзін прыклад неудачы і запіс пра вярнэнне да пачатковага стану перш чым расширваць масштаб. Спрыявайце гэты этап як кантракт межа вхіднымі дадзеннямі і паверыжанымі выходнымі рэзультатамі. Дайце назвы артыфактам, задаць критэрыя успеху і адмовіцеся ад тыхоўскага частковага завершэння. Раздзеліце політіку часткавага апранкавання дадзеных ад політіки ўзяць іх. Змена адной з яных не павінна прымусваць перапісванне другой, калі зменяюцыся паказнікі якосці. Этап «Практычныя вывыкі» працюе наўсёх краща, калі яго спрыяваць як мерыемую паверхню. Зафіксавайце адны ідеальны транскрыпт, адзін прыклад неудачы і запіс пра вярнэнне да пачатковага стану перш чым расширваць масштаб. Зберагаўце настройкі праза код аплікацыі. Файлы сяродавішча, хранільнікі секрэтных дадзеных і флагі функций павінны знаходзіцца ў адном месцы, якое аператары можаць пераглядаць без неабяжнага чытання всіх дадзеных.
Што ўсё ж неяўнае
Для стадіі «Што ўсё яшчэ неяўнае» неабходна прадзефінаваць вхідныя даны, адпаведальную за крок і крэтынія выходу пры змены коду. Аператары должны магчыма было перзапускаць крок з вядомай точкі контролю, не падозрываючы прыхованы стан. Неабходна аддзеўнаваць як шлях успеху, так і шлях вярнення. Перапрыбуткі, людзкіе перакрыцця і обработка некоректных паведамленняў є часткай продукту, а не чымсь, што дадаецца пазней. Паказваць трэба тыя часткі тексту, якія фактычна лежаць у падставе адпаведнай адказы. Без ціх цитатаў аператары не зможуць разлічыць галюцинацію ад прасоў у індэксаванні.
Давайце продовжым развівацца разам
Для стадіі «Давайте продолжаем вучыцца» неабяжна ўзначыць даннэ, адпаведальнага за крок і критэрыя завершэння пры перадзеіснаванні коду. Аператары должны магчымаць перзапуск кроку з вядомай точкі контролю, не падозрываючы прыхованы стан. Лепш выбіраць маленькія, тэставаныя елементы замест большых скрыптов. Калі крок не выйшоў, прычына неудачы должна вказываць на адну конкрэтную адпаведальнасць, а не на заплутаны процес. Прытамульвайце тыя часткі тексту, якія фактычна служылі падставай для адпаведнага адказу. Без ціх праменаванняў аператары не зможуць разлічыць галюцинацію ад працягу ў індэксаванні.
Чек-ліст для эксплуатацыі
Стадія чек-ліста для эксплуатацыі працюе найэфектывней, калі яе спрыяваць як мерыемую плошчу. Зберагачыце адна ідеальная транскрыпцыя, адзін прыклад неудачы і запіс пра вярнэнне да пачатковага стану перад расшырэнням масштаба.
Запісвайце часы выканання і вартасць токеноў або запытаванняя разам з функцыйнальнымі рэзултатамі. Відразувыя данні пра вартасці запобегаюць неспакоўным рахункам, калі процес пераходзіць з дэмовай среды ў спакульнаваныя сераўы.
Раздзеліце політыку часткавання ад політыке выявлення дадзейнаў. Змена адной з іх не павінна прымусваць перапісванне другой, калі зменяюцыся паказнікі якосці.
Дадзейснюйце тэст на працяванне критычнага маршруту ў системе CI з викорыстанням фіксатываючых элементаў, а не рэальных платных API, калі тое дазволяе бюджет.
Зберагаўце настройкі праза код аплікацыі. Файлы сяродавішча, хранільнікі секрэтных дадзейнаў і флагі функцый павінны знаходзіцца ў адном месцы, куды аператары можаць адбавляць контроль без неабходнасці чытання всей структуры.
Раздзеліце політыку часткавання ад політыке выявлення дадзейнаў. Змена адной з іх не павінна прымусваць перапісванне другой, калі зменяюцыся паказнікі якосці.
Перад апраноўкай всей структуры заморозьце версіі, зафіксавайце ідеальны варыянт дадзейнаў для критычнага маршруту і паказвайце способы адворачэння змян. У спільных сяродавішчах неабходны ліміты частоты запытоў, пераказы на адпаведнасць тэнанты і чысткі власнік для змены секрэтных дадзейнаў. Валіце простую надзяйніснасць працы ад крэатывных, але разовых дамах.
Запіскі для пакета b3567bc23c05: не трэба кантрацаваць ключы прадступніка ў репазітарыі, задаць максімальную кантэйнернае значэнне токена на сесію, а таксама зберагчы транскрыпціі празьледу самым фіксчарам дапамогі, каб пазнейшыя змены модэляў заставаліся пораўнанневымі.