Vector Retrieval Service for Milvus:Transkripsi Audio

Format perintah

{
  "model_name": "qwen3-asr-flash",
  "texts": ["<audio_url>"],
  "params": {
    "language": "zh",
    "enable_itn": true,
    "max_concurrency": 1
  }
}

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",
        },
    )
)

Parameter

Nilai kembalian

data.output.outputs mengembalikan teks transkripsi yang sesuai dengan setiap input audio secara berurutan. usage.audio_tokens menunjukkan jumlah token audio yang digunakan, dan usage.seconds menunjukkan durasi audio yang diproses.

{
  "code": 0,
  "data": {
    "output": {"outputs": ["<transcript>"]},
    "usage": {"audio_tokens": 256, "total_tokens": 256, "seconds": 45}
  }
}

Contoh: Transkripsi rekaman layanan pelanggan

Contoh berikut mengonversi pesan sambutan hotline menjadi teks untuk pengarsipan versi dan tinjauan manual apakah jam layanan, pemberitahuan rekaman, dan konten lainnya dinyatakan secara jelas. Bidang yang tidak muncul dalam transkripsi harus ditandai sebagai "unconfirmed" dan tidak boleh dilengkapi oleh model.

welcome.mp3 publik default adalah pesan sambutan dalam bahasa Tiongkok; contoh ini menggunakan language=zh. Saat mengganti AIFUNC_AUDIO_URL, perbarui juga AIFUNC_AUDIO_LANGUAGE ke bahasa sumber aktual. Contoh cURL memerlukan jq.

#!/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"
fi

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
    # Alibaba Cloud Milvus provides TEXTTRANSFORM as a hosted extension (function type value 9);
    # some pymilvus versions do not yet have this enum member built in, while Function(...) validates via 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]}],
)

Hasil yang diharapkan: data.output.outputs[0] adalah transkripsi pesan sambutan non-kosong dengan panjang tidak lebih dari 10.000 karakter, yang dapat ditulis ke bidang transcript arsip versi hotline (hasil pengujian aktual: 欢迎与使用阿里云。).

{"code": 0, "data": {"output": {"outputs": ["欢迎与使用阿里云。"]}}}

Untuk skenario Python, Anda dapat memberi nama bidang input audio_url dan bidang output transcript. Fungsi tersebut melakukan transkripsi secara otomatis saat insert.