MP4からAPIへ：AWS Bedrock上のMarengo & Pegasusによるエンドツーエンドのビデオ理解

このチュートリアルは、AWS環境内で動画理解の力を活用したいと考えている開発者やエンタープライズデータチームを対象に設計されています。私たちの目標は、Amazon Bedrockを通じてTwelveLabsの最先端の動画AIモデル（MarengoとPegasus）を統合するための、実践的でハンズオン形式のガイドを提供することです。このチュートリアルの最後まで進めると、強力な動画分析パイプラインが構築され、検索可能な動画ライブラリの作成や、豊かで説明的なメタデータの自動生成が可能になります。

2つの主要なユースケースを示す、2部構成のJupyter Notebookを構築します。

1. マルチモーダル検索：Marengo Embed 2.7モデルを使用して、動画ファイルからマルチモーダルベクトル埋め込み（embeddings）を生成します。これらの埋め込みはAmazon OpenSearch Serverlessにインデックスされ、動画コンテンツに対するスケーラブルで高速かつ正確な類似性検索エンジンを作成します。

2. ジェネレーティブ動画分析：Pegasus 1.2モデルを活用して、同じ動画セットから要約の作成、チャプターの生成、構造化されたメタデータの抽出などの動画テキスト化（video-to-text）タスクを実行します。

前提条件

このチュートリアルを進めるには、以下が必要です。

• Amazon S3、Amazon Bedrock、およびAmazon OpenSearch Serverlessの権限が設定されたIAMロールを持つAWSアカウント。

• 選択したAWSリージョンでTwelveLabs MarengoおよびPegasusへのモデルアクセスが有効になっていること。

• 動画ファイルとモデルの出力を保存するためのS3バケット。

• boto3およびその他の必要なライブラリがインストールされたPython 3.10+環境。

• 分析用の5〜10個の短いMP4動画ファイルの小さなコレクション。

1 - アーキテクチャと基本概念

このソリューションは、スケーラブルでコスト効率の高いサーバーレス、イベント駆動型のワークフローを中心に構築されています。マルチモーダル検索コンポーネントの主なデータフローは以下の通りです。

このアーキテクチャにより、大規模な動画ライブラリを効率的に処理できます。動画がS3バケットにアップロードされると、Marengoモデルを使用したBedrockへの非同期呼び出しが開始されます。Bedrockは動画を処理し、生成されたベクトル埋め込みを指定されたS3プレフィックスに書き戻します。そこから、低レイテンシーの類似性検索のために、埋め込みをAmazon OpenSearch Serverlessインデックスにロードできます。

コードに入る前に、使用する2つの異なるタイプのモデルを理解することが重要です。

• マルチモーダル埋め込み (Marengo)：埋め込みとは、動画、画像、テキストなどの複雑なデータを表す数値ベクトルです。Marengoは、さまざまな種類のメディアを単一の統一されたベクトル空間に変換します。これにより、シンプルなテキストクエリを使用して動画内の視覚的な瞬間を検索できるようになるため強力です。モデルは、クエリ内の単語と動画内のピクセルとの関係を理解します。これは検索アプリケーションの構築に使用します。

• 動画テキスト化生成 (Pegasus)：Pegasusは、動画を「視聴」し、それに関する人間が読めるテキストを生成するために設計された生成モデルです。Marengoとは異なり、その出力はベクトルではなく、要約、ハイライトのリスト、または動画のコンテンツに関する特定の質問への回答などの説明的なテキストです。

AWS開発者にとって実用上重要な考慮事項は、リージョンの可用性です。TwelveLabsモデルは特定のAWSリージョンにデプロイされています。AWSドキュメントのリージョン可用性リストを参照してください。例えば、Marengoはus-east-1で利用可能で、Pegasusはus-east-1とus-west-2の両方で利用可能です。重要な点として、開発者はまず特定のリージョンでモデルを有効にする必要がありますが、Bedrock APIを介して呼び出す場合は、クロスリージョン推論エンドポイントを使用する必要があります。デフォルトのBedrockモデルIDは、Pegasusで直接使用することはできません。コードは、必要に応じて必要な各リージョンに対して個別のBoto3クライアントを初期化し、クロスリージョン呼び出しを行うことで、これらの違いを処理できるように設計されるべきです。

2 - 環境のセットアップ

動画インテリジェンスパイプラインの構築を開始する前に、開発環境を準備する必要があります。これには、必要なPythonライブラリのインストール、Bedrockやその他のサービスと通信するためのAWS SDKの設定、ファイルを保存するための構造化されたS3バケットのセットアップが含まれます。

2.1 - 依存関係のインストール

まず、スクリプトが依存するいくつかのPythonパッケージをインストールする必要があります。これらのライブラリは、AWSサービスとの連携からデータの管理、計算の実行まで、あらゆる処理を担当します。

ターミナルで単一のpipコマンドを使用して、これらすべてをインストールできます。

各ライブラリの簡単な機能概要は以下の通りです。

• boto3：Python用の公式AWS SDK。Bedrock、S3、OpenSearchなどのサービスとのやり取りに使用します。

• opensearch-py：OpenSearchの公式Pythonクライアント。動画埋め込みのインデックス作成と検索に使用します。

• pandasおよびnumpy：データ操作と数値演算のための強力なライブラリ。Marengoから返される埋め込みベクトルの処理に役立ちます。

• scikit-learn：ベクトル間のコサイン類似度を計算する関数など、便利なツールを提供する機械学習ライブラリ。

• botocore：boto3の基盤となる低レベルライブラリ。認証情報の処理やAWSリクエストの署名などのコア機能を提供します。

なお、pandas、numpy、scikit-learnはここでは厳密には必須ではなく、埋め込みを直接操作してコサイン類似度を計算する場合にのみ必要となります。

2.2 - AWS SDKの設定

ライブラリがインストールされたら、次のステップはAWSアカウントと通信するようにスクリプトを設定することです。boto3を使用して、必要なサービスのクライアントオブジェクトを作成します。

import boto3
import json

# Bedrockモデルが有効になっているAWSリージョンを定義します
# Marengoの場合、これは "us-east-1" の可能性があります
# Pegasusの場合、これは "us-west-2" の可能性があります
region_name = "us-east-1" 

# デフォルトのAWSプロファイルでセッションを初期化します
session = boto3.Session(profile_name="default", region_name=region_name)

# Bedrock RuntimeとS3のクライアントを作成します
bedrock_runtime_client = session.client("bedrock-runtime")
s3_client = session.client("s3")

print("✅ AWS clients created successfully.")

このコードの動作：

このスクリプトはAWSへの接続をセットアップします。

• まず、boto3ライブラリをインポートします。

• 次に、「デフォルト」のAWS認証情報プロファイルを使用してsessionを確立します。これはAWSアカウントにログインするようなものです。また、TwelveLabsモデルを有効にしたregion_nameも指定します。

• 最後に、特化したclientオブジェクトを作成します。s3_clientはS3バケットとのやり取りに使用し、bedrock_runtime_clientはBedrockモデル自体の呼び出しに使用します。

開発者にとっての重要なポイントは、bedrockとbedrock-runtimeの区別です。bedrockクライアントは、モデルアクセスのリクエストやプロビジョニングされたスループットの管理などの管理タスク用です。ここで使用するbedrock-runtimeクライアントは、モデルを呼び出して推論（埋め込みやテキストの生成など）を取得するために特別に設計されています。

2.3 - S3バケットのレイアウト

整理されたS3バケットは、クリーンでスケーラブルなアーキテクチャにとって極めて重要です。このチュートリアルでは、未加工の動画、Marengoによって生成された埋め込み、および検索に使用する可能性のあるオプションの画像用に、個別のプレフィックス（フォルダのように機能）を作成することをおすすめします。

S3バケットの構造は以下のようになります。

このレイアウトはデータの整理に役立ち、権限の管理や、ワークフローのさまざまな部分に関連するコストの追跡を容易にします。以降のセクションでは、コードが/videos/プレフィックスに動画をアップロードし、Marengoが出力を/embeddings/プレフィックスに書き込むように設定します。

3 - パートA – Marengoによるマルチモーダル検索

環境がセットアップされたので、プロジェクトの最初のパートである、強力なマルチモーダル検索エンジンの構築に取り掛かりましょう。このセクションでは、Bedrock上のTwelveLabs Marengoモデルを使用して動画をベクトル埋め込みに変換します。その後、これらの埋め込みをAmazon OpenSearch Serverlessに保存し、テキストの説明またはサンプル画像を使用して動画内の瞬間を見つけることができる検索アプリケーションを構築します。

3.1 - サンプル動画のアップロード

まず、作業対象となるいくつかの動画が必要です。簡単に進めるために、ヘルパー関数を使用して、Netflix Open Contentライブラリからいくつかのサンプル動画を、先ほど設定したS3バケットに直接コピーします。

import re

# Netflixから公開されているサンプル動画のリスト
sample_videos = [
    's3://download.opencontent.netflix.com/TechblogAssets/CosmosLaundromat/encodes/CosmosLaundromat_2048x858_24fps_SDR.mp4',
    's3://download.opencontent.netflix.com/TechblogAssets/Meridian/encodes/Meridian_3840x2160_5994fps_SDR.mp4',
    's3://download.opencontent.netflix.com/TechblogAssets/Sparks/encodes/Sparks_4096x2160_5994fps_SDR.mp4'
]

# このクライアントは、公開されているS3バケットにアクセスするためにAWS認証情報を必要としません
public_s3_client = boto3.client('s3', config=botocore.client.Config(signature_version=botocore.UNSIGNED))

def parse_s3_uri(s3_uri: str) -> tuple[str, str]:
    """S3 URIをバケットとキーに分割するヘルパー関数。"""
    match = re.match(r'^s3://([^/]+)/(.+)$', s3_uri)
    if not match:
        raise ValueError(f"Invalid S3 URI format: {s3_uri}")
    return match.group(1), match.group(2)

def copy_public_s3_object(public_s3_uri: str, dest_bucket: str, dest_key: str):
    """公開S3バケットからプライベートバケットにオブジェクトをコピーします。"""
    source_bucket, source_key = parse_s3_uri(public_s3_uri)
    print(f"Downloading from {public_s3_uri}...")
    response = public_s3_client.get_object(Bucket=source_bucket, Key=source_key)
    
    print(f"Uploading to s3://{dest_bucket}/{dest_key} ...")
    s3_client.put_object(Bucket=dest_bucket, Key=dest_key, Body=response['Body'].read())
    print("✅ Copy completed successfully!")

# サンプル動画をループ接続し、S3バケットにコピーします
S3_BUCKET_NAME = "<your-bucket-name>" # あなたのバケット名に置き換えてください
S3_VIDEOS_PATH = "videos"

for video_uri in sample_videos:
    _, src_key = parse_s3_uri(video_uri)
    filename = src_key.split("/")[-1]
    dest_key = f"{S3_VIDEOS_PATH}/{filename}"
    copy_public_s3_object(
        public_s3_uri=video_uri,
        dest_bucket=S3_BUCKET_NAME,
        dest_key=dest_key
    )

このコードの動作：

このスクリプトはサンプルデータを取得するプロセスを自動化します。Netflixが提供する高品質な動画を指すS3 URIのリストを定義します。その後、このリストをループし、特別なboto3クライアントを使用して、公開場所から各動画をダウンロードし、自身のS3バケットの/videos/プレフィックスにアップロードします。

3.2 - Marengoによる同期埋め込みと非同期埋め込みの比較

Marengoは、大容量の動画ファイルから短いテキストクエリまで、さまざまなメディア入力を処理できるように設計されています。これらの異なるユースケースに合わせて最適化するために、Amazon Bedrockはモデルを呼び出す2つの方法を提供するようになりました。大きなメディア向けには非同期、リアルタイムクエリ向けには同期です。

大規模な動画処理のための非同期StartAsyncInvoke

動画や音声などの大きなファイルを埋め込むには、非同期のStartAsyncInvoke APIを使用することが引き続きベストプラクティスです。これにより、Bedrockにジョブを送信してバックグラウンドで処理させることができるため、接続を開いたままにすることなく、動画ライブラリ全体を一括処理するのに最適です。

これは写真現像所にフィルムを預けるようなものです。チケット（呼び出しARN）を受け取り、後でステータスを確認して現像された写真（埋め込み）を受け取ります。前のセクションのembed_video_async関数は引き続きこの方法を使用しており、動画に対して正しいアプローチです。

def embed_video_async(s3_uri: str, output_s3_bucket: str) -> str:
    """
    動画の埋め込みを生成する非同期ジョブを開始します。
    
    引数:
        s3_uri (str): 処理する動画のS3 URI。
        output_s3_bucket (str): 結果を保存するS3バケット。

    戻り値:
        str: ジョブIDのようなものである呼び出しARN。
    """
    response = bedrock_runtime_client.start_async_invoke(
        modelId="twelvelabs.marengo-embed-2-7-v1:0",
        modelInput={
            "inputType": "video",
            "mediaSource": {"s3Location": {"uri": s3_uri}}
        },
        outputDataConfig={"s3OutputDataConfig": {
            "s3Uri": f"s3://{output_s3_bucket}/embeddings/"
        }}
    )
    return response["invocationArn"]

このコードの動作：

embed_video_async関数は動画のS3ロケーションを受け取り、それをMarengoモデル（twelvelabs.marengo-embed-2-7-v1:0）に送信します。inputTypeがvideoであることを指定し、そのs3Locationを指します。また、出力を保存する場所（バケットの/embeddings/プレフィックス内のJSONファイル）をBedrockに指示します。関数は、ジョブのステータスを確認するために使用するinvocationArnを返します。

リアルタイムのテキストおよび画像クエリのための同期InvokeModel

ユーザーのテキスト検索クエリやサンプル画像の埋め込みなど、低レイテンシーが求められる小規模なタスクでは、同期されたInvokeModel APIを使用できるようになりました。これは、結果をポーリングすることなくAPIレスポンスで直接埋め込みを返すため、応答性の高いアプリケーションを構築する上で大幅な向上となります。

テキストクエリを同期的に埋め込む関数を作成する方法は以下の通りです。

def embed_text_sync(query_text: str) -> list:
    """
    同期InvokeModel APIを使用して、テキストクエリの埋め込みを生成します。

    引数:
        query_text (str): 埋め込むテキストクエリ。

    戻り値:
        list: テキストの埋め込みベクトル。
    """
    # 注意: bedrock_runtime_clientが、Marengoが利用可能なリージョン用に設定されていることを確認してください
    request_body = {
        "inputType": "text",
        "inputText": query_text
    }

    response = bedrock_runtime_client.invoke_model(
        modelId="twelvelabs.marengo-embed-2-7-v1:0",
        body=json.dumps(request_body),
        contentType="application/json",
        accept="application/json"
    )
    
    response_body = json.loads(response["body"].read())
    
    # 埋め込みはレスポンスで直接返されます
    return response_body["embedding"]

このコードの動作：

この関数はinvoke_modelを使用してMarengoモデルにリクエストを送信します。

• inputTypeをtextに指定し、inputTextを提供します。

• 非同期呼び出しとは異なり、invoke_modelからのレスポンスには埋め込みが直接含まれています。JSONレスポンスをパースし、embeddingベクトルを返します。

• この同期アプローチにより、個別のポーリング関数が不要になり、コードが簡素化され、アプリケーションの検索機能のレイテンシーが短縮されます。リクエストにmediaSourceを指定することで、画像に対しても同様の関数を使用できます。

バッチ動画処理にはStartAsyncInvoke、リアルタイムクエリにはInvokeModelといった適切なツールを使用することで、より効率的で応答性の高い動画検索アーキテクチャを構築できます。

3.3 - 埋め込みJSONのパース

Bedrockが動画の処理を完了すると、指定されたS3の場所にoutput.jsonファイルが配置されます。このファイルには埋め込みベクトルのリストが含まれており、各ベクトルは動画の短いセグメントを表します。

出力の構造は以下のようになります。

{
  "data": [
    {
      "embedding": [0.123, -0.045, 0.567, ...],
      "embeddingOption": "visual-text",
      "startSec": 0.0,
      "endSec": 5.0
    },
    {
      "embedding": [-0.234, 0.089, -0.112, ...],
      "embeddingOption": "visual-text",
      "startSec": 5.0,
      "endSec": 10.0
    }
    // ... 他のセグメント
  ]
}

data配列内の各オブジェクトには、埋め込みベクトル（1,024個の数値のリスト）、秒単位の動画セグメントの開始時間と終了時間（startSec、endSec）、および生成された埋め込みのタイプが含まれています。

3.4 - OpenSearch Serverlessインデックスの作成

これらの埋め込みを効率的に検索するには、ベクトルデータベースが必要です。Amazon OpenSearch Serverlessは、強力なk近傍（k-NN）検索機能を備えた、完全に管理されたソリューションを提供します。

ベクトル検索用に設定されたインデックスを作成してみましょう。

# 注意: 最初にOpenSearch Serverlessクライアント 'os_client' を作成する必要があります。
# AWSV4SignerAuthを使用してクライアントを作成する詳細については、ワークショップファイルを参照してください。

INDEX_NAME = "video-search-index"

index_body = {
    "settings": { "index": { "knn": True } },
    "mappings": {
        "properties": {
            "embedding": {
                "type": "knn_vector",
                "dimension": 1024,
                "method": {
                    "engine": "faiss",
                    "name": "hnsw",
                    "space_type": "cosinesimil"
                }
            },
            "start_time": {"type": "float"},
            "end_time": {"type": "float"},
            "video_s3_uri": {"type": "keyword"}
        }
    }
}

os_client.indices.create(index=INDEX_NAME, body=index_body)
print(f"✅ Index '{INDEX_NAME}' created successfully.")

このコードの動作：

このスクリプトは、検索インデックスのスキーマを定義します。

• "knn": Trueにより、k-NN機能が有効になります。

• embeddingフィールドは、Marengoの出力と一致するdimension1024のknn_vectorとして定義されます。

• ベクトル検索に非常に効率的なfaissエンジンとhnswアルゴリズムを指定します。

• space_typeはcosinesimilに設定されます。これにより、ベクトル間の角度に基づいて類似度を測定するようOpenSearchに指示します。これは、マルチモーダル埋め込みにとって優れた選択肢です。

3.5 - 動画セグメントの一括インデックス作成

インデックスの準備ができたら、S3から埋め込みファイルを処理し、OpenSearchにロードできます。これは通常、効率向上のために一括で実行されます。すべてのoutput.jsonファイルをリストし、パースして、データをバッチでOpenSearchに送信するスクリプトを作成します。

3.6 - テキストと画像によるクエリ

ここからが面白いところです。Marengoは統一されたベクトル空間に埋め込みを作成するため、テキストクエリや画像の埋め込みを生成し、それを使用して類似する動画セグメントを見つけることができます。

テキストから動画を検索する関数は以下の通りです。

def search_videos_by_text(query_text: str, top_k: int = 3) -> list:
    """テキストクエリを使用して動画セグメントを検索します。"""
    # 1. Marengoを使用してテキストクエリの埋め込みを生成する
    print(f"Generating embedding for query: '{query_text}'")
    query_embedding_response = # inputType: "text" でMarengoを呼び出します
    query_embedding = query_embedding_response[0]["embedding"]

    # 2. 埋め込みを使用してOpenSearchを検索する
    search_body = {
        "query": { "knn": { "embedding": { "vector": query_embedding, "k": top_k } } },
        "_source": ["video_s3_uri", "start_time", "end_time"]
    }
    
    response = os_client.search(index=INDEX_NAME, body=search_body)
    return response['hits']['hits']

このコードの動作：

この関数はまずMarengoを呼び出して、query_textをベクトルに変換します。次に、このベクトルを使用してOpenSearchインデックスでk-NN検索を実行し、最も類似している上位k個の動画セグメントをリクエストします。画像検索に対しても、MarengoをinputType: "image"で呼び出すことで、同様の関数を作成できます。

3.7 - デモ検索

実際に試してみましょう。動画ライブラリ内の特定のシーンを見つけたいとします。

query = "ネオンライトに照らされた夜 of カーチェイス"
search_results = search_videos_by_text(query)

# 最上位の結果を出力する
top_result = search_results[0]['_source']
video_url = top_result['video_s3_uri'] # ここで署名付きURLを生成します
start_time = top_result['start_time']

print(f"Top match found in {video_url} at {start_time} seconds.")
# Jupyter Notebookでは、'start_time'から再生を開始するように設定された
# HTML5動画プレイヤーを表示できます。

これにより、説明に最もよく一致する動画セグメントが即座に見つかり、マルチモーダル検索の威力を実感できます。

3.8 - まとめ

このセクションでは、完全にサーバーレスのアーキテクチャを使用して、高度な動画検索エンジンの構築に成功しました。主なポイントは以下の通りです。

• 統一されたベクトル空間：動画、テキスト、画像といった異なるモダリティを単一のベクトル空間に埋め込むMarengoの機能が、強力なクロスモーダル検索を実現します。キーワードだけでなく、文章で動画を検索できます。

• スケーラブルなアーキテクチャ：Bedrockの非同期APIとAmazon OpenSearch Serverlessを組み合わせることで、サーバーを管理することなく、ペタバイト規模の動画データを処理できるように拡張可能なパイプラインを実現できます。

4 - パートB – Pegasusによる詳細な説明の生成

チュートリアルの最初のパートでは、Marengoのベクトル埋め込みを使用して強力な検索アプリケーションを構築しました。ここでは、生成モデルであるTwelveLabs Pegasusモデルの機能を探求します。Pegasusは、動画を「視聴」して、要約、ハイライト、さらには構造化されたメタデータなど、人間が読める詳細なテキストを生成するように設計されています。これにより、単なる検索を超えて、動画コンテンツのより深く、文脈に沿った理解を実現できます。

4.1 - 分析用の動画の選択

パートAでS3バケットにアップロードしたのと同じ動画を使用して作業を続けます。まず、Pegasusで分析する動画を1つ選択しましょう。/videos/プレフィックスにあるオブジェクトをリストして、1つ選択します。

# s3_client と S3_BUCKET_NAME はすでに設定されていると仮定します
S3_VIDEOS_PATH = "videos"

# S3バケット内の動画をリストします
response = s3_client.list_objects_v2(
    Bucket=S3_BUCKET_NAME, 
    Prefix=f"{S3_VIDEOS_PATH}/"
)

if 'Contents' in response and len(response['Contents']) > 1:
    # 分析用に最初の動画ファイルを選択します（フォルダオブジェクトはスキップします）
    video_s3_key = response['Contents'][1]['Key']
    video_s3_uri = f"s3://{S3_BUCKET_NAME}/{video_s3_key}"
    print(f"✅ Selected for analysis: {video_s3_uri}")
else:
    print("❌ No videos found in the specified S3 path.")

このコードスニペットは、videosディレクトリ内のファイルをリストし、処理対象となる最初のファイルを選択して、確認のためにそのS3 URIを出力します。

4.2 - InvokeModelを使用したクイック分析の取得

Pegasusと連携する最も簡単な方法は、同期されたinvoke_model呼び出しを使用することです。これは、比較的短く、即座の回答が期待される簡単な質問に最適です。選択した動画の基本的な説明をPegasusに求めてみましょう。

# 注意: bedrock_runtime_clientが、Pegasusが利用可能なリージョン（例: 'us-west-2'）用に設定されていることを確認してください
prompt = "この動画は何についてですか？"

request_body = {
    "inputPrompt": prompt,
    "mediaSource": {
        "s3Location": {"uri": video_s3_uri}
    },
    "temperature": 0
}

# リージョンに適した正確なモデルIDを使用してください
response = bedrock_runtime_client.invoke_model(
    modelId="us.twelvelabs.pegasus-1-2-v1:0", 
    body=json.dumps(request_body),
    contentType="application/json",
    accept="application/json"
)

# レスポンスをパースして生成されたテキストを取得します
response_body = json.loads(response["body"].read())
print(response_body["message"])

このコードの動作：

この関数はPegasusモデルにリクエストを送信します。

• inputPromptに質問が含まれています。

• mediaSourceはS3内の動画の場所を指します。

• モデルができるだけファクトに基づいた決定論的な回答を提供するように、temperature: 0が設定されています。より高い温度（例：0.7）を設定すると、よりクリエイティブですが予測が難しい結果が生成されます。

• invoke_modelの呼び出しは、分析が完了するまでブロックされ、完全なレスポンスを返します。レスポンスをパースし、messageを展開して出力します。

より長い動画や、広範囲にわたるテキストを生成する場合は、invoke_model_with_response_streamを使用する方が適しています。このメソッドは、結果の生成に合わせてレスポンスをチャンク単位で返すため、結果をより早く表示してユーザー体験を向上させることができます。

4.3 - 要約、ハッシュタグ、ハイライトの生成

Pegasusは、シンプルで自然な言葉のプロンプトに基づいて、さまざまな種類の説明コンテンツを作成することに長けています。複雑な指示は必要ありません。望むものを尋ねるだけです。

役立つメタデータを生成するために使用できるプロンプトの例をいくつか紹介します。

• 要約用："動画を要約してください"

• ハッシュタグ用："動画に関連するハッシュタグを生成してください"

• ハイライト用："この動画のハイライトされた瞬間はどこですか？"

これらの各プロンプトをinvoke_modelまたはinvoke_model_with_response_stream呼び出しのrequest_bodyで使用することで、簡潔な要約、SEOに適したタグ、または動画内の主要なイベントの時系列リストを作成できます。

4.4 - Pegasusによる構造化メタデータの取得

Bedrock上のPegasusの強力な機能は、定義したJSON形式で構造化された出力を返すことができる点です。これは、コンテンツ管理システム（CMS）や動画データベースへの自動入力など、プログラムによるワークフローにおいて非常に便利です。

わかりやすさのためにリクエスト構造がアップデートされています。responseFormatオブジェクトに、目的の出力に対するschemaが直接含まれるようになりました。

prompt = """
以下のフィールドを持つ動画のメタデータを生成します。
- title: 動画にふさわしいクリエイティブなタイトル。
- description: 1段落程度の短い要約。
- mood: 動画の全体的な雰囲気や感情。
- genre: 動画コンテンツに最もよく当てはまるジャンル。
"""

# 目的とする出力構造のJSONスキーマを定義します
json_schema = {
    "type": "object",
    "properties": {
        "title": {"type": "string", "description": "The title of the video"},
        "description": {"type": "string", "description": "A short summary of the video"},
        "mood": {"type": "string", "description": "The overall mood of the video"},
        "genre": {"type": "string", "description": "The genre that best fits the video"}
    },
    "required": ["title", "description", "mood", "genre"]
}

request_body = {
    "inputPrompt": prompt,
    "mediaSource": {"s3Location": {"uri": video_s3_uri}},
    "temperature": 0.5,
    "responseFormat": {
        "schema": json_schema # スキーマが直接渡されるようになりました
    }
}

# リージョンに適した正確なモデルIDを使用してください
response = bedrock_runtime_client.invoke_model(
    modelId="us.twelvelabs.pegasus-1-2-v1:0", 
    body=json.dumps(request_body)
)

# レレスポンスボディには文字列化されたJSONオブジェクトが含まれるようになりました
response_body = json.loads(response.get("body").read())
structured_metadata = json.loads(response_body["message"])

# 整形されたJSON出力を印刷します
print(json.dumps(structured_metadata, indent=4))

このコードの動作：

このリクエストは、特定のメタデータを生成し、提供されたjson_schemaに従ってフォーマットするようPegasusに指示します。ここでの主な変更はresponseFormatオブジェクトにあり、スキーマがschemaキーの直下に直接渡されるようになったため、リクエストが簡素化されました。モデルは分析に基づいて、title、description、mood、およびgenreフィールドに値を入力します。レスポンスはクリーンで予測可能なJSONオブジェクトであり、アプリケーションワークフローで直接使用でき、壊れやすいテキストのパースが不要になります。

4.5 - 検索結果との組み合わせ

BedrockにおけるTwelveLabsモデルの本当の強みは、MarengoとPegasusの出力を組み合わせることで発揮されます。Pegasusによって生成された構造化メタデータと要約は、Marengoのベクトル埋め込みとともにOpenSearchインデックス内に直接保存できます。

ユーザーが検索を実行すると、アプリケーションは（Marengoのベクトル検索によって見つかった）関連する動画クリップを返せるだけでなく、（Pegasusによって生成された）コンテキストに応じた豊かな要約も表示できます。これにより、より詳細に情報を提供する完全なユーザー体験が構築され、かつ単なる検索結果が包括的なナレッジへと変わります。

5 - クリーンアップ

クラウド環境でのベストプラクティスとして、継続的なコストの発生を避けるために、作成したリソースを削除することが重要です。このセクションでは、チュートリアルでセットアップしたコンポーネントの廃止手順について説明します。

5.1 - OpenSearch Serverlessインデックスの削除

まず、動画の埋め込みを保存しているOpenSearch Serverlessインデックスを削除します。これにより、インデックスとそこに含まれるすべてのデータが完全に削除されます。OpenSearch Pythonクライアントを使用して、プログラムでこれを実行できます。

# 'os_client' と 'INDEX_NAME' はすでに定義されていると仮定します
try:
    response = os_client.indices.delete(index=INDEX_NAME)
    print(f"✅ Index '{INDEX_NAME}' deleted successfully.")
except Exception as e:
    print(f"❌ Error deleting index '{INDEX_NAME}': {e}")

このスクリプトは、指定されたインデックス名について、OpenSearch Serverlessコレクションにdeleteリクエストを送信します。

5.2 - S3バケットの空化

次に、サンプル動画、生成された埋め込み、およびアップロードした可能性のあるクエリの画像を削除して、S3バケットをクリーンアップしましょう。以下のスクリプトは、バケット内のすべてのオブジェクトをリストし、1回のバッチ操作でそれらを削除します。

# 's3_client' と 'S3_BUCKET_NAME' はすでに定義されていると仮定します
try:
    # バケット内のすべてのオブジェクトをリストします
    response = s3_client.list_objects_v2(Bucket=S3_BUCKET_NAME)
    
    if 'Contents' in response:
        # 削除するオブジェクトのリストを準備します
        objects_to_delete = [{'Key': obj['Key']} for obj in response['Contents']]
        
        # 削除操作を実行します
        s3_client.delete_objects(
            Bucket=S3_BUCKET_NAME,
            Delete={'Objects': objects_to_delete}
        )
        print(f"✅ Bucket '{S3_BUCKET_NAME}' emptied successfully.")
    else:
        print(f"✅ Bucket '{S3_BUCKET_NAME}' is already empty.")
except Exception as e:
    print(f"❌ Error emptying bucket '{S3_BUCKET_NAME}': {e}")

このコードの動作：

まずバケット内のすべてのファイルのリストを取得し、そのリストをdelete_objects API呼び出しに渡し、指定されたすべてのコンテンツを効率的に削除します。

5.3 - ノートブックインスタンスの停止

最後に、Amazon SageMakerノートブックインスタンスまたはEC2インスタンスでこのチュートリアルを実行していた場合は、追加料金を防ぐためにAWSマネジメントコンソールからインスタンスを停止または削除することを忘れないでください。

6 - 次のステップと拡張案

完全に機能する動画検索および分析パイプラインの構築、おめでとうございます。ここで学んだ概念は、さまざまな強力な実世界アプリケーションの基盤となります。プロジェクトをさらに発展させるためのいくつかのアイデアを紹介します。

• お使いの動画用のBedrockエージェントの構築：Amazon Bedrock Agentsを使用して、高度な対話型AIを作成します。MarengoとPegasusの両方への呼び出しをオーケストレーションするエージェントを構築できます。これにより、ユーザーは「昨年のカンファレンスで当社のCEOが登壇したすべてのシーンを表示し、その要点を要約してほしい。」といった、動画ライブラリとの自然な会話での複雑な質問が可能になります。優れたサンプルはこちらで確認できます：https://github.com/garystafford/twelvelabs-bedrock-search-agent

• Amazon Kendra による RAG パイプラインの強化：Marengoのベクトル埋め込みを、AWSのインテリジェント検索サービスであるAmazon Kendraと統合します。Marengoをカスタムの埋め込みプロバイダーとして使用することで、Kendraによる動画ライブラリ全体のセマンティック検索が可能になり、その結果を大規模言語モデル（LLM）に入力してRetrieval-Augmented Generation（RAG）システムを構築できます。これにより、動画アーカイブがクエリ可能な質問応答ナレッジベースに変わります。

• AWS Step Functionsによるコンテンツワークフローの自動化：AWS Step Functionsを使用して、サーバーレスワークフローを実行します。たとえば、新しい動画がS3にアップロードされるたびにトリガーされるステートマシンを設計できます。このワークフローは自動的に以下を実行します。

  1. Marengoによる埋め込みの生成。

  2. OpenSearch Serverlessへの埋め込みのインデックス作成。

  3. Pegasusによる要約と構造化メタデータの生成。

  4. 動画にリンクされたPegasusの出力をAmazon DynamoDBに保存。

• ダイナミックな動画体験の作成：Pegasusのハイライトおよびチャプター生成機能を使用して、新しい形式のメディアを作成します。たとえば、ハイライトのタイムスタンプをAWS Elemental MediaTailorに渡して広告やチャプターマーカーを動的に挿入したり、データを使用して「予告編」やダイジェストリールを自動生成したりできます。

このリポジトリで完全なコードを確認できます：https://github.com/twelvelabs-io/tl-solutions-samples/tree/main/Workshops/TwelveLabs_Bedrock_Workshop

重要なリソース

すぐに始める：

• Amazon Bedrock モデルアクセスのリクエスト - AWSアカウントでTwelveLabs Marengo 2.7およびPegasus 1.2を有効化する

• TwelveLabs on Amazon Bedrock - 公式の統合発表と機能紹介

実装ガイド：

• Amazon Bedrock ユーザガイド: TwelveLabs モデル - APIパラメータと使用例

• TwelveLabs モデルが Bedrock で利用可能になったことを発表 - AWS開発者ブログのエンドツーエンドのガイド

開発者リソース：

• TwelveLabs ドキュメント - 完全なAPIリファレンスと動画理解ガイド

• AWS SDK for Python (Boto3) - 実装例とコードサンプル