使用結構陣列的資料模型設計Compatible with Milvus 2.6.4+
現代的人工智能應用程式，尤其是在物聯網 (IoT) 和自動駕駛領域，通常會推理豐富且結構化的事件：包含時間戳記和向量嵌入的感測器讀數、包含錯誤代碼和音訊片段的診斷記錄，或是包含位置、速度和場景情境的行程片段。這些都需要資料庫原生支援嵌套資料的擷取與搜尋。
Milvus 並沒有要求使用者將原子結構事件轉換成平面資料模型，而是引進了結構陣列 (Array of Structs)，陣列中的每個結構都可以持有標量和向量，並保留語意完整性。
為什麼要使用結構陣列
現代人工智能應用程式，從自動駕駛到多模式檢索，越來越依賴嵌套的異質資料。傳統的平面資料模型難以表示複雜的關係，例如「一個文件包含許多註釋區塊」或「一個駕駛場景包含多個觀察到的動作」。這正是 Milvus 的 Array of Structs 資料類型的優點所在。
Structs 陣列允許您儲存一組有序的結構化元素，其中每個 Struct 包含其標量欄位和向量嵌入的組合。這使得它非常適用於
分層資料：具有多個子記錄的父實體，例如具有許多文字區塊的書籍，或具有許多註解畫格的視訊。
多模式嵌入：每個 Struct 都可以容納多個向量，例如文字嵌入加上影像嵌入，以及元資料。
時間或序列資料：Array 欄位中的 Struct 很自然地代表時間序列或逐步發生的事件。
與傳統儲存 JSON blob 或在多個集合中分割資料的解決方案不同，Structs 陣列在 Milvus 中提供原生模式強制執行、向量索引和有效率的儲存。
模式設計準則
除了在「搜尋的資料模型設計」中討論的所有指引外，在開始在資料模型設計中使用 Structs 陣列前，您還應該考慮下列事項。
定義 Struct 結構描述
在將 Array 欄位加入集合之前，請定義內部的 Struct 結構圖。Struct 中的每個欄位都必須是明確的類型，標量(VARCHAR、INT、BOOLEAN 等) 或向量(FLOAT_VECTOR)。
建議您只包含擷取或顯示時會用到的欄位，以保持 Struct 結構描述的精簡。避免臃腫的未使用元資料。
深思熟慮設定最大容量
每個 Array 欄位都有一個屬性，指定每個實體的 Array 欄位可以容納的最大元素數量。請根據您使用個案的上限來設定。例如，每個文件有 1,000 個文字區塊，或每個駕駛場景有 100 個操作。
過高的值會浪費記憶體，您需要做一些計算來決定 Array 欄位中 Struct 的最大數量。
在 Structs 中索引向量欄位
對於向量欄位，包括集合中的向量欄位和 Struct 中定義的向量欄位，索引是必須的。對於 Struct 中的向量欄位，您應該使用AUTOINDEX 或HNSW 作為索引類型，並使用MAX_SIM 系列作為度量類型。
有關所有適用限制的詳細資訊，請參閱限制。
一個真實世界的範例：為自動駕駛建立 CoVLA 資料集模型
由Turing Motors介紹並在 2025 年電腦視覺應用冬季會議 (Winter Conference on Applications of Computer Vision, WACV) 上接受的全面視覺-語言-動作 (Comprehensive Vision-Language-Action, CoVLA) 資料集，為訓練和評估自動駕駛中的視覺-語言-動作 (Vision-Language-Action, VLA) 模型提供了豐富的基礎。每個資料點（通常是視訊片段）不僅包含原始視覺輸入，還包含描述下列內容的結構化字幕：
自我車輛的行為(例如：「向左並線，同時讓開迎面駛來的車輛」)、
偵測到的存在物件(例如：前方車輛、行人、交通燈)，以及
畫面層級的場景說明。
這種分層、多模式的特性使其成為 Array of Structs 功能的理想候選。有關 CoVLA 資料集的詳細資訊，請參閱CoVLA 資料集網站。
步驟 1：將資料集映射為集合模式
CoVLA 資料集是一個大型、多模態的駕駛資料集，包含 10,000 個視訊片段，總長度超過 80 小時。它以 20Hz 的速率採樣畫面，並為每個畫面加上詳細的自然語言說明，以及車輛狀態和偵測到的物體座標等資訊。
資料集結構如下：
├── video_1 (VIDEO) # video.mp4
│ ├── video_id (INT)
│ ├── video_url (STRING)
│ ├── frames (ARRAY)
│ │ ├── frame_1 (STRUCT)
│ │ │ ├── caption (STRUCT) # captions.jsonl
│ │ │ │ ├── plain_caption (STRING)
│ │ │ │ ├── rich_caption (STRING)
│ │ │ │ ├── risk (STRING)
│ │ │ │ ├── risk_correct (BOOL)
│ │ │ │ ├── risk_yes_rate (FLOAT)
│ │ │ │ ├── weather (STRING)
│ │ │ │ ├── weather_rate (FLOAT)
│ │ │ │ ├── road (STRING)
│ │ │ │ ├── road_rate (FLOAT)
│ │ │ │ ├── is_tunnel (BOOL)
│ │ │ │ ├── is_tunnel_yes_rate (FLOAT)
│ │ │ │ ├── is_highway (BOOL)
│ │ │ │ ├── is_highway_yes_rate (FLOAT)
│ │ │ │ ├── has_pedestrain (BOOL)
│ │ │ │ ├── has_pedestrain_yes_rate (FLOAT)
│ │ │ │ ├── has_carrier_car (BOOL)
│ │ │ ├── traffic_light (STRUCT) # traffic_lights.jsonl
│ │ │ │ ├── index (INT)
│ │ │ │ ├── class (STRING)
│ │ │ │ ├── bbox (LIST<FLOAT>)
│ │ │ ├── front_car (STRUCT) # front_cars.jsonl
│ │ │ │ ├── has_lead (BOOL)
│ │ │ │ ├── lead_prob (FLOAT)
│ │ │ │ ├── lead_x (FLOAT)
│ │ │ │ ├── lead_y (FLOAT)
│ │ │ │ ├── lead_speed_kmh (FLOAT)
│ │ │ │ ├── lead_a (FLOAT)
│ │ ├── frame_2 (STRUCT)
│ │ ├── ... (STRUCT)
│ │ ├── frame_n (STRUCT)
├── video_2
├── ...
├── video_n
您可以發現 CoVLA 資料集的結構具有高度層次性，將收集到的資料分為多個.jsonl 檔案，以及.mp4 格式的視訊片段。
在 Milvus 中，您可以使用 JSON 欄位或 Array-of-Structs 欄位在集合模式中建立巢狀結構。當向量嵌入是巢狀格式的一部分時，只支援 Array-of-Structs 欄位。然而，Array 內的 Struct 本身無法包含其他巢狀結構。為了在保留基本關係的同時儲存 CoVLA 資料集，您需要移除不必要的層級結構，並將資料扁平化，使其符合 Milvus 集合模式。
下圖說明了我們如何使用下面的模式來建立這個資料集的模型：
資料集模式
上圖說明了視訊素材的結構，其中包含下列欄位：
video_id是主索引鍵，接受 INT64 類型的整數。states是原始 JSON 體，包含目前視訊中每一格的小我車輛狀態。captions是一個 Struct 陣列，每個 Struct 都有下列欄位：frame_id識別目前視訊中的特定幀。plain_caption是目前畫面的描述，不包含周遭環境，例如天氣、路況等，plain_cap_vector是其相對應的向量嵌入。rich_caption是有環境環境的目前畫面的描述，rich_cap_vector是其相對應的向量嵌入。risk是小我車輛在目前幀中所面臨風險的描述，risk_vector是其對應的向量嵌入，以及是該幀的所有其他屬性，例如
road,weather,is_tunnel,has_pedestrain, 等等...
traffic_lights是一個 JSON 體，包含目前框架中識別的所有交通燈號。front_cars也是一個 Structs 陣列，包含目前框架中識別出的所有領頭車。
步驟 2：初始化模式
首先，我們需要初始化標題 Struct、front_cars Struct 和集合的模式。
初始化 Caption Struct 的模式。
client = MilvusClient("http://localhost:19530") # create the schema for the caption struct schema_for_caption = client.create_struct_field_schema() schema_for_caption.add_field( field_name="frame_id", datatype=DataType.INT64, description="ID of the frame to which the ego vehicle's behavior belongs" ) schema_for_caption.add_field( field_name="plain_caption", datatype=DataType.VARCHAR, max_length=1024, description="plain description of the ego vehicle's behaviors" ) schema_for_caption.add_field( field_name="plain_cap_vector", datatype=DataType.FLOAT_VECTOR, dim=768, description="vectors for the plain description of the ego vehicle's behaviors" ) schema_for_caption.add_field( field_name="rich_caption", datatype=DataType.VARCHAR, max_length=1024, description="rich description of the ego vehicle's behaviors" ) schema_for_caption.add_field( field_name="rich_cap_vector", datatype=DataType.FLOAT_VECTOR, dim=768, description="vectors for the rich description of the ego vehicle's behaviors" ) schema_for_caption.add_field( field_name="risk", datatype=DataType.VARCHAR, max_length=1024, description="description of the ego vehicle's risks" ) schema_for_caption.add_field( field_name="risk_vector", datatype=DataType.FLOAT_VECTOR, dim=768, description="vectors for the description of the ego vehicle's risks" ) schema_for_caption.add_field( field_name="risk_correct", datatype=DataType.BOOL, description="whether the risk assessment is correct" ) schema_for_caption.add_field( field_name="risk_yes_rate", datatype=DataType.FLOAT, description="probability/confidence of risk being present" ) schema_for_caption.add_field( field_name="weather", datatype=DataType.VARCHAR, max_length=50, description="weather condition" ) schema_for_caption.add_field( field_name="weather_rate", datatype=DataType.FLOAT, description="probability/confidence of the weather condition" ) schema_for_caption.add_field( field_name="road", datatype=DataType.VARCHAR, max_length=50, description="road type" ) schema_for_caption.add_field( field_name="road_rate", datatype=DataType.FLOAT, description="probability/confidence of the road type" ) schema_for_caption.add_field( field_name="is_tunnel", datatype=DataType.BOOL, description="whether the road is a tunnel" ) schema_for_caption.add_field( field_name="is_tunnel_yes_rate", datatype=DataType.FLOAT, description="probability/confidence of the road being a tunnel" ) schema_for_caption.add_field( field_name="is_highway", datatype=DataType.BOOL, description="whether the road is a highway" ) schema_for_caption.add_field( field_name="is_highway_yes_rate", datatype=DataType.FLOAT, description="probability/confidence of the road being a highway" ) schema_for_caption.add_field( field_name="has_pedestrian", datatype=DataType.BOOL, description="whether there is a pedestrian present" ) schema_for_caption.add_field( field_name="has_pedestrian_yes_rate", datatype=DataType.FLOAT, description="probability/confidence of pedestrian presence" ) schema_for_caption.add_field( field_name="has_carrier_car", datatype=DataType.BOOL, description="whether there is a carrier car present" )初始化 Front Car Struct 的模式。
雖然前車不涉及向量嵌入，但由於資料大小超過 JSON 欄位的最大值，您仍需要將其納入 Struct 陣列。
schema_for_front_car = client.create_struct_field_schema() schema_for_front_car.add_field( field_name="frame_id", datatype=DataType.INT64, description="ID of the frame to which the ego vehicle's behavior belongs" ) schema_for_front_car.add_field( field_name="has_lead", datatype=DataType.BOOL, description="whether there is a leading vehicle" ) schema_for_front_car.add_field( field_name="lead_prob", datatype=DataType.FLOAT, description="probability/confidence of the leading vehicle's presence" ) schema_for_front_car.add_field( field_name="lead_x", datatype=DataType.FLOAT, description="x position of the leading vehicle relative to the ego vehicle" ) schema_for_front_car.add_field( field_name="lead_y", datatype=DataType.FLOAT, description="y position of the leading vehicle relative to the ego vehicle" ) schema_for_front_car.add_field( field_name="lead_speed_kmh", datatype=DataType.FLOAT, description="speed of the leading vehicle in km/h" ) schema_for_front_car.add_field( field_name="lead_a", datatype=DataType.FLOAT, description="acceleration of the leading vehicle" )為集合初始化模式
schema = client.create_schema() schema.add_field( field_name="video_id", datatype=DataType.VARCHAR, description="primary key", max_length=16, is_primary=True, auto_id=False ) schema.add_field( field_name="video_url", datatype=DataType.VARCHAR, max_length=512, description="URL of the video" ) schema.add_field( field_name="captions", datatype=DataType.ARRAY, element_type=DataType.STRUCT, struct_schema=schema_for_caption, max_capacity=600, description="captions for the current video" ) schema.add_field( field_name="traffic_lights", datatype=DataType.JSON, description="frame-specific traffic lights identified in the current video" ) schema.add_field( field_name="front_cars", datatype=DataType.ARRAY, element_type=DataType.STRUCT, struct_schema=schema_for_front_car, max_capacity=600, description="frame-specific leading cars identified in the current video" )
步驟 3：設定索引參數
所有向量欄位都必須建立索引。要索引元素 Struct 中的向量欄位，您需要使用AUTOINDEX 或HNSW 作為索引類型，並使用MAX_SIM 系列度量類型來衡量嵌入清單之間的相似性。
index_params = client.prepare_index_params()
index_params.add_index(
field_name="captions[plain_cap_vector]",
index_type="AUTOINDEX",
metric_type="MAX_SIM_COSINE",
index_name="captions_plain_cap_vector_idx", # mandatory for now
index_params={"M": 16, "efConstruction": 200}
)
index_params.add_index(
field_name="captions[rich_cap_vector]",
index_type="AUTOINDEX",
metric_type="MAX_SIM_COSINE",
index_name="captions_rich_cap_vector_idx", # mandatory for now
index_params={"M": 16, "efConstruction": 200}
)
index_params.add_index(
field_name="captions[risk_vector]",
index_type="AUTOINDEX",
metric_type="MAX_SIM_COSINE",
index_name="captions_risk_vector_idx", # mandatory for now
index_params={"M": 16, "efConstruction": 200}
)
建議您啟用 JSON 欄位的 JSON 切碎功能，以加速這些欄位內的篩選。
步驟 4：建立集合
一旦模式和索引準備就緒，您就可以按以下步驟建立目標集合：
client.create_collection(
collection_name="covla_dataset",
schema=schema,
index_params=index_params
)
步驟 5：插入資料
Turing Motos 將 CoVLA 資料集組織為多個檔案，包括原始視訊片段 (.mp4)、狀態 (states.jsonl)、字幕 (captions.jsonl)、交通燈 (traffic_lights.jsonl)，以及前車 (front_cars.jsonl)。
您需要合併這些檔案中每個視訊素材的資料片段，並插入資料。以下是合併特定視訊素材資料片段的腳本。
import json
from openai import OpenAI
openai_client = OpenAI(
api_key='YOUR_OPENAI_API_KEY',
)
video_id = "0a0fc7a5db365174" # represent a single video with 600 frames
# get all front car records in the specified video clip
entries = []
front_cars = []
with open('data/front_car/{}.jsonl'.format(video_id), 'r') as f:
for line in f:
entries.append(json.loads(line))
for entry in entries:
for key, value in entry.items():
value['frame_id'] = int(key)
front_cars.append(value)
# get all traffic lights identified in the specified video clip
entries = []
traffic_lights = []
frame_id = 0
with open('data/traffic_lights/{}.jsonl'.format(video_id), 'r') as f:
for line in f:
entries.append(json.loads(line))
for entry in entries:
for key, value in entry.items():
if not value or (value['index'] == 1 and key != '0'):
frame_id+=1
if value:
value['frame_id'] = frame_id
traffic_lights.append(value)
else:
value_dict = {}
value_dict['frame_id'] = frame_id
traffic_lights.append(value_dict)
# get all captions generated in the video clip and convert them into vector embeddings
entries = []
captions = []
with open('data/captions/{}.jsonl'.format(video_id), 'r') as f:
for line in f:
entries.append(json.loads(line))
def get_embedding(text, model="embeddinggemma:latest"):
response = openai_client.embeddings.create(input=text, model=model)
return response.data[0].embedding
# Add embeddings to each entry
for entry in entries:
# Each entry is a dict with a single key (e.g., '0', '1', ...)
for key, value in entry.items():
value['frame_id'] = int(key) # Convert key to integer and assign to frame_id
if "plain_caption" in value and value["plain_caption"]:
value["plain_cap_vector"] = get_embedding(value["plain_caption"])
if "rich_caption" in value and value["rich_caption"]:
value["rich_cap_vector"] = get_embedding(value["rich_caption"])
if "risk" in value and value["risk"]:
value["risk_vector"] = get_embedding(value["risk"])
captions.append(value)
data = {
"video_id": video_id,
"video_url": "https://your-storage.com/{}".format(video_id),
"captions": captions,
"traffic_lights": traffic_lights,
"front_cars": front_cars
}
一旦您對資料進行了相應的處理，您就可以按以下方式插入資料：
client.insert(
collection_name="covla_dataset",
data=[data]
)
# {'insert_count': 1, 'ids': ['0a0fc7a5db365174'], 'cost': 0}