作者：周弘懿（锦琛）
业务背景
在智能驾驶与自动驾驶的研发链路里，车端摄像头是数据的第一入口。一辆测试车通常搭载 6~12 路环视相机，以 20~30 FPS 持续采集行车画面。单车单日的路测就能产生数 TB 的视频，一支上百辆车的车队每天回传的原始视频动辄数十上百 TB。这些视频是训练感知模型、复现事故、迭代规控策略的核心资产，但它们有一个共同的特征：海量、非结构化、难以检索。
工程团队真正关心的，从来不是"某段视频"本身，而是视频里发生了什么。按帧（截帧）做场景理解，需要从每一帧画面中识别出：
- 交通参与者：行人、非机动车、前车、对向来车；
- 交通管制元素：红绿灯状态、车道线、交通标志、限速牌、地面箭头；
- 道路环境：路口、高速、隧道、施工区、匝道、收费站；
- 异常事件：加塞、闯红灯、急刹、逆行、事故、抛洒物、大车压线。
把这些理解结果结构化沉淀下来，才能撑起智能驾驶研发中几个高价值的场景：
- 数据回灌（Data Replay）：把真实路况帧回灌到仿真与训练管线，持续喂养感知模型。
- Corner Case 挖掘：从海量帧里精准捞出"雨夜隧道口的施工区"这类长尾场景，这正是模型最容易翻车、也最缺样本的地方。
- 自动标注（Auto-labeling）：用大模型对帧做粗标注，替代大量纯人工拉框打标，人工只做审核与精修。
- 路测报告：按场景、按事件聚合统计，快速产出"本周高速匝道加塞 132 次、隧道口误识别 7 次"这类可量化报告。
技术挑战
- 推理链路要自建、要运维。抽帧后要做场景理解，得自己拉起一套模型推理服务：GPU 资源、模型加载、批处理、扩缩容、故障恢复，一样不能少。对一个以"造好车"为目标的团队来说，维护一条推理链路是纯粹的成本项。
- 多套系统拼接，数据搬来搬去。典型架构是"抽帧服务 → 目标检测/多模态模型 → 向量库 → 元数据库"。帧数据在对象存储、推理集群、向量库之间反复搬运，链路长、时延高、故障点多，任何一环挂了整条流水线都会堵。
- 结构化标注成本高、口径不统一。人工打标不仅慢、贵，而且不同标注员对"施工区""加塞"的判定口径难以统一，导致标签噪声大、下游模型学偏。
解决方案
阿里云 Milvus 2.6 提供的 AI Function 把上面三件事收敛进了同一个向量数据库：推理即查询。模型推理不再是一条独立的外部链路，而是在写入（insert）和检索（search）时由 Milvus 内部自动触发的函数调用，数据全程不出 Milvus 实例。
针对智能驾驶截帧分析，我们把整条链路映射到四个 AI Function：
函数 |
作用 |
在智能驾驶截帧里 |
AI_EMBEDDING |
用 qwen3-vl-embedding 把截帧图片（或视频片段）转成 2560 维稠密向量入库；多模态模型让文本与图像映射到同一向量空间 |
支持以文搜帧与以图搜帧，「雨天施工区」这类语义需求能被向量检索命中 |
AI_CLASSIFY |
从预设标签集中为每帧选出最匹配的一项，写入分类字段 |
给每帧打上 路口/高速/隧道/施工区 等场景标签 |
AI_EXTRACT |
按指定标签从帧中抽取交通元素，以 JSON 写入结构化字段 |
抽取红绿灯状态、车道数、天气、异常事件，用于精确过滤 |
AI_ENTITY_EXTRACT |
识别帧中明确出现的命名实体 |
抽取地名、路名、限速数值，用于事件检索与索引 |
AI_RERANK |
向量召回 Top-N 后用 qwen3-vl-rerank 二次打分（精排） |
按主体、外观、构图一致性重排，提升 Corner Case 挖掘的精排质量 |
具体AI Function用法参考官方文档
整体架构如下：
实战操作
- 视频预处理（ffmpeg 抽帧）
车端相机采集的是连续视频，需先用 ffmpeg 抽帧，得到帧图片并上传到可被模型访问的 URL（如 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" -vsync 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
用 ffmpeg 对每段视频定时截帧（抽帧），各取 4 帧，合计 12 帧进入流水线。
- 核心代码
包含以下核心步骤：
- 建 Collection：一次建表挂 4 个 AI Function ——
embed_frame（多模态向量化，qwen3-vl-embedding、dim=2560）、classify_scene（场景分类：路口/高速/隧道/施工区/普通道路）、extract_traffic（结构化抽取：红绿灯/车道数/天气/异常事件）、extract_entities（命名实体：路名/地名/限速），向量索引用AUTOINDEX+COSINE。 - 截帧入库：只写
frame_url/clip_id/ts_ms，embedding/scene/attributes/entities由 AI Function 在写入时自动填充；写入后用query验证结构化结果。
12 帧全部insert成功，无需人工标注，Milvus 在写入时自动产出 scene/attributes。例如 dashcam 第 5s 帧被正确识别为路口 + 绿灯，traffic_cam 多帧识别出雨天： - 以文搜帧 / 以图搜帧：文本查询经同一多模态 embedding 映射到图像向量空间直接召回；可叠加
scene/attributes结构化过滤做 Corner Case 挖掘；把data换成图片 URL 即以图搜帧。 - AI_RERANK 重排（可选）：
search挂ranker用qwen3-vl-rerank对 Top-N 精排；或对已召回的一批帧 URL 走 REST/v2/vectordb/ai/rerank独立重排。
用 qwen3-vl-rerank 二次精排
完整示例代码
将MILVUS_HOST和MILVUS_TOKEN替换成自己的集群就可以直接看到效果
# ===== 步骤 1｜环境准备：实例地址、token、REST 辅助函数与 TEXTTRANSFORM 兼容封装 ===== 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")) TEXTTRANSFORM_FUNCTION_TYPE = 9 # 阿里云 Milvus 将 TEXTTRANSFORM 作为托管扩展暴露，函数类型值为 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) # ===== 步骤 2｜建 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) 结构化抽取：红绿灯/车道数/天气/异常事件 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；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) # ===== 步骤 3｜截帧入库：写入即推理，embedding/scene/attributes/entities 自动填充 ===== # 真实行车帧：由 assets/videos 下三段视频截帧后上传 OSS，URL 存于 real_frames_urls.json（签名URL） from pathlib import Path as _Path _frames_meta = json.loads(_Path(__file__).with_name("real_frames_urls.json").read_text(encoding="utf-8")) _frames_all = sorted(_frames_meta["frames"].values(), key=lambda x: (x["clip_id"], x["ts_ms"])) frames = [{"frame_url": f["url"], "clip_id": f["clip_id"], "ts_ms": f["ts_ms"]} for f in _frames_all] _url_by_clip = {} for f in _frames_all: _url_by_clip.setdefault(f["clip_id"], f["url"]) # qwen3-vl-embedding 多模态批量上限为 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(row) # ===== 步骤 4｜以文搜帧 / 以图搜帧：文本或图片查询走同一多模态 embedding 召回 ===== 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']} url={hit['entity']['frame_url']}") # 向量召回 + 结构化过滤（Corner Case 挖掘）：先按 scene 与异常事件过滤，再做语义排序 corner_query = "夜间隧道口，前车急刹" corner_search = client.search( collection_name=collection_name, data=[corner_query], anns_field="embedding", limit=10, filter='scene == "隧道" and attributes["anomaly_event"] != "none"', output_fields=["frame_url", "scene", "attributes"]) # 以图搜帧：用一张真实行车帧（dashcam 首帧）作为查询图 image_query_url = _url_by_clip.get("clip_dashcam", 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"]) # ===== 步骤 5｜AI_RERANK 多模态重排（可选）===== # 方式一：search 挂 ranker，向量召回 Top-N 后由 qwen3-vl-rerank 精排 QUERY = "高速匝道处的加塞行为" 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"score={hit['distance']:.4f} url={hit['entity']['frame_url']}") # 方式二：只对已召回的一批帧 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": "高速匝道处的加塞行为", "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}}) rest_rerank = [] if status == 200 and data.get("code") == 0: for item in sorted(data["data"]["output"]["results"], key=lambda x: x["relevance_score"], reverse=True): print(item["index"], item["relevance_score"]) rest_rerank.append({"url": rest_docs[item["index"]], "score": item["relevance_score"]}) # ===== 步骤 6｜生成可视化 HTML：把图/视频和检索结果直接看到 ===== import html as _html import webbrowser from pathlib import Path OUTPUT_HTML = Path(__file__).with_name("test3_report.html") VIDEO_EXTS = (".mp4", ".webm", ".mov", ".m3u8") def media_tag(url: str) -> str: safe = _html.escape(url, quote=True) lower = url.split("?", 1)[0].lower() if lower.endswith(VIDEO_EXTS): return f'<video src="{safe}" controls muted loop playsinline></video>' return f'<img src="{safe}" loading="lazy" alt="frame">' def fmt_attrs(attributes: Any) -> str: if not isinstance(attributes, dict): return "" parts = [f"{k}={v}" for k, v in attributes.items() if v not in (None, "none", "", [])] return " · ".join(parts) def fmt_entities(entities: Any) -> str: if isinstance(entities, dict): entities = entities.get("entities", entities) if not entities: return "" return _html.escape(json.dumps(entities, ensure_ascii=False)) def card(url: str, badges: list[str], meta: str = "") -> str: chips = "".join(f'<span class="chip">{_html.escape(b)}</span>' for b in badges if b) meta_html = f'<div class="meta">{_html.escape(meta)}</div>' if meta else "" short = url.split("/")[-1].split("?")[0] return ( '<div class="card">' f'<div class="thumb">{media_tag(url)}</div>' f'<div class="chips">{chips}</div>' f'{meta_html}' f'<a class="url" href="{_html.escape(url, quote=True)}" target="_blank">{_html.escape(short)}</a>' '</div>' ) sections: list[str] = [] # 入库帧 cards = [] for r in ingested_rows: badges = [f"场景 {r.get('scene')}", f"{r.get('clip_id')} @ {r.get('ts_ms')}ms"] ent = fmt_entities(r.get("entities")) meta = fmt_attrs(r.get("attributes")) if ent: meta = (meta + " · " if meta else "") + "实体 " + ent cards.append(card(r["frame_url"], badges, meta)) sections.append( f'<section><h2>步骤 3 · 截帧入库（写入即推理）</h2>' f'<p class="desc">写入时由 4 个 AI Function 自动生成向量、场景分类、结构化属性与命名实体。共 {len(ingested_rows)} 帧。</p>' f'<div class="grid">{"".join(cards)}</div></section>') def search_section(title: str, desc: str, hits, score_label: str = "相似度") -> str: cards = [] for rank, hit in enumerate(hits, 1): e = hit["entity"] badges = [f"#{rank}", f"{score_label} {hit['distance']:.4f}"] if e.get("scene"): badges.append(f"场景 {e['scene']}") cards.append(card(e["frame_url"], badges, fmt_attrs(e.get("attributes")))) body = f'<div class="grid">{"".join(cards)}</div>' if cards else '<p class="empty">无命中结果</p>' return f'<section><h2>{title}</h2><p class="desc">{_html.escape(desc)}</p>{body}</section>' sections.append(search_section( "步骤 4 · 以文搜帧", f'查询："{query}"', text_search[0])) sections.append(search_section( "步骤 4 · Corner Case 过滤检索", f'查询："{corner_query}"，过滤：scene==\"隧道\" 且 anomaly_event!=\"none\"', corner_search[0])) # 以图搜帧：先展示查询图，再展示结果 img_cards = [card(image_query_url, ["查询图"], "以图搜帧的输入")] for rank, hit in enumerate(image_search[0], 1): e = hit["entity"] img_cards.append(card(e["frame_url"], [f"#{rank}", f"相似度 {hit['distance']:.4f}", f"场景 {e.get('scene')}"])) sections.append( '<section><h2>步骤 4 · 以图搜帧</h2>' '<p class="desc">用一张参考图 URL 作为查询，走同一多模态 embedding 召回。</p>' f'<div class="grid">{"".join(img_cards)}</div></section>') sections.append(search_section( "步骤 5 · AI_RERANK 精排（search 内挂 ranker）", f'查询："{QUERY}"，向量召回后由 qwen3-vl-rerank 精排', rerank_search[0], score_label="重排分")) # REST 即时重排 rest_cards = [ card(item["url"], [f"#{rank}", f"相关度 {item['score']:.4f}"]) for rank, item in enumerate(rest_rerank, 1) ] rest_body = f'<div class="grid">{"".join(rest_cards)}</div>' if rest_cards else '<p class="empty">REST 重排未返回结果</p>' sections.append( '<section><h2>步骤 5 · REST 即时重排 /v2/vectordb/ai/rerank</h2>' f'<p class="desc">对给定的一批帧 URL 独立重排，查询："{_html.escape(QUERY)}"</p>{rest_body}</section>') page = f"""<!DOCTYPE html> <html lang="zh-CN"><head><meta charset="utf-8"> <meta name="viewport" content="width=device-width, initial-scale=1"> <title>test3 多模态截帧检索 · 可视化报告</title> <style> :root{{color-scheme:dark}} *{{box-sizing:border-box}} body{{margin:0;font-family:-apple-system,BlinkMacSystemFont,"Segoe UI","PingFang SC",sans-serif;background:#0f1115;color:#e6e8eb}} header{{padding:28px 32px;border-bottom:1px solid #23262d;background:#151821}} header h1{{margin:0 0 6px;font-size:20px}} header p{{margin:0;color:#8b929e;font-size:13px}} section{{padding:24px 32px;border-bottom:1px solid #1c1f27}} section h2{{margin:0 0 4px;font-size:16px;color:#f2f4f7}} .desc{{margin:0 0 16px;color:#8b929e;font-size:13px}} .empty{{color:#6b7280;font-size:13px}} .grid{{display:grid;grid-template-columns:repeat(auto-fill,minmax(220px,1fr));gap:16px}} .card{{background:#171a21;border:1px solid #262a33;border-radius:10px;overflow:hidden;display:flex;flex-direction:column}} .thumb{{aspect-ratio:4/3;background:#0b0d11;display:flex;align-items:center;justify-content:center;overflow:hidden}} .thumb img,.thumb video{{width:100%;height:100%;object-fit:cover;display:block}} .chips{{display:flex;flex-wrap:wrap;gap:6px;padding:10px 10px 4px}} .chip{{font-size:11px;background:#232735;color:#a9c7ff;border:1px solid #2f3b5a;padding:2px 8px;border-radius:999px}} .meta{{padding:2px 10px 6px;color:#c4cad3;font-size:12px;line-height:1.5}} .url{{padding:6px 10px 12px;color:#5b7fff;font-size:11px;text-decoration:none;word-break:break-all}} .url:hover{{text-decoration:underline}} </style></head> <body> <header> <h1>多模态截帧检索 · 可视化报告</h1> <p>Collection: {collection_name} ｜ 模型: {EMBED_MODEL} / {VLM_MODEL} ｜ 实例: {_html.escape(MILVUS_URI)}</p> </header> {"".join(sections)} </body></html>""" OUTPUT_HTML.write_text(page, encoding="utf-8") print(f"\n✅ 可视化报告已生成：{OUTPUT_HTML}") try: webbrowser.open(OUTPUT_HTML.as_uri()) except Exception: pass
实战效果
1）以文搜帧
查询 "雨天施工区，路面有锥形桶和施工牌"，Top-3 召回（示意）：
一句自然语言即可召回相关帧：
2）图搜帧
把参考图 URL 作为 query，参考图自身相似度 = 1.000，验证图像编码一致性：
3）rerank精排
经 qwen3-vl-rerank 精排后，天气与场景同时匹配的第 1 帧被进一步拉开与其余候选的差距，误召回的"普通道路"帧被压到末位。
自建方案 vs AI Function 对比
维度 |
传统自建方案 |
Milvus AI Function |
开发周期 |
抽帧+推理+向量库+元数据库多系统联调，数周起 |
一次建表挂函数， |
推理运维 |
自建 GPU 推理集群，需扩缩容/故障恢复 |
由 Milvus 托管调用，业务侧零运维 |
数据流转 |
帧在对象存储/推理集群/向量库间多次搬运 |
写入即推理，数据不出 Milvus 实例 |
合规风险 |
中间向量/原始帧易出库，评审难过 |
凭据由管理员配置，请求体不含凭据，数据不出库 |
多模态检索 |
需自建对齐文本-图像向量空间 |
|
标注口径 |
人工标注慢、贵、口径不一 |
|
总结
回到最初的痛点：智能驾驶截帧分析难，难在要把"推理、存储、检索"三件事分别用三套系统实现，还要保证敏感的行车数据在它们之间安全流转。阿里云 Milvus AI Function 的做法，是把推理内聚进向量数据库——写入即向量化与结构化理解，检索即多模态语义召回，凭据与模型服务由平台托管，数据全程不出库。
对智能驾驶团队而言，这套方案的直接价值是：
- Corner Case 挖掘提速：以文搜帧 + 结构化过滤 + 多模态重排，长尾场景可用一句话捞出。
- 自动标注降本：
AI_CLASSIFY/AI_EXTRACT/AI_ENTITY_EXTRACT在写入时统一口径产出标签，人工只做审核。
让 AI 落地，不必再从复杂链路开始
阿里云 Milvus 是目前唯一支持 AI Native 的 Milvus 云服务。它不仅提供高性能、全托管的向量检索能力，更将 AI Function 与 AI-Gateway 原生嵌入数据链路，让模型能力真正成为数据库的一部分。
从数据进入阿里云Milvus 的那一刻起，理解、加工、检索和生成便可以在一条原生链路中完成。阿里云 Milvus 正在把“接入 AI”变成“原生拥有 AI”。
