作者：周弘懿（锦琛）
业务背景
在智能驾驶与自动驾驶的研发链路里，车端摄像头是数据的第一入口。一辆测试车通常搭载 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 挖掘的精排质量 |
","margin":true,"hideBorder":false,"id":"lx7QA"}">具体AI Function用法参考官方文档
整体架构如下：
实战操作
视频预处理（ffmpeg 抽帧）
车端相机采集的是连续视频，需先用 ffmpeg 抽帧，得到帧图片并上传到可被模型访问的 URL（如 OSS，生产建议短时效签名地址），再把 frame_url 交给下面的核心代码入库。抽帧时同步记录每帧的 clip_id 与 ts_ms（视频片段 ID 帧时间戳），便于后续检索定位。
用 ffmpeg 对每段视频定时截帧（抽帧），各取 4 帧，合计 12 帧进入流水线。
核心代码
包含以下核心步骤：
建 Collection：一次建表挂 4 个 AI Function ——
embed_frame（多模态向量化，
qwen3-vl-embedding、dim=2560）、
classify_scene（场景分类：路口/高速/隧道/施工区/普通道路）、
extract_traffic（结构化抽取：红绿灯/车道数/天气/异常事件）、
extract_entities（命名实体：路名/地名/限速），向量索引用
AUTOINDEXCOSINE。
截帧入库：只写 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替换成自己的集群就可以直接看到效果
tuple[int, dict[str, Any]]:n request = Request(n f"{MILVUS_URI.rstrip('/')}{path}",n data=json.dumps(body, ensure_ascii=False).encode("utf-8"),n headers={"Authorization": f"Bearer {MILVUS_TOKEN}", "Content-Type": "application/json"},n method="POST",n )n try:n with urlopen(request, timeout=timeout) as response:n return response.status, json.loads(response.read().decode("utf-8"))n except HTTPError as exc:n return exc.code, json.loads(exc.read().decode("utf-8"))nnnTEXTTRANSFORM_FUNCTION_TYPE = 9 # 阿里云 Milvus 将 TEXTTRANSFORM 作为托管扩展暴露，函数类型值为 9nnndef texttransform_function_type() -> Any:n for type_name in ("TEXTTRANSFORM", "TEXT_TRANSFORM", "TextTransform"):n function_type = getattr(FunctionType, type_name, None)n if function_type is not None:n return function_typen existing = getattr(FunctionType, "_value2member_map_", {}).get(TEXTTRANSFORM_FUNCTION_TYPE)n if existing is not None:n return existingn extension = int.__new__(FunctionType, TEXTTRANSFORM_FUNCTION_TYPE)n extension._name_ = "TEXTTRANSFORM"n extension._value_ = TEXTTRANSFORM_FUNCTION_TYPEn FunctionType._value2member_map_[TEXTTRANSFORM_FUNCTION_TYPE] = extensionn FunctionType._member_map_["TEXTTRANSFORM"] = extensionn return extensionnnnVECTOR_DIM = 2560nEMBED_MODEL = "qwen3-vl-embedding"nVLM_MODEL = "qwen3.7-plus" # 多模态理解（分类/抽取/实体）模型nclient = MilvusClient(uri=MILVUS_URI, token=MILVUS_TOKEN)nn# ===== 步骤 2｜建 Collection：一次建表挂 4 个 AI Function =====ncollection_name = "driving_frames"nif client.has_collection(collection_name):n client.drop_collection(collection_name)nnschema = MilvusClient.create_schema(auto_id=True, enable_dynamic_field=False)nschema.add_field("id", DataType.INT64, is_primary=True)nschema.add_field("frame_url", DataType.VARCHAR, max_length=4096) # 截帧图片地址nschema.add_field("clip_id", DataType.VARCHAR, max_length=128) # 所属视频片段 IDnschema.add_field("ts_ms", DataType.INT64) # 帧时间戳(ms)nschema.add_field("embedding", DataType.FLOAT_VECTOR, dim=VECTOR_DIM)nschema.add_field("scene", DataType.VARCHAR, max_length=64) # AI_CLASSIFY 输出nschema.add_field("attributes", DataType.JSON) # AI_EXTRACT 输出nschema.add_field("entities", DataType.JSON) # AI_ENTITY_EXTRACT 输出nn# 1) 多模态向量化：帧图片 -> 2560 维向量nschema.add_function(Function(n name="embed_frame", function_type=FunctionType.TEXTEMBEDDING,n input_field_names=["frame_url"], output_field_names=["embedding"],n params={"provider": "aliyun_milvus", "model_name": EMBED_MODEL, "dim": VECTOR_DIM, "is_multimodal": "true"}))nn# 2) 场景分类：路口/高速/隧道/施工区/普通道路nschema.add_function(Function(n name="classify_scene", function_type=texttransform_function_type(),n input_field_names=["frame_url"], output_field_names=["scene"],n params={"provider": "aliyun_milvus", "model_name": VLM_MODEL, "task": "ai_classify", "media_type": "image",n "labels": "路口,高速,隧道,施工区,普通道路", "prompt": "根据行车画面所处的道路环境分类。", "temperature": "0"}))nn# 3) 结构化抽取：红绿灯/车道数/天气/异常事件nschema.add_function(Function(n name="extract_traffic", function_type=texttransform_function_type(),n input_field_names=["frame_url"], output_field_names=["attributes"],n params={"provider": "aliyun_milvus", "model_name": VLM_MODEL, "task": "ai_extract", "media_type": "image",n "labels": "traffic_light,lane_count,weather,anomaly_event",n "prompt": "traffic_light 取值 red/green/yellow/none；anomaly_event 描述加塞、闯红灯、事故等异常，无则填 none。", "temperature": "0"}))nn# 4) 命名实体：路名/地名/限速数值nschema.add_function(Function(n name="extract_entities", function_type=texttransform_function_type(),n input_field_names=["frame_url"], output_field_names=["entities"],n params={"provider": "aliyun_milvus", "model_name": VLM_MODEL, "task": "ai_entity_extract", "media_type": "image",n "entity_types": "LOCATION,PRODUCT",n "prompt": "仅抽取画面中交通标志牌上明确出现的地名、路名与限速数值，不要根据外观猜测。", "temperature": "0"}))nnindex_params = client.prepare_index_params()nindex_params.add_index(field_name="embedding", index_type="AUTOINDEX", metric_type="COSINE")nclient.create_collection(collection_name=collection_name, schema=schema, index_params=index_params)nn# ===== 步骤 3｜截帧入库：写入即推理，embedding/scene/attributes/entities 自动填充 =====n# 真实行车帧：由 assets/videos 下三段视频截帧后上传 OSS，URL 存于 real_frames_urls.json（签名URL）nfrom pathlib import Path as _Pathnn_frames_meta = json.loads(_Path(__file__).with_name("real_frames_urls.json").read_text(encoding="utf-8"))n_frames_all = sorted(_frames_meta["frames"].values(), key=lambda x: (x["clip_id"], x["ts_ms"]))nframes = [{"frame_url": f["url"], "clip_id": f["clip_id"], "ts_ms": f["ts_ms"]} for f in _frames_all]n_url_by_clip = {}nfor f in _frames_all:n _url_by_clip.setdefault(f["clip_id"], f["url"])n# qwen3-vl-embedding 多模态批量上限为 10，按批写入n_BATCH = 8nfor _i in range(0, len(frames), _BATCH):n client.insert(collection_name, frames[_i:_i + _BATCH])nclient.flush(collection_name)nclient.load_collection(collection_name)nn# 写入完成后直接 query 出结构化结果验证ningested_rows = client.query(n collection_name, filter="",n output_fields=["frame_url", "clip_id", "ts_ms", "scene", "attributes", "entities"], limit=100)nfor row in ingested_rows:n print(row)nn# ===== 步骤 4｜以文搜帧 / 以图搜帧：文本或图片查询走同一多模态 embedding 召回 =====nquery = "雨天施工区，路面有锥形桶和施工牌"ntext_search = client.search(n collection_name=collection_name, data=[query], anns_field="embedding",n limit=10, output_fields=["frame_url", "scene", "attributes"])nfor hit in text_search[0]:n print(f"score={hit['distance']:.4f} scene={hit['entity']['scene']} url={hit['entity']['frame_url']}")nn# 向量召回 + 结构化过滤（Corner Case 挖掘）：先按 scene 与异常事件过滤，再做语义排序ncorner_query = "夜间隧道口，前车急刹"ncorner_search = client.search(n collection_name=collection_name, data=[corner_query], anns_field="embedding",n limit=10, filter='scene == "隧道" and attributes["anomaly_event"] != "none"',n output_fields=["frame_url", "scene", "attributes"])nn# 以图搜帧：用一张真实行车帧（dashcam 首帧）作为查询图nimage_query_url = _url_by_clip.get("clip_dashcam", frames[0]["frame_url"])nimage_search = client.search(n collection_name=collection_name,n data=[image_query_url],n anns_field="embedding", limit=10, output_fields=["frame_url", "scene"])nn# ===== 步骤 5｜AI_RERANK 多模态重排（可选）=====n# 方式一：search 挂 ranker，向量召回 Top-N 后由 qwen3-vl-rerank 精排nQUERY = "高速匝道处的加塞行为"nreranker = Function(n name="rerank_frames", function_type=FunctionType.RERANK, input_field_names=["frame_url"],n params={"reranker": "model", "provider": "aliyun_milvus", "model_name": "qwen3-vl-rerank",n "queries": [QUERY], "is_multimodal": "true",n "instruct": "Rank candidate frames by relevance to the query, prioritizing subject, action, scene and fine-grained visual details.",n "timeout_sec": 10})nrerank_search = client.search(n collection_name=collection_name, data=[QUERY], anns_field="embedding",n limit=20, output_fields=["frame_url", "scene"], ranker=reranker)nfor hit in rerank_search[0]:n print(f"score={hit['distance']:.4f} url={hit['entity']['frame_url']}")nn# 方式二：只对已召回的一批帧 URL 列表做独立重排，走 REST /v2/vectordb/ai/reranknrest_docs = [f["frame_url"] for f in frames[:3]]nstatus, data = post_json("/v2/vectordb/ai/rerank", {n "model_name": "qwen3-vl-rerank", "query": "高速匝道处的加塞行为",n "documents": rest_docs,n "params": {"is_multimodal": True,n "instruct": "Rank candidate frames by relevance to the query, prioritizing subject, action, scene and fine-grained visual details.",n "timeout_sec": 10}})nrest_rerank = []nif status == 200 and data.get("code") == 0:n for item in sorted(data["data"]["output"]["results"], key=lambda x: x["relevance_score"], reverse=True):n print(item["index"], item["relevance_score"])n rest_rerank.append({"url": rest_docs[item["index"]], "score": item["relevance_score"]})nn# ===== 步骤 6｜生成可视化 HTML：把图/视频和检索结果直接看到 =====nimport html as _htmlnimport webbrowsernfrom pathlib import PathnnOUTPUT_HTML = Path(__file__).with_name("test3_report.html")nVIDEO_EXTS = (".mp4", ".webm", ".mov", ".m3u8")nnndef media_tag(url: str) -> str:n safe = _html.escape(url, quote=True)n lower = url.split("?", 1)[0].lower()n if lower.endswith(VIDEO_EXTS):n return f''n return f''nnndef fmt_attrs(attributes: Any) -> str:n if not isinstance(attributes, dict):n return ""n parts = [f"{k}={v}" for k, v in attributes.items() if v not in (None, "none", "", [])]n return " · ".join(parts)nnndef fmt_entities(entities: Any) -> str:n if isinstance(entities, dict):n entities = entities.get("entities", entities)n if not entities:n return ""n return _html.escape(json.dumps(entities, ensure_ascii=False))nnndef card(url: str, badges: list[str], meta: str = "") -> str:n chips = "".join(f'{_html.escape(b)}' for b in badges if b)n meta_html = f'{_html.escape(meta)}
' if meta else ""n short = url.split("/")[-1].split("?")[0]n return (n ''n f'
{media_tag(url)}
'n f'{chips}
'n f'{meta_html}'n f'{_html.escape(short)}'n ''n )nnnsections: list[str] = []nn# 入库帧ncards = []nfor r in ingested_rows:n badges = [f"场景 {r.get('scene')}", f"{r.get('clip_id')} @ {r.get('ts_ms')}ms"]n ent = fmt_entities(r.get("entities"))n meta = fmt_attrs(r.get("attributes"))n if ent:n meta = (meta + " · " if meta else "") + "实体 " + entn cards.append(card(r["frame_url"], badges, meta))nsections.append(n f'步骤 3 · 截帧入库（写入即推理）
'n f'写入时由 4 个 AI Function 自动生成向量、场景分类、结构化属性与命名实体。共 {len(ingested_rows)} 帧。
'n f'{"".join(cards)}
')nnndef search_section(title: str, desc: str, hits, score_label: str = "相似度") -> str:n cards = []n for rank, hit in enumerate(hits, 1):n e = hit["entity"]n badges = [f"#{rank}", f"{score_label} {hit['distance']:.4f}"]n if e.get("scene"):n badges.append(f"场景 {e['scene']}")n cards.append(card(e["frame_url"], badges, fmt_attrs(e.get("attributes"))))n body = f'{"".join(cards)}
' if cards else '无命中结果
'n return f'{title}
{_html.escape(desc)}
{body}'nnnsections.append(search_section(n "步骤 4 · 以文搜帧", f'查询："{query}"', text_search[0]))nsections.append(search_section(n "步骤 4 · Corner Case 过滤检索",n f'查询："{corner_query}"，过滤：scene=="隧道" 且 anomaly_event!="none"', corner_search[0]))nn# 以图搜帧：先展示查询图，再展示结果nimg_cards = [card(image_query_url, ["查询图"], "以图搜帧的输入")]nfor rank, hit in enumerate(image_search[0], 1):n e = hit["entity"]n img_cards.append(card(e["frame_url"], [f"#{rank}", f"相似度 {hit['distance']:.4f}", f"场景 {e.get('scene')}"]))nsections.append(n '步骤 4 · 以图搜帧
'n '用一张参考图 URL 作为查询，走同一多模态 embedding 召回。
'n f'{"".join(img_cards)}
')nnsections.append(search_section(n "步骤 5 · AI_RERANK 精排（search 内挂 ranker）",n f'查询："{QUERY}"，向量召回后由 qwen3-vl-rerank 精排', rerank_search[0], score_label="重排分"))nn# REST 即时重排nrest_cards = [n card(item["url"], [f"#{rank}", f"相关度 {item['score']:.4f}"])n for rank, item in enumerate(rest_rerank, 1)n]nrest_body = f'{"".join(rest_cards)}
' if rest_cards else 'REST 重排未返回结果
'nsections.append(n '步骤 5 · REST 即时重排 /v2/vectordb/ai/rerank
'n f'对给定的一批帧 URL 独立重排，查询："{_html.escape(QUERY)}"
{rest_body}')nnpage = f"""nnntest3 多模态截帧检索 · 可视化报告nn:root{{color-scheme:dark}}n*{{box-sizing:border-box}}nbody{{margin:0;font-family:-apple-system,BlinkMacSystemFont,"Segoe UI","PingFang SC",sans-serif;background:#0f1115;color:#e6e8eb}}nheader{{padding:28px 32px;border-bottom:1px solid #23262d;background:#151821}}nheader h1{{margin:0 0 6px;font-size:20px}}nheader p{{margin:0;color:#8b929e;font-size:13px}}nsection{{padding:24px 32px;border-bottom:1px solid #1c1f27}}nsection h2{{margin:0 0 4px;font-size:16px;color:#f2f4f7}}n.desc{{margin:0 0 16px;color:#8b929e;font-size:13px}}n.empty{{color:#6b7280;font-size:13px}}n.grid{{display:grid;grid-template-columns:repeat(auto-fill,minmax(220px,1fr));gap:16px}}n.card{{background:#171a21;border:1px solid #262a33;border-radius:10px;overflow:hidden;display:flex;flex-direction:column}}n.thumb{{aspect-ratio:4/3;background:#0b0d11;display:flex;align-items:center;justify-content:center;overflow:hidden}}n.thumb img,.thumb video{{width:100%;height:100%;object-fit:cover;display:block}}n.chips{{display:flex;flex-wrap:wrap;gap:6px;padding:10px 10px 4px}}n.chip{{font-size:11px;background:#232735;color:#a9c7ff;border:1px solid #2f3b5a;padding:2px 8px;border-radius:999px}}n.meta{{padding:2px 10px 6px;color:#c4cad3;font-size:12px;line-height:1.5}}n.url{{padding:6px 10px 12px;color:#5b7fff;font-size:11px;text-decoration:none;word-break:break-all}}n.url:hover{{text-decoration:underline}}nnnn{"".join(sections)}n"""nnOUTPUT_HTML.write_text(page, encoding="utf-8")nprint(f"n✅ 可视化报告已生成：{OUTPUT_HTML}")ntry:n webbrowser.open(OUTPUT_HTML.as_uri())nexcept Exception:n pass","id":"OJFbO"}">实战效果
1）以文搜帧
查询 "雨天施工区，路面有锥形桶和施工牌"，Top-3 召回（示意）：
一句自然语言即可召回相关帧：
2）图搜帧
把参考图 URL 作为 query，参考图自身相似度 = 1.000，验证图像编码一致性：
3）rerank精排
经 qwen3-vl-rerank 精排后，天气与场景同时匹配的第 1 帧被进一步拉开与其余候选的差距，误召回的"普通道路"帧被压到末位。
自建方案 vs AI Function 对比
维度 | 传统自建方案 | Milvus AI Function |
开发周期 | 抽帧+推理+向量库+元数据库多系统联调，数周起 | 一次建表挂函数，insert/search 即用，天级 |
推理运维 | 自建 GPU 推理集群，需扩缩容/故障恢复 | 由 Milvus 托管调用，业务侧零运维 |
数据流转 | 帧在对象存储/推理集群/向量库间多次搬运 | 写入即推理，数据不出 Milvus 实例 |
合规风险 | 中间向量/原始帧易出库，评审难过 | 凭据由管理员配置，请求体不含凭据，数据不出库 |
多模态检索 | 需自建对齐文本-图像向量空间 | qwen3-vl-embedding 原生支持以文/以图搜帧
|
标注口径 | 人工标注慢、贵、口径不一 | prompt 统一判定规则，标签口径一致可复现
|
","margin":true,"hideBorder":false,"id":"g0yJp"}">总结
回到最初的痛点：智能驾驶截帧分析难，难在要把"推理、存储、检索"三件事分别用三套系统实现，还要保证敏感的行车数据在它们之间安全流转。阿里云 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”。