向量检索服务 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": ["您好，我的手机号是 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": "您好，我的手机号是 13812345678，麻烦帮我查下工单。"}])
client.flush(collection_name)
for row in client.query(collection_name, filter="", output_fields=["masked"], limit=1):
    print(row["masked"])

preserve_length 为 true 时掩码后保持原字符长度，便于在不暴露原值的前提下保留格式特征。实测输出：

您好，我的手机号是 ***********，身份证 ******************，麻烦帮我查下工单。

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 各自填充，写入一条数据即同时得到情感标签与工单类目。