Vector Retrieval Service for Milvus：通過阿里雲Milvus融合地理檢索與AI Function構建本地生活榜單

GEOMETRY 與 WKT

如果把經緯度存成兩個 DOUBLE 欄位，資料庫只會把它當兩個數字，無法判斷「誰在誰裡面」「兩塊地區是否相交」。GEOMETRY 是 Milvus 2.6 提供的地理空間類型，它把點、線、面作為幾何Object Storage Service，從而支援空間關係運算，並可通過地理位置索引加速查詢。

GEOMETRY 欄位的值使用 WKT（Well-Known Text）格式表示，常用類型如下：

ST_ 空間關係函數

以 ST_ 開頭的函數用於判斷兩個幾何對象之間的空間關係，寫在 filter 運算式中，返回真或假。本文主要用到以下三個：

此外還支援以下函數，可按需選用：

準備公用代碼

以下程式碼封裝含串連配置、REST 介面封裝以及把半徑圓近似成多邊形的工具函數，後續各步驟共用。請將 MILVUS_HOST 與 MILVUS_TOKEN 替換為實際執行個體資訊。

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

# ==================== 串連配置 ====================
MILVUS_HOST = "http://c-xxx.milvus.aliyuncs.com"
MILVUS_TOKEN = "root:xxx"
# RESTful 介面與 gRPC 同在 19530 連接埠，需顯式帶連接埠，否則預設 80 連接埠會連線逾時
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]]:
    """AI_RERANK / AI_SENTIMENT / AI_SUMMARIZE 的 REST 介面統一封裝。"""
    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:
    """把以 (lng, lat) 為圓心、radius_km 為半徑的圓近似成正多邊形的 WKT。
    注意：經緯度是角度，此處按平面近似換算，緯度越高誤差越大。"""
    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)}))"

步驟一：建立帶地理欄位與向量化 Function 的 Collection

location 欄位使用 DataType.GEOMETRY 儲存門店座標點；review 欄位掛載 AI_EMBEDDING Function，寫入時自動產生 1024 維向量。同時為向量欄位建立 HNSW 索引、為地理欄位建立 R-Tree 索引。

# ==================== 步驟 1：建立 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)                          # 門店經緯度，WKT: POINT(lng lat)
schema.add_field("review", DataType.VARCHAR, max_length=4096)            # 評價原文，作為 embedding 輸入
schema.add_field("review_embedding", DataType.FLOAT_VECTOR, dim=1024)    # 評價向量

# 寫入時自動把 review 向量化
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},
)
# 為 GEOMETRY 欄位建立 R-Tree 地理位置索引，加速空間查詢
index_params.add_index(
    field_name="location",
    index_type="RTREE",
)
client.create_collection(
    collection_name=collection_name,
    schema=schema,
    index_params=index_params,
)

步驟二：寫入門店與評價資料

寫入門店座標與評價原文，review_embedding 由 Function 自動產生。

# ==================== 步驟 2：寫入門店與評價資料 ====================
# review_embedding 由 Function 自動產生，無需手工傳入向量
rows = [
    {"shop_id": 1001, "shop_name": "巴蜀人家·親子川菜", "category": "川菜",
     "location": "POINT(116.4612 39.9088)",
     "review": "帶娃來的，有兒童座椅和小碗餐具，毛血旺不算太辣，服務員很有耐心，環境乾淨。"},
    {"shop_id": 1001, "shop_name": "巴蜀人家·親子川菜", "category": "川菜",
     "location": "POINT(116.4612 39.9088)",
     "review": "周末人多要等位，但菜品穩定，適合一家人聚餐，孩子吃得開心。"},
    {"shop_id": 1002, "shop_name": "麻辣江湖", "category": "川菜",
     "location": "POINT(116.4501 39.9101)",
     "review": "重口味愛好者天堂，水煮魚夠味，但空間偏小、比較吵，不太適合帶小孩。"},
    {"shop_id": 1003, "shop_name": "靜巷咖啡", "category": "咖啡",
     "location": "POINT(116.4550 39.9075)",
     "review": "環境安靜適合談事，手沖不錯，不過跟川菜需求無關。"},
]
client.insert(collection_name, rows)
client.flush(collection_name)
client.load_collection(collection_name)

步驟三：地理範圍過濾

通過 ST_WITHIN 判斷門店是否落在商圈多邊形內，或是否落在指定半徑的近似圓內。半徑圓由 circle_to_wkt 近似成多邊形。

# ==================== 步驟 3：地理範圍過濾 ====================
# 3.1 查詢某商圈多邊形範圍內的門店
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("商圈範圍內的評價記錄：")
for h in hits:
    print(f"  · {h['shop_name']}（{h['category']}）  {h['location']}")

# 3.2 附近 3km 半徑查詢：把圓近似成多邊形，再判斷門店是否落在圈內
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"中心點 ({center_lng}, {center_lat}) 附近 3km 內的評價記錄：")
for n in nearby:
    print(f"  · {n['shop_name']}（{n['category']}）  {n['location']}")

樣本資料中 4 條評價記錄分屬 3 家門店（巴蜀人家有 2 條評價），因此上述查詢返回 4 條記錄。如需按門店維度輸出，請在應用側按 shop_id 去重。

步驟四：語義檢索與地理過濾融合

這是本方案的核心：把「就近」（filter 中的空間函數）與「對味」（ANN 向量檢索）融合進同一次 search() 調用。

# ==================== 步驟 4：語義檢索與地理過濾融合 ====================
# ANN 檢索評價向量負責「對味」，filter 中的空間函數負責「就近」，一次請求聯合求解
QUERY = "附近適合親子的川菜館"
geo_filter = f"category == '川菜' && 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}")
for rank, hit in enumerate(results[0], 1):
    e = hit["entity"]
    print(f"  {rank}. [相似性 {hit['distance']:.4f}]  {e['shop_name']} ｜ {e['review'][:24]}")

返回結果樣本如下，「巴蜀人家·親子川菜」因評價中「兒童座椅」「服務員很有耐心」等內容與「適合親子」語義高度相關而排在首位：

  1. [相似性 0.6926]  巴蜀人家·親子川菜 ｜ 帶娃來的，有兒童座椅和小碗餐具，
  2. [相似性 0.6591]  麻辣江湖 ｜ 重口味愛好者天堂，水煮魚夠味，但空間偏小、比較吵
  3. [相似性 0.6091]  巴蜀人家·親子川菜 ｜ 周末人多要等位，但菜品穩定，適合一家人聚餐，孩子

步驟五：使用 AI_RERANK 精排候選

向量相似性反映的是語義接近程度，與「和查詢有多相關」並不完全等價。通過 AI_RERANK 對召回候選做二次精排，可以把更貼合查詢意圖的門店排到前面。下面給出兩種用法：在 search() 中掛載 ranker，以及調用無狀態 REST 介面。

# ==================== 步驟 5：AI_RERANK 重排候選 ====================
RERANK_MODEL = "qwen3-rerank"

# 5.1 在 search() 中掛載 RERANK ranker，ANN 召回後由模型二次精排
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}. [重排分 {hit['distance']:.4f}]  {e['shop_name']} ｜ {e['review'][:24]}")

# 5.2 也可使用無狀態的 REST 介面 /v2/vectordb/ai/rerank
rerank_docs = [
    "帶娃來的，有兒童座椅和小碗餐具，毛血旺不算太辣，服務員很有耐心，環境乾淨。",
    "重口味愛好者天堂，水煮魚夠味，但空間偏小、比較吵，不太適合帶小孩。",
]
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}. [相關度 {item['relevance_score']:.4f}]  {rerank_docs[item['index']][:28]}")

精排後的輸出樣本如下，「巴蜀人家·親子川菜」的兩條評價均被排到「麻辣江湖」之前：

  1. [重排分 0.6986]  巴蜀人家·親子川菜 ｜ 帶娃來的，有兒童座椅和小碗餐具，
  2. [重排分 0.5389]  巴蜀人家·親子川菜 ｜ 周末人多要等位，但菜品穩定，適合一家人聚餐，孩子
  3. [重排分 0.5214]  麻辣江湖 ｜ 重口味愛好者天堂，水煮魚夠味，但空間偏小、比較吵

REST 介面的返回結構為 data.output.results，其中每個候選對應一項，包含 index（候選在入參 documents 中的下標）與 relevance_score（相關性得分）。按得分降序排列即為精排結果：

  1. [相關度 0.6986]  帶娃來的，有兒童座椅和小碗餐具，毛血旺不算太辣，服務員很有耐心，環境乾淨。
  2. [相關度 0.5214]  重口味愛好者天堂，水煮魚夠味，但空間偏小、比較吵，不太適合帶小孩。

候選文本需覆蓋查詢的全部語義維度。本例的查詢「附近適合親子的川菜館」同時包含「親子」與「川菜」兩個維度，候選文本缺失任一維度都會顯著掉分。以第一條候選為例：

僅僅補上一個川菜菜名，得分就從 0.4723 升到 0.6666。而作為對照的「重口味愛好者天堂，水煮魚夠味……不太適合帶小孩。」因含「水煮魚」而在川菜維度得分，相關性穩定為 0.5214。因此如果第一條候選缺失川菜資訊，它反而會被排到後面。

步驟六：計算好評率並產生評價摘要

用 AI_SENTIMENT 對門店的全部評價做情感分類，在應用側彙總出好評率；再用 AI_SUMMARIZE 把評價壓縮成一句話點評。

# ==================== 步驟 6：情感分析算好評率 + 摘要產生點評 ====================
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 情感分類，彙總出好評率
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]              # 空值計入未判定，不參與彙總
positive = sum(1 for x in valid if x == "positive")
good_rate = positive / len(valid) if valid else 0.0
print(f"好評率：{good_rate:.0%}   （有效樣本 {len(valid)}/{len(labels)}）")

# 6.2 產生門店評價摘要
status, summ = post_json("/v2/vectordb/ai/summarize", {
    "model_name": "qwen3.7-max",
    "texts": ["\n".join(reviews)],
    "params": {"max_words": 40, "temperature": 0,
               "prompt": "用中文概括這家店的整體評價，突出招牌菜、適合人群與環境。"},
})
assert status == 200 and summ.get("code") == 0, summ
print(f"店鋪摘要：{summ['data']['output']['outputs'][0]}")

以 shop_id 1001 為例，兩條評價均被判定為 positive，好評率為 100%，產生的摘要為「環境乾淨，毛血旺溫和適口。服務耐心且備有兒童餐具，適合帶娃家庭聚餐，周末需等位。」