向量化 - 向量检索服务 Milvus 版

AI_EMBEDDING 函数可将文本、图片及视频转换为稠密向量，支持语义检索、RAG 及聚类。通过 params.dim指定维度，或配置 FunctionType.TEXT_EMBEDDING 实现写入自动向量化；多模态数据需设置 is_multimodal。

命令格式

REST 接口

POST /v2/vectordb/ai/embedding
Content-Type: application/json

{
  "model_name": "<模型名称>",
  "texts": ["<文本或媒体引用>"],
  "params": {"dim": 1024}
}

Python

schema = MilvusClient.create_schema(auto_id=True, enable_dynamic_field=False)
schema.add_field("id", DataType.INT64, is_primary=True)
schema.add_field("content", DataType.VARCHAR, max_length=4096)
schema.add_field("embedding", DataType.FLOAT_VECTOR, dim=1024)
schema.add_function(
    Function(
        name="embed_content",
        function_type=FunctionType.TEXTEMBEDDING,
        input_field_names=["content"],
        output_field_names=["embedding"],
        params={
            "provider": "aliyun_milvus",
            "model_name": "<模型名称>",
            "dim": 1024,
        },
    )
)

参数说明

参数	说明
`model_name`	必填。模型名称；文本可用 `text-embedding-v4`、`text-embedding-v3`，图片或视频可用 `qwen3-vl-embedding`、`tongyi-embedding-vision-plus`。
`texts`	REST 必填。非空字符串数组；多模态模型可传文本、图片 Base64、图片 URL 或视频 URL。
`dim`	可选。输出向量维度；`0` 表示使用模型默认维度。Collection 场景必须与输出向量字段维度一致。
`is_multimodal`	可选。图片或视频向量化时设为 `true`；文本模型默认 `false`。
`max_client_batch_size`	可选。单次发送给模型的最大输入数，必须大于 `0`。建议与所用模型在百炼向量化文档中标明的批量大小一致。
`max_concurrency`	可选。并发数，范围 `1`~`64`，默认 `1`。
`provider`	仅 Collection Function 必填，固定为 `aliyun_milvus`。凭据和服务地址由管理员在 `milvus.yaml` 配置，不要写入请求体。

性能建议：为获得较好的写入性能，请先查看所用模型在百炼向量化官方文档中标明的批量大小；每次调用 insert 时传入的数组大小应与该批量大小一致（最后一批不足一批时除外）。批量大小因模型而异。

返回值说明

成功时，data.output.embeddings 返回与输入对应的向量列表。每项的 text_index 是原输入下标，embedding 是向量；data.usage 记录本次模型用量。

{
  "code": 0,
  "data": {
    "output": {
      "embeddings": [
        {"text_index": 0, "embedding": [0.012, -0.084, "..."]}
      ]
    },
    "usage": {"total_tokens": 22}
  }
}

示例一：知识库文章入库与语义召回（文本）

企业知识库需要将文章正文和向量一起写入 Collection，后续根据用户问题召回相关文章。输入是两段文章正文，关键参数为 model_name=text-embedding-v4 与 dim=1024。

REST 接口

#!/usr/bin/env bash
set -euo pipefail

MILVUS_REST_BASE_URL="http://c-xxxx.milvus.aliyuncs.com:19530"
MILVUS_AUTH_TOKEN="<yourUsername>:<yourPassword>"

post_json() {
  local path="$1"
  local body="$2"
  curl -X POST "$MILVUS_REST_BASE_URL$path" \
    -H "Authorization: Bearer $MILVUS_AUTH_TOKEN" \
    -H "Content-Type: application/json" \
    -d "$body"
}

BODY=$(cat <<'JSON'
{
  "model_name": "text-embedding-v4",
  "texts": [
    "Milvus Serverless 适合构建企业知识库。",
    "AI Function 可以在写入时自动调用模型。"
  ],
  "params": {
    "dim": 1024,
    "max_client_batch_size": 2,
    "max_concurrency": 1
  }
}
JSON
)

post_json "/v2/vectordb/ai/embedding" "$BODY"

Python

from __future__ import annotations
from typing import Any

from pymilvus import DataType, Function, FunctionType, MilvusClient

MILVUS_URI = "http://c-xxxx.milvus.aliyuncs.com:19530"
MILVUS_TOKEN = "<yourUsername>:<yourPassword>"

def add_id(schema: Any) -> None:
    schema.add_field("id", DataType.INT64, is_primary=True)

client = MilvusClient(uri=MILVUS_URI, token=MILVUS_TOKEN)

collection_name = "simple_ai_embedding"
if client.has_collection(collection_name):
    client.drop_collection(collection_name)

schema = MilvusClient.create_schema(auto_id=True, enable_dynamic_field=False)
add_id(schema)
schema.add_field("content", DataType.VARCHAR, max_length=4096)
schema.add_field("embedding", DataType.FLOAT_VECTOR, dim=1024)
schema.add_function(
    Function(
        name="embed_content",
        function_type=FunctionType.TEXTEMBEDDING,
        input_field_names=["content"],
        output_field_names=["embedding"],
        params={
            "provider": "aliyun_milvus",
            "model_name": "text-embedding-v4",
            "dim": 1024,
            # 百炼官方文档中 text-embedding-v4 的批量大小为 10。
            "max_client_batch_size": 10,
            "max_concurrency": 1,
        },
    )
)

index_params = client.prepare_index_params()
index_params.add_index(
    field_name="embedding",
    index_type="HNSW",
    metric_type="COSINE",
    params={"M": 16, "efConstruction": 200},
)
client.create_collection(
    collection_name=collection_name,
    schema=schema,
    index_params=index_params,
)

client.insert(
    collection_name,
    [
        {"content": "Milvus Serverless 适合构建企业知识库。"},
        {"content": "AI Function 可以在写入时自动调用模型。"},
        {"content": "关系型数据库适合处理事务数据。"},
        {"content": "向量检索可以按语义相似度召回内容。"},
        {"content": "RAG 通过检索外部知识辅助模型生成答案。"},
        {"content": "语义搜索能够识别不同表述中的相近含义。"},
        {"content": "文档切分后可以分别生成向量。"},
        {"content": "检索结果可以使用重排模型进一步优化。"},
        {"content": "向量索引可以提高大规模数据的查询效率。"},
        {"content": "知识库应定期更新已入库的业务内容。"},
    ],
)
client.flush(collection_name)
client.load_collection(collection_name)

result = client.search(
    collection_name=collection_name,
    data=["怎么构建知识库问答？"],
    anns_field="embedding",
    limit=2,
    output_fields=["content"],
)
for hit in result[0]:
    print(f"score={hit['distance']:.4f} content={hit['entity']['content']}")

返回两个 embeddings 项，text_index 分别为 0 和 1。业务侧将每项向量与对应文章一起写入 Collection；搜索时使用同一模型和维度生成查询向量。

示例二：商品图片向量化（图片）

商品图片库需要将每张主图转换为向量，以支持“搜索蓝色外套”等图文检索。输入是模型可访问的图片 URL，关键参数是 model_name=qwen3-vl-embedding、is_multimodal=true 和 dim=2560。

REST 接口

#!/usr/bin/env bash
set -euo pipefail

MILVUS_REST_BASE_URL="http://c-xxxx.milvus.aliyuncs.com:19530"
MILVUS_AUTH_TOKEN="<yourUsername>:<yourPassword>"

post_json() {
  local path="$1"
  local body="$2"
  curl -X POST "$MILVUS_REST_BASE_URL$path" \
    -H "Authorization: Bearer $MILVUS_AUTH_TOKEN" \
    -H "Content-Type: application/json" \
    -d "$body"
}

MODEL_NAME="qwen3-vl-embedding"
IMAGE_URL="${IMAGE_URL:-https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20260415/hynnff/wan-video-edit-clothes.webp}"

BODY=$(cat <<JSON
{
  "model_name": "$MODEL_NAME",
  "texts": ["$IMAGE_URL"],
  "params": {"is_multimodal": true}
}
JSON
)

RESPONSE_BODY="$(post_json "/v2/vectordb/ai/embedding" "$BODY")"
if command -v jq >/dev/null 2>&1; then
  echo "$RESPONSE_BODY" | jq .
  [ "$(echo "$RESPONSE_BODY" | jq -r '.code // -1')" = "0" ] || exit 1
else
  echo "$RESPONSE_BODY"
fi

Python

from __future__ import annotations
import os
from typing import Any

from pymilvus import DataType, Function, FunctionType, MilvusClient

MILVUS_URI = "http://c-xxxx.milvus.aliyuncs.com:19530"
MILVUS_TOKEN = "<yourUsername>:<yourPassword>"

def add_id(schema: Any) -> None:
    schema.add_field("id", DataType.INT64, is_primary=True)

MODEL_NAME = "qwen3-vl-embedding"
VECTOR_DIM = int(os.getenv("AIFUNC_EMBEDDING_DIM", "2560"))
IMAGE_URL = os.getenv("IMAGE_URL", "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20260415/hynnff/wan-video-edit-clothes.webp")
client = MilvusClient(uri=MILVUS_URI, token=MILVUS_TOKEN)

collection_name = "simple_ai_embedding_image"
if client.has_collection(collection_name):
    client.drop_collection(collection_name)

schema = MilvusClient.create_schema(auto_id=True, enable_dynamic_field=False)
add_id(schema)
schema.add_field("image_url", DataType.VARCHAR, max_length=4096)
schema.add_field("embedding", DataType.FLOAT_VECTOR, dim=VECTOR_DIM)
schema.add_function(
    Function(
        name="embed_media",
        function_type=FunctionType.TEXTEMBEDDING,
        input_field_names=["image_url"],
        output_field_names=["embedding"],
        params={"provider": "aliyun_milvus", "model_name": MODEL_NAME, "dim": VECTOR_DIM, "is_multimodal": "true"},
    )
)
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)
client.insert(collection_name, [{"image_url": IMAGE_URL}])
client.flush(collection_name)
client.load_collection(collection_name)
for hit in client.search(collection_name=collection_name, data=[IMAGE_URL], anns_field="embedding", limit=1, output_fields=["image_url"])[0]:
    print(f"score={hit['distance']:.4f} image_url={hit['entity']['image_url']}")

返回 text_index=0 的 2560 维图片向量。将该向量写入商品向量字段后，可用同一多模态模型将文字查询或图片查询转换为向量并执行检索。

示例三：视频素材向量化（视频）

视频素材库需要为短视频生成向量，供运营人员按语义检索。输入是可被模型访问的视频 URL；模型与图片场景相同，资源类型由视频 URL 识别。

REST 接口

#!/usr/bin/env bash
set -euo pipefail

MILVUS_REST_BASE_URL="http://c-xxxx.milvus.aliyuncs.com:19530"
MILVUS_AUTH_TOKEN="<yourUsername>:<yourPassword>"

post_json() {
  local path="$1"
  local body="$2"
  curl -X POST "$MILVUS_REST_BASE_URL$path" \
    -H "Authorization: Bearer $MILVUS_AUTH_TOKEN" \
    -H "Content-Type: application/json" \
    -d "$body"
}

MODEL_NAME="qwen3-vl-embedding"
VIDEO_URL="${VIDEO_URL:-https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20260409/dozxak/Wan_Video_Edit_33_1.mp4}"

BODY=$(cat <<JSON
{
  "model_name": "$MODEL_NAME",
  "texts": ["$VIDEO_URL"],
  "params": {"is_multimodal": true}
}
JSON
)

RESPONSE_BODY="$(post_json "/v2/vectordb/ai/embedding" "$BODY")"
if command -v jq >/dev/null 2>&1; then
  echo "$RESPONSE_BODY" | jq .
  [ "$(echo "$RESPONSE_BODY" | jq -r '.code // -1')" = "0" ] || exit 1
else
  echo "$RESPONSE_BODY"
fi

Python

from __future__ import annotations
import os
from typing import Any

from pymilvus import DataType, Function, FunctionType, MilvusClient

MILVUS_URI = "http://c-xxxx.milvus.aliyuncs.com:19530"
MILVUS_TOKEN = "<yourUsername>:<yourPassword>"

def add_id(schema: Any) -> None:
    schema.add_field("id", DataType.INT64, is_primary=True)

MODEL_NAME = "qwen3-vl-embedding"
VECTOR_DIM = int(os.getenv("AIFUNC_EMBEDDING_DIM", "2560"))
VIDEO_URL = os.getenv("VIDEO_URL", "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20260409/dozxak/Wan_Video_Edit_33_1.mp4")
client = MilvusClient(uri=MILVUS_URI, token=MILVUS_TOKEN)

collection_name = "simple_ai_embedding_video"
if client.has_collection(collection_name):
    client.drop_collection(collection_name)

schema = MilvusClient.create_schema(auto_id=True, enable_dynamic_field=False)
add_id(schema)
schema.add_field("video_url", DataType.VARCHAR, max_length=4096)
schema.add_field("embedding", DataType.FLOAT_VECTOR, dim=VECTOR_DIM)
schema.add_function(
    Function(
        name="embed_media",
        function_type=FunctionType.TEXTEMBEDDING,
        input_field_names=["video_url"],
        output_field_names=["embedding"],
        params={"provider": "aliyun_milvus", "model_name": MODEL_NAME, "dim": VECTOR_DIM, "is_multimodal": "true"},
    )
)
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)
client.insert(collection_name, [{"video_url": VIDEO_URL}])
client.flush(collection_name)
client.load_collection(collection_name)
for hit in client.search(collection_name=collection_name, data=[VIDEO_URL], anns_field="embedding", limit=1, output_fields=["video_url"])[0]:
    print(f"score={hit['distance']:.4f} video_url={hit['entity']['video_url']}")

返回视频的向量结果。业务侧将向量与 video_url 一起保存，即可将视频检索接入内容运营或素材推荐流程。