Vector Retrieval Service for Milvus:Frame analysis and multimodal retrieval for autonomous driving with Alibaba Cloud Milvus

Step 1: Extract frames from video

In-vehicle cameras capture continuous video, so you first extract frames with ffmpeg, upload the frame images to an address that the model can access, and then pass frame_url to the ingestion code. During extraction, record the clip_id and ts_ms of each frame (the video clip ID and the frame timestamp) so that you can trace a search hit back to the original video.

# Time-based extraction: take one frame every second and scale it to a width of 960
ffmpeg -i clip001.mp4 -vf "fps=1,scale=960:-1" -q:v 3 frames/clip001_%04d.jpg

# Keyframes only (I-frames): more information and less redundancy
ffmpeg -i clip001.mp4 -vf "select='eq(pict_type,I)',scale=960:-1" -fps_mode vfr -q:v 3 frames/clip001_key_%04d.jpg

# A single frame at a specific point in time (for example, 5.2s, which corresponds to ts_ms=5200)
ffmpeg -ss 5.2 -i clip001.mp4 -frames:v 1 -vf scale=960:-1 -q:v 3 frames/clip001_5200ms.jpg

Choose the extraction strategy based on your goal:

• Time-based extraction — Regular sampling for building training and replay datasets.

• Keyframes only (I-frames) — More information with less redundancy. Recommended for large-scale corner case mining, where reducing the frame count controls cost.

• A single frame at a specific time — Pinpoint a known moment, such as reproducing an accident at a specific timestamp.

  The -1 in scale=960:-1 calculates the height automatically from the original aspect ratio. Use -fps_mode vfr to take I-frames. The older -vsync vfr syntax still runs but reports a deprecation warning.

Step 2: Prepare the shared connection code

The following code contains the connection settings and a compatibility wrapper for the TEXTTRANSFORM function type. Replace MILVUS_URI and MILVUS_TOKEN with the values of your instance.

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 exposes TEXTTRANSFORM as a managed extension whose function type value is 9.
# Some pymilvus versions do not include this member in the FunctionType enum, so the wrapper below adds it.
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"          # Multimodal understanding model (classification, extraction, entities)
client = MilvusClient(uri=MILVUS_URI, token=MILVUS_TOKEN)

AI_CLASSIFY, AI_EXTRACT, and AI_ENTITY_EXTRACT are all functions of the TEXTTRANSFORM type (function type value 9) and are distinguished by the task parameter. Some pymilvus versions do not include this member in the FunctionType enum, which is why the compatibility wrapper is required.

If you change EMBED_MODEL, the vector dimension may change too. Keep VECTOR_DIM and the schema in Step 3 in sync with the model you choose, because reindexing an existing collection is a high-cost operation.

Step 3: Create the collection and attach the AI functions

The vector field serves semantic retrieval, and the other three fields hold the results of classification, structured extraction, and entity recognition. All four functions take frame_url as input and run automatically on write.

# ===== Create the collection: attach four AI Functions in a single definition =====
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)   # Frame image address
schema.add_field("clip_id", DataType.VARCHAR, max_length=128)      # ID of the source video clip
schema.add_field("ts_ms", DataType.INT64)                          # Frame timestamp (ms)
schema.add_field("embedding", DataType.FLOAT_VECTOR, dim=VECTOR_DIM)
schema.add_field("scene", DataType.VARCHAR, max_length=64)         # AI_CLASSIFY output
schema.add_field("attributes", DataType.JSON)                      # AI_EXTRACT output
schema.add_field("entities", DataType.JSON)                        # AI_ENTITY_EXTRACT output

# 1) Multimodal embedding: frame image -> 2,560-dimensional vector
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) Scene classification: intersection/highway/tunnel/construction zone/ordinary road
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": "Classify the frame by the road environment it shows.", "temperature": "0"}))

# 3) Structured extraction: traffic light/lane count/weather/abnormal event
#    Note: define a default value for every label to prevent the model from returning null (see the description below)
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 takes red/green/yellow/none; "
                      "weather takes sunny/rainy/cloudy/night/unknown, use unknown when it cannot be determined; "
                      "lane_count takes an integer, use 0 when it cannot be determined; "
                      "anomaly_event describes abnormal events such as cut-ins, red light running, or accidents, use none if there are none.",
            "temperature": "0"}))

# 4) Named entities: road names/place names/speed limit values
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": "Extract only the place names, road names, and speed limit values that explicitly appear on traffic signs in the frame. Do not guess from appearance.",
            "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)

The main function parameters are:

• provider — The model provider. This tutorial uses aliyun_milvus.

• model_name — The model that backs the function.

• task — Selects the operation for a TEXTTRANSFORM function: ai_classify, ai_extract, or ai_entity_extract.

• media_type — Set to image because the input is a frame image.

• labels — The comma-separated label set for classification or extraction.

• entity_types — The comma-separated entity types to recognize. This example maps place and road names to LOCATION and speed limit values to PRODUCT.

• temperature — Set to 0 for deterministic output.

• is_multimodal and dim — See Limitations and considerations. A multimodal function must set is_multimodal, and the vector field dim must match the function dim (2560).

  In the SDK, params values are passed as strings (for example, "is_multimodal": "true" and "temperature": "0"). In the REST payload used in Step 6, params values use native JSON types (for example, "is_multimodal": true).

Define a default value for every extraction label. The prompt above assigns a default for each label so that the model avoids returning null. This matters because a null JSON field is not matched by a != condition, which silently drops frames from filtered results. Defaults reduce, but do not always eliminate, missing values: as the ingestion results in Step 4 show, the tunnel frame still has no weather value. Design your filters to account for missing values, and see Limitations and considerations.

Successful ingestion in Step 4 (fields populated automatically) confirms that the collection and its four functions are attached correctly.

Step 4: Ingest frames with inference on write

You write only the three fields frame_url, clip_id, and ts_ms. The AI Functions populate the other four fields automatically, with no manual annotation.

# ===== Ingest frames: inference on write populates embedding/scene/attributes/entities automatically =====
# Replace frame_url with your own image address that the model can access over the Internet.
# clip_id and ts_ms record the source clip and timestamp of the frame, so you can trace a hit back to the original video.
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},
]

# Keep each batch at or below the multimodal batch limit of 10 (see Limitations and considerations).
_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)

# After the write completes, query the structured results directly to verify them
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)}")

The write loop keeps each batch at 8, which stays within the multimodal batch limit of 10. After the write, you must call flush(); otherwise, a search that immediately follows may return empty results. The verification query uses limit=100, so it returns up to 100 rows as a spot check. For larger frame sets, page through the results or run a count to confirm the full write.

In tests, writing six frames plus flush and load took about 12 seconds. The following table shows the reference output for the example frame set. The exact values depend on your own frames, so focus on whether each field is auto-populated and whether the classification matches the image.

The scene classification of all six frames matched the image content, and the extracted traffic light status, weather, and lane count were consistent with the images for the frames where those fields were populated. Some fields were still absent for certain frames (for example, weather for the tunnel frame and traffic_light/lane_count for the ordinary-road frame), which is why you must account for missing values when you filter. Named entities were extracted only for frames that contained readable sign text, and the model did not infer place names from appearance, which matches the "do not guess from appearance" constraint in the prompt.

Step 5: Search frames

The vectors and structured fields produced during the write stage combine at the retrieval stage: vectors handle semantic recall, and structured fields handle precise filtering.

To search by text

The query text is mapped into the image vector space by the same multimodal model and recalled directly.

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']}")

To combine vector recall with structured filtering (corner case mining)

The structured fields come from AI_EXTRACT and AI_CLASSIFY at write time and can be used in filter directly.

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

Structured filtering supports string equality and numeric comparison. For example, attributes["lane_count"] >= 4 selects frames with four or more lanes. When you combine conditions, remember that a != condition does not match a row whose JSON field value is null, which can silently drop long-tail samples. Combined conditions must match the actual content of your frame library: scene == "tunnel" and attributes["anomaly_event"] != "none" requires both a tunnel scene and an abnormal event. If no such frame exists, the query returns 0 rows, which is itself a valid corner case mining conclusion, not a malfunction. During debugging, confirm that the pipeline works with a single condition first, then add conditions to narrow the range step by step.

To search by image

Replace data with an image URL and keep everything else unchanged.

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']}")

The following table shows reference results for the example frame set. Focus on the relative ordering, not on the absolute scores, which depend on your own frames.

In image-to-frame search, the query image scores 1.0000 against itself. This is a reproducible check that you can use to verify the consistency of image encoding.

Step 6: Rerank results with the rerank function (optional)

Vector similarity measures semantic closeness, which is not fully equivalent to relevance to the query intent. You can use qwen3-vl-rerank to fine-rank the recalled frames and improve ranking quality for long-tail scenarios. Choose one of two patterns before you read the code:

• Attach a ranker in search — Use this when you want recall and fine-ranking in a single call.

• Call the REST rerank interface — Use this when you already have a candidate frame list and only want to rerank it.

  Both patterns produce identical scores for the same frame, so you can choose either one based on your engineering needs.

Pattern 1 attaches a ranker in search, so qwen3-vl-rerank fine-ranks the top-N vector recall results.

# ===== Multimodal reranking with AI_RERANK (optional) =====
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": "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']}")

Pattern 2 reranks only a set of already recalled frame URLs through the REST endpoint /v2/vectordb/ai/rerank. The following helper posts a JSON request; define it before the REST call.

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

The assert statement is an example-only check. In production, replace it with explicit error handling that inspects status and the response code, and log the response instead of raising.

Multimodal reranking requires is_multimodal in params, and you can use instruct to specify the ranking focus, such as prioritizing subject, action, scene, and fine-grained visual details. The REST documents parameter takes an array of frame URL strings and does not support top_n, so it returns one score for each candidate.

The following table shows reference reranking results for the test query "cut-in behavior at a highway ramp." Focus on the relative ordering, not on the absolute scores.

The highway frame ranked first, which matches the query semantics.

Step 7: Clean up resources

After you finish the tutorial, delete the resources you created to avoid unnecessary charges.

1. Drop the collection:

   client.drop_collection(collection_name)

2. Confirm the collection is deleted. The following command should return False:

   print(client.has_collection(collection_name))

3. Delete the frame images you uploaded to OSS for the tutorial, if you no longer need them.

4. (Optional) If you enabled public network access only for this tutorial, disable Public Network Access on the Security Configuration tab of the instance details page.