Vector Retrieval Service for Milvus：通過阿里雲Milvus構建客服語音分析與FAQ智能匹配鏈路

準備公用代碼

以下程式碼封裝含串連配置、REST 調用封裝以及 TEXTTRANSFORM 函數類型的相容兜底，供後續步驟共用。請將 MILVUS_URI 與 MILVUS_TOKEN 替換為實際執行個體資訊。

from __future__ import annotations

import json
import time
from typing import Any
from urllib.error import HTTPError
from urllib.request import Request, urlopen

from pymilvus import DataType, Function, FunctionType, MilvusClient

# ==================== 串連配置 ====================
MILVUS_URI = "http://c-xxx.milvus.aliyuncs.com:19530"  # 連接埠必須寫 19530
MILVUS_TOKEN = "root:xxx"
MILVUS_REST_BASE_URL = MILVUS_URI

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

# AI_AUDIO_TRANSCRIBE / AI_PII_MASK / AI_SENTIMENT / AI_CLASSIFY 都屬於 TEXTTRANSFORM
# 類型的 Function（函數類型值 9），通過 task 參數區分具體任務。
# 部分 pymilvus 版本的 FunctionType 枚舉中沒有該成員，下面做一次相容兜底。
TEXTTRANSFORM_FUNCTION_TYPE = 9


def texttransform_function_type() -> Any:
    for type_name in ("TEXTTRANSFORM", "TEXT_TRANSFORM", "TextTransform"):
        ft = getattr(FunctionType, type_name, None)
        if ft is not None:
            return ft
    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 post_json(path: str, body: dict[str, Any], timeout: int = 120,
              retries: int = 3) -> tuple[int, dict[str, Any]]:
    """REST 即時介面統一封裝（走大模型，加重試更穩）。"""
    last: tuple[int, dict[str, Any]] | None = None
    for _ in range(retries):
        request = Request(
            f"{MILVUS_REST_BASE_URL.rstrip('/')}{path}",
            data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
            headers={"Authorization": f"Bearer {MILVUS_TOKEN}",
                     "Content-Type": "application/json"},
            method="POST",
        )
        try:
            with urlopen(request, timeout=timeout) as response:
                status, data = response.status, json.loads(response.read().decode("utf-8"))
        except HTTPError as exc:
            status, data = exc.code, json.loads(exc.read().decode("utf-8"))
        last = (status, data)
        if status == 200 and data.get("code") == 0:
            return status, data
        time.sleep(1)
    return last

步驟一：錄音轉寫

把通話錄音批量轉寫成文本。將 AI_AUDIO_TRANSCRIBE 掛在 Collection 上後，寫入音頻地址時自動產出轉寫文本，無需應用側先調用 ASR 服務。

# ==================== 步驟 1：AI_AUDIO_TRANSCRIBE 錄音轉寫 ====================
collection_name = "cs_transcribe"
if client.has_collection(collection_name):
    client.drop_collection(collection_name)

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)  # 轉寫輸出欄位
# Collection 必須至少包含一個向量欄位。本步驟不做向量檢索，用 2 維佔位欄位滿足約束；
# 聲明為 nullable=True 後，insert 時無需再傳該欄位。
schema.add_field("dummy_vector", DataType.FLOAT_VECTOR, dim=2, nullable=True)
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",     # 正常化口語數字，如「一三八」轉為 138
        },
    )
)

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)

# audio_url 必須是服務端可下載的真真實位址，生產環境用自有 OSS 的短時效簽名 URL
audio_urls = [
    "https://<your-bucket>.oss-cn-hangzhou.aliyuncs.com/calls/call_0001.mp3",
    "https://<your-bucket>.oss-cn-hangzhou.aliyuncs.com/calls/call_0002.wav",
]
client.insert(collection_name, [{"audio_url": u} for u in audio_urls])
client.flush(collection_name)

for row in client.query(collection_name, filter="",
                       output_fields=["audio_url", "transcript"], limit=10):
    print(f"{row['audio_url'].split('/')[-1]} -> {row['transcript']}")

enable_itn 開啟逆文本正常化，會把口語化的數字念法轉成規範寫法，便於後續檢索與結構化處理。

步驟二：PII 脫敏

轉寫文本常包含手機號、社會安全號碼、銀行卡號等個人敏感資訊。在文本進入知識庫或分析庫之前完成脫敏，把合規環節前置，而不是留到鏈路末端。提供兩種用法：REST 即時介面，以及寫入型 Collection。

# ==================== 步驟 2：AI_PII_MASK 脫敏 ====================
# 2.1 REST 即時介面：文本進庫前先脫敏
status, data = post_json(
    "/v2/vectordb/ai/pii_mask",
    {
        "model_name": "qwen3.7-max",
        "texts": ["您好，My Phone號是 13812345678，身份證 110101199003071234，麻煩幫我查下工單。"],
        "params": {
            "pii_types": ["PERSON", "PHONE", "ID_CARD"],
            "mask_char": "*",
            "preserve_length": True,   # 掩碼後保持原長度
            "temperature": 0,
        },
    },
)
assert status == 200 and data.get("code") == 0, data
for item in data["data"]["output"]["outputs"]:
    print(item)

# 2.2 寫入型 Collection：content 入庫時自動產生脫敏欄位 masked
collection_name = "cs_pii_mask"
if client.has_collection(collection_name):
    client.drop_collection(collection_name)

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("masked", DataType.VARCHAR, max_length=4096)
schema.add_field("dummy_vector", DataType.FLOAT_VECTOR, dim=2, nullable=True)
schema.add_function(
    Function(
        name="mask_pii",
        function_type=texttransform_function_type(),
        input_field_names=["content"],
        output_field_names=["masked"],
        params={
            "provider": "aliyun_milvus",
            "model_name": "qwen3.7-max",
            "task": "ai_pii_mask",
            "pii_types": "PERSON,PHONE,ID_CARD",
            "mask_char": "*",
            "preserve_length": "true",
            "temperature": "0",
        },
    )
)
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,
              [{"content": "您好，My Phone號是 13812345678，麻煩幫我查下工單。"}])
client.flush(collection_name)
for row in client.query(collection_name, filter="", output_fields=["masked"], limit=1):
    print(row["masked"])

preserve_length 為 true 時掩碼後保持原字元長度，便於在不暴露原值的前提下保留格式特徵。實測輸出：

您好，My Phone號是 ***********，身份證 ******************，麻煩幫我查下工單。

11 位手機號掩碼為 11 個星號，18 位身份證掩碼為 18 個星號。兩種用法的區別是：REST 即時介面適合在資料流入前做一次性處理，寫入型 Collection 適合讓脫敏結果與原文一併留存、便於審計。

步驟三：建 FAQ 知識庫

把 FAQ 問題與標準答案寫入 Collection，AI_EMBEDDING 在寫入時自動完成向量化，應用側無需先調用 embedding 模型。

# ==================== 步驟 3：AI_EMBEDDING 建 FAQ 知識庫 ====================
collection_name = "cs_faq_kb"
if client.has_collection(collection_name):
    client.drop_collection(collection_name)

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)   # FAQ 問題文本
schema.add_field("answer", 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,
            "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)

faqs = [
    {"content": "如何開啟 Serverless Milvus 執行個體的公網訪問？",
     "answer": "在控制台執行個體詳情頁開啟公網並配置白名單。"},
    {"content": "忘記控制台登入密碼怎麼辦？", "answer": "通過帳號中心的找回密碼流程重設。"},
    {"content": "賬單為什麼比預期高？", "answer": "查看用量明細，重點關注計算與儲存用量。"},
    {"content": "如何建立 Collection 並寫入向量？",
     "answer": "使用 create_collection 定義 schema 後 insert。"},
    {"content": "執行個體擴容會影響線上業務嗎？", "answer": "擴容為線上操作，通常不中斷服務。"},
]
client.insert(collection_name, faqs)
client.flush(collection_name)
client.load_collection(collection_name)

步驟四與步驟五：語義召回與大模型重排

使用者的口語化問題先經向量檢索召回 Top-N 候選，再由重排模型按與查詢的真實相關性精排。這兩步合起來才能穩定命中標準 FAQ。

# ==================== 步驟 4：語義檢索召回 FAQ ====================
query = "我的執行個體怎麼才能讓外網連上？"
result = client.search(
    collection_name="cs_faq_kb",
    data=[query],
    anns_field="embedding",
    limit=3,
    output_fields=["content", "answer"],
)
for rank, hit in enumerate(result[0], 1):
    print(f"{rank}. [相似性 {hit['distance']:.4f}] {hit['entity']['content']}")


# ==================== 步驟 5：AI_RERANK 重排 ====================
# 5.1 REST 即時介面：對給定的一批候選獨立重排
status, data = post_json(
    "/v2/vectordb/ai/rerank",
    {
        "model_name": "qwen3-rerank",
        "query": query,
        "documents": [
            "如何開啟 Serverless Milvus 執行個體的公網訪問？",
            "執行個體擴容會影響線上業務嗎？",
            "如何建立 Collection 並寫入向量？",
        ],
        "params": {"max_concurrency": 2, "timeout_sec": 10},
    },
)
assert status == 200 and data.get("code") == 0, data
ranked = sorted(data["data"]["output"]["results"],
                key=lambda x: x["relevance_score"], reverse=True)
for rank, item in enumerate(ranked, 1):
    print(f"{rank}. [相關度 {item['relevance_score']:.4f}] 候選 index={item['index']}")

# 5.2 在 search 時掛載 ranker，召回與重排一次完成
reranker = Function(
    name="rerank_faq",
    function_type=FunctionType.RERANK,
    input_field_names=["content"],
    params={
        "reranker": "model",
        "provider": "aliyun_milvus",
        "model_name": "qwen3-rerank",
        "queries": [query],
        "max_concurrency": 2,
        "timeout_sec": 10,
    },
)
result = client.search(
    collection_name="cs_faq_kb",
    data=[query],
    anns_field="embedding",
    limit=3,
    output_fields=["content", "answer"],
    ranker=reranker,
)
for rank, hit in enumerate(result[0], 1):
    e = hit["entity"]
    print(f"{rank}. [重排分 {hit['distance']:.4f}] {e['content']} ｜ 答：{e['answer']}")

以查詢「我的執行個體怎麼才能讓外網連上？」為例，實測資料如下。這組前後對比很能說明重排的價值：

向量召回把「忘記控制台登入密碼怎麼辦？」排到了第二位（0.5613），但它與「讓外網連上」在意圖上並不相關——這正是純向量檢索「召回但不精準」的典型表現：文本表面相似性不低，實際意圖相去甚遠。經重排後該條被壓到第三位（0.2607），語義更接近的「執行個體擴容」升到第二位。對客服機器人而言，Top-1 的準確性直接決定應答品質，因此重排環節不可省略。

步驟六：情感識別與工單分類

對脫敏後的會話文本同時掛載兩個 Function：一個判斷情感傾向用於質檢與輿情預警，一個歸類到工單類目用於自動分流。兩者都在寫入時自動完成。

# ==================== 步驟 6：AI_SENTIMENT 情感識別 + AI_CLASSIFY 工單分類 ====================
collection_name = "cs_qc"
if client.has_collection(collection_name):
    client.drop_collection(collection_name)

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("category", DataType.VARCHAR, max_length=64)      # 工單類目輸出
schema.add_field("dummy_vector", DataType.FLOAT_VECTOR, dim=2, nullable=True)
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": "qwen3.7-max",
                "task": "ai_sentiment", "categories": "positive,negative,neutral",
                "temperature": "0"},
    )
)
schema.add_function(
    Function(
        name="classify_ticket",
        function_type=texttransform_function_type(),
        input_field_names=["content"], output_field_names=["category"],
        params={"provider": "aliyun_milvus", "model_name": "qwen3.7-max",
                "task": "ai_classify", "labels": "帳號,諮詢,故障,計費",
                "prompt": "根據客戶問題所屬主題分類。", "temperature": "0"},
    )
)
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, [
    {"content": "你們這個問題我已經打了三次電話了，到現在還沒解決，太讓人失望了！"},
    {"content": "請問 Serverless Milvus 執行個體怎麼開啟公網訪問？"},
])
client.flush(collection_name)
for row in client.query(collection_name, filter="",
                        output_fields=["content", "sentiment", "category"], limit=10):
    print(f"情感={row['sentiment']:<10} 類目={row['category']:<6} ｜ {row['content'][:28]}")

實測結果：

一個 Collection 上可以掛載多個 Function，只要它們的輸出欄位不衝突。本例中 sentiment 與 category 由兩個 Function 各自填充，寫入一條資料即同時得到情感標籤與工單類目。