Pesquisa avançada de vídeo: Tirar partido do Twelve Labs e do Milvus para a recuperação semântica

Introdução

Bem-vindo a este tutorial abrangente sobre a implementação da pesquisa semântica de vídeo usando a API de incorporação do Twelve Labs e o Milvus. Neste guia, vamos explorar como aproveitar o poder das incorporações multimodais avançadas do Twelve Labs e da base de dados vetorial eficiente do Milvus para criar uma solução de pesquisa de vídeo robusta. Ao integrar estas tecnologias, os programadores podem desbloquear novas possibilidades na análise de conteúdos de vídeo, permitindo aplicações como a recuperação de vídeo baseada em conteúdos, sistemas de recomendação e motores de pesquisa sofisticados que compreendem as nuances dos dados de vídeo.

Este tutorial irá guiá-lo por todo o processo, desde a configuração do seu ambiente de desenvolvimento até a implementação de um aplicativo de pesquisa de vídeo semântico funcional. Abordaremos conceitos-chave como a geração de embeddings multimodais a partir de vídeos, armazenando-os de forma eficiente no Milvus e realizando pesquisas de similaridade para recuperar conteúdo relevante. Quer esteja a construir uma plataforma de análise de vídeo, uma ferramenta de descoberta de conteúdo ou a melhorar as suas aplicações existentes com capacidades de pesquisa de vídeo, este guia irá fornecer-lhe o conhecimento e os passos práticos para aproveitar os pontos fortes combinados do Twelve Labs e do Milvus nos seus projectos.

Pré-requisitos

Antes de começarmos, certifique-se de que possui o seguinte:

Uma chave de API do Twelve Labs (inscreva-se em https://api.twelvelabs.io se ainda não tiver uma) Python 3.7 ou posterior instalado no seu sistema

Configurando o ambiente de desenvolvimento

Crie um novo diretório para o seu projeto e navegue até ele:

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

Configure um ambiente virtual (opcional, mas recomendado):

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

Instale as bibliotecas Python necessárias:

pip install twelvelabs pymilvus

Crie um novo ficheiro Python para o seu projeto:

touch video_search.py

Este arquivo video_search.py será o script principal que usaremos para o tutorial. Em seguida, configure sua chave de API do Twelve Labs como uma variável de ambiente para segurança:

export TWELVE_LABS_API_KEY='your_api_key_here'

Conexão com o Milvus

Para estabelecer uma conexão com o Milvus, usaremos a classe MilvusClient. Esta abordagem simplifica o processo de ligação e permite-nos trabalhar com uma instância local do Milvus baseada em ficheiros, o que é perfeito para o nosso tutorial.

from pymilvus import MilvusClient

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

print("Successfully connected to Milvus")

Esse código cria uma nova instância de cliente Milvus que armazenará todos os dados em um arquivo chamado milvus_twelvelabs_demo.db. Esta abordagem baseada em ficheiros é ideal para fins de desenvolvimento e teste.

Criando uma coleção do Milvus para incorporação de vídeos

Agora que estamos ligados ao Milvus, vamos criar uma coleção para armazenar as nossas incorporações de vídeo e os metadados associados. Vamos definir o esquema da coleção e criar a coleção, caso ainda não exista.

# 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")

Neste código, primeiro verificamos se a coleção já existe e, se existir, eliminamo-la. Isto garante que começamos com um quadro limpo. Criamos a coleção com uma dimensão de 1024, que corresponde à dimensão de saída dos embeddings do Twelve Labs.

Geração de embeddings com a API de embeddings do Twelve Labs

Para gerar embeddings para nossos vídeos usando a API de incorporação do Twelve Labs, usaremos o SDK Python do Twelve Labs. Este processo envolve a criação de uma tarefa de incorporação, aguardando a sua conclusão e recuperando os resultados. Veja como implementar isso:

Primeiro, certifique-se de que tem o SDK do Twelve Labs instalado e importe os módulos necessários:

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')

Inicialize o cliente Twelve Labs:

twelvelabs_client = TwelveLabs(api_key=TWELVE_LABS_API_KEY)

Crie uma função para gerar embeddings para um determinado URL de vídeo:

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

Utilize a função para gerar embeddings para os seus 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 implementação permite-lhe gerar embeddings para qualquer URL de vídeo utilizando a API de incorporação do Twelve Labs. A função generate_embedding trata de todo o processo, desde a criação da tarefa até à obtenção dos resultados. Ela retorna uma lista de dicionários, cada um contendo um vetor de incorporação junto com seus metadados (intervalo de tempo e escopo). Lembre-se de lidar com possíveis erros, como problemas de rede ou limites de API, em um ambiente de produção. Também pode querer implementar novas tentativas ou um tratamento de erros mais robusto, dependendo do seu caso de utilização específico.

Inserindo embeddings no Milvus

Depois de gerar embeddings usando a API Embed do Twelve Labs, a próxima etapa é inserir esses embeddings junto com seus metadados na coleção do Milvus. Este processo permite-nos armazenar e indexar os nossos embeddings de vídeo para uma pesquisa de semelhança eficiente mais tarde.

Veja como inserir os embeddings no 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 função prepara os dados para inserção, incluindo todos os metadados relevantes, como o vetor de incorporação, o intervalo de tempo e o URL do vídeo de origem. Em seguida, ela usa o cliente Milvus para inserir esses dados na coleção especificada.

Realização de pesquisa de similaridade

Assim que tivermos nossos embeddings armazenados no Milvus, podemos realizar pesquisas de similaridade para encontrar os segmentos de vídeo mais relevantes com base em um vetor de consulta. Veja como implementar essa funcionalidade:

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 implementação faz o seguinte:

1. Define uma função perform_similarity_search que recebe um vetor de consulta e procura por embeddings similares na coleção do Milvus.
2. Usa o método de pesquisa do cliente Milvus para encontrar os vectores mais semelhantes.
3. Especifica os campos de saída que queremos obter, incluindo metadados sobre os segmentos de vídeo correspondentes.
4. Fornece um exemplo de como usar essa função com um vídeo de consulta, primeiro gerando sua incorporação e depois usando-a para pesquisar.
5. Imprime os resultados da pesquisa, incluindo metadados relevantes e pontuações de similaridade.

Ao implementar essas funções, você criou um fluxo de trabalho completo para armazenar as incorporações de vídeo no Milvus e realizar pesquisas de similaridade. Essa configuração permite a recuperação eficiente de conteúdo de vídeo semelhante com base nos embeddings multimodais gerados pela API Embed do Twelve Labs.

Otimização do desempenho

Muito bem, vamos levar esta aplicação para o próximo nível! Ao lidar com colecções de vídeo em grande escala, o desempenho é fundamental. Para otimizar, devemos implementar o processamento em lote para a geração e inserção de embedding no Milvus. Desta forma, podemos lidar com vários vídeos em simultâneo, reduzindo significativamente o tempo de processamento global. Além disso, poderíamos aproveitar o recurso de particionamento do Milvus para organizar nossos dados de forma mais eficiente, talvez por categorias de vídeo ou períodos de tempo. Isto aceleraria as consultas, permitindo-nos pesquisar apenas as partições relevantes.

Outro truque de otimização é a utilização de mecanismos de cache para os embeddings ou resultados de pesquisa frequentemente acedidos. Isto pode melhorar drasticamente os tempos de resposta para consultas populares. Não se esqueça de afinar os parâmetros de indexação do Milvus com base no seu conjunto de dados específico e nos padrões de consulta - um pequeno ajuste aqui pode ajudar muito a melhorar o desempenho da pesquisa.

Funcionalidades avançadas

Agora, vamos adicionar algumas funcionalidades interessantes para destacar a nossa aplicação! Podemos implementar uma pesquisa híbrida que combine consultas de texto e vídeo. De facto, a API de incorporação do Twelve Labs também pode gerar incorporações de texto para as suas consultas de texto. Imagine permitir que os utilizadores introduzam uma descrição de texto e um clip de vídeo de amostra - geraríamos embeddings para ambos e efectuaríamos uma pesquisa ponderada em Milvus. Isto dar-nos-ia resultados muito precisos.

Outra adição espetacular seria a pesquisa temporal nos vídeos. Poderíamos dividir vídeos longos em segmentos mais pequenos, cada um com o seu próprio embedding. Desta forma, os utilizadores poderiam encontrar momentos específicos nos vídeos e não apenas clips inteiros. E porque não incluir uma análise de vídeo básica? Poderíamos utilizar as incorporações para agrupar segmentos de vídeo semelhantes, detetar tendências ou mesmo identificar valores atípicos em grandes colecções de vídeo.

Tratamento de erros e registo

Sejamos realistas, as coisas podem correr mal e, quando correm, temos de estar preparados. A implementação de um tratamento de erros robusto é crucial. Devemos envolver as nossas chamadas API e operações de base de dados em blocos try-except, fornecendo mensagens de erro informativas aos utilizadores quando algo falha. Para problemas relacionados com a rede, a implementação de novas tentativas com backoff exponencial pode ajudar a lidar com falhas temporárias de forma graciosa.

Quanto ao registo, é o nosso melhor amigo para depuração e monitorização. Devemos usar o módulo de registo do Python para registar eventos importantes, erros e métricas de desempenho ao longo da nossa aplicação. Vamos configurar diferentes níveis de registo - DEBUG para desenvolvimento, INFO para funcionamento geral e ERROR para problemas críticos. E não se esqueça de implementar a rotação de registos para gerir o tamanho dos ficheiros. Com o registo adequado, poderemos identificar e resolver rapidamente os problemas, garantindo que a nossa aplicação de pesquisa de vídeo funciona sem problemas, mesmo quando aumenta de escala.

Conclusão

Parabéns! Você criou um poderoso aplicativo de pesquisa de vídeo semântico usando a API Embed do Twelve Labs e o Milvus. Essa integração permite processar, armazenar e recuperar conteúdo de vídeo com precisão e eficiência sem precedentes. Ao tirar partido dos embeddings multimodais, criou um sistema que compreende as nuances dos dados de vídeo, abrindo possibilidades interessantes para a descoberta de conteúdos, sistemas de recomendação e análise avançada de vídeo.

À medida que continua a desenvolver e a aperfeiçoar a sua aplicação, lembre-se de que a combinação da geração de incorporação avançada do Twelve Labs e do armazenamento vetorial escalável do Milvus fornece uma base robusta para enfrentar desafios de compreensão de vídeo ainda mais complexos. Incentivamo-lo a experimentar as funcionalidades avançadas discutidas e a ultrapassar os limites do que é possível na pesquisa e análise de vídeo.