空间 + 向量混合检索助手（Hybrid Spatial-Vector Retrieval）

一、什么时候用我

地点 / POI 搜索需要"语义相关 + 距离近"同时排序（例：找"适合安静看书"的咖啡店，且在 1km 内）
RAG 系统需要融合关键词召回与向量召回两路结果
需要 Haversine / 余弦相似度 / RRF 的公式与落地代码
数据库（如 OceanBase、PostGIS、Elasticsearch）中设计混合检索查询

二、三路召回

| 通路 | 计算方式 | 说明 | |---|---|---| | 关键词路 | BM25 / FTS / LIKE | 命中显式词，精度高、召回窄 | | 向量路 | cosine similarity = 1 − cosine_distance | 语义泛化好，对专有名词弱 | | 空间路 | Haversine 大圆距离 | 先过滤（如 ≤1km）再参与排序，降成本 |

Haversine 公式（地球半径 R ≈ 6371 km）：

a = sin²(Δφ/2) + cos(φ1)·cos(φ2)·sin²(Δλ/2)
d = 2R·asin(√a)

三、融合策略

方案 A：RRF（Reciprocal Rank Fusion）——推荐，免调参

score(d) = Σ_i  1 / (k + rank_i(d))      # k 常取 60

对每一路的排名取倒数和，天然免归一化、对异常分数鲁棒。

方案 B：加权线性——需要调参

score(d) = w1·norm(kw) + w2·norm(vec) + w3·norm(geo)

各路先 min-max 归一化到 [0,1]；地理路建议用 1 − d/d_max 反向（越近分越高）。

四、Python 最小实现（零依赖）

import math

def haversine_km(lat1, lon1, lat2, lon2):
    """大圆距离，单位 km。"""
    r = math.pi / 180
    dphi = (lat2 - lat1) * r
    dlam = (lon2 - lon1) * r
    a = (math.sin(dphi / 2) ** 2
         + math.cos(lat1 * r) * math.cos(lat2 * r) * math.sin(dlam / 2) ** 2)
    return 2 * 6371.0 * math.asin(math.sqrt(a))


def cosine_sim(v1, v2):
    dot = sum(a * b for a, b in zip(v1, v2))
    n1 = math.sqrt(sum(a * a for a in v1))
    n2 = math.sqrt(sum(b * b for b in v2))
    return dot / (n1 * n2) if n1 and n2 else 0.0


def rrf_fuse(rank_lists, k=60):
    """rank_lists: [[doc_id 按该路得分降序], ...] -> {doc_id: rrf_score}

    所有列表必须使用同一套 doc_id（同类型、同取值空间），否则融合会静默失效。
    """
    scores = {}
    for ranking in rank_lists:
        for rank, doc_id in enumerate(ranking):
            scores[doc_id] = scores.get(doc_id, 0.0) + 1.0 / (k + rank + 1)
    return dict(sorted(scores.items(), key=lambda x: -x[1]))


def keyword_rank(terms, docs):
    """关键词路的最小占位实现：按命中词次排序。生产环境请换 FTS / BM25。"""
    def hit(d):
        text = (d.get("text") or "").lower()
        return sum(text.count(t) for t in terms)

    scored = [(d["id"], hit(d)) for d in docs]
    return [did for did, s in sorted(scored, key=lambda x: -x[1]) if s > 0]


def hybrid_search(query_vec, doc_vecs, docs, center, max_km=1.0,
                  topk=3, query_terms=None, require_geo=True):
    """三路混合检索。

    docs:      [{"id","name","lat","lon","text"}]，id 必须唯一
    doc_vecs:  与 docs 一一对应的向量
    center:    (lat, lon) 查询中心
    require_geo: True 时把 max_km 当硬约束（融合后再兜一次过滤）
    """
    by_id = {d["id"]: d for d in docs}
    dist = {d["id"]: haversine_km(center[0], center[1], d["lat"], d["lon"])
            for d in docs}

    # 通路 1：向量 —— 必须先把下标映射成 id，否则与其余两路无法对齐
    order = sorted(range(len(docs)), key=lambda i: -cosine_sim(query_vec, doc_vecs[i]))
    vec_ids = [docs[i]["id"] for i in order]

    # 通路 2：空间 —— 距离只算一次，先过滤再升序
    geo_ids = sorted((did for did, km in dist.items() if km <= max_km),
                     key=lambda did: dist[did])

    # 通路 3：关键词 —— 可选
    rank_lists = [vec_ids, geo_ids]
    if query_terms:
        rank_lists.append(keyword_rank(query_terms, docs))

    fused = rrf_fuse(rank_lists)

    picked = [did for did in fused
              if not require_geo or dist[did] <= max_km]
    return [by_id[did] for did in picked[:topk]]

五、使用示例

输入："找离我 1km 内、语义上最像「适合安静看书」的咖啡店，推荐 3 家。"

执行步骤：

向量路：把候选店描述向量化，与查询向量算 cosine，按相似度降序
空间路：Haversine 过滤 1km 内的店，按距离升序
关键词路（可选）：对查询做分词，按命中词次排序
RRF 融合三路排名，按 require_geo 决定是否硬性剔除 1km 外，输出 Top3

六、注意事项

空间过滤先于排序：先用距离圈缩小候选集，再算向量，成本能降一个量级
三路必须用同一套 doc_id：RRF 最常见的翻车点是某一路返回下标、另一路返回字符串 id，两边永远加不到一起，而且不报错、只是融合失效
距离上限若为硬约束，融合后要再兜一次过滤；否则向量路单边命中的远处结果会混进推荐列表
距离只算一次再排序：把 haversine 放进 sorted 的 key 里会退化成 O(N log N) 次三角函数调用
文本归一化（大小写、分词、去停用词）对关键词路影响显著
RRF 的 k=60 是经验值；有验证集时可网格搜索 k ∈ {10, 20, 60, 100}
向量维度必须与查询向量一致；不同 embedding 模型的向量不可混用
排序结果建议保留各路子分（可解释性），方便排查"为什么这条排前面"