向量检索服务 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，即使查询与素材完全无关也会返回结果。实测对比：