向量检索服务 Milvus 版：通过阿里云Milvus融合地理检索与AI Function构建本地生活榜单

GEOMETRY 与 WKT

如果把经纬度存成两个 DOUBLE 字段，数据库只会把它当两个数字，无法判断「谁在谁里面」「两块区域是否相交」。GEOMETRY 是 Milvus 2.6 提供的地理空间类型，它把点、线、面作为几何对象存储，从而支持空间关系运算，并可通过地理位置索引加速查询。

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%，生成的摘要为「环境干净，毛血旺温和适口。服务耐心且备有儿童餐具，适合带娃家庭聚餐，周末需等位。」