AI_AUDIO_TRANSCRIBE 函數可將音頻轉寫為文本，適用於客服錄音、會議紀要、語音信箱和播客內容索引等情境。
命令格式
REST 介面
{
"model_name": "qwen3-asr-flash",
"texts": ["<audio_url>"],
"params": {
"language": "zh",
"enable_itn": true,
"max_concurrency": 1
}
}Python
schema = MilvusClient.create_schema(auto_id=True, enable_dynamic_field=False)
schema.add_field("id", DataType.INT64, is_primary=True)
schema.add_field("audio_url", DataType.VARCHAR, max_length=4096)
schema.add_field("transcript", DataType.VARCHAR, max_length=4096)
schema.add_field("dummy_vector", DataType.FLOAT_VECTOR, dim=2)
schema.add_function(
Function(
name="transcribe_audio",
function_type=texttransform_function_type(),
input_field_names=["audio_url"],
output_field_names=["transcript"],
params={
"provider": "aliyun_milvus",
"model_name": "qwen3-asr-flash",
"task": "ai_audio_transcribe",
"language": "zh",
"enable_itn": "true",
},
)
)參數說明
|
參數 |
說明 |
|
|
必填，使用 |
|
|
REST 必填。音頻 URL 數組，不能包含Null 字元串；結果與輸入順序一致。 |
|
|
可選，源語言代碼，例如 |
|
|
可選，布爾值；用於將口語數字等正常化為書面表達。 |
|
|
可選，分別控制調用逾時和多音頻並發處理數。 |
|
|
可省略或設定為 |
|
其他 |
不支援 |
傳回值說明
data.output.outputs 返回與音頻輸入一一對應的轉寫文本；usage.audio_tokens 表示音頻用量，usage.seconds 表示已處理音頻時間長度。
{
"code": 0,
"data": {
"output": {"outputs": ["<transcript>"]},
"usage": {"audio_tokens": 256, "total_tokens": 256, "seconds": 45}
}
}樣本：客服熱線歡迎語產生可檢索版本檔案
客服平台將熱線歡迎語轉為文本，用於版本歸檔，以及人工複核服務時間、錄音告知等內容是否表達清晰；轉寫中未出現的欄位應標記為“未確認”，不能由模型補寫。預設公開 welcome.mp3 是中文歡迎語，樣本使用 language=zh；替換 AIFUNC_AUDIO_URL 時，應同步將 AIFUNC_AUDIO_LANGUAGE 設定為實際源語言。cURL 樣本需要 jq。
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-asr-flash"
AUDIO_URL="${AIFUNC_AUDIO_URL:-https://dashscope.oss-cn-beijing.aliyuncs.com/audios/welcome.mp3}"
BODY=$(cat <<JSON
{
"model_name": "$MODEL_NAME",
"texts": ["$AUDIO_URL"],
"params": {"language": "zh", "enable_itn": true, "max_concurrency": 1}
}
JSON
)
RESPONSE_BODY="$(post_json "/v2/vectordb/ai/audio_transcribe" "$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"
fiPython
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)
MODEL_NAME = "qwen3-asr-flash"
client = MilvusClient(uri=MILVUS_URI, token=MILVUS_TOKEN)
run_texttransform_example(
client=client,
collection_name="simple_ai_audio_transcribe_schema",
input_fields=[("audio_url", DataType.VARCHAR, 4096)],
output_field=("transcript", DataType.VARCHAR, 4096),
function_name="transcribe_audio",
function_params={"provider": "aliyun_milvus", "model_name": MODEL_NAME, "task": "ai_audio_transcribe", "language": "zh", "enable_itn": "true"},
rows=[{"audio_url": "https://dashscope.oss-cn-beijing.aliyuncs.com/audios/welcome.mp3", "dummy_vector": [0.1, 0.2]}],
)預期結果：data.output.outputs[0] 為非空且不超過 10000 字元的歡迎語轉寫文本，可寫入熱線版本檔案的 transcript 欄位（實測返回 歡迎與使用阿里雲。）。
{"code": 0, "data": {"output": {"outputs": ["歡迎與使用阿里雲。"]}}}Python 情境可將輸入欄位命名為 audio_url、輸出欄位命名為 transcript，由 Function 自動轉寫。
對自有錄音進行轉寫前，請確認已履行錄音告知並獲得必要授權。建議使用短時效、最小許可權的媒體 URL，並按客服資料保留原則限制原始音頻和轉寫文本的儲存時間。