Búsqueda de video y audio por significado: una implementación multimodal de RAG con Gemini Embedding 2
RAG multimodal con Gemini Embedding 2: un espacio vectorial para texto, audio, video y PDFs, por lo que no hay necesidad de paso de transcripción, además de fragmentación, ChromaDB y reclasificación.
Fuentes: *My RAG System Missed 80% of My Data. This One Change Fixed It.*
Considere una base de conocimientos creada para un cliente: decenas de miles de documentos de la empresa, indexados y buscables, respaldados por un eficiente proceso de recuperación que destacó en las demostraciones. Todos estaban satisfechos hasta que alguien preguntó por qué no podía responder a preguntas extraídas de las grabaciones semanales de toda la plantilla. La respuesta honesta resultó incómoda: el sistema simplemente no podía ver esas grabaciones.
Gran parte del conocimiento más valioso de esa empresa se encontraba en videos. Se grababan explicaciones del producto en cada ciclo de desarrollo. Los ingenieros explicaban el esquema de la base de datos señalándolo en la pantalla. Los gerentes de producto navegaban por la interfaz mientras hablaban sobre los problemas de los clientes. Nada de eso era accesible, ya que todos los modelos de integración disponibles solo comprendían texto.
La solución habitual es la transcripción: pasar el audio por Whisper, gastar horas de procesamiento y aceptar una factura de nube considerable. Esto revela una falla más profunda. Una transcripción elimina el contexto que da valor a las grabaciones. Frases como “Como pueden ver aquí” se convierten en oraciones vacías. Un boceto en pizarra se transforma en silencio. El tono con el que un ingeniero dice “Esto está bien por ahora” desaparece. Las palabras se indexan, pero su significado se descarta.
Por qué es importante un espacio de embedding compartido
Antes de analizar el código, es útil comprender por qué Gemini Embedding 2 es diferente en esencia y no simplemente una versión mejor de lo mismo.
Los sistemas de recuperación multimodal enfrentan un problema fundamental: cada modalidad opera en su propio espacio vectorial. Tanto la incrustación de texto como la de audio son listas de números, pero estos no poseen un significado común, y la similitud coseno entre vectores provenientes de modelos diferentes entrenados con objetivos distintos no aporta información útil. La solución temporal consistió en utilizar pipelines separados a los que se añadía una heurística de puntuación: una búsqueda de texto, una búsqueda de audio y, quizás, una búsqueda de imágenes, seguida de un intento por clasificar los resultados combinados. Ese diseño fracasó de dos maneras graves.
Una de ellas fue la pérdida de las relaciones intermodales. Cuando un video muestra lo que describe un documento, ambos elementos no tienen ninguna conexión matemática al encontrarse en espacios separados, por lo que una consulta podría mostrar el documento pero omitir el video, o viceversa.
El otro problema era operativo: tres índices que mantener, tres pipelines de ingesta que monitorear y tres conjuntos de fallos que depurar. La mayoría de los equipos renunciaron a la recuperación de audio y video y se conformaron con búsquedas solo de texto.
Con Gemini Embedding 2, el texto, las imágenes, el video, el audio y los documentos se mapean todos en un mismo espacio. Una consulta de texto y un clip de video sobre el mismo tema aparecen cerca uno del otro, por lo que cualquiera de ellos puede usarse para encontrar al otro. Esa es la parte realmente nueva.
Qué acepta el modelo
Cada modalidad tiene un límite que determina su diseño:
- Texto: 8,192 tokens de entrada por llamada
- Imágenes: como máximo seis por llamada, en formato PNG o JPEG
- Video: fragmentos de no más de 128 segundos, en formato MP4 o MOV
- Audio: fragmentos de no más de 80 segundos, en formato MP3 o WAV
- PDFs: seis páginas por llamada
Elegir entre los ocho tipos de tarea no es algo superficial. Son SEMANTIC_SIMILARITY, CLASSIFICATION y CLUSTERING; RETRIEVAL_DOCUMENT, RETRIEVAL_QUERY y CODE_RETRIEVAL_QUERY; además de QUESTION_ANSWERING y FACT_VERIFICATION. El tipo de tarea indicado en el momento de la incrustación indica al modelo en qué dirección dar forma al vector. Un vector ajustado para la recuperación de información no es lo mismo que uno ajustado para la clasificación, incluso con entradas idénticas.
El precio es de $0.20 por millón de tokens de texto, con un 50% de descuento a través de la API por lotes para trabajos que no requieren respuestas en tiempo real. Para la ingesta masiva de una gran biblioteca de medios, el método por lotes es casi siempre la opción más adecuada.
Una advertencia sobre la migración: los vectores de gemini-embedding-001 (el modelo GA solo de texto) no pueden mezclarse con los de gemini-embedding-2-preview, ya que estos dos entornos son incompatibles y no existe forma de convertir los vectores antiguos. Actualizar un sistema existente implica presupuestar para una reindexación completa.
Arquitectura, componente por componente
La mayoría de los tutoriales se limitan a mostrar un diagrama de flujo. Es más útil explicar cada parte, por qué existe y qué falla sin ella. El sistema cuenta con dos pipelines independientes: uno de ingestión, que funciona sin conexión y procesa los medios, y otro de consultas, que opera en tiempo real y proporciona respuestas.
Ingestión
El chunker se ejecuta localmente en ffmpeg y divide los medios en segmentos que se ajustan a los límites del modelo: menos de 128 segundos para video y menos de 80 para audio. También superpone los segmentos adyacentes para que una conversación que se encuentra en un límite mantenga su significado en ambos lados.
El embedder llama a la API de Gemini y convierte cada fragmento en un vector. El video y el audio primero deben cargarse en la API de Files de Google y esperar su procesamiento antes de poder llamar al punto de extremo de incrustación. El texto va directamente a la API sin necesidad de carga previa.
El store es la base de datos de vectores. Almacena cada incrustación junto con metadatos: la ruta del archivo de origen, marcas de tiempo, modalidad y una breve cadena de vista previa. Esos metadatos son los que permiten que las respuestas citen fuentes y hagan enlaces al registro original.
Consultas
El lado de la consulta es deliberadamente simple, ya que la complejidad reside en la fase de ingestión. En el momento de la consulta, la pregunta se incluye con el tipo de tarea RETRIEVAL_QUERY, se ejecuta una búsqueda por similitud coseno, se recopilan los mejores resultados independientemente de su modalidad, se formatean en un bloque de contexto y todo se envía a un modelo generativo.
La respuesta la proporciona Gemini 2.5 Pro, un modelo capaz de trabajar con descripciones de texto y con referencias a segmentos de video en el mismo contexto. Entiende que un fragmento etiquetado como [VIDEO a 02:14] significa algo diferente de uno etiquetado como [TEXT p.3], y cita cada uno en consecuencia.
Separar los dos componentes
Ejecutar la ingestión y las consultas en un mismo proceso es un error común. La ingestión es lenta: convertir una hora de video en 40 fragmentos, enviar cada uno a través de la Files API y esperar a que se generen los embeddings consume de 15 a 30 minutos, mucho más tiempo del que puede esperar un endpoint de consultas. Además, la ingestión permite reintentar las operaciones. Si un fragmento falla, solo ese fragmento debe reintentarse, sin afectar al proceso de consultas, y mezclar ambos tipos de operaciones lo complica aún más. Por lo tanto, la ingestión debe realizarse en un trabajo por lotes fuera de línea, ya sea mediante un script, una entrada de cron o una etapa de pipeline, mientras que las consultas deben realizarse mediante una API sin estado. La base de datos vectorial funciona como el contrato entre ambos.
Desarrollo del lado de ingestión
Entorno
Primero, instale las dependencias:
pip install google-genai chromadb google-generativeai python-dotenv ffmpeg-python
También se requiere ffmpeg a nivel del sistema, mediante brew install ffmpeg en una Mac o apt install ffmpeg en Ubuntu. Luego, un pequeño módulo de configuración configura al cliente:
# 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)
La ingesta de producción debe manejar las tres modalidades, gestionar las subidas de medios de gran tamaño, almacenar metadatos significativos y funcionar de manera adecuada en caso de fallo. Capa por capa:
El cliente de incrustación
El cliente es una capa sencilla que rodea la API de Gemini. Un detalle importante: el video y el audio deben subirse antes de realizar la incrustación, a diferencia de las imágenes, que pueden enviarse directamente.
# 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
El complemento _wait_for_file es fácil de omitir, pero resulta costoso hacerlo. Las subidas se procesan de forma asíncrona, y incluirlo antes de que el estado del archivo sea ACTIVO genera un error. Un intervalo de consulta de cinco segundos es un valor conservador. Los archivos pequeños se procesan más rápido, mientras que los videos grandes pueden tardar hasta un minuto completo.
El fragmentador
Todo aquí es local. ffmpeg divide los archivos en segmentos con una zona de superposición en cada límite. La razón de esta superposición es fácil de comprender. Imagine una discusión de 30 segundos que comienza a los 85 segundos de un video. Con segmentos de 90 segundos sin superposición, sus primeros 5 segundos quedarían en el segmento 0 y los otros 25 en el segmento 1, por lo que una consulta sobre ese tema podría obtener un segmento que solo contenga un fragmento. Con 10 segundos de superposición, ambos segmentos incluirán la parte relevante, y la consulta probablemente devolverá el segmento con la mayor parte de la discusión.
# 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
El almacén de vectores
ChromaDB es adecuado para este ejemplo porque no se necesita configurar nada alrededor de él. No hay servidor que ejecutar, ni contenedor ni cuenta en la nube, ya que funciona dentro del proceso de Python y guarda los datos en una carpeta del disco. En un sistema de producción con millones de fragmentos, se utilizarían Qdrant, Pinecone o Weaviate, con una interfaz prácticamente idéntica.
# 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()
El ejecutor
Un script lo integra todo y procesa un directorio con medios mixtos. Un esquema de identificación basado en hash hace que la incorporación sea idempotente: ejecutar un archivo dos veces o bien falla debido a un ID duplicado, lo cual se puede detectar y omitir, o bien se evita mediante una verificación antes de incorporarlo.
# 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")
Construcción del lado de consultas
La ingesta es la parte más difícil; una vez que existe el almacenamiento, realizar consultas resulta sencillo. El flujo consiste en incluir la pregunta, buscar en el almacenamiento, formatear los resultados y generar una respuesta.
# 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)
}
El parámetro similarity_threshold merece atención. Si se establece en un valor demasiado bajo, el contenido irrelevante se filtra en el contexto y confunde al generador. Si se establece en un valor demasiado alto, se descartan los resultados válidos. Un rango inicial de 0.6 a 0.7 es razonable, y debe ajustarse mediante la ejecución de consultas reales en un conjunto de prueba etiquetado.
Mejorar la precisión de la recuperación
Tener un sistema básico que funcione es una cosa, pero un sistema de producción útil necesita más. Las mejoras se logran en tres etapas: calidad de la ingesta, formulación de la consulta y filtrado después de la recuperación.
Adapte siempre al tipo de tarea
Este es el factor más importante. RETRIEVAL_QUERY en el momento de la consulta y RETRIEVAL_DOCUMENT durante la incorporación de datos no son opcionales. Estos dos vectores actúan en direcciones opuestas, ajustados a la asimetría de la búsqueda, donde las consultas cortas apuntan a documentos largos. En las pruebas, al utilizar RETRIEVAL_DOCUMENT para ambos casos, el valor predeterminado causó una pérdida de 13 puntos porcentuales en la precisión de la recuperación de datos. El error no genera ninguna alerta y es fácil de pasar por alto.
Amplíe primero la consulta
Las consultas cortas y ambiguas son perjudiciales. Frases como “Límites de la API: revisión por ingenieros” son inadecuadas. Una pregunta más completa, como “¿Qué decisiones sobre los límites de velocidad de la API se tomaron en la reunión de revisión por ingenieros y por qué?”, funciona mucho mejor. Una llamada rápida a un LLM antes del paso de incrustación puede automatizar la reescritura de la consulta:
# 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")
En un conjunto de preguntas extraídas de videos, la expansión por sí sola aumentó la tasa de recordación entre los 5 mejores resultados del 74 % al 81 %.
Ejecutar varias formulaciones en paralelo
En lugar de una única consulta expandida, genere tres formulaciones y combine los resultados. Diferentes formas de expresar la consulta revelan diferentes fragmentos de información, y la unión de estos fragmentos permite recordar más datos.
# 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)
La búsqueda con múltiples consultas añade aproximadamente dos a tres llamadas a la API por pregunta, lo que genera mayor latencia y costos adicionales. Suelen valer la pena en casos en que perder un clip relevante tiene un alto costo. Para preguntas menos importantes, basta con una sola consulta ampliada.
Reordenar los candidatos
La búsqueda basada en embeddings es eficaz para encontrar fragmentos semánticamente relacionados, pero es menos precisa a la hora de distinguir entre lo que es “altamente relevante para esta pregunta” y lo que está “vagamente relacionado con el mismo tema”. Un sistema de reordenamiento cierra esa brecha.
# 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
El patrón habitual obtiene de 15 a 20 candidatos mediante búsquedas incrustadas, los pasa todos por el re-clasificador y mantiene a los 5 mejores. El re-clasificador es un codificador cruzado local que puede ejecutarse sin problemas y termina en menos de 100 ms en una CPU, por lo que la latencia apenas cambia.
Usar un umbral de similitud
Cuando la base de conocimientos no contiene nada relevante, la recuperación sigue devolviendo resultados, ya que siempre muestra las coincidencias más cercanas incluso si no están relacionadas. El umbral en el pipeline de consultas se encarga de eso. Los valores por debajo de una similitud coseno de aproximadamente 0.6 casi nunca son relevantes, y pasarlos al generador produce respuestas falsas. Los rangos varían según el dominio. El contenido técnico con terminología precisa obtiene puntuaciones más altas para coincidencias reales, mientras que el audio conversacional obtiene puntuaciones más bajas incluso cuando es relevante; por lo tanto, los umbrales deben calibrarse con un conjunto de pruebas etiquetado a partir de datos reales.
Resultados frente a la configuración anterior
En comparación con el sistema anterior, la transcripción de Whisper combinada con RAG de texto puro mostró tres aspectos destacados. La precisión mejoró de manera más significativa en las preguntas que requerían contexto visual proveniente de videos.
- Precisión en la recuperación de información para preguntas basadas en videos. El proceso anterior alcanzó un 61% de precisión en la identificación de las 5 respuestas principales, mientras que el nuevo sistema logró el 84%. La diferencia se debe casi en su totalidad a las preguntas respondidas mediante elementos visuales: explicaciones en pantalla, diagramas en pizarra y interacciones con la interfaz. El sistema basado en transcripciones generaba frases como “como se puede ver aquí” sin ningún contexto adicional, mientras que los embeddings multimodales capturaban directamente los elementos visuales.
- Tiempo de ingestión. En el caso de un video de 60 minutos, el proceso anterior tardaba de cuatro a cinco minutos en transcribirlo antes de que pudiera comenzar la indexación. El nuevo sistema genera los fragmentos indexados tan pronto como se carga e incrusta el primer segmento de 90 segundos, aproximadamente en dos minutos. El tiempo total de ingestión es ligeramente mayor debido a la carga adicional del envío, pero el índice queda listo para usarlo más pronto.
- Costo por hora de video. La transcripción con Whisper costaba alrededor de 0,36 dólares por hora de audio. Incrustar 40 fragmentos de video, según los precios actuales, cuesta aproximadamente entre 0,09 y 0,14 dólares, dependiendo de la duración. El método multimodal es tanto más económico como más preciso.
Dimensiones e infraestructura de Matryoshka
El Aprendizaje por Representación Matryoshka (MRL) es una de las características más interesantes. El modelo se entrena de tal manera que la información semántica más importante se encuentra en las primeras dimensiones del vector, lo que significa que el vector completo de 3072 dimensiones puede truncarse con muy poca pérdida en la calidad de la recuperación. El tamaño de la salida se elige una vez al momento de la ingesta y se mantiene, ya que no se admite mezclar dimensiones dentro de una misma colección.
Para la mayoría de los sistemas, 1536 es el valor predeterminado adecuado: la mitad del almacenamiento, consultas más rápidas y solo una pérdida de calidad de aproximadamente el 2%. 3072 solo tiene sentido en sistemas de alta precisión donde la recuperación es crítica y el costo no representa una limitación. Bajar a menos de 1536 solo está justificado con decenas de millones de fragmentos, cuando el costo de almacenamiento se vuelve significativo. Incluso con 256 dimensiones, la calidad se mantiene alrededor del 88%, lo cual es perfectamente aceptable para muchos usos.
Qué hace posible esto
- Bases de conocimiento institucionales donde las grabaciones de incorporación, los videos de capacitación y las guías paso a paso de ingeniería se encuentran junto con la documentación escrita. Un nuevo empleado que pregunta cómo funciona el pipeline de despliegue recibe tanto el manual de operaciones como la guía de ocho minutos.
- Sistemas de cumplimiento y auditoría que deben sincronizar las pruebas entre diferentes formatos. Un equipo legal que pregunta dónde se discutió la política de devoluciones con los clientes puede encontrar los documentos de la política y las grabaciones de las llamadas con una sola búsqueda.
- Asistentes de soporte que muestran el clip exacto de 40 segundos de una demostración del producto junto al artículo de ayuda. Ver a alguien realizar una tarea suele ser más efectivo que leer sobre ella.
Reglas prácticas
- Comience con el contenido protegido de mayor valor: grabaciones que contienen conocimientos que actualmente nadie puede encontrar.
- Utilice 1536 dimensiones, y vuelva a analizarlas solo si el costo de almacenamiento se convierte en una verdadera limitación.
- Aplicue el umbral de similitud desde el primer día. Sin él, el sistema genera respuestas erróneas cuando la base de conocimientos carece de contenido relevante. 0.65 es un punto de partida razonable, ajustado según consultas reales.
- Amplíe las consultas antes de realizar el procesamiento. Una llamada a un LLM cuesta una fracción de centavo y mejora significativamente la capacidad de recuperación en videos, donde los términos de búsqueda rara vez coinciden exactamente con las palabras pronunciadas.
- Mantenga el trabajo de ingestión y el servidor de consultas completamente separados, compartiendo únicamente la base de datos vectorial. El estado compartido entre ellos genera problemas de fiabilidad.
- Procure que los metadatos de los fragmentos sean detallados. Las representaciones vectoriales se encargan de la recuperación semántica, mientras que los metadatos se ocupan del resto: filtrado por fecha, orador, tipo de grabación o nivel de confianza.
Limitaciones en la versión preliminar
El modelo se encuentra en versión preliminar pública, y aplican algunas limitaciones:
- El límite de 80 segundos para el audio resulta incómodo para podcasts o conferencias de varias horas, ya que genera muchos fragmentos pequeños y numerosas llamadas a la API. Probablemente se elevará cuando esté disponible de forma general.
- La clase GoogleGenerativeAIEmbeddings de LangChain solo admite texto. Para el embedding de audio y video se necesita directamente el SDK google-genai, que luego se integra en un pipeline de LangChain; funciona, pero no es muy elegante.
- Durante la versión de prueba no existe un plan gratuito, por lo que los desarrolladores individuales deben pagar por cada llamada de incrustación. El descuento del 50% en la API por lotes ayuda, pero requiere estructurar la ingesta como una tarea en segundo plano, y no como un proceso en tiempo real.
- No se exponen pesos de atención ni otros indicadores de interpretabilidad. Es imposible determinar qué frames influyeron más en una incrustación, por lo que depurar problemas de recuperación es más difícil que con el texto, ya que solo se dispone de puntuaciones de similitud globales sin conocer las causas que las generaron.