高级视频搜索：利用 Twelve Labs 和 Milvus 进行语义检索

介绍

欢迎学习这个关于使用 Twelve Labs 嵌入 API 和 Milvus 实现语义视频搜索的综合教程。在本指南中，我们将探索如何利用 Twelve Labs 的高级多模态嵌入和 Milvus 的高效向量数据库来创建强大的视频搜索解决方案。通过集成这些技术，开发人员可以释放视频内容分析的新可能性，实现基于内容的视频检索、推荐系统和理解视频数据细微差别的复杂搜索引擎等应用。

本教程将引导您完成整个过程，从设置开发环境到实现功能性语义视频搜索应用程序。我们将涵盖关键概念，如从视频生成多模态嵌入、在 Milvus 中高效存储以及执行相似性搜索以检索相关内容。无论您是在构建视频分析平台、内容发现工具，还是为现有应用程序增强视频搜索功能，本指南都将为您提供在项目中利用 Twelve Labs 和 Milvus 的综合优势所需的知识和实践步骤。

先决条件

在开始之前，请确保您拥有以下条件：

一个 Twelve Labs API 密钥（如果没有，请在 https://api.twelvelabs.io 注册）
在您的系统上安装 Python 3.7 或更高版本

设置开发环境

为您的项目创建一个新目录并导航到其中：

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

设置虚拟环境（可选但推荐）：

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

安装所需的 Python 库：

pip install twelvelabs pymilvus

为您的项目创建一个新的 Python 文件：

touch video_search.py

这个 video_search.py 文件将是我们在教程中使用的主要脚本。接下来，为了安全起见，将您的 Twelve Labs API 密钥设置为环境变量：

export TWELVE_LABS_API_KEY='your_api_key_here'

连接到 Milvus

要建立与 Milvus 的连接，我们将使用 MilvusClient 类。这种方法简化了连接过程，并允许我们使用基于本地文件的 Milvus 实例，这对我们的教程来说是完美的。

from pymilvus import MilvusClient

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

print("Successfully connected to Milvus")

此代码创建一个新的 Milvus 客户端实例，该实例将把所有数据存储在名为 milvus_twelvelabs_demo.db 的文件中。这种基于文件的方法非常适合开发和测试目的。

为视频嵌入创建 Milvus Collection

现在我们已连接到 Milvus，让我们创建一个 Collection 来存储我们的视频嵌入和相关元数据。我们将定义 Collection 模式并在其不存在时创建 Collection。

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

在此代码中，我们首先检查 Collection 是否已存在，如果存在则删除它。这确保我们从一个干净的状态开始。我们创建的 Collection 具有 1024 的维度，这与 Twelve Labs 嵌入的输出维度匹配。

使用 Twelve Labs 嵌入 API 生成嵌入

要使用 Twelve Labs 嵌入 API 为我们的视频生成嵌入，我们将使用 Twelve Labs Python SDK。此过程涉及创建嵌入任务、等待其完成和检索结果。以下是如何实现这一点：

首先，确保您已安装 Twelve Labs SDK 并导入必要的模块：

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

初始化 Twelve Labs 客户端：

twelvelabs_client = TwelveLabs(api_key=TWELVE_LABS_API_KEY)

创建一个函数为给定的视频 URL 生成嵌入：

def generate_embedding(video_url):
	"""
    使用 Twelve Labs API 为给定的视频 URL 生成嵌入。

    此函数使用 Marengo-retrieval-2.6 引擎为指定的视频 URL 创建嵌入任务。
    它监控任务进度并等待完成。完成后，它检索任务结果并提取嵌入及其关联的元数据。

    Args:
        video_url (str): 要生成嵌入的视频的 URL。

    Returns:
        tuple: 包含两个元素的元组：
            1. list: 字典列表，每个字典包含：
                - 'embedding': 作为浮点数列表的嵌入向量。
                - 'start_offset_sec': 段开始时间（秒）。
                - 'end_offset_sec': 段结束时间（秒）。
                - 'embedding_scope': 嵌入的范围（例如，'shot'、'scene'）。
            2. EmbeddingsTaskResult: 来自 Twelve Labs API 的完整任务结果对象。

    Raises:
        在任务创建、执行或检索期间由 Twelve Labs API 引发的任何异常。
    """

    # 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

使用该函数为您的视频生成嵌入：

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

此实现允许您使用 Twelve Labs 嵌入 API 为任何视频 URL 生成嵌入。generate_embedding 函数处理整个过程，从创建任务到检索结果。它返回一个字典列表，每个字典包含一个嵌入向量及其元数据（时间范围和范围）。请记住在生产环境中处理潜在错误，如网络问题或 API 限制。您可能还希望根据您的特定用例实现重试或更强大的错误处理。

将嵌入插入 Milvus

使用 Twelve Labs 嵌入 API 生成嵌入后，下一步是将这些嵌入及其元数据插入我们的 Milvus Collection。此过程允许我们存储和索引视频嵌入，以便稍后进行高效的相似性搜索。

以下是如何将嵌入插入 Milvus：

def insert_embeddings(milvus_client, collection_name, task_result, video_url):
    """
    将嵌入插入 Milvus Collection。

    Args:
        milvus_client: Milvus 客户端实例。
        collection_name (str): 要插入的 Milvus Collection 的名称。
        task_result (EmbeddingsTaskResult): 包含视频嵌入的任务结果。
        video_url (str): 与嵌入关联的视频的 URL。

    Returns:
        MutationResult: 插入操作的结果。

    此函数从任务结果中获取视频嵌入并将其插入到指定的 Milvus Collection 中。
    每个嵌入都与其他元数据一起存储，包括其范围、开始和结束时间以及关联的视频 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)

此函数准备用于插入的数据，包括所有相关元数据，如嵌入向量、时间范围和源视频 URL。然后，它使用 Milvus 客户端将此数据插入指定的 Collection。

执行相似性搜索

一旦我们在 Milvus 中存储了嵌入，我们就可以执行相似性搜索，根据查询向量找到最相关的视频段。以下是如何实现此功能：

def perform_similarity_search(milvus_client, collection_name, query_vector, limit=5):
    """
    在 Milvus Collection 上执行相似性搜索。

    Args:
        milvus_client: Milvus 客户端实例。
        collection_name (str): 要搜索的 Milvus Collection 的名称。
        query_vector (list): 要搜索相似嵌入的查询向量。
        limit (int, optional): 要返回的最大结果数。默认为 5。

    Returns:
        list: 搜索结果列表，其中每个结果是包含匹配实体的元数据和相似性分数的字典。

    此函数在指定的 Milvus Collection 中搜索与给定查询向量相似的嵌入。
    它返回顶部匹配结果，包括每个匹配的元数据，如嵌入范围、时间范围和关联的视频 URL。
    """
    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()

此实现执行以下操作：

定义一个 perform_similarity_search 函数，该函数接受查询向量并在 Milvus Collection 中搜索相似的嵌入。
使用 Milvus 客户端的搜索方法查找最相似的向量。
指定我们要检索的输出 Field，包括有关匹配视频段的元数据。
提供如何使用此函数与查询视频的示例，首先生成其嵌入，然后使用它进行搜索。
打印搜索结果，包括相关元数据和相似性分数。

通过实现这些函数，您已经创建了一个完整的工作流程，用于在 Milvus 中存储视频嵌入并执行相似性搜索。此设置允许基于 Twelve Labs 嵌入 API 生成的多模态嵌入高效检索相似的视频内容。

性能优化

好的，让我们将这个应用程序提升到一个新的水平！在处理大规模视频Collection时，性能是关键。为了优化，我们应该实现嵌入生成和插入 Milvus 的批处理。这样，我们可以同时处理多个视频，显著减少总体处理时间。此外，我们可以利用 Milvus 的分区功能更有效地组织我们的数据，也许按视频类别或时间段分组。这将通过允许我们仅搜索相关分区来加速查询。

另一个优化技巧是为频繁访问的嵌入或搜索结果使用缓存机制。这可以大大改善热门查询的响应时间。不要忘记根据您的特定数据集和查询模式微调 Milvus 的索引参数 - 这里的一点调整可以大大提升搜索性能。

高级功能

现在，让我们添加一些酷炫的功能来让我们的应用程序脱颖而出！我们可以实现结合文本和视频查询的混合搜索。事实上，Twelve Labs 嵌入 API 也可以为您的文本查询生成文本嵌入。想象一下允许用户输入文本描述和样本视频片段 - 我们将为两者生成嵌入并在 Milvus 中执行加权搜索。这将给我们超精确的结果。

另一个很棒的补充将是视频内的时间搜索。我们可以将长视频分解为较小的段，每个段都有自己的嵌入。这样，用户可以找到视频中的特定时刻，而不仅仅是整个片段。嘿，为什么不加入一些基本的视频分析呢？我们可以使用嵌入来聚类相似的视频段，检测趋势，甚至识别大视频Collection中的异常值。

错误处理和日志记录

说实话，事情可能会出错，当它们出错时，我们需要做好准备。实现健壮的错误处理至关重要。我们应该将 API 调用和数据库操作包装在 try-except 块中，在出现问题时为用户提供信息性错误消息。对于网络相关问题，实现带指数退避的重试可以帮助优雅地处理临时故障。

至于日志记录，它是我们调试和监控的最佳朋友。我们应该使用 Python 的日志模块来跟踪整个应用程序中的重要事件、错误和性能指标。让我们设置不同的日志级别 - 开发用 DEBUG，一般操作用 INFO，关键问题用 ERROR。不要忘记实现日志轮转来管理文件大小。有了适当的日志记录，我们将能够快速识别和解决问题，确保我们的视频搜索应用程序即使在扩展时也能平稳运行。

结论

恭喜！您现在已经使用 Twelve Labs 的嵌入 API 和 Milvus 构建了一个强大的语义视频搜索应用程序。这种集成允许您以前所未有的准确性和效率处理、存储和检索视频内容。通过利用多模态嵌入，您已经创建了一个理解视频数据细微差别的系统，为内容发现、推荐系统和高级视频分析开辟了令人兴奋的可能性。

当您继续开发和完善应用程序时，请记住 Twelve Labs 的高级嵌入生成和 Milvus 的可扩展向量存储的结合为应对更复杂的视频理解挑战提供了强大的基础。我们鼓励您尝试讨论的高级功能，并推动视频搜索和分析中可能性的边界。

介绍​

先决条件​

设置开发环境​

连接到 Milvus​

为视频嵌入创建 Milvus Collection​

使用 Twelve Labs 嵌入 API 生成嵌入​

初始化 Twelve Labs 客户端：​

将嵌入插入 Milvus​

执行相似性搜索​

性能优化​

高级功能​

错误处理和日志记录​

结论​

介绍