Vector Retrieval Service for Milvus：通過阿里雲Milvus構建短劇素材生產與多模態檢索流水線

準備公用代碼

以下程式碼封裝含串連配置與 REST 調用工具，供步驟 1~3 共用。請將 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_REST_BASE_URL


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

步驟一：主要畫面格生圖

拿一張分鏡參考圖，按分鏡文案做背景替換或風格調整，產出一張主要畫面格圖片，作為後續圖生視頻的首幀。

# ========== 步驟 1：主要畫面格生圖 AI_IMAGE_EDIT (wan2.7-image) ==========
IMAGE_URL = "https://dashscope.oss-cn-beijing.aliyuncs.com/images/dog_and_girl.jpeg"
PROMPT = (
    "雨夜，一位女子撐著傘站在便利店門口，暖色的店內燈光打在她臉上，"
    "電影級構圖，保留畫面主體。"
)
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"主要畫面格圖片 URL：{keyframe_url}")

步驟二：圖生視頻

把主要畫面格作為首幀產生一段運動鏡頭。視頻產生是長耗時非同步任務，需要先建立任務拿到 task_id，再輪詢查詢任務狀態。

# ========== 步驟 2：圖生視頻 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": "鏡頭緩緩推進，雨絲飄落，暖光在傘面上輕輕閃動，保留首幀圖中的主體。",
        "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_id = {task_id}")

# 輪詢查詢：tasks/describe 一次只查一個 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"  輪詢 #{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"輪詢達到本地上限，請稍後憑 task_id={task_id} 繼續查詢"
print(f"運動鏡頭 video_url：{video_url}")

輪詢返回的 task_status 取值說明：

步驟三：素材理解，產生描述與標籤

在素材入庫前，用多模態大模型給圖片和視頻各產生一句客觀的檢索描述，作為標題、簡介或標籤來源，隨素材一起在步驟四入庫。

# ========== 步驟 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",
    "用一句中文客觀描述這張主要畫面格圖片的主體、情境和色調，便於以文搜圖。",
)
video_caption = describe_media(
    video_url, "video",
    "用一句中文客觀描述這段視頻的主體、動作和畫面氛圍，便於按鏡頭檢索。",
)
print(f"[圖片描述] {image_caption}")
print(f"[視頻描述] {video_caption}")

圖片用 media_type=image，視頻用 media_type=video，prompt 為必填項。實測產生的描述樣本：

[圖片描述] 雨夜街頭，一位穿格子襯衫的女子撐著傘站在便利店門口，身旁坐著一隻金毛犬，
          整體色調偏冷藍，燈光溫暖映襯濕漉地面，營造靜謐而略帶憂鬱的氛圍。
[視頻描述] 這段視頻呈現了雨夜街頭，一名身穿格子襯衫的女子撐著透明雨傘與一隻黃狗站在亮燈的
          店鋪外，鏡頭由全景逐漸推近至女子面部的特寫，整體畫面色調偏冷且帶有濕潤的反光。

步驟四：多模態向量入庫

建一個「寫入即向量化」的多模態素材 Collection，把圖片 URL、視頻 URL 連同描述、標籤寫入，Milvus 會自動為素材產生向量並落庫。

# ========== 步驟 4：建多模態 Collection 並向量入庫 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": "主要畫面格,產生圖"},
        {"media_type": "video", "media_ref": video_url,
         "caption": video_caption, "tags": "運動鏡頭,產生視頻"},
    ],
)
client.flush(collection_name)
client.load_collection(collection_name)
print(f"圖片與視頻已向量化入庫，Collection：{collection_name}")

產生物 URL 是臨時簽名地址，入庫前建議轉存。生圖與生視頻返回的 URL 均帶 OSS 簽名，實測有效期間約 24 小時。這會帶來兩類影響：

步驟五：內容搜尋

素材入庫後，用同一個多模態模型把查詢映射到同一向量空間，檢索最相似的素材實現複用。由於 Collection 已綁定 qwen3-vl-embedding 的 Function，search 的 data 裡直接傳原始文本或媒體 URL 即可，Milvus 會自動向量化查詢。

# ========== 步驟 5：內容搜尋 以文搜圖 / 以文搜視頻 / 以圖搜視頻 ==========
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}. [相似性 {hit['distance']:.4f}] [{e['media_type']}] {e['caption'][:42]}")
        print(f"       {url}")


# 以文搜圖：限定只搜圖片
print("[5.1] 以文搜圖：")
search_assets("雨夜撐傘的女子站在便利店門口", media_type="image")

# 以文搜視頻：限定只搜視頻
print("[5.2] 以文搜視頻：")
search_assets("雨夜街頭鏡頭緩慢推進的運動鏡頭", media_type="video")

# 以圖搜視頻：用一張主要畫面格圖片 URL 去召迴風格相近的視頻素材
print("[5.3] 以圖搜視頻（用主要畫面格圖召回同源視頻）：")
search_assets(keyframe_url, media_type="video")

# 跨模態混合檢索：去掉 filter，在圖片與視頻混合庫中檢索
print("[5.4] 跨模態混合檢索：")
search_assets("雨夜便利店暖光鏡頭")

實測三路檢索的召回結果與相似性：

其中以圖搜視頻達到 0.8715，是四種方式中最高的——該視頻正是由這張主要畫面格產生的，同源素材在同一向量空間下相似性極高。這也直觀體現了「圖片和視頻共用同一向量空間」的價值：把查詢文案換成一張圖片 URL，同一個 search 調用即可完成以圖搜圖與以圖搜視頻；去掉 filter 則在圖片與視頻混合庫中跨模態檢索。

查詢文案要與素材主題相關，並配合相似性閾值。多模態檢索按相似性排序返回 Top-K，即使查詢與素材完全無關也會返回結果。實測對比：