Vector Retrieval Service for Milvus:Alibaba Cloud Milvus によるショートドラマアセット制作とマルチモーダル検索パイプラインの構築

ソリューション概要

AIGC ショートドラマ、縦型短編ドラマ、ブランドショート動画は、高速なペースで制作されます。チームは週に数十本から数百本の完成動画を制作し、各完成動画には大量のキーフレーム、ストーリーボード、候補アセットの生成とスクリーニングが必要です。蓄積されたアセットライブラリに対するセマンティック検索がない場合、過去のアセットを特定するには、ファイル名と記憶に頼るしかありません。

Alibaba Cloud Milvus は、生成と検索の 2 つのフェーズと 5 つのステップにおいて、4 つの AI 機能でこのパイプラインを連携させます。次の表は、一般的な制作パイプラインの 4 つの段階をこれらのステップにマッピングしたものです。

これらのステップで使用される 4 つの AI 機能は次のとおりです。

• AI_IMAGE_EDIT (ステップ 1) — 画像生成と編集。単一画像の編集と複数の参照画像の融合をサポートします。ストーリーボード参照画像を台本に基づいてキーフレーム画像に変換します。

• AI_VIDEO_EDIT (ステップ 2) — 画像から動画への生成およびテキストから動画への生成。非同期タスクとして実行されます。キーフレームを最初のフレームとしてモーションショットを生成します。

• AI_MULTI_MODAL_GENERATE (ステップ 3) — マルチモーダルコンテンツ理解。画像と動画のタイトル、キャプション、タグを生成します。検索に使用される客観的なキャプションを生成します。

• AI_EMBEDDING (ステップ 4 および 5) — 書き込み時のベクトル化。テキスト、画像、動画を同じベクトル空間にマッピングし、クロスモーダル検索をサポートします。

  このパイプラインを支える 3 つのメカニズムは次のとおりです。

• 生成時のインジェスト — キーフレームとモーションショットは、キャプションとタグと共にインジェストされるため、生成されたすべてのアイテムには構造化された情報とベクトルが紐付けられます。

• 単一のベクトル空間 — 画像と動画は同じマルチモーダルモデルと同じベクトルフィールドを共有するため、テキストによる画像検索や画像による動画検索などのクロスモーダル検索が可能になります。

• 書き込み時のベクトル化とクエリ時の推論 — アプリケーションは埋め込みモデルを直接呼び出す必要がありません。

前提条件

• Milvus 2.6 インスタンスが作成されていること。AI 機能には 2.6 カーネルが必要です。インスタンスの作成後、モデルサービスを個別にバインドする必要はありません。

• pymilvus がインストールされていること。このチュートリアルの例は、pymilvus 3.0.0 で検証されています。

• インターネット経由でインスタンスにアクセスするには、インスタンス詳細ページの [セキュリティ設定] タブで [パブリックネットワークアクセス] を有効にし、クライアントのエグレス IP アドレスをパブリックネットワークアクセスホワイトリストに追加する必要があります。

• (本番パイプラインでは推奨) 生成された画像と動画を取り込み前に永続化するために使用する、独自の Object Storage Service (OSS) バケットが必要です。

注意事項

パイプラインを実行する前に、次の点を考慮してください：

• エンドポイントポート： RESTful API と gRPC はポート 19530 を共有します。エンドポイントにはポートを明示的に指定する必要があります。例： http://c-xxx.milvus.aliyuncs.com:19530。ポートを省略すると、リクエストはデフォルトでポート 80 に送信され、接続がタイムアウトします。

• 生成された URL の有効期間：画像生成および動画生成で返される URL は OSS の署名付き URL で、テスト環境での計測では約 24 時間有効です。インジェスト前に生成されたアセットを独自の OSS に永続化し、長期間有効な URL を media_ref に書き込んでください。

• クエリ側のメディア URL：クエリ側で渡されるメディア URL にアクセスできることを確認してください。

• 1 つのベクトルフィールド、1 つのモデル：同じベクトル空間でクロスモーダル検索を実行するには、画像と動画が同じベクトルフィールドに格納され、同じモデルを使用する必要があります。

• マルチモーダル関数のパラメータ：マルチモーダル関数には、文字列の値である "is_multimodal": "true" を含める必要があります。ベクトルフィールドの dim は、関数パラメータ内の dim と一致している必要があります。

• 検索前のフラッシュ：データを書き込んだ後に flush() を呼び出す必要があります。そうしないと、直後に実行した検索で空の結果が返される場合があります。

• 計測値：このチュートリアルに記載されている所要時間、類似度スコア、およびキャプションは、テスト環境で計測したものです。製品仕様ではなく参考値として扱ってください。

共有コードの準備

このチュートリアルでは、2 つのアクセス方法を使用します。ステップ 1 から 3 では、post_json ユーティリティを介して /v2/vectordb/ai/* の RESTful API を呼び出し、生成および理解関数を実行します。ステップ 4 と 5 では、pymilvus の MilvusClient を使用して、コレクションの作成、データインジェスト、検索を行います。

このチュートリアルの Python コードブロックは、単一の Python セッションで順番に実行する 1 つのスクリプトを構成します。後のステップでは、前のステップで生成された変数 (keyframe_url、video_url、image_caption、video_caption、client、collection_name) を使用します。単一のステップを個別に実行するには、これらの上流の変数の値を自分で指定する必要があります。

次のコードには、接続設定と REST 呼び出しユーティリティが含まれています。MILVUS_ENDPOINT と MILVUS_TOKEN を、ご自身のインスタンスの情報に置き換えてください。

from __future__ import annotations

import json
import time
from typing import Any
from urllib.error import HTTPError
from urllib.request import Request, urlopen

from pymilvus import DataType, Function, FunctionType, MilvusClient

# ==== グローバル設定 (REST と gRPC はポート 19530 を共有) ====
MILVUS_ENDPOINT = "c-xxx.milvus.aliyuncs.com:19530"
MILVUS_TOKEN = "root:xxx"
MILVUS_REST_BASE_URL = f"http://{MILVUS_ENDPOINT}"
MILVUS_URI = MILVUS_ENDPOINT

def post_json(path: str, body: dict[str, Any], timeout: int = 200) -> tuple[int, dict[str, Any]]:
    """ステップ 1 から 3 で共通の REST 呼び出しユーティリティ。"""
    request = Request(
        f"{MILVUS_REST_BASE_URL.rstrip('/')}{path}",
        data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
        headers={"Authorization": f"Bearer {MILVUS_TOKEN}", "Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urlopen(request, timeout=timeout) as response:
            return response.status, json.loads(response.read().decode("utf-8"))
    except HTTPError as exc:
        return exc.code, json.loads(exc.read().decode("utf-8"))

このブロックのスコープは要素によって異なります。接続設定の MILVUS_ENDPOINT、MILVUS_TOKEN、および MILVUS_URI は、ステップ 4 の MilvusClient を含む 5 つすべてのステップで使用されます。post_json 関数は、ステップ 1 から 3 でのみ使用されます。timeout=200 は、各 REST 呼び出しのクライアント側 HTTP タイムアウトを秒単位で設定します。

ステップ 1：キーフレーム画像の生成

ストーリーボードの参照画像を用意し、ストーリーボードスクリプトに基づいて背景を置き換えるかスタイルを調整して、画像から動画を生成する際に最初のフレームとして使用するキーフレーム画像を作成します。ストーリーボードスクリリプトは手動で入力します。次のコードでは、params.prompt の値がそれに該当します。

# ========== ステップ 1：キーフレーム画像の生成 AI_IMAGE_EDIT (wan2.7-image) ==========
IMAGE_URL = "https://dashscope.oss-cn-beijing.aliyuncs.com/images/dog_and_girl.jpeg"
PROMPT = (
    "On a rainy night, a woman stands under an umbrella at the entrance of a convenience store, "
    "with warm light from inside the store falling on her face, cinematic composition, keep the main subject."
)
status, data = post_json(
    "/v2/vectordb/ai/image_edit",
    {
        "model_name": "wan2.7-image",
        "texts": [IMAGE_URL],
        "params": {"prompt": PROMPT, "n": 1, "size": "1024*1024",
                   "watermark": False, "timeout_sec": 180},
    },
)
assert status == 200 and data.get("code") == 0, data
keyframe_url = data["data"]["output"]["outputs"][0]
print(f"Keyframe image URL: {keyframe_url}")

単一画像を入力する場合は、参照画像 URL を texts に渡し、編集指示を params.prompt に記述します。n は生成する画像の数を指定します。この例では 1 です。

この例のプロンプトは末尾が keep the main subject となっているため、参照画像の主要な被写体たちがキーフレーム内に保持されます。参照画像には女性と犬が含まれているため、ステップ 3 で生成されるキャプションには、プロンプト自体に記述されていない犬が含まれます。

ステップ2：キーフレームからのモーションショット生成

キーフレームを最初のフレームとして使用し、モーションショットを生成します。動画生成は実行時間の長い非同期タスクです。タスクを作成して task_id を取得し、その後、タスクステータスをポーリングします。

AI_VIDEO_EDIT は Image-to-Video と Text-to-Video の両方の生成をサポートします。このチュートリアルでは Image-to-Video のみを扱うため、キーフレームは media に渡し、type は first_frame に設定します。

# ========== ステップ2：Image-to-Video 生成 AI_VIDEO_EDIT (happyhorse-1.1-i2v)、非同期タスク ==========
VIDEO_MODEL = "happyhorse-1.1-i2v"  # 作成リクエストとクエリリクエストでは、同じモデル名を使用する必要があります

# ステップ 1 のキーフレームを最初のフレームとして使用します (media.type=first_frame)
status, data = post_json(
    "/v2/vectordb/ai/video_edit",
    {
        "model_name": VIDEO_MODEL,
        "prompt": "The shot slowly pushes in, rain falls, warm light flickers gently on the umbrella, keep the main subject from the first frame.",
        "media": [{"type": "first_frame", "url": keyframe_url}],
        "params": {"resolution": "720P", "audio_setting": "none",
                   "watermark": False, "timeout_sec": 180},
    },
)
assert status == 200 and data.get("code") == 0, data
task_id = data["data"]["output"]["task_id"]
print(f"Task created  task_id = {task_id}")

# 結果をポーリングします： tasks/describe は 1 回の呼び出しで 1 つの task_id をクエリします。 provider と、作成リクエストと同じ model_name を指定する必要があります
video_url = None
for attempt in range(60):  # ローカルのポーリング上限。上限に達した場合は task_id のみを記録し、サーバーサイドの失敗として扱わないでください
    status, data = post_json(
        "/v2/vectordb/ai/tasks/describe",
        {"provider": "aliyun_milvus", "model_name": VIDEO_MODEL, "task_id": task_id},
    )
    task_status = data.get("data", {}).get("output", {}).get("task_status")
    print(f"  Poll #{attempt:<2} ->  {task_status}")
    if task_status == "SUCCEEDED":
        video_url = data["data"]["output"]["video_url"]
        break
    if task_status in ("FAILED", "CANCELED"):
        raise RuntimeError(f"task ended in {task_status}: {json.dumps(data, ensure_ascii=False)}")
    time.sleep(10)
assert video_url, f"Local polling limit reached. Query again later with task_id={task_id}"
print(f"Motion shot video_url: {video_url}")

tasks/describe へのリクエストには、作成リクエストと同じ model_name と provider を含める必要があり、1 回の呼び出しで 1 つの task_id をクエリします。レスポンスボディには task_id、task_status、video_url の 3 つのフィールドが含まれます。次の表では、ポーリングで返される task_status の値について説明します。

テスト環境では、720P、5秒の Image-to-Video タスクは約 2～3分 かかり、その間 task_status は UNKNOWN を返し続けます。この例のポーリングループは10秒間隔で最大60回試行します。これは約10分のローカル上限であり、サーバーサイドのタイムアウトではありません。

このローカル上限に達した場合でも、タスクはサーバーサイドで実行中です。中断を失敗として扱わないでください。作成呼び出しで出力された task_id を保持し、同じ provider、同じ model_name、およびその task_id を指定して、後で /v2/vectordb/ai/tasks/describe を再度呼び出し、video_url を取得してください。

params.resolution は、固定の寸法ではなく、ピクセル数の段階を指定します。実際の出力寸法は、最初のフレームのアスペクト比によって決まります。この例では、最初のフレームが 1024×1024 の正方形であるため、720P を指定すると、実際の出力は 960×960 (総ピクセル数は 1280×720 に相当) となり、長さは約 5 秒になります。16:9 の横長の出力には、対応するアスペクト比の最初のフレームを使用してください。

ステップ 3： アセットのキャプションとタグの生成

アセットをインジェストする前に、マルチモーダル大規模モデルを使用して、各画像と動画に対して客観的な検索用のキャプションを 1 つ生成します。そのキャプションをタイトル、要約、またはタグのソースとして使用し、ステップ 4 でアセットと一緒にインジェストします。

# ========== ステップ 3： アセットの理解 AI_MULTI_MODAL_GENERATE (qwen3.7-plus) ==========
def describe_media(url: str, media_type: str, prompt: str) -> str:
    status, data = post_json(
        "/v2/vectordb/ai/multi_modal_generate",
        {
            "model_name": "qwen3.7-plus",
            "texts": [url],
            "params": {"media_type": media_type, "prompt": prompt, "temperature": 0},
        },
    )
    assert status == 200 and data.get("code") == 0, data
    return data["data"]["output"]["outputs"][0]

image_caption = describe_media(
    keyframe_url, "image",
    "Describe the subject, scene, and color tone of this keyframe image in one objective sentence, for text-to-image search.",
)
video_caption = describe_media(
    video_url, "video",
    "Describe the subject, action, and visual mood of this video in one objective sentence, for shot-level retrieval.",
)
print(f"[Image caption] {image_caption}")
print(f"[Video caption] {video_caption}")

画像には media_type=image を、動画には media_type=video を使用します。prompt は必須です。以下のキャプションは、テスト環境で生成されたものです。

[Image caption] On a rainy street at night, a woman in a plaid shirt stands under an umbrella at the
                entrance of a convenience store, with a golden retriever sitting beside her. The overall
                tone is cool blue, and the warm light reflects off the wet ground, creating a quiet,
                slightly melancholy mood.
[Video caption] The video shows a rainy street at night, where a woman in a plaid shirt holding a
                transparent umbrella stands with a yellow dog outside a lit storefront. The shot moves
                from a wide view to a close-up of the woman's face, and the overall tone is cool with
                wet reflections.

キャプションの品質は、後のテキストによる画像検索の再現率を直接決定します。プロンプトでは、被写体、シーン、色調、アクションといった要素の客観的な説明や、検索時に使用するその他の次元を明示的に要求し、主観的な判断を含むキャプションは避けてください。

ステップ 4：マルチモーダルコレクションへのアセットのインジェスト

書き込み時のベクトル化を使用してマルチモーダルアセットコレクションを作成し、画像 URL と動画 URL をキャプションやタグと一緒に書き込みます。Milvus はアセットのベクトルを自動的に生成し、それらをインジェストします。

この例では、コレクションを作成する前に、同じ名前の既存のコレクションを削除します。したがって、スクリプトを再実行すると、コレクションが最初から再作成され、前回の実行でインジェストされたアセットは削除されます。

# ========== ステップ 4：マルチモーダルコレクションの作成とベクトルのインジェスト AI_EMBEDDING (qwen3-vl-embedding) ==========
EMBED_MODEL = "qwen3-vl-embedding"
VECTOR_DIM = 2560  # qwen3-vl-embedding のマルチモーダル次元。ベクトルフィールドの次元は dim と一致している必要があります

client = MilvusClient(uri=MILVUS_URI, token=MILVUS_TOKEN)

collection_name = "aigc_assets_multimodal"
if client.has_collection(collection_name):
    client.drop_collection(collection_name)

schema = MilvusClient.create_schema(auto_id=True, enable_dynamic_field=False)
schema.add_field("id", DataType.INT64, is_primary=True)
schema.add_field("media_type", DataType.VARCHAR, max_length=16)     # image / video
schema.add_field("media_ref", DataType.VARCHAR, max_length=4096)    # ベクトル化に使用される画像または動画の URL
schema.add_field("caption", DataType.VARCHAR, max_length=2048)      # ステップ 3 で生成されたキャプション
schema.add_field("tags", DataType.VARCHAR, max_length=512)
schema.add_field("embedding", DataType.FLOAT_VECTOR, dim=VECTOR_DIM)

# media_ref が書き込まれると、qwen3-vl-embedding が自動的に呼び出され、2560 次元のベクトルが生成されます
schema.add_function(
    Function(
        name="embed_media",
        function_type=FunctionType.TEXTEMBEDDING,
        input_field_names=["media_ref"],
        output_field_names=["embedding"],
        params={
            "provider": "aliyun_milvus",
            "model_name": EMBED_MODEL,
            "dim": VECTOR_DIM,
            "is_multimodal": "true",
        },
    )
)

index_params = client.prepare_index_params()
index_params.add_index(field_name="embedding", index_type="AUTOINDEX", metric_type="COSINE")
client.create_collection(collection_name=collection_name, schema=schema,
                        index_params=index_params)

# ステップ 1 〜 3 の出力 (url + caption + tags) をベクトル化してインジェストします
client.insert(
    collection_name,
    [
        {"media_type": "image", "media_ref": keyframe_url,
         "caption": image_caption, "tags": "keyframe,generated image"},
        {"media_type": "video", "media_ref": video_url,
         "caption": video_caption, "tags": "motion shot,generated video"},
    ],
)
client.flush(collection_name)
client.load_collection(collection_name)
print(f"Images and videos are vectorized and ingested. Collection: {collection_name}")

関数パラメータでは、is_multimodal が "true" に、dim が 2560 に設定されており、これは embedding フィールドの次元と一致します。画像と動画は、同じモデルを介して同じ embedding フィールドに書き込まれ、スクリプトはコレクションをロードする前に flush() を呼び出します。これら 3 つの設定の背後にある制約については、「注意事項」をご参照ください。

生成された URL の有効期限切れによる影響：「注意事項」で説明したように、ステップ 1 と 2 で生成される URL は一時的な署名付き URL です。このような URL を media_ref に書き込むと、有効期限が切れた場合に 2 つの影響があります。

したがって、本番環境では、インジェストする前に生成されたアセットを独自の OSS に永続化し、長期的な URL を media_ref に書き込むことを推奨します。

期待される結果：スクリプトは Images and videos are vectorized and ingested. Collection: aigc_assets_multimodal を出力します。この時点で、コレクションはロードされ、1 つの画像と 1 つの動画の 2 つのエンティティを保持しています。この出力が表示されるまで、ステップ 5 に進まないでください。後でステップ 5 が空の結果を返した場合は、検索の前に flush() が実行されたことを確認してください。

ステップ 5：アセットライブラリの検索

アセットがインジェストされた後、同じマルチモーダルモデルを使用してクエリを同じベクトル空間にマッピングし、再利用のために最も類似度の高いアセットを取得します。コレクションはすでに qwen3-vl-embedding 関数にバインドされているため、search の data パラメーターに生のテキストまたはメディア URL を直接渡すことができ、Milvus はクエリを自動的にベクトル化します。

# ========== ステップ 5：コンテンツ検索、Text-to-Image / Text-to-Video / Image-to-Video ==========
def search_assets(query: str, top_k: int = 5, media_type: str | None = None,
                  min_score: float = 0.0):
    """ query は、中国語または英語のテキスト説明、あるいはクエリ画像の URL です。
    min_score： 類似度のしきい値。この値を下回る結果は無関係と見なされ、破棄されます。"""
    filter_expr = f'media_type == "{media_type}"' if media_type else ""
    results = client.search(
        collection_name=collection_name,
        data=[query],                 # テキストまたは画像の URL、両方とも qwen3-vl-embedding によってベクトル化されます
        anns_field="embedding",
        limit=top_k,
        filter=filter_expr,
        output_fields=["media_type", "media_ref", "caption", "tags"],
    )
    for rank, hit in enumerate(results[0], 1):
        if hit["distance"] < min_score:
            continue
        e = hit["entity"]
        url = e["media_ref"].split("?")[0]   # 表示をすっきりさせるため、OSS 署名パラメーターを削除します
        print(f"  {rank}. [Similarity {hit['distance']:.4f}] [{e['media_type']}] {e['caption'][:42]}")
        print(f"       {url}")

# Text-to-Image 検索：検索を画像に制限します
print("[5.1] Text-to-image search:")
search_assets("A woman under an umbrella at the entrance of a convenience store on a rainy night", media_type="image")

# Text-to-Video 検索：検索を動画に制限します
print("[5.2] Text-to-video search:")
search_assets("A motion shot that slowly pushes in on a rainy street at night", media_type="video")

# Image-to-Video 検索：キーフレーム画像の URL を使用して、類似したスタイルの動画アセットを取得します
print("[5.3] Image-to-video search (recall videos from the same source with a keyframe image):")
search_assets(keyframe_url, media_type="video")

# クロスモーダルハイブリッド検索：フィルターを削除して、画像と動画の混合コレクションを検索します
print("[5.4] Cross-modal hybrid search:")
search_assets("A warm-light shot of a convenience store on a rainy night")

次の表は、テスト環境における 4 つの検索方法の呼び出し結果と類似度スコアを示しています。

画像から動画への検索は 0.8715 に達し、4 つの方法の中で最も高い値となりました。その理由は、動画がこのキーフレームから生成されており、同じソースのアセットは同一のベクトル空間内で非常に高い類似度を持つためです。画像と動画は 1 つのベクトル空間を共有するため、1 回の search 呼び出しで 4 つすべての方法に対応できます。メディアクエリからアセットを呼び出すにはクエリテキストを画像 URL に置き換え、画像と動画の混合コレクションを検索するには filter を削除します。

クエリテキストはアセットのトピックに関連させ、類似度のしきい値を適用してください。マルチモーダル検索は類似度でソートされた Top-K の結果を返すため、クエリがアセットと全く無関係であっても結果を返します。以下の比較は、テスト環境で測定されたものです。

その差はほぼ 7 倍です。したがって、類似度のしきい値 (この例では search_assets の min_score パラメーター) を使用してアプリケーション側で結果をフィルタリングし、0.07 のようなスコアが有効な呼び出しとしてユーザーに表示されないように、明らかに無関係なロングテールの結果を破棄してください。正確なしきい値は、ビジネスアセットの分布に基づいて調整してください。

この例の 2 つの詳細は、表示される内容と再利用すべき内容に影響します：

• min_score のデフォルトは 0.0 です — 4 つのサンプル呼び出しのいずれも min_score を渡していないため、結果は破棄されず、0.07 のような低い類似度のヒットが出力に表示されます。検索をアプリケーションに統合する際は、独自のしきい値を渡してください。フィルタリングは Top-K の選択後に行われ、この例ではフィルタリングされた行を再番号付けせずにスキップするため、表示されるランク番号が連続しない場合があります。

• 出力される URL は切り捨てられています — この例では、出力を見やすくするために media_ref から OSS 署名パラメーターを削除しています。アプリケーションでは、切り捨てられた形式ではなく、完全な media_ref の値を使用してください。

トラブルシューティング

クリーンアップ

チュートリアルを完了したら、ステップ 4 で作成したコレクションを削除してください。

client.drop_collection(collection_name)

ソリューションのメリット

次の表では、このパイプラインと、各生成モデルをゼロから統合して検索システムを自社で構築する方法を比較します。

生成された各アイテムは、キャプション、タグ、ベクターとともに保存されるため、生成結果は一度きりの出力ではなく、検索可能なアセットになります。新しいスクリプトで類似したショットが必要になった場合は、まず過去のアセットライブラリに対してセマンティック検索を実行し、検索で見つかったアセットを再利用したうえで、不足分のみを生成します。