高度なビデオ検索：Twelve LabsとMilvusを活用したセマンティック検索

はじめに

Twelve Labs Embed 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コレクションの作成

Milvus に接続できたので、動画の埋め込みと関連するメタデータを保存するコレクションを作成しましょう。コレクション スキーマを定義し、コレクションがまだ存在しない場合は作成します。

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

このコードでは、まずコレクションがすでに存在するかどうかをチェックし、存在する場合は削除します。これにより、まっさらな状態から開始できます。これはTwelve Labsのエンベッディングの出力次元と一致する。

Twelve Labs Embed API を使用してエンベッディングを生成する

Twelve Labs Embed 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):
    """
    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

この関数を使用して、動画の埋め込みを生成します：

# 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 Embed API を使用して任意の動画 URL の埋め込みを生成できます。generate_embedding 関数は、タスクの作成から結果の取得までのプロセス全体を処理します。この関数は、埋め込みベクトルとそのメタデータ（時間範囲とスコープ）を含む辞書のリストを返します。本番環境では、ネットワークの問題や API の制限などの潜在的なエラーを処理することを忘れないでください。また、特定のユースケースによっては、リトライやより堅牢なエラーハンドリングを実装したくなるかもしれません。

エンベッディングをMilvusに挿入する

Twelve Labs Embed APIを使用してエンベッディングを生成した後、次のステップはこれらのエンベッディングをメタデータと共にMilvusコレクションに挿入することです。このプロセスにより、後で効率的に類似検索を行うために、動画の埋め込みを保存し、インデックス化することができます。

以下は埋め込みデータを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)

この関数は、埋め込みベクトル、時間範囲、埋め込み元動画URLなど、関連するすべてのメタデータを含む埋め込み用データを準備します。その後、Milvusクライアントを使用して、このデータを指定されたコレクションに挿入します。

類似検索の実行

埋め込みデータが Milvus に格納されたら、クエリベクトルに基づいて最も関連性の高い動画セグメントを見つけるために類似検索を実行できます。この機能の実装方法を以下に示します：

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

この実装では以下のことを行います：

1. クエリベクトルを受け取り、Milvusコレクションから類似の埋め込みを検索する関数 perform_similarity_search を定義します。
2. Milvusクライアントの検索メソッドを使用して、最も類似したベクトルを検索します。
3. 一致する動画セグメントに関するメタデータを含む、取得したい出力フィールドを指定します。
4. この関数をクエリ動画で使用する方法の例として、まず埋め込みを生成し、次にそれを検索に使用する方法を示します。
5. 関連するメタデータや類似度のスコアを含む検索結果を出力する。

これらの関数を実装することで、動画の埋め込みを Milvus に保存し、類似検索を実行するための完全なワークフローが構築されます。このセットアップにより、Twelve Labs の Embed API で生成されたマルチモーダル埋め込みに基づく類似動画コンテンツの効率的な検索が可能になります。

パフォーマンスの最適化

さて、このアプリを次のレベルに持っていこう！大規模な動画コレクションを扱う場合、パフォーマンスが鍵となる。最適化するには、埋め込み生成とMilvusへの挿入のバッチ処理を実装する必要があります。そうすれば、複数の動画を同時に扱うことができ、全体の処理時間を大幅に短縮できます。さらに、Milvusのパーティショニング機能を活用して、より効率的にデータを整理することもできます。これにより、関連するパーティションだけを検索できるようになり、クエリのスピードが上がる。

もう一つの最適化のトリックは、頻繁にアクセスされる埋め込みや検索結果に対してキャッシュ機構を使うことです。これにより、よく使われるクエリに対する応答時間を劇的に改善することができます。特定のデータセットとクエリーパターンに基づいてMilvusのインデックスパラメータを微調整することを忘れないでください。

高度な機能

さて、アプリを際立たせるためにクールな機能を追加してみよう！テキストクエリとビデオクエリを組み合わせたハイブリッド検索を実装することができる。実は、Twelve Labs Embed API は、テキストクエリ用にテキスト埋め込みを生成することもできます。ユーザがテキストの説明とサンプルのビデオクリップの両方を入力できるようにすることを想像してみてください - 私たちは両方の埋め込みを生成し、Milvusで重み付け検索を実行します。そうすれば、非常に正確な検索結果が得られるでしょう。

もう一つの素晴らしい追加機能は、動画内の時間的検索です。長い動画を小さなセグメントに分割し、それぞれに埋め込みを持たせることができます。こうすることで、ユーザーはクリップ全体ではなく、動画内の特定の瞬間を見つけることができる。そして、基本的な動画分析も加えてはどうだろう？エンベッディングを使って、似たような動画セグメントをクラスタリングしたり、トレンドを検出したり、あるいは大規模な動画コレクションの異常値を特定したりすることができます。

エラー処理とログ

率直に言って、物事はうまくいかないことがあります。堅牢なエラー処理を実装することは非常に重要です。APIの呼び出しやデータベースの操作をtry-exceptブロックで囲み、何か失敗したときにユーザーに有益なエラーメッセージを提供する必要があります。ネットワーク関連の問題については、指数関数的バックオフによる再試行を実装することで、一時的な不具合を優雅に処理することができる。

ロギングに関しては、デバッグとモニタリングのための最良の友です。Pythonのロギングモジュールを使って、アプリケーション全体の重要なイベント、エラー、パフォーマンスメトリックスを追跡しましょう。開発用のDEBUG、一般的な操作用のINFO、重大な問題用のERRORなど、異なるログレベルを設定しましょう。そして、ファイルサイズを管理するために、ログのローテーションを実装することを忘れないでください。適切なロギングを行うことで、問題を素早く特定して解決することができ、動画検索アプリがスケールアップしてもスムーズに動作するようになります。

結論

おめでとうございます！Twelve Labs の Embed API と Milvus を使用して、強力なセマンティック動画検索アプリケーションを構築できました。この統合により、これまでにない精度と効率で動画コンテンツを処理、保存、検索することができます。マルチモーダルエンベッディングを活用することで、動画データのニュアンスを理解し、コンテンツディスカバリー、レコメンデーションシステム、高度な動画分析の可能性を広げるシステムを構築することができます。

アプリケーションの開発と改良を続ける際には、Twelve Labsの高度なエンベッディング生成とMilvusのスケーラブルなベクトルストレージの組み合わせが、さらに複雑なビデオ理解の課題に取り組むための強固な基盤を提供することを覚えておいてください。ぜひ、今回ご紹介した先進的な機能を試して、動画検索・解析の可能性の限界に挑戦してみてください。