向量化 - Vector Retrieval Service for 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 必填。非Null 字元串數組；多模態模型可傳文本、圖片 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 一起儲存，即可將視頻檢索接入內容營運或素材推薦流程。