Vector Retrieval Service for Milvus：通過阿里雲Milvus構建智能駕駛截幀分析與多模態檢索鏈路

步驟一：視頻抽幀

車端相機採集的是連續視頻，需先用 ffmpeg 抽幀，得到幀圖片並上傳到模型可訪問的地址（如 OSS），再把 frame_url 交給後續代碼入庫。抽幀時同步記錄每幀的 clip_id 與 ts_ms（視頻片段 ID 與幀時間戳記），便於檢索命中後定位回原視頻。

# 定時截幀：每隔 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.2s，對應 ts_ms=5200）
ffmpeg -ss 5.2 -i clip001.mp4 -frames:v 1 -vf scale=960:-1 -q:v 3 frames/clip001_5200ms.jpg

步驟二：準備公用代碼

以下程式碼封裝含串連配置、REST 調用工具與 TEXTTRANSFORM 函數類型的相容封裝。請將 MILVUS_URI 與 MILVUS_TOKEN 替換為實際執行個體資訊。

from __future__ import annotations

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

from pymilvus import DataType, Function, FunctionType, MilvusClient

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


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


# 阿里雲 Milvus 將 TEXTTRANSFORM 作為託管擴充暴露，函數類型值為 9。
# 部分 pymilvus 版本的 FunctionType 枚舉中沒有該成員，下面做相容封裝。
TEXTTRANSFORM_FUNCTION_TYPE = 9


def texttransform_function_type() -> Any:
    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)

步驟三：建 Collection，一次掛載 4 個 AI Function

向量欄位用於語義檢索，另外三個欄位分別承接分類、結構化抽取與實體識別的結果。四個 Function 都以 frame_url 為輸入，在寫入時自動執行。

# ===== 建 Collection：一次建表掛 4 個 AI Function =====
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) 多模態向量化：幀圖片 -> 2560 維向量
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": "路口,高速,隧道,施工區,普通道路",
            "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)

務必為每個抽取標籤規定預設取值。實測中若未規定，模型在無法判斷時會返回 null（例如隧道內的幀因看不到天空，weather 被判為 null）。而 JSON 欄位為 null 時，!= 條件不會匹配該行——例如 6 幀資料中執行 attributes["weather"] != "rainy" 只返回 4 條，null 那行被靜默排除。在 Corner Case 挖掘情境下這類靜默漏幀的後果很嚴重，因此上面的 prompt 為每個標籤都給出了預設值。

步驟四：截幀入庫（寫入即推理）

只需寫入 frame_url、clip_id、ts_ms 三個欄位，其餘四個欄位由 AI Function 自動填滿，無需人工標註。

# ===== 截幀入庫：寫入即推理，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},
]

# qwen3-vl-embedding 多模態批量上限為 10，超過會報 image batch size can should be [1, 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)

# 寫入完成後直接 query 出結構化結果驗證
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)}")

實測 6 幀寫入 + flush + load 共約 12 秒，結構化結果樣本：

六幀的情境分類與畫面內容全部一致，紅綠燈狀態、天氣、車道數的抽取也與畫面相符。具名實體只在出現可讀標誌牌文字的幀上抽到結果，模型未憑畫面外觀臆測地名，符合 prompt 中「不要根據外觀猜測」的約束。

步驟五：以文搜幀 / 結構化過濾 / 以圖搜幀

寫入階段產出的向量與結構化欄位，在檢索階段可以組合使用：向量負責語義召回，結構化欄位負責精確過濾。

# ===== 以文搜幀 / 結構化過濾 / 以圖搜幀 =====
# 1) 以文搜幀：查詢文本經同一多模態模型映射到映像向量空間，直接召回
query = "雨天施工區，路面有錐形桶和施工牌"
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']}")

# 2) 向量召回 + 結構化過濾（Corner Case 挖掘）
#    結構化欄位來自寫入時的 AI_EXTRACT / AI_CLASSIFY，可直接參与 filter
corner_query = "夜間道路上的例外狀況事件"
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"命中 {len(corner_search[0])} 條長尾情境")

# 3) 以圖搜幀：把 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，可用於驗證映像編碼的一致性。

Corner Case 挖掘的組合條件要與素材庫實際情況匹配。例如 scene == "隧道" and attributes["anomaly_event"] != "none" 要求同時滿足「隧道情境」且「存在例外狀況事件」，如果素材庫中尚無這類幀，查詢會正常返回 0 條。返回空說明當前庫內沒有該長尾情境，這本身就是 Corner Case 挖掘的有效結論，而不是功能異常；調試階段建議先用單一條件確認鏈路通暢，再逐步疊加條件收窄範圍。

步驟六：AI_RERANK 多模態重排（可選）

向量相似性衡量的是語義接近程度，與「和查詢意圖有多相關」並不完全等價。可用 qwen3-vl-rerank 對召回的幀做二次精排，提升長尾情境的排序品質。提供兩種用法：在 search 中掛 ranker，或對已有的一批幀 URL 走 REST 介面獨立重排。

# ===== AI_RERANK 多模態重排（可選）=====
QUERY = "高速匝道處的加塞行為"

# 方式一：search 掛 ranker，向量召回 Top-N 後由 qwen3-vl-rerank 精排
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": "Rank candidate frames by relevance to the query, "
                        "prioritizing subject, action, scene and fine-grained visual details.",
            "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']}")

# 方式二：只對已召回的一批幀 URL 做獨立重排，走 REST /v2/vectordb/ai/rerank
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": "Rank candidate frames by relevance to the query, "
                           "prioritizing subject, action, scene and fine-grained visual details.",
               "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}")

實測查詢「高速匝道處的加塞行為」的重排結果：

高速幀排在首位，符合查詢語義。兩種調用方式對同一幀給出的分數完全一致，可按工程需要選擇：需要「召回 + 精排」一步完成時用 ranker，已經拿到候選幀列表、只想重排時用 REST 介面。