Vector Retrieval Service for Milvus:Construa um pipeline de análise de voz no atendimento ao cliente e correspondência inteligente de FAQ com o Alibaba Cloud Milvus

Prepare o código compartilhado

O código a seguir contém a configuração de conexão, o wrapper de chamada REST e um fallback de compatibilidade para o tipo de função TEXTTRANSFORM, que são compartilhados pelas etapas subsequentes. Substitua MILVUS_URI e MILVUS_TOKEN pelas informações reais da sua instância.

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

# ==================== Connection configuration ====================
MILVUS_URI = "http://c-xxx.milvus.aliyuncs.com:19530"  # The port must be written as 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 are all Functions of the
# TEXTTRANSFORM type (function type value 9), distinguished by the task parameter.
# Some pymilvus versions do not have this member in the FunctionType enum, so a compatibility fallback is provided below.
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]]:
    """Unified wrapper for the REST real-time APIs (these call large models, so retries make them more stable)."""
    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

Etapa 1: Transcreva as gravações

Transcreva gravações de chamadas em texto em massa. Após anexar o AI_AUDIO_TRANSCRIBE a uma coleção, o texto transcrito é gerado automaticamente quando você escreve os endereços de áudio, sem necessidade de chamar um service de ASR no lado da aplicação primeiro.

# ==================== Step 1: AI_AUDIO_TRANSCRIBE audio transcription ====================
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)   # Audio input field
schema.add_field("transcript", DataType.VARCHAR, max_length=4096)  # Transcription output field
# A collection must contain at least one vector field. This step does not perform vector retrieval,
# so a 2-dimensional placeholder field is used to satisfy the constraint;
# after it is declared as nullable=True, you no longer need to pass this field on 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",     # Normalizes spoken numbers, for example "one three eight" becomes 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 must be a real address that the server side can download from; in production, use a short-lived signed URL from your own OSS bucket
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']}")

O enable_itn ativa a normalização inversa de texto, que converte leituras coloquiais de números na forma escrita padrão, facilitando a recuperação e o processamento estruturado subsequentes.

Etapa 2: Mascaramento de PII

O texto transcrito geralmente contém informações pessoais sensíveis, como números de celular, números de documentos de identidade e números de cartão bancário. Conclua o mascaramento antes de o texto entrar na base de conhecimento ou no banco de dados de análise, antecipando a conformidade no pipeline em vez de deixá-la para o final. São fornecidos dois usos: a API REST em tempo real e uma coleção no momento da escrita.

# ==================== Step 2: AI_PII_MASK masking ====================
# 2.1 REST real-time API: mask text before it is ingested
status, data = post_json(
    "/v2/vectordb/ai/pii_mask",
    {
        "model_name": "qwen3.7-max",
        "texts": ["Hello, my mobile phone number is 13812345678 and my ID card number is 110101199003071234. Please help me check my ticket."],
        "params": {
            "pii_types": ["PERSON", "PHONE", "ID_CARD"],
            "mask_char": "*",
            "preserve_length": True,   # Keep the original length after masking
            "temperature": 0,
        },
    },
)
assert status == 200 and data.get("code") == 0, data
for item in data["data"]["output"]["outputs"]:
    print(item)

# 2.2 Write-time collection: the masked field is automatically generated when content is ingested
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": "Hello, my mobile phone number is 13812345678. Please help me check my ticket."}])
client.flush(collection_name)
for row in client.query(collection_name, filter="", output_fields=["masked"], limit=1):
    print(row["masked"])

Quando o preserve_length é true, o comprimento original de caracteres é preservado após o mascaramento, o que retém as características de formato sem expor os valores originais. Saída real:

Hello, my mobile phone number is *********** and my ID card number is ******************. Please help me check my ticket.

O número de celular de 11 dígitos é mascarado com 11 asteriscos, e o número do documento de identidade de 18 dígitos é mascarado com 18 asteriscos. A diferença entre os dois usos é: a API REST em tempo real é adequada para processamento único antes do fluxo de entrada de dados, enquanto a coleção no momento da escrita é ideal para reter o resultado mascarado junto com o texto original para auditoria.

Etapa 3: Construa a base de conhecimento de FAQ

Escreva as perguntas de FAQ e as respostas padrão em uma coleção. O AI_EMBEDDING executa a vetorização automaticamente na escrita, portanto, o lado da aplicação não precisa chamar um modelo de embedding primeiro.

# ==================== Step 3: AI_EMBEDDING builds the FAQ knowledge base ====================
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 question text
schema.add_field("answer", DataType.VARCHAR, max_length=4096)    # Standard answer
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": "How do I enable Internet access for a Serverless Milvus instance?",
     "answer": "Enable Internet access and configure the whitelist on the instance details page in the console."},
    {"content": "What should I do if I forget my console logon password?", "answer": "Reset it through the password retrieval process in Account Center."},
    {"content": "Why is my bill higher than expected?", "answer": "Check the usage details, focusing on compute and storage usage."},
    {"content": "How do I create a collection and write vectors?",
     "answer": "Use create_collection to define the schema and then insert."},
    {"content": "Does scaling up an instance affect online business?", "answer": "Scaling is an online operation and usually does not interrupt services."},
]
client.insert(collection_name, faqs)
client.flush(collection_name)
client.load_collection(collection_name)

Etapas 4 e 5: Recuperação semântica e reranking de modelo grande

A pergunta coloquial de um usuário passa primeiro pela recuperação vetorial para recordar os candidatos Top-N e, em seguida, um modelo de reranking executa uma classificação de granularidade fina pela relevância real em relação à consulta. Somente as duas etapas combinadas podem atingir o FAQ padrão de forma confiável.

# ==================== Step 4: Semantic retrieval to recall FAQs ====================
query = "How can I make my instance reachable from the Internet?"
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}. [similarity {hit['distance']:.4f}] {hit['entity']['content']}")

# ==================== Step 5: AI_RERANK reranking ====================
# 5.1 REST real-time API: independently rerank a given batch of candidates
status, data = post_json(
    "/v2/vectordb/ai/rerank",
    {
        "model_name": "qwen3-rerank",
        "query": query,
        "documents": [
            "How do I enable Internet access for a Serverless Milvus instance?",
            "Does scaling up an instance affect online business?",
            "How do I create a collection and write vectors?",
        ],
        "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}. [relevance {item['relevance_score']:.4f}] candidate index={item['index']}")

# 5.2 Attach a ranker during search to complete recall and reranking in one call
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}. [rerank score {hit['distance']:.4f}] {e['content']} | answer: {e['answer']}")

Tome como exemplo a consulta "How can I make my instance reachable from the Internet?". Os dados medidos são os seguintes. Esta comparação de antes e depois ilustra claramente o valor do reranking:

A recuperação vetorial classificou "What should I do if I forget my console logon password?" em segundo lugar (0,5613), mas não é relevante em intenção para "torná-la acessível pela Internet". Este é exatamente o comportamento típico da recuperação puramente vetorial ser "recuperada, mas não precisa": a similaridade superficial do texto não é baixa, mas a intenção real está distante. Após o reranking, esta entrada é empurrada para o terceiro lugar (0,2607), e a entrada semanticamente mais próxima de "dimensionamento de instância" sobe para o segundo lugar. Para um bot de atendimento ao cliente, a precisão do Top-1 determina diretamente a qualidade da resposta, portanto, a etapa de reranking não deve ser omitida.

Etapa 6: Reconhecimento de sentimento e classificação de chamados

Anexe duas Functions ao texto da conversa mascarada ao mesmo tempo: uma determina a tendência de sentimento para inspeção de qualidade e alerta de opinião pública, e a outra classifica a conversa em uma categoria de chamado para roteamento automático. Ambas são concluídas automaticamente na escrita.

# ==================== Step 6: AI_SENTIMENT sentiment recognition + AI_CLASSIFY ticket classification ====================
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)     # Masked conversation text
schema.add_field("sentiment", DataType.VARCHAR, max_length=64)     # Sentiment output
schema.add_field("category", DataType.VARCHAR, max_length=64)      # Ticket category output
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": "Account,Inquiry,Fault,Billing",
                "prompt": "Classify based on the topic that the customer question belongs to.", "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": "I have already called three times about this problem of yours, and it still has not been resolved. This is so disappointing!"},
    {"content": "How do I enable Internet access for a Serverless Milvus instance?"},
])
client.flush(collection_name)
for row in client.query(collection_name, filter="",
                        output_fields=["content", "sentiment", "category"], limit=10):
    print(f"sentiment={row['sentiment']:<10} category={row['category']:<6} | {row['content'][:28]}")

Resultados medidos:

Várias Functions podem ser anexadas a uma única coleção, desde que seus campos de saída não entrem em conflito. Neste exemplo, sentiment e category são preenchidos por uma das duas Functions, portanto, escrever um único registro gera tanto o rótulo de sentimento quanto a categoria do chamado.