Vector Retrieval Service for Milvus:Analisis frame dan pengambilan multimodal untuk kendaraan otonom dengan Alibaba Cloud Milvus

Langkah 1: Ekstraksi frame dari video

Kamera dalam kendaraan merekam video kontinu, sehingga Anda pertama-tama mengekstraksi frame dengan ffmpeg, mengunggah gambar frame ke alamat yang dapat diakses model, lalu meneruskan frame_url ke kode ingest. Selama ekstraksi, catat clip_id dan ts_ms setiap frame (ID klip video dan timestamp frame) agar Anda dapat melacak hasil pencarian kembali ke video aslinya.

# Ekstraksi berbasis waktu: ambil satu frame setiap detik dan ubah skalanya ke lebar 960
ffmpeg -i clip001.mp4 -vf "fps=1,scale=960:-1" -q:v 3 frames/clip001_%04d.jpg

# Hanya keyframe (I-frame): lebih banyak informasi dan lebih sedikit redundansi
ffmpeg -i clip001.mp4 -vf "select='eq(pict_type,I)',scale=960:-1" -fps_mode vfr -q:v 3 frames/clip001_key_%04d.jpg

# Satu frame pada titik waktu tertentu (misalnya, 5,2 detik, yang sesuai dengan ts_ms=5200)
ffmpeg -ss 5.2 -i clip001.mp4 -frames:v 1 -vf scale=960:-1 -q:v 3 frames/clip001_5200ms.jpg

Pilih strategi ekstraksi berdasarkan tujuan Anda:

• Ekstraksi berbasis waktu — Sampling reguler untuk membangun dataset pelatihan dan replay.

• Hanya keyframe (I-frame) — Lebih banyak informasi dengan lebih sedikit redundansi. Direkomendasikan untuk penambangan kasus sudut skala besar, di mana pengurangan jumlah frame mengontrol biaya.

• Satu frame pada waktu tertentu — Menargetkan momen spesifik, seperti mereproduksi kecelakaan pada timestamp tertentu.

  Nilai -1 dalam scale=960:-1 menghitung tinggi secara otomatis berdasarkan rasio aspek asli. Gunakan -fps_mode vfr untuk mengambil I-frame. Sintaks lama -vsync vfr masih dapat dijalankan tetapi menghasilkan peringatan deprekasi.

Langkah 2: Siapkan kode koneksi bersama

Kode berikut berisi pengaturan koneksi dan wrapper kompatibilitas untuk jenis fungsi TEXTTRANSFORM. Ganti MILVUS_URI dan MILVUS_TOKEN dengan nilai instans Anda.

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 mengekspos TEXTTRANSFORM sebagai ekstensi terkelola yang nilai function type-nya adalah 9.
# Beberapa versi pymilvus tidak menyertakan anggota ini dalam enum FunctionType, sehingga diperlukan wrapper di bawah ini.
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"          # Model pemahaman multimodal (klasifikasi, ekstraksi, entitas)
client = MilvusClient(uri=MILVUS_URI, token=MILVUS_TOKEN)

AI_CLASSIFY, AI_EXTRACT, dan AI_ENTITY_EXTRACT semuanya merupakan fungsi jenis TEXTTRANSFORM (nilai function type 9) dan dibedakan berdasarkan parameter task. Beberapa versi pymilvus tidak menyertakan anggota ini dalam enum FunctionType, sehingga diperlukan wrapper kompatibilitas.

Jika Anda mengganti EMBED_MODEL, dimensi vektor juga mungkin berubah. Pastikan VECTOR_DIM dan skema di Langkah 3 tetap sinkron dengan model yang Anda pilih, karena pengindeksan ulang collection yang sudah ada merupakan operasi berbiaya tinggi.

Langkah 3: Buat collection dan sambungkan fungsi AI

Bidang vektor digunakan untuk pengambilan semantik, sedangkan tiga bidang lainnya menyimpan hasil klasifikasi, ekstraksi terstruktur, dan pengenalan entitas. Keempat fungsi tersebut menggunakan frame_url sebagai input dan dijalankan secara otomatis saat write.

# ===== Buat collection: sambungkan empat AI Functions dalam satu definisi =====
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)   # Alamat gambar frame
schema.add_field("clip_id", DataType.VARCHAR, max_length=128)      # ID klip video sumber
schema.add_field("ts_ms", DataType.INT64)                          # Timestamp frame (ms)
schema.add_field("embedding", DataType.FLOAT_VECTOR, dim=VECTOR_DIM)
schema.add_field("scene", DataType.VARCHAR, max_length=64)         # Output AI_CLASSIFY
schema.add_field("attributes", DataType.JSON)                      # Output AI_EXTRACT
schema.add_field("entities", DataType.JSON)                        # Output AI_ENTITY_EXTRACT

# 1) Embedding multimodal: gambar frame -> vektor berdimensi 2.560
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) Klasifikasi adegan: persimpangan/jalan tol/terowongan/zona konstruksi/jalan biasa
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": "Klasifikasikan frame berdasarkan lingkungan jalan yang ditunjukkan.", "temperature": "0"}))

# 3) Ekstraksi terstruktur: lampu lalu lintas/jumlah lajur/cuaca/event abnormal
#    Catatan: tentukan nilai default untuk setiap label untuk mencegah model mengembalikan null (lihat penjelasan di bawah)
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 bernilai red/green/yellow/none; "
                      "weather bernilai sunny/rainy/cloudy/night/unknown, gunakan unknown jika tidak dapat ditentukan; "
                      "lane_count bernilai integer, gunakan 0 jika tidak dapat ditentukan; "
                      "anomaly_event menggambarkan event abnormal seperti cut-in, pelanggaran lampu merah, atau kecelakaan, gunakan none jika tidak ada.",
            "temperature": "0"}))

# 4) Entitas bernama: nama jalan/nama tempat/nilai batas kecepatan
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": "Ekstrak hanya nama tempat, nama jalan, dan nilai batas kecepatan yang secara eksplisit muncul pada rambu lalu lintas dalam frame. Jangan menebak dari penampilan.",
            "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)

Parameter utama fungsi adalah:

• provider — Penyedia model. Tutorial ini menggunakan aliyun_milvus.

• model_name — Model yang mendukung fungsi tersebut.

• task — Memilih operasi untuk fungsi TEXTTRANSFORM: ai_classify, ai_extract, atau ai_entity_extract.

• media_type — Diatur ke image karena input berupa gambar frame.

• labels — Kumpulan label yang dipisahkan koma untuk klasifikasi atau ekstraksi.

• entity_types — Jenis entitas yang dipisahkan koma untuk dikenali. Contoh ini memetakan nama tempat dan jalan ke LOCATION dan nilai batas kecepatan ke PRODUCT.

• temperature — Diatur ke 0 untuk output deterministik.

• is_multimodal dan dim — Lihat Batasan dan pertimbangan. Fungsi multimodal harus mengatur is_multimodal, dan dim bidang vektor harus sesuai dengan dim fungsi (2560).

  Dalam SDK, nilai params dilewatkan sebagai string (misalnya, "is_multimodal": "true" dan "temperature": "0"). Dalam payload REST yang digunakan di Langkah 6, nilai params menggunakan tipe JSON native (misalnya, "is_multimodal": true).

Tentukan nilai default untuk setiap label ekstraksi. Prompt di atas memberikan nilai default untuk setiap label agar model menghindari mengembalikan null. Hal ini penting karena bidang JSON bernilai null tidak akan dicocokkan oleh kondisi !=, yang secara diam-diam menghilangkan frame dari hasil yang difilter. Nilai default mengurangi, tetapi tidak selalu menghilangkan, nilai yang hilang: seperti yang ditunjukkan pada hasil ingest di Langkah 4, frame terowongan masih tidak memiliki nilai weather. Rancang filter Anda untuk memperhitungkan nilai yang hilang, dan lihat Batasan dan pertimbangan.

Ingest yang berhasil di Langkah 4 (bidang terisi secara otomatis) mengonfirmasi bahwa collection dan keempat fungsinya telah tersambung dengan benar.

Langkah 4: Ingest frame dengan inference on write

Anda hanya menulis tiga bidang frame_url, clip_id, dan ts_ms. Fungsi AI mengisi keempat bidang lainnya secara otomatis, tanpa anotasi manual.

# ===== Ingest frame: inference on write mengisi embedding/scene/attributes/entities secara otomatis =====
# Ganti frame_url dengan alamat gambar Anda sendiri yang dapat diakses model melalui Internet.
# clip_id dan ts_ms mencatat klip sumber dan timestamp frame, sehingga Anda dapat melacak hasil kembali ke video asli.
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},
]

# Pertahankan setiap batch pada atau di bawah batas batch multimodal yaitu 10 (lihat Batasan dan pertimbangan).
_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)

# Setelah write selesai, kueri langsung hasil terstruktur untuk memverifikasinya
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)}")

Loop write menjaga ukuran setiap batch pada 8, yang berada di bawah batas batch multimodal 10. Setelah write, Anda harus memanggil flush(); jika tidak, pencarian yang dilakukan segera setelahnya mungkin mengembalikan hasil kosong. Kueri verifikasi menggunakan limit=100, sehingga mengembalikan hingga 100 baris sebagai pemeriksaan spot. Untuk kumpulan frame yang lebih besar, lakukan paging melalui hasil atau jalankan count untuk mengonfirmasi write lengkap.

Dalam pengujian, menulis enam frame ditambah flush dan load memakan waktu sekitar 12 detik. Tabel berikut menunjukkan output referensi untuk kumpulan frame contoh. Nilai pasti bergantung pada frame Anda sendiri, jadi fokuslah pada apakah setiap bidang terisi otomatis dan apakah klasifikasi sesuai dengan gambar.

Klasifikasi adegan keenam frame sesuai dengan konten gambar, dan status lampu lalu lintas, cuaca, serta jumlah lajur yang diekstraksi konsisten dengan gambar untuk frame yang memiliki bidang tersebut terisi. Beberapa bidang masih tidak ada untuk frame tertentu (misalnya, weather untuk frame terowongan dan traffic_light/lane_count untuk frame jalan biasa), sehingga Anda harus memperhitungkan nilai yang hilang saat melakukan filter. Entitas bernama hanya diekstraksi untuk frame yang berisi teks rambu yang dapat dibaca, dan model tidak menyimpulkan nama tempat dari penampilan, yang sesuai dengan batasan "jangan menebak dari penampilan" dalam prompt.

Langkah 5: Cari frame

Vektor dan bidang terstruktur yang dihasilkan selama tahap write digabungkan pada tahap pengambilan: vektor menangani recall semantik, sedangkan bidang terstruktur menangani filtering yang tepat.

Untuk mencari berdasarkan teks

Teks kueri dipetakan ke ruang vektor gambar oleh model multimodal yang sama dan direcall secara langsung.

query = "zona konstruksi saat hujan, dengan kerucut lalu lintas dan rambu konstruksi di jalan"
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']}")

Untuk menggabungkan recall vektor dengan filtering terstruktur (corner case mining)

Bidang terstruktur berasal dari AI_EXTRACT dan AI_CLASSIFY saat write dan dapat digunakan langsung dalam filter.

corner_query = "event abnormal di jalan pada malam hari"
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"Menemukan {len(corner_search[0])} adegan long-tail")

Filtering terstruktur mendukung kesamaan string dan perbandingan numerik. Misalnya, attributes["lane_count"] >= 4 memilih frame dengan empat lajur atau lebih. Saat menggabungkan kondisi, ingat bahwa kondisi != tidak akan mencocokkan baris yang nilai bidang JSON-nya null, yang dapat secara diam-diam menghilangkan sampel long-tail. Kondisi gabungan harus sesuai dengan konten aktual perpustakaan frame Anda: scene == "tunnel" and attributes["anomaly_event"] != "none" memerlukan kedua kondisi yaitu adegan terowongan dan event abnormal. Jika tidak ada frame seperti itu, kueri mengembalikan 0 baris, yang merupakan kesimpulan valid dalam penambangan kasus sudut, bukan kegagalan. Saat debugging, pastikan pipeline berfungsi dengan satu kondisi terlebih dahulu, lalu tambahkan kondisi secara bertahap untuk mempersempit cakupan.

Untuk mencari berdasarkan gambar

Ganti data dengan URL gambar dan pertahankan sisanya tidak berubah.

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

Tabel berikut menunjukkan hasil referensi untuk kumpulan frame contoh. Fokuslah pada urutan relatif, bukan skor absolut, yang bergantung pada frame Anda sendiri.

Dalam pencarian gambar-ke-frame, gambar kueri mencetak 1,0000 terhadap dirinya sendiri. Ini adalah pemeriksaan yang dapat direproduksi untuk memverifikasi konsistensi encoding gambar.

Langkah 6: Susun ulang hasil dengan fungsi rerank (opsional)

Ukuran kemiripan vektor mengukur kedekatan semantik, yang tidak sepenuhnya setara dengan relevansi terhadap maksud kueri. Anda dapat menggunakan qwen3-vl-rerank untuk fine-rank frame yang direcall dan meningkatkan kualitas peringkat untuk skenario long-tail. Pilih salah satu dari dua pola berikut sebelum membaca kode:

• Sambungkan ranker dalam search — Gunakan ini jika Anda ingin recall dan fine-ranking dalam satu panggilan.

• Panggil antarmuka rerank REST — Gunakan ini jika Anda sudah memiliki daftar kandidat frame dan hanya ingin menyusun ulangnya.

  Kedua pola menghasilkan skor identik untuk frame yang sama, sehingga Anda dapat memilih salah satunya berdasarkan kebutuhan teknis Anda.

Pola 1 menyambungkan ranker dalam search, sehingga qwen3-vl-rerank melakukan fine-rank terhadap hasil recall vektor top-N.

# ===== Penyusunan ulang multimodal dengan AI_RERANK (opsional) =====
QUERY = "perilaku cut-in di jalan layang jalan tol"

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": "Susun ulang kandidat frame berdasarkan relevansi terhadap kueri, "
                        "memprioritaskan subjek, aksi, adegan, dan detail visual detail halus.",
            "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']}")

Pola 2 hanya menyusun ulang kumpulan URL frame yang sudah direcall melalui endpoint REST /v2/vectordb/ai/rerank. Helper berikut mengirim permintaan JSON; definisikan sebelum panggilan REST.

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": "Susun ulang kandidat frame berdasarkan relevansi terhadap kueri, "
                           "memprioritaskan subjek, aksi, adegan, dan detail visual detail halus.",
               "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}")

Pernyataan assert adalah contoh pemeriksaan saja. Di lingkungan produksi, gantilah dengan penanganan error eksplisit yang memeriksa status dan code respons, serta mencatat log respons alih-alih menaikkan exception.

Penyusunan ulang multimodal memerlukan is_multimodal dalam params, dan Anda dapat menggunakan instruct untuk menentukan fokus peringkat, seperti memprioritaskan subjek, aksi, adegan, dan detail visual detail halus. Parameter REST documents menerima array string URL frame dan tidak mendukung top_n, sehingga mengembalikan satu skor untuk setiap kandidat.

Tabel berikut menunjukkan hasil referensi penyusunan ulang untuk kueri uji "perilaku cut-in di jalan layang jalan tol." Fokuslah pada urutan relatif, bukan skor absolut.

Frame jalan tol berada di peringkat pertama, yang sesuai dengan semantik kueri.

Langkah 7: Bersihkan sumber daya

Setelah menyelesaikan tutorial, hapus sumber daya yang Anda buat untuk menghindari biaya yang tidak perlu.

1. Hapus collection:

   client.drop_collection(collection_name)

2. Konfirmasi collection telah dihapus. Perintah berikut harus mengembalikan False:

   print(client.has_collection(collection_name))

3. Hapus gambar frame yang Anda unggah ke OSS untuk tutorial ini, jika tidak lagi diperlukan.

4. (Opsional) Jika Anda hanya mengaktifkan akses jaringan publik untuk tutorial ini, nonaktifkan Public Network Access di tab Security Configuration halaman detail instans.