情感分析 - Vector Retrieval Service for Milvus

AI_SENTIMENT 函數用於從文本、圖片或視頻中選擇最匹配的情感類別，適用於使用者評價分析、宣傳素材標註和內容營運看板等情境。

命令格式

REST 介面

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

{
  "model_name": "<模型名>",
  "texts": ["<文本或媒體 URL>"],
  "params": {"categories": ["positive", "negative", "neutral"]}
}

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("sentiment", DataType.VARCHAR, max_length=64)
schema.add_field("dummy_vector", DataType.FLOAT_VECTOR, dim=2)
schema.add_function(
    Function(
        name="analyze_sentiment",
        function_type=texttransform_function_type(),
        input_field_names=["content"],
        output_field_names=["sentiment"],
        params={
            "provider": "aliyun_milvus",
            "model_name": "<模型名>",
            "task": "ai_sentiment",
            "categories": "positive,negative,neutral",
            "temperature": "0",
        },
    )
)

參數說明

參數	說明
`model_name`	必填。模型名稱；圖片/視頻須使用已配置多模態模型（如 `qwen3.7-plus`）。
`texts`	REST 必填。待分析內容，或可被模型訪問的圖片/視頻 URL。
`categories`	選填。情感類別，支援數組或逗號分隔字串，數量 1～10。預設 `positive`、`negative`、`neutral`、`mixed`。
`prompt`	選填。補充判斷規則，最大 5000 字元，不支援 `${...}`。
`media_type`	選填。取值 `image` 或 `video`。
`temperature` / `max_concurrency` / `timeout_sec`	選填。模型穩定性、並發和逾時設定。
`provider` / `task`	僅 Collection Function 必填，固定為 `aliyun_milvus` 與 `ai_sentiment`。

傳回值說明

data.output.outputs 返回與輸入同序的情感類別。每個非空結果應是請求 categories 中的一項；無法判斷時可返回 null。Schema 會將結果寫入目標文字欄位。

樣本一：產生售後評價情感看板（文本）

售後團隊需要將評價歸入指定類別再彙總到看板。每項結果要麼屬於指定類別，要麼為合法的 null。

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": "qwen3.7-max",
  "texts": [
    "The food was excellent and the service was thoughtful.",
    "The product arrived late and the package was damaged."
  ],
  "params": {"temperature": 0}
}
JSON
)
RESPONSE_BODY="$(post_json "/v2/vectordb/ai/sentiment" "$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

from typing import Any

from pymilvus import DataType, Function, FunctionType, MilvusClient

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

DUMMY_VECTOR_DIM = 2
TEXTTRANSFORM_FUNCTION_TYPE = 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
    # 阿里雲 Milvus 將 TEXTTRANSFORM 作為託管擴充(函數類型值 9)提供;
    # 部分 pymilvus 版本尚未內建該枚舉成員,而 Function(...) 通過 FunctionType(...) 校正。
    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

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

def add_dummy_vector(schema: Any) -> None:
    schema.add_field("dummy_vector", DataType.FLOAT_VECTOR, dim=DUMMY_VECTOR_DIM)

def run_texttransform_example(*, client, collection_name, input_fields, output_field, function_name, function_params, rows) -> None:
    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)
    for name, data_type, max_length in input_fields:
        field_params = {"max_length": max_length} if max_length is not None else {}
        schema.add_field(name, data_type, **field_params)
    output_name, output_data_type, output_max_length = output_field
    output_params = {"max_length": output_max_length} if output_max_length is not None else {}
    schema.add_field(output_name, output_data_type, **output_params)
    add_dummy_vector(schema)
    schema.add_function(
        Function(
            name=function_name,
            function_type=texttransform_function_type(),
            input_field_names=[name for name, _, _ in input_fields],
            output_field_names=[output_name],
            params=function_params,
        )
    )
    index_params = client.prepare_index_params()
    index_params.add_index(field_name="dummy_vector", index_type="AUTOINDEX", metric_type="COSINE")
    client.create_collection(collection_name=collection_name, schema=schema, index_params=index_params)
    client.insert(collection_name, rows)
    client.flush(collection_name)
    fields = [name for name, _, _ in input_fields] + [output_name]
    for row in client.query(collection_name, filter="", output_fields=fields, limit=len(rows)):
        print(row)

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

run_texttransform_example(
    client=client,
    collection_name="simple_ai_sentiment_text",
    input_fields=[("content", DataType.VARCHAR, 4096)],
    output_field=("sentiment", DataType.VARCHAR, 64),
    function_name="analyze_sentiment",
    function_params={"provider": "aliyun_milvus", "model_name": "qwen3.7-max", "task": "ai_sentiment", "categories": "positive,negative,neutral", "temperature": "0"},
    rows=[{"content": "The food was excellent and the service was thoughtful.", "dummy_vector": [0.1, 0.2]}],
)

預期結果：保持 2 項結果及輸入順序；每項為聲明類別之一或 null。業務側僅彙總非空類別，null 計入未判定並進入人工複核。

樣本二：標註服飾宣傳圖的情感傾向（圖片）

營運先讓模型給出三類情感標籤，再結合品牌調性人工複核；不假設一定屬於某一類，也接受返回 null。

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":"qwen3.7-plus","texts":["https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20260415/hynnff/wan-video-edit-clothes.webp"],"params":{"media_type":"image","categories":["positive","negative","neutral"],"temperature":0}}
JSON
)
RESPONSE_BODY="$(post_json "/v2/vectordb/ai/sentiment" "$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

from typing import Any

from pymilvus import DataType, Function, FunctionType, MilvusClient

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

DUMMY_VECTOR_DIM = 2
TEXTTRANSFORM_FUNCTION_TYPE = 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
    # 阿里雲 Milvus 將 TEXTTRANSFORM 作為託管擴充(函數類型值 9)提供;
    # 部分 pymilvus 版本尚未內建該枚舉成員,而 Function(...) 通過 FunctionType(...) 校正。
    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

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

def add_dummy_vector(schema: Any) -> None:
    schema.add_field("dummy_vector", DataType.FLOAT_VECTOR, dim=DUMMY_VECTOR_DIM)

def run_texttransform_example(*, client, collection_name, input_fields, output_field, function_name, function_params, rows) -> None:
    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)
    for name, data_type, max_length in input_fields:
        field_params = {"max_length": max_length} if max_length is not None else {}
        schema.add_field(name, data_type, **field_params)
    output_name, output_data_type, output_max_length = output_field
    output_params = {"max_length": output_max_length} if output_max_length is not None else {}
    schema.add_field(output_name, output_data_type, **output_params)
    add_dummy_vector(schema)
    schema.add_function(
        Function(
            name=function_name,
            function_type=texttransform_function_type(),
            input_field_names=[name for name, _, _ in input_fields],
            output_field_names=[output_name],
            params=function_params,
        )
    )
    index_params = client.prepare_index_params()
    index_params.add_index(field_name="dummy_vector", index_type="AUTOINDEX", metric_type="COSINE")
    client.create_collection(collection_name=collection_name, schema=schema, index_params=index_params)
    client.insert(collection_name, rows)
    client.flush(collection_name)
    fields = [name for name, _, _ in input_fields] + [output_name]
    for row in client.query(collection_name, filter="", output_fields=fields, limit=len(rows)):
        print(row)

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

run_texttransform_example(
    client=client,
    collection_name="simple_ai_sentiment_image",
    input_fields=[("image_url", DataType.VARCHAR, 4096)],
    output_field=("sentiment", DataType.VARCHAR, 64),
    function_name="analyze_image_sentiment",
    function_params={"provider": "aliyun_milvus", "model_name": "qwen3.7-plus", "task": "ai_sentiment", "media_type": "image", "categories": "positive,negative,neutral", "temperature": "0"},
    rows=[{"image_url": "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20260415/hynnff/wan-video-edit-clothes.webp", "dummy_vector": [0.1, 0.2]}],
)

預期結果：結果為 positive、negative、neutral 之一或 null。情感標籤不是對圖中人物心理狀態的事實判定。

樣本三：標註擬人角色短視頻的情感傾向（視頻）

內容平台對短視頻做初始標註，供編輯檢索複核。只驗證結果屬於聲明類別或為 null。

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":"qwen3.7-plus","texts":["https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20260409/dozxak/Wan_Video_Edit_33_1.mp4"],"params":{"media_type":"video","categories":["positive","negative","neutral"],"temperature":0}}
JSON
)
RESPONSE_BODY="$(post_json "/v2/vectordb/ai/sentiment" "$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

from typing import Any

from pymilvus import DataType, Function, FunctionType, MilvusClient

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

DUMMY_VECTOR_DIM = 2
TEXTTRANSFORM_FUNCTION_TYPE = 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
    # 阿里雲 Milvus 將 TEXTTRANSFORM 作為託管擴充(函數類型值 9)提供;
    # 部分 pymilvus 版本尚未內建該枚舉成員,而 Function(...) 通過 FunctionType(...) 校正。
    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

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

def add_dummy_vector(schema: Any) -> None:
    schema.add_field("dummy_vector", DataType.FLOAT_VECTOR, dim=DUMMY_VECTOR_DIM)

def run_texttransform_example(*, client, collection_name, input_fields, output_field, function_name, function_params, rows) -> None:
    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)
    for name, data_type, max_length in input_fields:
        field_params = {"max_length": max_length} if max_length is not None else {}
        schema.add_field(name, data_type, **field_params)
    output_name, output_data_type, output_max_length = output_field
    output_params = {"max_length": output_max_length} if output_max_length is not None else {}
    schema.add_field(output_name, output_data_type, **output_params)
    add_dummy_vector(schema)
    schema.add_function(
        Function(
            name=function_name,
            function_type=texttransform_function_type(),
            input_field_names=[name for name, _, _ in input_fields],
            output_field_names=[output_name],
            params=function_params,
        )
    )
    index_params = client.prepare_index_params()
    index_params.add_index(field_name="dummy_vector", index_type="AUTOINDEX", metric_type="COSINE")
    client.create_collection(collection_name=collection_name, schema=schema, index_params=index_params)
    client.insert(collection_name, rows)
    client.flush(collection_name)
    fields = [name for name, _, _ in input_fields] + [output_name]
    for row in client.query(collection_name, filter="", output_fields=fields, limit=len(rows)):
        print(row)

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

run_texttransform_example(
    client=client,
    collection_name="simple_ai_sentiment_video",
    input_fields=[("video_url", DataType.VARCHAR, 4096)],
    output_field=("sentiment", DataType.VARCHAR, 64),
    function_name="analyze_video_sentiment",
    function_params={"provider": "aliyun_milvus", "model_name": "qwen3.7-plus", "task": "ai_sentiment", "media_type": "video", "categories": "positive,negative,neutral", "temperature": "0"},
    rows=[{"video_url": "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20260409/dozxak/Wan_Video_Edit_33_1.mp4", "dummy_vector": [0.1, 0.2]}],
)

預期結果：結果為三個聲明類別之一或 null。僅非空標籤進入彙總；高影響動作前仍應人工複核。