向量检索服务 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 接口。