Búsqueda avanzada de vídeos: Aprovechamiento de Twelve Labs y Milvus para la recuperación semántica

Introducción

Bienvenido a este completo tutorial sobre la implementación de la búsqueda semántica de vídeos mediante la API de incrustación de Twelve Labs y Milvus. En esta guía, exploraremos cómo aprovechar el poder de las incrustaciones multimodales avanzadas de Twelve Labs y la eficiente base de datos vectorial de Milvus para crear una solución robusta de búsqueda de vídeo. Mediante la integración de estas tecnologías, los desarrolladores pueden desbloquear nuevas posibilidades en el análisis de contenido de vídeo, permitiendo aplicaciones como la recuperación de vídeo basada en contenido, sistemas de recomendación y sofisticados motores de búsqueda que entienden los matices de los datos de vídeo.

Este tutorial le guiará a través de todo el proceso, desde la configuración de su entorno de desarrollo hasta la implementación de una aplicación funcional de búsqueda semántica de vídeo. Cubriremos conceptos clave como la generación de incrustaciones multimodales a partir de vídeos, su almacenamiento eficiente en Milvus y la realización de búsquedas por similitud para recuperar contenido relevante. Tanto si está construyendo una plataforma de análisis de vídeo, una herramienta de descubrimiento de contenidos o mejorando sus aplicaciones existentes con capacidades de búsqueda de vídeo, esta guía le proporcionará los conocimientos y pasos prácticos para aprovechar las fortalezas combinadas de Twelve Labs y Milvus en sus proyectos.

Requisitos previos

Antes de empezar, asegúrese de que dispone de lo siguiente

Una clave API de Twelve Labs (regístrese en https://api.twelvelabs.io si no tiene una) Python 3.7 o posterior instalado en su sistema

Configuración del entorno de desarrollo

Cree un nuevo directorio para su proyecto y navegue hasta él:

mkdir video-search-tutorial
cd video-search-tutorial

Configura un entorno virtual (opcional pero recomendado):

python -m venv venv
source venv/bin/activate  # On Windows, use `venv\Scripts\activate`

Instala las librerías Python necesarias:

pip install twelvelabs pymilvus

Crea un nuevo archivo Python para tu proyecto:

touch video_search.py

Este archivo video_search.py será el script principal que utilizaremos para el tutorial. A continuación, configure su clave de API de Twelve Labs como variable de entorno por motivos de seguridad:

export TWELVE_LABS_API_KEY='your_api_key_here'

Conexión a Milvus

Para establecer una conexión con Milvus, utilizaremos la clase MilvusClient. Este enfoque simplifica el proceso de conexión y nos permite trabajar con una instancia de Milvus basada en un archivo local, lo que es perfecto para nuestro tutorial.

from pymilvus import MilvusClient

# Initialize the Milvus client
milvus_client = MilvusClient("milvus_twelvelabs_demo.db")

print("Successfully connected to Milvus")

Este código crea una nueva instancia de cliente Milvus que almacenará todos los datos en un archivo llamado milvus_twelvelabs_demo.db. Este enfoque basado en archivos es ideal para fines de desarrollo y pruebas.

Creación de una colección Milvus para incrustaciones de vídeo

Ahora que estamos conectados a Milvus, vamos a crear una colección para almacenar nuestras incrustaciones de vídeo y los metadatos asociados. Definiremos el esquema de la colección y crearemos la colección si aún no existe.

# Initialize the collection name
collection_name = "twelvelabs_demo_collection"

# Check if the collection already exists and drop it if it does
if milvus_client.has_collection(collection_name=collection_name):
    milvus_client.drop_collection(collection_name=collection_name)

# Create the collection
milvus_client.create_collection(
    collection_name=collection_name,
    dimension=1024  # The dimension of the Twelve Labs embeddings
)

print(f"Collection '{collection_name}' created successfully")

En este código, primero comprobamos si la colección ya existe y la eliminamos en caso afirmativo. Así nos aseguramos de empezar de cero. Creamos la colección con una dimensión de 1024, que coincide con la dimensión de salida de las incrustaciones de Twelve Labs.

Generación de incrustaciones con la API de incrustación de Twelve Labs

Para generar incrustaciones para nuestros vídeos mediante la API de incrustación de Twelve Labs, utilizaremos el SDK de Python de Twelve Labs. Este proceso implica crear una tarea de incrustación, esperar a que se complete y recuperar los resultados. A continuación te explicamos cómo hacerlo:

En primer lugar, asegúrese de que tiene instalado el SDK de Twelve Labs e importe los módulos necesarios:

from twelvelabs import TwelveLabs
from twelvelabs.models.embed import EmbeddingsTask
import os

# Retrieve the API key from environment variables
TWELVE_LABS_API_KEY = os.getenv('TWELVE_LABS_API_KEY')

Inicialice el cliente de Twelve Labs:

twelvelabs_client = TwelveLabs(api_key=TWELVE_LABS_API_KEY)

Cree una función para generar incrustaciones para una URL de vídeo dada:

def generate_embedding(video_url):
    """
    Generate embeddings for a given video URL using the Twelve Labs API.

    This function creates an embedding task for the specified video URL using
    the Marengo-retrieval-2.6 engine. It monitors the task progress and waits
    for completion. Once done, it retrieves the task result and extracts the
    embeddings along with their associated metadata.

    Args:
        video_url (str): The URL of the video to generate embeddings for.

    Returns:
        tuple: A tuple containing two elements:
            1. list: A list of dictionaries, where each dictionary contains:
                - 'embedding': The embedding vector as a list of floats.
                - 'start_offset_sec': The start time of the segment in seconds.
                - 'end_offset_sec': The end time of the segment in seconds.
                - 'embedding_scope': The scope of the embedding (e.g., 'shot', 'scene').
            2. EmbeddingsTaskResult: The complete task result object from Twelve Labs API.

    Raises:
        Any exceptions raised by the Twelve Labs API during task creation,
        execution, or retrieval.
    """

    # Create an embedding task
    task = twelvelabs_client.embed.task.create(
        engine_name="Marengo-retrieval-2.6",
        video_url=video_url
    )
    print(f"Created task: id={task.id} engine_name={task.engine_name} status={task.status}")

    # Define a callback function to monitor task progress
    def on_task_update(task: EmbeddingsTask):
        print(f"  Status={task.status}")

    # Wait for the task to complete
    status = task.wait_for_done(
        sleep_interval=2,
        callback=on_task_update
    )
    print(f"Embedding done: {status}")

    # Retrieve the task result
    task_result = twelvelabs_client.embed.task.retrieve(task.id)

    # Extract and return the embeddings
    embeddings = []
    for v in task_result.video_embeddings:
        embeddings.append({
            'embedding': v.embedding.float,
            'start_offset_sec': v.start_offset_sec,
            'end_offset_sec': v.end_offset_sec,
            'embedding_scope': v.embedding_scope
        })
    
    return embeddings, task_result

Utilice la función para generar incrustaciones para sus vídeos:

# Example usage
video_url = "https://example.com/your-video.mp4"

# Generate embeddings for the video
embeddings, task_result = generate_embedding(video_url)

print(f"Generated {len(embeddings)} embeddings for the video")
for i, emb in enumerate(embeddings):
    print(f"Embedding {i+1}:")
    print(f"  Scope: {emb['embedding_scope']}")
    print(f"  Time range: {emb['start_offset_sec']} - {emb['end_offset_sec']} seconds")
    print(f"  Embedding vector (first 5 values): {emb['embedding'][:5]}")
    print()

Esta implementación permite generar incrustaciones para cualquier URL de vídeo utilizando la API de incrustación de Twelve Labs. La función generate_embedding se encarga de todo el proceso, desde la creación de la tarea hasta la recuperación de los resultados. Devuelve una lista de diccionarios, cada uno de los cuales contiene un vector de incrustación junto con sus metadatos (intervalo de tiempo y ámbito). También es posible que desee implementar reintentos o una gestión de errores más robusta en función de su caso de uso específico.

Insertar incrustaciones en Milvus

Después de generar incrustaciones utilizando la API Embed de Twelve Labs, el siguiente paso es insertar estas incrustaciones junto con sus metadatos en nuestra colección Milvus. Este proceso nos permite almacenar e indexar nuestras incrustaciones de vídeo para una búsqueda eficiente de similitudes más adelante.

He aquí cómo insertar las incrustaciones en Milvus:

def insert_embeddings(milvus_client, collection_name, task_result, video_url):
    """
    Insert embeddings into the Milvus collection.

    Args:
        milvus_client: The Milvus client instance.
        collection_name (str): The name of the Milvus collection to insert into.
        task_result (EmbeddingsTaskResult): The task result containing video embeddings.
        video_url (str): The URL of the video associated with the embeddings.

    Returns:
        MutationResult: The result of the insert operation.

    This function takes the video embeddings from the task result and inserts them
    into the specified Milvus collection. Each embedding is stored with additional
    metadata including its scope, start and end times, and the associated video URL.
    """
    data = []

    for i, v in enumerate(task_result.video_embeddings):
        data.append({
            "id": i,
            "vector": v.embedding.float,
            "embedding_scope": v.embedding_scope,
            "start_offset_sec": v.start_offset_sec,
            "end_offset_sec": v.end_offset_sec,
            "video_url": video_url
        })

    insert_result = milvus_client.insert(collection_name=collection_name, data=data)
    print(f"Inserted {len(data)} embeddings into Milvus")
    return insert_result

# Usage example
video_url = "https://example.com/your-video.mp4"

# Assuming this function exists from previous step
embeddings, task_result = generate_embedding(video_url)

# Insert embeddings into the Milvus collection
insert_result = insert_embeddings(milvus_client, collection_name, task_result, video_url)
print(insert_result)

Esta función prepara los datos para su inserción, incluyendo todos los metadatos relevantes como el vector de incrustación, el intervalo de tiempo y la URL del vídeo de origen. A continuación, utiliza el cliente Milvus para insertar estos datos en la colección especificada.

Búsqueda por similitud

Una vez que tenemos nuestras incrustaciones almacenadas en Milvus, podemos realizar búsquedas por similitud para encontrar los segmentos de vídeo más relevantes basándonos en un vector de consulta. He aquí cómo implementar esta funcionalidad:

def perform_similarity_search(milvus_client, collection_name, query_vector, limit=5):
    """
    Perform a similarity search on the Milvus collection.

    Args:
        milvus_client: The Milvus client instance.
        collection_name (str): The name of the Milvus collection to search in.
        query_vector (list): The query vector to search for similar embeddings.
        limit (int, optional): The maximum number of results to return. Defaults to 5.

    Returns:
        list: A list of search results, where each result is a dictionary containing
              the matched entity's metadata and similarity score.

    This function searches the specified Milvus collection for embeddings similar to
    the given query vector. It returns the top matching results, including metadata
    such as the embedding scope, time range, and associated video URL for each match.
    """
    search_results = milvus_client.search(
        collection_name=collection_name,
        data=[query_vector],
        limit=limit,
        output_fields=["embedding_scope", "start_offset_sec", "end_offset_sec", "video_url"]
    )

    return search_results
    
# define the query vector
# We use the embedding inserted previously as an example. In practice, you can replace it with any video embedding you want to query.
query_vector = task_result.video_embeddings[0].embedding.float

# Perform a similarity search on the Milvus collection
search_results = perform_similarity_search(milvus_client, collection_name, query_vector)

print("Search Results:")
for i, result in enumerate(search_results[0]):
    print(f"Result {i+1}:")
    print(f"  Video URL: {result['entity']['video_url']}")
    print(f"  Time Range: {result['entity']['start_offset_sec']} - {result['entity']['end_offset_sec']} seconds")
    print(f"  Similarity Score: {result['distance']}")
    print()

Esta implementación hace lo siguiente

1. Define una función perform_similarity_search que toma un vector de consulta y busca incrustaciones similares en la colección Milvus.
2. Utiliza el método de búsqueda del cliente Milvus para encontrar los vectores más similares.
3. Especifica los campos de salida que queremos recuperar, incluyendo metadatos sobre los segmentos de vídeo coincidentes.
4. Proporciona un ejemplo de cómo utilizar esta función con un vídeo de consulta, generando primero su incrustación y utilizándola después para buscar.
5. Imprime los resultados de la búsqueda, incluidos los metadatos relevantes y las puntuaciones de similitud.

Al implementar estas funciones, ha creado un flujo de trabajo completo para almacenar incrustaciones de vídeo en Milvus y realizar búsquedas de similitud. Esta configuración permite una recuperación eficiente de contenidos de vídeo similares basados en las incrustaciones multimodales generadas por la Embed API de Twelve Labs.

Optimización del rendimiento

Muy bien, ¡vamos a llevar esta aplicación al siguiente nivel! Cuando se trata de colecciones de vídeos a gran escala, el rendimiento es clave. Para optimizarlo, deberíamos implementar el procesamiento por lotes para la generación de incrustaciones y su inserción en Milvus. De esta forma, podemos manejar múltiples vídeos simultáneamente, reduciendo significativamente el tiempo total de procesamiento. Además, podríamos aprovechar la función de partición de Milvus para organizar nuestros datos de forma más eficiente, quizás por categorías de vídeo o periodos de tiempo. Esto aceleraría las consultas al permitirnos buscar sólo en las particiones relevantes.

Otro truco de optimización consiste en utilizar mecanismos de caché para las incrustaciones o los resultados de búsqueda a los que se accede con frecuencia. Esto podría mejorar drásticamente los tiempos de respuesta de las consultas más frecuentes. No olvide ajustar los parámetros de índice de Milvus en función de su conjunto de datos específico y de sus patrones de consulta: un pequeño ajuste puede contribuir en gran medida a mejorar el rendimiento de la búsqueda.

Funciones avanzadas

Ahora vamos a añadir algunas funciones interesantes para que nuestra aplicación destaque. Podríamos implementar una búsqueda híbrida que combine consultas de texto y vídeo. De hecho, Twelve Labs Embed API también puede generar incrustaciones de texto para sus consultas de texto. Imagine que permite a los usuarios introducir tanto una descripción de texto como un clip de vídeo de muestra: generaríamos incrustaciones para ambos y realizaríamos una búsqueda ponderada en Milvus. Esto nos daría resultados muy precisos.

Otro añadido impresionante sería la búsqueda temporal dentro de los vídeos. Podríamos dividir los vídeos largos en segmentos más pequeños, cada uno con su propia incrustación. De esta manera, los usuarios podrían encontrar momentos específicos dentro de los vídeos, no sólo clips enteros. ¿Y por qué no añadir un poco de análisis de vídeo básico? Podríamos utilizar las incrustaciones para agrupar segmentos de vídeo similares, detectar tendencias o incluso identificar valores atípicos en grandes colecciones de vídeos.

Gestión y registro de errores

Admitámoslo, las cosas pueden salir mal, y cuando lo hacen, tenemos que estar preparados. Implementar una gestión de errores sólida es crucial. Debemos envolver las llamadas a la API y las operaciones de base de datos en bloques try-except, proporcionando mensajes de error informativos a los usuarios cuando algo falla. Para los problemas relacionados con la red, la implementación de reintentos con retroceso exponencial puede ayudar a gestionar los fallos temporales con elegancia.

En cuanto al registro, es nuestro mejor amigo para la depuración y la monitorización. Deberíamos usar el módulo de registro de Python para rastrear eventos importantes, errores y métricas de rendimiento a lo largo de nuestra aplicación. Vamos a establecer diferentes niveles de registro - DEBUG para el desarrollo, INFO para el funcionamiento general, y ERROR para problemas críticos. Y no olvides aplicar la rotación de registros para gestionar el tamaño de los archivos. Con un registro adecuado, seremos capaces de identificar y resolver rápidamente los problemas, asegurando que nuestra aplicación de búsqueda de vídeo se ejecuta sin problemas, incluso a medida que se escala.

Conclusión

Enhorabuena. Ya ha creado una potente aplicación de búsqueda semántica de vídeo utilizando la API de incrustación de Twelve Labs y Milvus. Esta integración le permite procesar, almacenar y recuperar contenido de vídeo con una precisión y eficacia sin precedentes. Al aprovechar las incrustaciones multimodales, ha creado un sistema que comprende los matices de los datos de vídeo, abriendo interesantes posibilidades para el descubrimiento de contenidos, los sistemas de recomendación y el análisis avanzado de vídeo.

A medida que continúe desarrollando y perfeccionando su aplicación, recuerde que la combinación de la generación de incrustación avanzada de Twelve Labs y el almacenamiento vectorial escalable de Milvus proporciona una base sólida para abordar retos de comprensión de vídeo aún más complejos. Le animamos a experimentar con las funciones avanzadas comentadas y a ampliar los límites de lo que es posible en la búsqueda y el análisis de vídeo.