Vector Retrieval Service for Milvus:Crie um ranking de serviços locais com o Alibaba Cloud Milvus combinando busca geoespacial e AI Function

GEOMETRY e WKT

Se longitude e latitude forem armazenadas como dois campos DOUBLE, o banco de dados os tratará apenas como números, sem capacidade de determinar relações como "qual está dentro de qual" ou "se duas áreas se intersectam". O tipo GEOMETRY foi introduzido no Milvus 2.6 e armazena pontos, linhas e polígonos como objetos geométricos, permitindo operações de relação espacial e aceleração de consultas por meio de índices geoespaciais.

Os valores de um campo GEOMETRY são expressos no formato WKT (Well-Known Text). Os tipos mais comuns são:

Funções de relação espacial ST_

Funções iniciadas com ST_ determinam a relação espacial entre dois objetos geométricos. Elas são inseridas na expressão filter e retornam verdadeiro ou falso. Este tópico utiliza principalmente as três seguintes:

Outras funções também são suportadas e podem ser usadas conforme necessário:

Prepare o código comum

O código abaixo contém a configuração de conexão, um wrapper para a interface REST e uma função utilitária que aproxima um círculo de raio como um polígono. Ele é compartilhado por todas as etapas subsequentes. Substitua MILVUS_HOST e MILVUS_TOKEN pelas informações da sua instância real.

from __future__ import annotations

import json
import math
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_HOST = "http://c-xxx.milvus.aliyuncs.com"
MILVUS_TOKEN = "root:xxx"
# The RESTful interface shares port 19530 with gRPC. The port must be specified explicitly;
# otherwise, port 80 is used by default and the connection times out.
MILVUS_REST_BASE_URL = f"{MILVUS_HOST}:19530"

client = MilvusClient(uri=MILVUS_HOST, token=MILVUS_TOKEN)
collection_name = "shop_reviews"

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 interfaces of AI_RERANK / AI_SENTIMENT / AI_SUMMARIZE."""
    last = 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

def circle_to_wkt(lng: float, lat: float, radius_km: float, num_seg: int = 32) -> str:
    """Approximate the circle centered at (lng, lat) with a radius of radius_km as the WKT of a regular polygon.
    Note: longitude and latitude are angles. The conversion here is a planar approximation,
    and the error grows as the latitude increases."""
    d_lat = radius_km / 111.0
    d_lng = radius_km / (111.0 * math.cos(math.radians(lat)))
    pts = []
    for i in range(num_seg + 1):
        theta = 2 * math.pi * i / num_seg
        pts.append(f"{lng + d_lng * math.cos(theta):.6f} {lat + d_lat * math.sin(theta):.6f}")
    return f"POLYGON(({', '.join(pts)}))"

Etapa 1: Crie uma Collection com campo geoespacial e Function de vetorização

O campo location usa DataType.GEOMETRY para armazenar o ponto de coordenadas do estabelecimento. O campo review está associado a uma Function AI_EMBEDDING, que gera automaticamente um vetor de 1024 dimensões no momento da escrita. Adicionalmente, um índice HNSW é criado para o campo vetorial e um índice R-Tree para o campo geoespacial.

# ==================== Step 1: Create the Collection ====================
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("shop_id", DataType.INT64)
schema.add_field("shop_name", DataType.VARCHAR, max_length=256)
schema.add_field("category", DataType.VARCHAR, max_length=64)
schema.add_field("location", DataType.GEOMETRY)                          # Shop longitude and latitude, WKT: POINT(lng lat)
schema.add_field("review", DataType.VARCHAR, max_length=4096)            # Original review text, used as the embedding input
schema.add_field("review_embedding", DataType.FLOAT_VECTOR, dim=1024)    # Review vector

# Automatically vectorize review at write time
schema.add_function(
    Function(
        name="embed_review",
        function_type=FunctionType.TEXTEMBEDDING,
        input_field_names=["review"],
        output_field_names=["review_embedding"],
        params={
            "provider": "aliyun_milvus",
            "model_name": "qwen3.7-text-embedding",
            "dim": 1024,
            "max_client_batch_size": 20,
            "max_concurrency": 1,
        },
    )
)

index_params = client.prepare_index_params()
index_params.add_index(
    field_name="review_embedding",
    index_type="HNSW",
    metric_type="COSINE",
    params={"M": 16, "efConstruction": 200},
)
# Create an R-Tree geospatial index for the GEOMETRY field to accelerate spatial queries
index_params.add_index(
    field_name="location",
    index_type="RTREE",
)
client.create_collection(
    collection_name=collection_name,
    schema=schema,
    index_params=index_params,
)

Etapa 2: Insira dados de estabelecimentos e avaliações

Insira as coordenadas dos estabelecimentos e o texto original das avaliações. O campo review_embedding é gerado automaticamente pela Function.

# ==================== Step 2: Write shop and review data ====================
# review_embedding is generated automatically by the Function; there is no need to pass in vectors manually
rows = [
    {"shop_id": 1001, "shop_name": "Bashu Home - Kid-Friendly Sichuan", "category": "Sichuan",
     "location": "POINT(116.4612 39.9088)",
     "review": "Came with the kids. They have high chairs and small bowls, the Mao Xue Wang is not too spicy, the staff are very patient, and the place is clean."},
    {"shop_id": 1001, "shop_name": "Bashu Home - Kid-Friendly Sichuan", "category": "Sichuan",
     "location": "POINT(116.4612 39.9088)",
     "review": "Crowded on weekends and you have to wait for a table, but the dishes are consistently good. Great for a family meal, and the kids enjoyed it."},
    {"shop_id": 1002, "shop_name": "Spicy Jianghu", "category": "Sichuan",
     "location": "POINT(116.4501 39.9101)",
     "review": "A paradise for lovers of strong flavors, the boiled fish in chili oil is full of flavor, but the space is small and quite noisy, so it is not very suitable for bringing children."},
    {"shop_id": 1003, "shop_name": "Quiet Lane Coffee", "category": "Coffee",
     "location": "POINT(116.4550 39.9075)",
     "review": "A quiet place suitable for talking business, and the pour-over coffee is good, although it has nothing to do with a demand for Sichuan food."},
]
client.insert(collection_name, rows)
client.flush(collection_name)
client.load_collection(collection_name)

Etapa 3: Filtre por alcance geográfico

Utilize ST_WITHIN para verificar se um estabelecimento está dentro do polígono de um bairro comercial ou dentro de um círculo aproximado de raio definido. O círculo de raio é aproximado como polígono pela função circle_to_wkt.

# ==================== Step 3: Geographic range filtering ====================
# 3.1 Query the shops inside a business district polygon
cbd_polygon = ("POLYGON((116.448 39.905, 116.465 39.905, 116.465 39.914, "
               "116.448 39.914, 116.448 39.905))")
hits = client.query(
    collection_name=collection_name,
    filter=f"ST_WITHIN(location, '{cbd_polygon}')",
    output_fields=["shop_name", "category", "location"],
    limit=100,
)
print("Review records inside the business district:")
for h in hits:
    print(f"  - {h['shop_name']} ({h['category']})  {h['location']}")

# 3.2 Query within a 3 km radius: approximate the circle as a polygon, then check whether the shop falls inside it
center_lng, center_lat = 116.4600, 39.9090
circle = circle_to_wkt(center_lng, center_lat, radius_km=3.0)
nearby = client.query(
    collection_name=collection_name,
    filter=f"ST_WITHIN(location, '{circle}')",
    output_fields=["shop_name", "category", "location"],
    limit=200,
)
print(f"Review records within 3 km of the center point ({center_lng}, {center_lat}):")
for n in nearby:
    print(f"  - {n['shop_name']} ({n['category']})  {n['location']}")

Nos dados de exemplo, os 4 registros de avaliação pertencem a 3 estabelecimentos (Bashu Home possui 2 avaliações), portanto a consulta acima retorna 4 registros. Para obter resultados no nível do estabelecimento, faça a deduplicação por shop_id na aplicação.

Etapa 4: Combine busca semântica com filtragem geográfica

Esta é a parte central da solução: unir "proximidade" (funções espaciais no filter) e "preferência" (busca vetorial ANN) em uma única chamada search().

# ==================== Step 4: Fuse semantic retrieval with geographic filtering ====================
# ANN retrieval over review vectors handles "on taste", the spatial function in filter handles "nearby",
# and both are solved jointly in a single request
QUERY = "Kid-friendly Sichuan restaurants nearby"
geo_filter = f"category == 'Sichuan' && ST_WITHIN(location, '{circle}')"
results = client.search(
    collection_name=collection_name,
    data=[QUERY],
    anns_field="review_embedding",
    filter=geo_filter,
    limit=20,
    output_fields=["shop_id", "shop_name", "category", "location", "review"],
)
print(f"Query: {QUERY}")
for rank, hit in enumerate(results[0], 1):
    e = hit["entity"]
    print(f"  {rank}. [similarity {hit['distance']:.4f}]  {e['shop_name']} | {e['review'][:24]}")

Um exemplo de resultado retornado é mostrado abaixo. "Bashu Home - Kid-Friendly Sichuan" aparece em primeiro lugar porque trechos de suas avaliações como "high chairs" e "the staff are very patient" têm alta relevância semântica para "kid-friendly":

  1. [similarity 0.6926]  Bashu Home - Kid-Friendly Sichuan | Came with the kids. They have
  2. [similarity 0.6591]  Spicy Jianghu | A paradise for lovers of strong flavors, the boil
  3. [similarity 0.6091]  Bashu Home - Kid-Friendly Sichuan | Crowded on weekends and you

Etapa 5: Use AI_RERANK para refinar a ordenação dos candidatos

A similaridade vetorial reflete proximidade semântica, mas não equivale totalmente a "quão relevante algo é para a consulta". Aplicar AI_RERANK para uma segunda rodada de refinamento nos candidatos recuperados permite mover para o topo os estabelecimentos que melhor correspondem à intenção da consulta. Abaixo são mostrados dois modos de uso: anexando um ranker em search() e chamando a interface REST stateless.

# ==================== Step 5: Rerank candidates with AI_RERANK ====================
RERANK_MODEL = "qwen3-rerank"

# 5.1 Attach a RERANK ranker in search(); after ANN recall, the model performs a second round of fine ranking
reranker = Function(
    name="rerank_reviews",
    function_type=FunctionType.RERANK,
    input_field_names=["review"],
    params={
        "reranker": "model",
        "provider": "aliyun_milvus",
        "model_name": RERANK_MODEL,
        "queries": [QUERY],
        "timeout_sec": 10,
    },
)
results = client.search(
    collection_name=collection_name,
    data=[QUERY],
    anns_field="review_embedding",
    filter=geo_filter,
    limit=20,
    output_fields=["shop_id", "shop_name", "review"],
    ranker=reranker,
)
for rank, hit in enumerate(results[0], 1):
    e = hit["entity"]
    print(f"  {rank}. [rerank score {hit['distance']:.4f}]  {e['shop_name']} | {e['review'][:24]}")

# 5.2 You can also use the stateless REST interface /v2/vectordb/ai/rerank
rerank_docs = [
    "Came with the kids. They have high chairs and small bowls, the Mao Xue Wang is not too spicy, the staff are very patient, and the place is clean.",
    "A paradise for lovers of strong flavors, the boiled fish in chili oil is full of flavor, but the space is small and quite noisy, so it is not very suitable for bringing children.",
]
status, data = post_json(
    "/v2/vectordb/ai/rerank",
    {
        "model_name": RERANK_MODEL,
        "query": QUERY,
        "documents": rerank_docs,
        "params": {"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}]  {rerank_docs[item['index']][:28]}")

Um exemplo da saída após o refinamento é apresentado a seguir. Ambas as avaliações de "Bashu Home - Kid-Friendly Sichuan" aparecem antes de "Spicy Jianghu":

  1. [rerank score 0.6986]  Bashu Home - Kid-Friendly Sichuan | Came with the kids. They have
  2. [rerank score 0.5389]  Bashu Home - Kid-Friendly Sichuan | Crowded on weekends and you
  3. [rerank score 0.5214]  Spicy Jianghu | A paradise for lovers of strong flavors, the boil

A estrutura de retorno da interface REST é data.output.results, onde cada candidato corresponde a um item contendo index (o índice do candidato no array de entrada documents) e relevance_score (a pontuação de relevância). Ordenar por pontuação em ordem decrescente produz o resultado refinado:

  1. [relevance 0.6986]  Came with the kids. They have high chairs and small bowls, the Mao Xue Wang is not too spicy, the staff are very patient, and the place is clean.
  2. [relevance 0.5214]  A paradise for lovers of strong flavors, the boiled fish in chili oil is full of flavor, but the space is small and quite noisy, so it is not very suitable for bringing children.

O texto do candidato deve abranger todas as dimensões semânticas da consulta. A consulta deste exemplo, "Kid-friendly Sichuan restaurants nearby", contém tanto a dimensão "adequado para crianças" quanto "culinária chinesa", e um texto candidato que omita qualquer uma dessas dimensões perde uma parcela significativa da pontuação. Tome o primeiro candidato como exemplo:

Simplesmente adicionar o nome de um prato chinês eleva a pontuação de 0.4723 para 0.6666. Em contraste, "A paradise for lovers of strong flavors, the boiled fish in chili oil is full of flavor ... not very suitable for bringing children." pontua na dimensão culinária chinesa por mencionar "boiled fish in chili oil", mantendo sua relevância estável em 0.5214. Portanto, se o primeiro candidato não trouxesse informação sobre culinária chinesa, ficaria posicionado abaixo.

Etapa 6: Calcule a taxa de avaliações positivas e gere um resumo das avaliações

Utilize AI_SENTIMENT para classificar o sentimento de todas as avaliações de um estabelecimento e agregue a taxa de avaliações positivas na aplicação; em seguida, use AI_SUMMARIZE para condensar as avaliações em um comentário resumido.

# ==================== Step 6: Sentiment analysis for the positive review rate + summarization for the comment ====================
shop_id = 1001
reviews = [r["review"] for r in client.query(
    collection_name=collection_name,
    filter=f"shop_id == {shop_id}",
    output_fields=["review"],
    limit=1000,
)]

# 6.1 Sentiment classification, aggregated into a positive review rate
status, sent = post_json("/v2/vectordb/ai/sentiment", {
    "model_name": "qwen3.7-max",
    "texts": reviews,
    "params": {"categories": ["positive", "negative", "neutral"], "temperature": 0},
})
assert status == 200 and sent.get("code") == 0, sent
labels = sent["data"]["output"]["outputs"]
valid = [x for x in labels if x]              # Empty values count as undetermined and are excluded from the aggregation
positive = sum(1 for x in valid if x == "positive")
good_rate = positive / len(valid) if valid else 0.0
print(f"Positive review rate: {good_rate:.0%}   (valid samples {len(valid)}/{len(labels)})")

# 6.2 Generate a summary of the shop's reviews
status, summ = post_json("/v2/vectordb/ai/summarize", {
    "model_name": "qwen3.7-max",
    "texts": ["\n".join(reviews)],
    "params": {"max_words": 40, "temperature": 0,
               "prompt": "Summarize the overall reviews of this shop in English, highlighting its signature dishes, the audience it suits, and its environment."},
})
assert status == 200 and summ.get("code") == 0, summ
print(f"Shop summary: {summ['data']['output']['outputs'][0]}")

Tomando shop_id 1001 como exemplo, ambas as avaliações são classificadas como positivas, resultando em uma taxa de avaliações positivas de 100%. O resumo gerado é: "Ambiente limpo, com Mao Xue Wang suave e agradável. Atendimento paciente com talheres infantis disponíveis, adequado para refeições em família com crianças; espere fila nos fins de semana."