Cómo la IA multimodal resuelve definitivamente la búsqueda de vídeos

Permíteme ser duro por un segundo: la mayoría de las canalizaciones RAG de vídeo están anticuadas. Probablemente estés incrustando fotogramas individualmente con CLIP, almacenando vectores de imagen y texto por separado, y rezando para que la similitud del coseno en una imagen estática o un fragmento de transcripción sea suficiente para encontrar el momento que estás buscando.

Ese enfoque era lo mejor que podíamos hacer en 2024, e incluso entonces sería puntero. Ya no lo es.

El juego cambió silenciosamente a principios de 2025 con la llegada de modelos de incrustación multimodal listos para la producción. Modelos como voyage-multimodal-3 de Voyage AI (que utilizaremos aquí, aunque están surgiendo otros). No se trata sólo de incrustadores de imágenes o de texto atornillados. Ingieren simultáneamente los fotogramas visuales y la transcripción del texto correspondiente, y escupen un único vector denso (a menudo de 384 o 1024 dimensiones) que representa toda la escena: los elementos visuales, las palabras, el contexto, todo fusionado.

¿A qué se debe este cambio? Porque el vídeo es multimodal. Un fotograma que muestra un gráfico no significa nada sin la explicación del orador. La frase «como puede ver aquí» es inútil sin el elemento visual al que apunta. Al integrarlos, captamos esa relación. De repente, buscar «la diapositiva que explica el Teorema Central del Límite» puede llevarte al segmento de 30 segundos en el que la fórmula aparece en pantalla y se habla de la definición.

Este artículo es el plan para construir ese canal RAG de vídeo moderno. Tomaremos un vídeo en bruto de YouTube, lo dividiremos en escenas semánticamente significativas (fotogramas + transcripción), incrustaremos esas escenas utilizando el modelo multimodal de Voyage, almacenaremos los vectores en KDB.AI (porque el rendimiento es importante) y, a continuación, utilizaremos un LLM de visión como GPT-4o-mini para responder a las preguntas con un contexto visual y textual completo. Voy a caminar a través de los conceptos básicos, los fragmentos de código esenciales, y las gotchas que se encontrará en el camino.

Hay que tener en cuenta que la búsqueda de vídeos sigue siendo un área de investigación activa, por lo que aún no se han establecido las mejores prácticas. Existen pocos puntos de referencia y muchas de las estrategias existentes están obsoletas. He preguntado a expertos en búsqueda cuál es la mejor manera de buscar vídeos, y la respuesta más común es «no lo sé». Esto es algo que cambiará a lo largo del año que viene, y mi objetivo era escribir la mejor guía técnica de Internet sobre RAG de vídeo.

Nuestro pipeline completo tendrá este aspecto:
El código completo y ejecutable se encuentra en este cuaderno Colab – consíguelo si quieres seguirnos o simplemente robar el producto final.

Conceptos básicos: Por qué funciona ahora

1. Incrustaciones multimodales: Fusión de imagen y sonido

Es probable que conozcas las incrustaciones. Las incrustaciones de texto asignan frases a vectores numéricos; las frases similares aterrizan cerca unas de otras en un espacio de alta dimensión. Las incrustaciones de imágenes hacen lo mismo con los elementos visuales.

Un modelo de incrustación multimodal hace algo más sofisticado: toma varios tipos de datos a la vez (por ejemplo, un fragmento de texto y varias imágenes relacionadas) y los asigna a un único vector que representa su significado combinado.

Aprende que las palabras «Esta fórmula muestra…» pertenecen al fotograma que muestra E=mc². Esto permite que la búsqueda por similitud encuentre momentos en los que la información visual y auditiva coinciden con la consulta, lo que es imposible con incrustaciones separadas. Fundamentalmente, mezclar los fotogramas con la transcripción nos va a proporcionar incrustaciones más ricas. El modelo de Voyage AI es un buen ejemplo de esta nueva capacidad accesible a través de una API. Este es el eje central; sin él, volvemos a tratar el vídeo como partes inconexas. Sólo existen unos pocos modelos de incrustación verdaderamente multimodales, y los mejores son de código cerrado en el momento de escribir este artículo. Esto podría cambiar en un futuro próximo.

2. KDB.AI: Almacenamiento de vectores a gran escala y velocidad

Necesitas una base de datos diseñada para encontrar rápidamente los vectores más cercanos entre millones o miles de millones. KDB.AI, creada por el equipo kdb+ (conocido por sus sistemas financieros de baja latencia), ofrece indexación vectorial especializada. En este ejemplo, utilizamos qHNSW, un índice de aproximación al vecino más cercano (RNA) ideal para grandes conjuntos de datos en los que es aceptable una ligera reducción de la precisión a cambio de un aumento masivo de la velocidad. Admite el almacenamiento de vectores en disco, lo que reduce drásticamente los costes de memoria en comparación con los índices totalmente en memoria.

Construcción de la canalización: Etapas clave y lógica del código

Desglosemos el proceso, centrándonos en el código impactante.

Etapa 1: Preparación de datos – Vídeo a trozos de escena
En primer lugar, tome el vídeo, extraiga fotogramas periódicamente (por ejemplo, 1 cada 5 segundos) y obtenga la transcripción.

# ... [For download video code - see Colab] ...
with VideoFileClip(VIDEO_PATH) as clip:
    # Extract frames at 0.2 FPS. This rate is a tunable hyperparameter.
    # Lower FPS = fewer frames, less cost, potentially less visual detail.
    # Higher FPS = more detail, more cost, diminishing returns after a point.
    clip.write_images_sequence(os.path.join(FRAMES_DIR, "frame%04d.png"), fps=0.2, logger=None)
    clip.audio.write_audiofile(AUDIO_PATH, codec="libmp3lame", bitrate="192k", logger=None)

# Transcribe using Whisper
openai_client = OpenAI()

with open(AUDIO_PATH, "rb") as audio_file:
    # Requesting segments gives us text blocks with rough start/end times
    transcription = openai_client.audio.transcriptions.create(
        model="whisper-1", file=audio_file, response_format="verbose_json",
        timestamp_granularities=["segment"]
    )

Explicación: Obtenemos fotogramas numerados secuencialmente y un objeto de transcripción estructurado que contiene segmentos de texto con marcas de tiempo. fps=0,2 es nuestro punto de partida para equilibrar detalle y coste.

Es importante que establezcamos timestamp_granularities en ‘segment’. Esta es la única forma en que pude obtener marcas de tiempo y puntuación con la API OpenAI Whisper. OpenAI también tiene una API de transcripción basada en LLM, pero la falta de marcas de tiempo la hace inútil para nuestra tarea.

Ahora, la parte crítica: crear escenas multimodales. Agrupamos los fotogramas y alineamos las frases transcritas basándonos en las marcas de tiempo. Esto utiliza NLTK para dividir mejor las frases e intenta alinearlas con las horas de los fotogramas. Puedes ignorar la mayor parte de esto, lo importante es que tomamos trozos de ~30s de vídeo y los convertimos en 6 fotogramas y alrededor de 30s de transcripción, asegurándonos de que cada sección de transcripción termina en puntuación con nltk. Esto significa que cada sección no será exactamente 30s.

# --- 1.d  Create Video Chunks (NLTK, sentence‑aware, true punctuation) ------
import os, re, math, pandas as pd
from IPython.display import display
import nltk

##############################################################################
# Configuration
##############################################################################
FRAMES_DIR       = frames_dir     # from step 1.b
FRAME_FPS        = 0.2            # write_images_sequence fps
TARGET_CHUNK_SEC = 30             # desired chunk length
SLACK_FACTOR     = 1.20           # allow up to 20 % over‑run before forcing cut
##############################################################################

print("\n--- Building sentence‑aligned ~30 s chunks ---")

# ---------------------------------------------------------------------------
# 0. Make sure the Punkt model is present
# ---------------------------------------------------------------------------
try:
    nltk.data.find("tokenizers/punkt")
except LookupError:
    nltk.download("punkt", quiet=True)
from nltk.tokenize import sent_tokenize

# ---------------------------------------------------------------------------
# 1. Frame paths and helper
# ---------------------------------------------------------------------------
frame_paths = sorted(
    [os.path.join(FRAMES_DIR, f) for f in os.listdir(FRAMES_DIR) if f.endswith(".png")],
    key=lambda p: int(re.search(r"frame(\d+)\.png", os.path.basename(p)).group(1))
)
def idx_from_time(t):   return int(round(t * FRAME_FPS))

# ---------------------------------------------------------------------------
# 2. Build per‑sentence list with *estimated* timestamps
# ---------------------------------------------------------------------------
sentences = []   # list of dict(start, end, sentence)
for seg in transcription_result.segments:
    seg_start, seg_end, seg_text = seg.start, seg.end, seg.text
    seg_sents = sent_tokenize(seg_text)

    # Distribute the segment’s duration across its sentences by char length
    seg_dur    = seg_end - seg_start
    char_total = sum(len(s) for s in seg_sents)
    running_t  = seg_start
    for s in seg_sents:
        char_frac  = len(s) / char_total
        sent_end   = running_t + char_frac * seg_dur
        sentences.append({"start": running_t, "end": sent_end, "sentence": s})
        running_t = sent_end

print(f"  • {len(sentences):,} sentences total")

# ---------------------------------------------------------------------------
# 3. Pack sentences into chunks  (guarantee ending punctuation)
# ---------------------------------------------------------------------------
chunks, cur_sents = [], []
cur_start, cur_end = None, None

def ends_with_stop(txt): return txt[-1] in ".?!"

for sent in sentences:
    if not cur_sents:                            # start new chunk
        cur_sents  = [sent]
        cur_start, cur_end = sent["start"], sent["end"]
        continue

    prospective_end  = sent["end"]
    prospective_span = prospective_end - cur_start

    # Decide if we should append sentence to current chunk
    if prospective_span <= TARGET_CHUNK_SEC * SLACK_FACTOR or not ends_with_stop(cur_sents[-1]["sentence"]):
        cur_sents.append(sent)
        cur_end = prospective_end
    else:
        chunks.append({"start": cur_start, "end": cur_end, "sentences": cur_sents})
        cur_sents  = [sent]
        cur_start, cur_end = sent["start"], sent["end"]

if cur_sents:
    chunks.append({"start": cur_start, "end": cur_end, "sentences": cur_sents})

print(f"  • {len(chunks)} chunks produced "
      f"(avg {sum(c['end']-c['start'] for c in chunks)/len(chunks):.1f}s)")

# ---------------------------------------------------------------------------
# 4. Attach frames to each chunk
# ---------------------------------------------------------------------------
records = []
total_imgs = 0
for idx, ch in enumerate(chunks):
    start_t, end_t = ch["start"], ch["end"]

    first_idx = idx_from_time(start_t)
    last_idx  = max(idx_from_time(end_t) - 1, first_idx)   # inclusive
    imgs      = frame_paths[first_idx : last_idx + 1]
    total_imgs += len(imgs)

    chunk_text = " ".join(s["sentence"] for s in ch["sentences"]).strip()

    records.append(
        {
            "section":    idx,
            "start_time": round(start_t, 2),
            "end_time":   round(end_t,   2),
            "images":     imgs,
            "text":       chunk_text,
        }
    )

print(f"  • {total_imgs} total images linked")

# ---------------------------------------------------------------------------
# 5. DataFrame
# ---------------------------------------------------------------------------
df_aligned = pd.DataFrame(records)
pd.set_option("display.max_colwidth", None)
display(df_aligned.head(10))
# ---------------------------------------------------------------------------

Output:

Explicación:

• Este chunking es más sofisticado que una simple división de tamaño fijo. Utiliza las marcas de tiempo de los segmentos de Whisper y el tokenizador de frases de NLTK para crear trozos que tengan una longitud aproximada de TARGET_CHUNK_SEC pero que también intenten terminar en los límites adecuados de las frases. Esto mejora la coherencia semántica de cada trozo.
• A continuación, asigna el intervalo de tiempo calculado (start_t, end_t) de cada fragmento de texto a los archivos de fotogramas correspondientes en función de FRAME_FPS.
• Cada fila de scene_df representa ahora una escena multimodal de unos 30 segundos, la unidad fundamental que vamos a incrustar.

Etapa 2: Incrustación multimodal
Introduce cada escena (cadena de texto + lista de imágenes PIL) en Voyage AI. Cada escena consta de 6 fotogramas y un párrafo de texto en un periodo aproximado de 30 segundos.

# Generating the Multimodal Embedding Vector
import voyageai
from PIL import Image

voyage = voyageai.Client() # Reads API key from env
model_name = "voyage-multimodal-3"
def embed_scene(scene_data):
    text = scene_data["text"]
    # Load images associated with this scene
    pil_images = [Image.open(p) for p in scene_data["images"] if os.path.exists(p)]
    if not pil_images: return None # Skip if no valid images
    # Critical: Input must be [text_string, Image_obj1, Image_obj2, ...]
    input_payload = [text] + pil_images
    try:
        response = voyage.multimodal_embed(
            inputs=[input_payload], # Embed one scene at a time
            model=model_name,
            input_type="document", # Optimize for retrieval
            truncation=True        # Auto-handle long inputs
        )
        return response.embeddings[0] # Get the single vector
    except Exception as e:
        print(f"Error embedding scene {scene_data.get('id', 'N/A')}: {e}")
        return None
# Apply the embedding function to each row
scene_df['embedding'] = scene_df.apply(embed_scene, axis=1)
# Drop rows where embedding failed
scene_df.dropna(subset=['embedding'], inplace=True)

Explicación:

• La clave es la estructura de input_payload. Voyage requiere primero el texto, luego las imágenes.
• input_type=«document» es crucial – le dice a Voyage que estás incrustando contenido para almacenamiento/búsqueda, no haciendo una consulta. El modelo optimiza la representación vectorial en consecuencia.
• Aquí incrustamos cada escena individualmente para mayor claridad, pero Voyage admite la agrupación (inputs=[scene1_payload, scene2_payload, …]) para un mayor rendimiento en producción.

Etapa 3: Almacenamiento de vectores en KDB.AI
Define el esquema de la tabla utilizando los tipos KDB.AI adecuados y crea un índice optimizado para grandes conjuntos de datos.

# SNIPPET 4: KDB.AI Schema and Table Creation
import kdbai_client as kdbai
import json
import numpy as np

session = kdbai.Session(endpoint=os.getenv("KDBAI_ENDPOINT"), api_key=os.getenv("KDBAI_API_KEY"))
db = session.database("default")
TABLE_NAME = "video_multimodal_scenes"
embedding_dim = len(scene_df['embedding'].iloc[0])
schema = [
    {"name": "id",          "type": "str"},
    {"name": "text_bytes",  "type": "bytes"},    # Store text efficiently as bytes
    {"name": "image_paths", "type": "str"},      # JSON string list of frame paths
    {"name": "embeddings",  "type": "float32s", "pytype": f"{embedding_dim}f"}, # Vector
]
# Use qHNSW for large datasets: Approximate search, disk-based storage
indexes = [{
    "type": "qHNSW", # Changed from qFlat for scalability
    "name": "idx_emb",
    "column": "embeddings",
    "params": {"dims": embedding_dim, "metric": "CS"}
}]
table = db.create_table(TABLE_NAME, schema=schema, indexes=indexes)
# Prepare and insert data
table.insert(insert_payload) # insert_payload is the formatted

Explicación:

• Almacenamos el texto como bytes (codificado UTF-8)
• Las rutas de las imágenes se almacenan como una cadena JSON str.
• La columna embeddings utiliza float32s.
• Elección del índice: Utilizamos qHNSW. ¿Por qué? qHNSW construye una estructura de grafos que permite una búsqueda aproximada pero extremadamente rápida, incluso cuando los vectores se almacenan principalmente en disco. Esto permite escalar a miles de millones de vectores sin necesidad de terabytes de RAM, lo que lo hace rentable para grandes archivos de vídeo. CS (Cosine Similarity) es estándar para vectores semánticos.

Etapa 4: La consulta RAG – Buscar, fusionar, preguntar

Esta función lo une todo: incrustar la consulta, encontrar escenas similares, preparar el contexto (con imágenes fusionadas) y consultar el VLM.

La primera vez que desarrollé esta función, me topé rápidamente con los límites de la API debido al número de tokens de las imágenes. Dado que obtenemos 6 imágenes por sección de vídeo recuperada, si recuperamos 4 secciones son 24 imágenes, ¡o cientos de miles de tokens!

Un truco para reducir esto es crear un sprite: un sprite es una imagen larga fusionada compuesta de muchas subimágenes. Esto se utiliza a menudo para reducir el tiempo de carga de una página web, cargando un montón de imágenes a la vez como una gran imagen, pero en este caso reducirá nuestro uso de tokens drásticamente.

from PIL import Image

def merge_images(paths):
    images = [Image.open(p) for p in paths]
    widths, heights = zip(*(img.size for img in images))
    total_width = sum(widths)
    max_height = max(heights)

    merged = Image.new('RGB', (total_width, max_height))
    x_offset = 0
    for img in images:
        merged.paste(img, (x_offset, 0))
        x_offset += img.size[0]

    return merged

def multimodal_rag(query: str, k: int = 3) -> str:
    global table

    q_emb = voyage.multimodal_embed([[query]], model="voyage-multimodal-3", input_type="query", truncation=True).embeddings[0]
    retrieved = table.search(vectors={"idx_emb": [q_emb]}, n=k)[0]

    context = [{"type": "text", "text": f"Answer the query based only on the following video segments.\n\nQuestion: {query}\n"}]
    preview = []

    for i, row in retrieved.iterrows():
        tid = row.get('id', f'Retrieved_{i}')
        txt = row['text_bytes'].decode('utf-8')
        context += [{"type": "text", "text": f"\n--- Segment {tid} Text ---"}, {"type": "text", "text": txt}]
        preview.append({"type": "input_text", "text": f"\n--- Segment {tid} Text ---\n{txt}"})

        img_paths = json.loads(row.get('image_paths', '[]'))
        if img_paths:
            merged_img = merge_images(img_paths)
            merged_img_path = "/tmp/merged_segment_image.jpg"
            merged_img.save(merged_img_path, format="JPEG")

            base64_img = encode_base64(merged_img_path)
            context += [{"type": "image_url", "image_url": {"url": base64_img}}]

            preview.append({"type": "input_image", "image_url": base64_img})
            display(merged_img)

    context.append({"type": "text", "text": "\n--- End of Retrieved Context ---"})

    display_rag_preview(preview)

    res = openai.chat.completions.create(model="gpt-4o-mini", messages=[{"role": "user", "content": context}], max_tokens=500)
    out = res.choices[0].message.content

    display(Markdown(out))
    return out

Y ahora podemos consultar nuestra RAG pipeline:

question1 = "What is the central limit theorem?"
print(f"\nExecuting RAG for query: '{question1}'")
response1 = multimodal_rag(query=question1, k=4)

Resultado:

Explicación:

• Incrustación de consultas: Es muy importante utilizar input_type=«query» al incrustar la pregunta del usuario. Voyage optimiza los vectores de consulta de forma diferente a los vectores de documentos para una mejor recuperación.
• Búsqueda KDB.AI: table.search utiliza idx_emb (nuestro índice HNSW) para recuperar rápidamente los k vectores de escena más similares.
• Fusión de imágenes (ahorro de fichas): La llamada merge_images_to_sprite es vital. En lugar de enviar k * 6 = 24 imágenes separadas (si k=4), enviamos sólo k=4 sprites fusionados. Esto reduce drásticamente el número de entradas de imágenes y los costes de token asociados para la llamada VLM, haciendo que el enfoque sea económicamente viable.
• Estructura del prompt: El llm_context intercala cuidadosamente texto ({«type»: «text», …}) y el correspondiente sprite de imagen fusionada ({«type»: «image_url», …}). Este formato es necesario para modelos como GPT-4o.
• Elección LLM: gpt-4o-mini ofrece un buen equilibrio entre capacidad y coste para este tipo de Q&A en tierra. Gemini es mucho más barato, pero he utilizado 4o por conveniencia ya que ya estamos utilizando OpenAI para la API Whisper.

Por qué es la nueva norma (y qué hay que mejorar)

Este enfoque RAG multimodal basado en escenas es una mejora fundamental porque respeta la naturaleza inherente del vídeo: la fusión de la vista y el sonido a lo largo del tiempo.

Qué se ha resuelto:

• Preservación del contexto: Las incrustaciones capturan el vínculo entre los fotogramas y la transcripción.
• Recuperación relevante: La búsqueda encuentra escenas semánticamente relevantes, no sólo fotogramas aislados o fragmentos de texto.
• Generación fundamentada: Los VLM pueden responder a preguntas utilizando pruebas visuales y textuales.
• Gestión de costes: La fusión de imágenes hace que las llamadas VLM sean asequibles.

Lo que necesita optimización (Las oportunidades Alfa):

• Chunking Granularity: Encontrar los fps y FRAMES_PER_CHUNK óptimos para distintos tipos de vídeo sigue siendo un arte. Se necesitan más estudios y puntos de referencia. La fragmentación inteligente basada en cortes de escena reales será superior a los intervalos fijos.
• Selección de fotogramas: ¿Podemos utilizar la similitud/agrupación de imágenes para descartar los fotogramas redundantes antes de incrustarlos? Esto podría reducir aún más el coste y el ruido.
• Modelos de código abierto: El ecosistema necesita incrustadores multimodales de código abierto maduros para reducir la dependencia de los proveedores y permitir implantaciones totalmente privadas.
• Modelos de vídeo real: Los modelos actuales procesan secuencias de fotogramas. Los modelos futuros podrían ingerir secuencias de vídeo directamente, comprendiendo mejor el movimiento y los patrones temporales.

Incluso con estas fronteras, el plan que aquí se presenta es práctico hoy en día. Puedes construir un sistema que convierta tu archivo de vídeo de un problema de almacenamiento pasivo en una base de conocimientos activa y consultable.

Constrúyelo

Por fin las herramientas están a la altura de la visión. Incrustaciones multimodales, bases de datos vectoriales rápidas, VLM capaces: todos son accesibles a través de API o bibliotecas sencillas.

Obtén la implementación completa del Colab, dirígela a tu propio contenido de vídeo (conferencias, reuniones, demostraciones de productos) y descubre qué información puedes obtener cuando tu IA por fin pueda ver y escuchar.

Puedes continuar en: Cuaderno Colab completo