Vector Retrieval Service for Milvus:Alibaba Cloud Milvus を使用した自動運転のためのフレーム分析とマルチモーダル検索

ステップ 1：ビデオからフレームを抽出

車載カメラは連続したビデオを撮影するため、まず ffmpeg を使用してフレームを抽出し、フレーム画像をモデルがアクセスできるアドレスにアップロードしてから、frame_url を取り込みコードに渡します。抽出中に、各フレームの clip_id と ts_ms (ビデオクリップ ID とフレームのタイムスタンプ) を記録して、検索ヒットを元のビデオにトレースできるようにします。

# 時間ベースの抽出：1 秒ごとに 1 フレームを取得し、幅を 960 にスケーリングします
ffmpeg -i clip001.mp4 -vf "fps=1,scale=960:-1" -q:v 3 frames/clip001_%04d.jpg

# キーフレームのみ (I フレーム)：より多くの情報と少ない冗長性
ffmpeg -i clip001.mp4 -vf "select='eq(pict_type,I)',scale=960:-1" -fps_mode vfr -q:v 3 frames/clip001_key_%04d.jpg

# 特定の時点の単一フレーム (例：5.2 秒、ts_ms=5200 に対応)
ffmpeg -ss 5.2 -i clip001.mp4 -frames:v 1 -vf scale=960:-1 -q:v 3 frames/clip001_5200ms.jpg

目的に応じて抽出戦略を選択します。

• 時間ベースの抽出 — トレーニングおよびリプレイデータセットを構築するための定期的なサンプリング。

• キーフレームのみ (I フレーム) — より多くの情報と少ない冗長性。フレーム数を減らしてコストを制御できるため、大規模なコーナーケースマイニングに推奨されます。

• 特定の時間の単一フレーム — 特定のタイムスタンプでの事故の再現など、既知の瞬間を特定します。

  scale=960:-1 の -1 は、元の縦横比から高さを自動的に計算します。I フレームを取得するには -fps_mode vfr を使用します。古い -vsync vfr 構文も実行できますが、非推奨の警告が表示されます。

ステップ 2：共有接続コードの準備

以下のコードには、接続設定と TEXTTRANSFORM 関数タイプの互換性ラッパーが含まれています。MILVUS_URI と MILVUS_TOKEN を、ご利用のインスタンスの値に置き換えてください。

from __future__ import annotations

import json

from pymilvus import DataType, Function, FunctionType, MilvusClient

MILVUS_URI = "http://c-xxxx.milvus.aliyuncs.com:19530"
MILVUS_TOKEN = "root:xxx"

# Alibaba Cloud Milvus は、関数タイプの値が 9 のマネージド拡張機能として TEXTTRANSFORM を公開します。
# 一部の pymilvus バージョンでは、このメンバーが FunctionType enum に含まれていないため、以下のラッパーで追加します。
TEXTTRANSFORM_FUNCTION_TYPE = 9

def texttransform_function_type() -> FunctionType:
    for type_name in ("TEXTTRANSFORM", "TEXT_TRANSFORM", "TextTransform"):
        function_type = getattr(FunctionType, type_name, None)
        if function_type is not None:
            return function_type
    existing = getattr(FunctionType, "_value2member_map_", {}).get(TEXTTRANSFORM_FUNCTION_TYPE)
    if existing is not None:
        return existing
    extension = int.__new__(FunctionType, TEXTTRANSFORM_FUNCTION_TYPE)
    extension._name_ = "TEXTTRANSFORM"
    extension._value_ = TEXTTRANSFORM_FUNCTION_TYPE
    FunctionType._value2member_map_[TEXTTRANSFORM_FUNCTION_TYPE] = extension
    FunctionType._member_map_["TEXTTRANSFORM"] = extension
    return extension

VECTOR_DIM = 2560
EMBED_MODEL = "qwen3-vl-embedding"
VLM_MODEL = "qwen3.7-plus"          # マルチモーダル理解モデル (分類、抽出、エンティティ)
client = MilvusClient(uri=MILVUS_URI, token=MILVUS_TOKEN)

AI_CLASSIFY、AI_EXTRACT、および AI_ENTITY_EXTRACT はすべて TEXTTRANSFORM タイプ (関数タイプの値 9) の関数であり、task パラメーターによって区別されます。一部の pymilvus バージョンでは、このメンバーが FunctionType enum に含まれていないため、互換性ラッパーが必要です。

EMBED_MODEL を変更すると、ベクトルディメンションも変更される可能性があります。既存のコレクションのインデックス再作成はコストの高い操作であるため、VECTOR_DIM とステップ 3 のスキーマを、選択したモデルと同期させてください。

ステップ 3：コレクションの作成と AI 関数のアタッチ

ベクトルフィールドはセマンティック検索に使用され、他の 3 つのフィールドは分類、構造化抽出、エンティティ認識の結果を保持します。4 つの関数はすべて frame_url を入力として受け取り、書き込み時に自動的に実行されます。

# ===== コレクションの作成：単一の定義で 4 つの AI 関数をアタッチ =====
collection_name = "driving_frames"
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("frame_url", DataType.VARCHAR, max_length=4096)   # フレーム画像のアドレス
schema.add_field("clip_id", DataType.VARCHAR, max_length=128)      # ソースビデオクリップの ID
schema.add_field("ts_ms", DataType.INT64)                          # フレームのタイムスタンプ (ms)
schema.add_field("embedding", DataType.FLOAT_VECTOR, dim=VECTOR_DIM)
schema.add_field("scene", DataType.VARCHAR, max_length=64)         # AI_CLASSIFY の出力
schema.add_field("attributes", DataType.JSON)                      # AI_EXTRACT の出力
schema.add_field("entities", DataType.JSON)                        # AI_ENTITY_EXTRACT の出力

# 1) マルチモーダル埋め込み：フレーム画像 -> 2,560 次元のベクトル
schema.add_function(Function(
    name="embed_frame", function_type=FunctionType.TEXTEMBEDDING,
    input_field_names=["frame_url"], output_field_names=["embedding"],
    params={"provider": "aliyun_milvus", "model_name": EMBED_MODEL,
            "dim": VECTOR_DIM, "is_multimodal": "true"}))

# 2) シーン分類：交差点/高速道路/トンネル/工事区域/一般道
schema.add_function(Function(
    name="classify_scene", function_type=texttransform_function_type(),
    input_field_names=["frame_url"], output_field_names=["scene"],
    params={"provider": "aliyun_milvus", "model_name": VLM_MODEL,
            "task": "ai_classify", "media_type": "image",
            "labels": "intersection,highway,tunnel,construction zone,ordinary road",
            "prompt": "表示されている道路環境に基づいてフレームを分類してください。", "temperature": "0"}))

# 3) 構造化抽出：信号機/車線数/天候/異常イベント
#    注：モデルが null を返すのを防ぐため、すべてのラベルにデフォルト値を定義します (下記の説明を参照)
schema.add_function(Function(
    name="extract_traffic", function_type=texttransform_function_type(),
    input_field_names=["frame_url"], output_field_names=["attributes"],
    params={"provider": "aliyun_milvus", "model_name": VLM_MODEL,
            "task": "ai_extract", "media_type": "image",
            "labels": "traffic_light,lane_count,weather,anomaly_event",
            "prompt": "traffic_light は red/green/yellow/none のいずれかを取ります。"
                      "weather は sunny/rainy/cloudy/night/unknown のいずれかを取り、判断できない場合は unknown を使用します。"
                      "lane_count は整数を取り、判断できない場合は 0 を使用します。"
                      "anomaly_event は、割り込み、信号無視、事故などの異常イベントを記述し、何もない場合は none を使用します。",
            "temperature": "0"}))

# 4) 名前付きエンティティ：道路名/地名/制限速度の値
schema.add_function(Function(
    name="extract_entities", function_type=texttransform_function_type(),
    input_field_names=["frame_url"], output_field_names=["entities"],
    params={"provider": "aliyun_milvus", "model_name": VLM_MODEL,
            "task": "ai_entity_extract", "media_type": "image",
            "entity_types": "LOCATION,PRODUCT",
            "prompt": "フレーム内の交通標識に明示的に表示されている地名、道路名、制限速度の値のみを抽出してください。見た目から推測しないでください。",
            "temperature": "0"}))

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)

主な関数パラメーターは次のとおりです。

• provider — モデルプロバイダー。このチュートリアルでは aliyun_milvus を使用します。

• model_name — 関数を支えるモデル。

• task — TEXTTRANSFORM 関数の操作を選択します：ai_classify、ai_extract、または ai_entity_extract。

• media_type — 入力がフレーム画像であるため、image に設定します。

• labels — 分類または抽出のための、カンマ区切りのラベルセット。

• entity_types — 認識するカンマ区切りのエンティティタイプ。この例では、地名と道路名を LOCATION に、制限速度の値を PRODUCT にマッピングします。

• temperature — 決定論的な出力を得るために 0 に設定します。

• is_multimodal と dim — 「制限事項と考慮事項」をご参照ください。マルチモーダル関数は is_multimodal を設定する必要があり、ベクトルフィールドの dim は関数の dim (2560) と一致する必要があります。

  SDK では、params の値は文字列として渡されます (例："is_multimodal": "true"、"temperature": "0")。ステップ 6 で使用される REST ペイロードでは、params の値はネイティブ JSON 型を使用します (例："is_multimodal": true)。

すべての抽出ラベルにデフォルト値を定義します。 上記のプロンプトでは、モデルが null を返すのを避けるために、各ラベルにデフォルト値を割り当てています。これは、null の JSON フィールドは != 条件に一致せず、フィルタリングされた結果からフレームが静かにドロップされるため重要です。デフォルト値は欠損値を減らしますが、必ずしも排除するわけではありません。ステップ 4 の取り込み結果が示すように、トンネルのフレームにはまだ weather の値がありません。フィルターを設計する際には欠損値を考慮し、「制限事項と考慮事項」をご参照ください。

ステップ 4 での取り込みが成功する (フィールドが自動的に入力される) ことで、コレクションとその 4 つの関数が正しくアタッチされていることが確認されます。

ステップ 4：書き込み時推論によるフレームの取り込み

frame_url、clip_id、ts_ms の 3 つのフィールドのみを書き込みます。AI 関数は、手動アノテーションなしで他の 4 つのフィールドを自動的に入力します。

# ===== フレームの取り込み：書き込み時推論により embedding/scene/attributes/entities が自動的に入力されます =====
# frame_url を、モデルがインターネット経由でアクセスできるご自身の画像アドレスに置き換えてください。
# clip_id と ts_ms は、フレームのソースクリップとタイムスタンプを記録するため、ヒットを元のビデオにトレースできます。
frames = [
    {"frame_url": "https://<your-bucket>.oss-cn-hangzhou.aliyuncs.com/frames/clip_dashcam_5000.jpg",
     "clip_id": "clip_dashcam", "ts_ms": 5000},
    {"frame_url": "https://<your-bucket>.oss-cn-hangzhou.aliyuncs.com/frames/clip_dashcam_12000.jpg",
     "clip_id": "clip_dashcam", "ts_ms": 12000},
    {"frame_url": "https://<your-bucket>.oss-cn-hangzhou.aliyuncs.com/frames/clip_highway_3000.jpg",
     "clip_id": "clip_highway", "ts_ms": 3000},
    {"frame_url": "https://<your-bucket>.oss-cn-hangzhou.aliyuncs.com/frames/clip_highway_9000.jpg",
     "clip_id": "clip_highway", "ts_ms": 9000},
    {"frame_url": "https://<your-bucket>.oss-cn-hangzhou.aliyuncs.com/frames/clip_urban_2000.jpg",
     "clip_id": "clip_urban", "ts_ms": 2000},
    {"frame_url": "https://<your-bucket>.oss-cn-hangzhou.aliyuncs.com/frames/clip_urban_8000.jpg",
     "clip_id": "clip_urban", "ts_ms": 8000},
]

# 各バッチをマルチモーダルバッチ制限の 10 以下に保ちます (「制限事項と考慮事項」を参照)。
_BATCH = 8
for _i in range(0, len(frames), _BATCH):
    client.insert(collection_name, frames[_i:_i + _BATCH])
client.flush(collection_name)
client.load_collection(collection_name)

# 書き込みが完了したら、構造化された結果を直接クエリして検証します
ingested_rows = client.query(
    collection_name, filter="",
    output_fields=["frame_url", "clip_id", "ts_ms", "scene", "attributes", "entities"],
    limit=100)
for row in ingested_rows:
    print(f"{row['clip_id']}@{row['ts_ms']}ms  scene={row['scene']}  "
          f"attributes={json.dumps(row['attributes'], ensure_ascii=False)}  "
          f"entities={json.dumps(row['entities'], ensure_ascii=False)}")

書き込みループは各バッチを 8 に保ち、マルチモーダルバッチ制限の 10 以内に収めます。書き込み後、flush() を呼び出す必要があります。そうしないと、直後の検索で空の結果が返される可能性があります。検証クエリは limit=100 を使用するため、スポットチェックとして最大 100 行を返します。より大きなフレームセットの場合は、結果をページングするか、カウントを実行して完全な書き込みを確認してください。

テストでは、6 フレームの書き込みに加えてフラッシュとロードに約 12 秒かかりました。次の表は、サンプルフレームセットのリファレンス出力を示しています。正確な値はご自身のフレームに依存するため、各フィールドが自動入力されているか、分類が画像と一致しているかに注目してください。

6 つのフレームすべてのシーン分類は画像の内容と一致し、抽出された信号機の状態、天候、車線数は、それらのフィールドが入力されたフレームの画像と一致していました。一部のフレームでは、特定のフィールドがまだ欠落していました (例：トンネルフレームの weather、一般道フレームの traffic_light/lane_count)。これが、フィルタリング時に欠損値を考慮する必要がある理由です。名前付きエンティティは、読み取り可能な標識テキストを含むフレームに対してのみ抽出され、モデルは見た目から地名を推測しませんでした。これは、プロンプトの「見た目から推測しない」という制約と一致します。

ステップ 5：フレームの検索

書き込み段階で生成されたベクトルと構造化フィールドは、検索段階で組み合わされます。ベクトルはセマンティック検索を処理し、構造化フィールドは正確なフィルタリングを処理します。

テキストで検索するには

クエリテキストは、同じマルチモーダルモデルによって画像ベクトル空間にマッピングされ、直接検索されます。

query = "construction zone in the rain, with traffic cones and construction signs on the road"
text_search = client.search(
    collection_name=collection_name, data=[query], anns_field="embedding",
    limit=10, output_fields=["frame_url", "scene", "attributes"])
for hit in text_search[0]:
    print(f"score={hit['distance']:.4f} scene={hit['entity']['scene']}")

ベクトル検索と構造化フィルタリングを組み合わせるには (コーナーケースマイニング)

構造化フィールドは、書き込み時に AI_EXTRACT と AI_CLASSIFY から取得され、filter で直接使用できます。

corner_query = "abnormal event on the road at night"
corner_search = client.search(
    collection_name=collection_name, data=[corner_query], anns_field="embedding",
    limit=10, filter='attributes["anomaly_event"] != "none"',
    output_fields=["frame_url", "scene", "attributes"])
print(f"Matched {len(corner_search[0])} long-tail scenes")

構造化フィルタリングは、文字列の等価性および数値比較をサポートします。たとえば、attributes["lane_count"] >= 4 は、4 車線以上のフレームを選択します。条件を組み合わせる場合、!= 条件は JSON フィールド値が null の行に一致せず、ロングテールサンプルが静かにドロップされる可能性があることを覚えておいてください。組み合わせた条件は、フレームライブラリの実際の内容と一致する必要があります。scene == "tunnel" and attributes["anomaly_event"] != "none" は、トンネルシーンと異常イベントの両方を必要とします。そのようなフレームが存在しない場合、クエリは 0 行を返しますが、これは故障ではなく、有効なコーナーケースマイニングの結論です。デバッグ中は、まず単一の条件でパイプラインが機能することを確認し、次に条件を段階的に追加して範囲を絞り込んでください。

画像で検索するには

data を画像 URL に置き換え、他はすべて変更しません。

image_query_url = frames[0]["frame_url"]
image_search = client.search(
    collection_name=collection_name, data=[image_query_url],
    anns_field="embedding", limit=10, output_fields=["frame_url", "scene"])
for hit in image_search[0]:
    print(f"score={hit['distance']:.4f} scene={hit['entity']['scene']}")

次の表は、サンプルフレームセットのリファレンス結果を示しています。絶対的なスコアではなく、相対的な順序に注目してください。スコアはご自身のフレームに依存します。

画像からフレームへの検索では、クエリ画像は自身に対して 1.0000 のスコアになります。これは、画像エンコーディングの一貫性を検証するために使用できる再現可能なチェックです。

ステップ 6：再ランキング関数による結果の再ランキング (オプション)

ベクトル類似度はセマンティックな近さを測定しますが、これはクエリの意図への関連性と完全に同等ではありません。qwen3-vl-rerank を使用して、検索されたフレームを詳細にランキングし、ロングテールシナリオのランキング品質を向上させることができます。コードを読む前に、2 つのパターンのいずれかを選択してください。

• search でランカーをアタッチする — 1 回の呼び出しでリコールと詳細ランキングを行いたい場合に使用します。

• REST 再ランキングインターフェイスを呼び出す — 既に候補フレームリストがあり、それを再ランキングしたいだけの場合に使用します。

  どちらのパターンも同じフレームに対して同一のスコアを生成するため、エンジニアリングのニーズに応じてどちらかを選択できます。

パターン 1 は search でランカーをアタッチするため、qwen3-vl-rerank は上位 N 件のベクトルリコール結果を詳細にランキングします。

# ===== AI_RERANK によるマルチモーダル再ランキング (オプション) =====
QUERY = "cut-in behavior at a highway ramp"

reranker = Function(
    name="rerank_frames", function_type=FunctionType.RERANK,
    input_field_names=["frame_url"],
    params={"reranker": "model", "provider": "aliyun_milvus",
            "model_name": "qwen3-vl-rerank", "queries": [QUERY],
            "is_multimodal": "true",
            "instruct": "クエリへの関連性に基づいて候補フレームをランク付けし、"
                        "主題、アクション、シーン、および詳細な視覚的ディテールを優先します。",
            "timeout_sec": 10})
rerank_search = client.search(
    collection_name=collection_name, data=[QUERY], anns_field="embedding",
    limit=20, output_fields=["frame_url", "scene"], ranker=reranker)
for hit in rerank_search[0]:
    print(f"rerank_score={hit['distance']:.4f} scene={hit['entity']['scene']}")

パターン 2 は、REST エンドポイント /v2/vectordb/ai/rerank を介して、既に検索されたフレーム URL のセットのみを再ランキングします。以下のヘルパーは JSON リクエストを POST します。REST 呼び出しの前に定義してください。

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

def post_json(path: str, body: dict[str, Any], timeout: int = 120) -> tuple[int, dict[str, Any]]:
    request = Request(
        f"{MILVUS_URI.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"))

rest_docs = [f["frame_url"] for f in frames[:3]]
status, data = post_json("/v2/vectordb/ai/rerank", {
    "model_name": "qwen3-vl-rerank", "query": QUERY,
    "documents": rest_docs,
    "params": {"is_multimodal": True,
               "instruct": "クエリへの関連性に基づいて候補フレームをランク付けし、"
                           "主題、アクション、シーン、および詳細な視覚的ディテールを優先します。",
               "timeout_sec": 10}})
assert status == 200 and data.get("code") == 0, data
for item in sorted(data["data"]["output"]["results"],
                   key=lambda x: x["relevance_score"], reverse=True):
    print(f"index={item['index']} relevance_score={item['relevance_score']:.4f}")

assert 文は例専用のチェックです。本番環境では、status とレスポンスの code を検査する明示的なエラー処理に置き換え、レスポンスをログに記録して例外を発生させないようにしてください。

マルチモーダル再ランキングには params に is_multimodal が必要であり、instruct を使用して、主題、アクション、シーン、詳細な視覚的ディテールを優先するなど、ランキングの焦点を指定できます。REST の documents パラメーターはフレーム URL 文字列の配列を受け取り、top_n をサポートしていないため、候補ごとに 1 つのスコアを返します。

次の表は、「高速道路のランプでの割り込み行為」というテストクエリのリファレンス再ランキング結果を示しています。絶対的なスコアではなく、相対的な順序に注目してください。

高速道路のフレームが 1 位にランク付けされ、クエリのセマンティクスと一致しました。

ステップ 7：リソースのクリーンアップ

チュートリアルが完了したら、不要な課金を避けるために作成したリソースを削除します。

1. コレクションを削除します。

   client.drop_collection(collection_name)

2. コレクションが削除されたことを確認します。次のコマンドは False を返すはずです。

   print(client.has_collection(collection_name))

3. チュートリアルのために OSS にアップロードしたフレーム画像が不要になった場合は、削除します。

4. (オプション) このチュートリアルのためだけにパブリックネットワークアクセスを有効にした場合は、インスタンス詳細ページの [セキュリティ設定] タブで [パブリックネットワークアクセス] を無効にします。