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

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 settings ====================
MILVUS_URI = "http://c-xxx.milvus.aliyuncs.com:19530"  # the port must be 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) and are distinguished by the task parameter.
# The FunctionType enum of some pymilvus versions has no such member, so add a fallback.
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]]:
    """Shared wrapper for the synchronous REST APIs. These calls go through an LLM, 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

# ==================== Step 1: AI_AUDIO_TRANSCRIBE 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)  # transcript output field
# A Collection must contain at least one vector field. This step performs no vector search,
# so a 2-dimensional placeholder field satisfies the constraint. Declaring it as
# nullable=True means the field can be omitted at insert time.
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",     # normalize spoken numbers, such as one-three-eight to 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 the server 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']}")

# ==================== Step 2: AI_PII_MASK masking ====================
# 2.1 Synchronous REST API: mask the text before it enters the database
status, data = post_json(
    "/v2/vectordb/ai/pii_mask",
    {
        "model_name": "qwen3.7-max",
        "texts": ["Hello, my phone number is 13800138000 and my ID card number is 110101199001010000. Please 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)

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

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

# ==================== 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 public network access for a Serverless Milvus instance?",
     "answer": "Enable public network access on the instance details page in the console and configure the whitelist."},
    {"content": "What do I do if I forget the console logon password?", "answer": "Reset it through the password recovery process in Account Center."},
    {"content": "Why is my bill higher than expected?", "answer": "Check the usage details, with a focus on compute and storage usage."},
    {"content": "How do I create a Collection and insert vectors?",
     "answer": "Define the schema with create_collection, then call insert."},
    {"content": "Does scaling up an instance affect online workloads?", "answer": "Scaling up is an online operation and usually does not interrupt service."},
]
client.insert(collection_name, faqs)
client.flush(collection_name)
client.load_collection(collection_name)

# ==================== Step 4: retrieve FAQs by semantic search ====================
query = "How can I get my instance connected 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 Synchronous REST API: rerank a given batch of candidates independently
status, data = post_json(
    "/v2/vectordb/ai/rerank",
    {
        "model_name": "qwen3-rerank",
        "query": query,
        "documents": [
            "How do I enable public network access for a Serverless Milvus instance?",
            "Does scaling up an instance affect online workloads?",
            "How do I create a Collection and insert 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 to search so that retrieval and reranking finish 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']}")

# ==================== Step 6: AI_SENTIMENT sentiment analysis + 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 by the topic of the customer question.", "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 called three times about this problem and it is still not fixed. This is very disappointing!"},
    {"content": "How do I enable public network 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]}")

for name in ("cs_transcribe", "cs_pii_mask", "cs_faq_kb", "cs_qc"):
    if client.has_collection(name):
        client.drop_collection(name)