企业官网建设流程全解析

一、背景与目标：为什么要做向量 + GEO 的混合检索？

现在不少搜索/推荐场景同时具备两个特点：

• 用户表达是「自然语言语义」：

  • 比如「附近 3 公里适合约会的咖啡店」

  • 「新宿氛围好一点的居酒屋」

• 结果必须满足「地理约束」：

  • 距离用户位置必须在一定范围内

  • 甚至要求特定区域、商圈、地铁口

如果只用关键词/SQL 做查询，几乎无法理解「适合约会」「氛围好」这样的语义；如果只用向量检索，又无法强约束地理范围。

混合检索的目标，就是用一条检索链路同时满足两件事：

• 用向量检索理解自然语言语义，召回“语义上相似”的候选；

• 用 GEO 约束过滤/排序，保证返回结果在指定空间范围内。

本文会从工程落地角度，给出一套「向量 + GEO」混合搜索引擎的架构设计和核心源码，方便你直接改造成自己项目中的基础组件。

二、整体架构：向量召回 + GEO Filter + 排序

1. 整体流程

可以把整个查询链路抽象成四步：

1. Query 解析：

   • 解析用户 query 文本

   • 解析用户位置（经纬度）和半径等参数

2. 向量召回：

   • 使用 Embedding 模型对 query 编码

   • 在向量索引中做 TopK 近邻搜索，得到候选集合

3. GEO 约束过滤：

   • 对候选集合按经纬度计算距离

   • 过滤出在指定半径内的文档

4. 评分与排序：

   • 综合语义相似度、距离、评分/热度等形成 final score

   • 返回排序好的 TopN

2. 模块划分

从代码架构上，可以拆成如下几个模块（服务）：

• EmbeddingService：负责文本 → 向量编码；

• VectorIndex：负责向量索引构建与 ANN 查询；

• GeoService：负责经纬度存储、距离计算和 GEO 过滤；

• HybridSearchService：协调向量召回、GEO 过滤和排序的统一入口。

接下来所有示例代码都用 Python 写伪实现，你可以很容易改成 Go/TS/Java。

三、数据模型设计：文档、向量与 GEO 信息

假设我们要做的是「本地生活店铺搜索」，每条文档（店铺）可以定义成：

from dataclasses import dataclass @dataclass class ShopDoc: id: str title: str desc: str lat: float lon: float rating: float # 用户评分 hot: float # 热度或曝光

四、Embedding 与向量索引（简化实现）

1. Embedding 服务（示意）

这里我们用一个假的 embedding 函数做结构示例，实际你可以接任意 Embedding 模型

import numpy as np class EmbeddingService: def __init__(self, dim: int = 384): self.dim = dim def embed(self, text: str) -> np.ndarray: # 实际场景接模型，这里用随机向量占位 rng = np.random.default_rng(abs(hash(text)) % (2**32)) vec = rng.normal(size=self.dim) # 做一次归一化，便于计算余弦相似度 return vec / np.linalg.norm(vec)

2. 向量索引（基于简单暴力 + 替换接口）

为了把重点放在“混合检索”和 GEO 约束上，索引部分用一个简单版本：

class SimpleVectorIndex: def __init__(self, dim: int): self.dim = dim self.vectors = [] # list[np.ndarray] self.doc_ids = [] # list[str] def add(self, doc_id: str, vec: np.ndarray): assert vec.shape[0] == self.dim self.vectors.append(vec) self.doc_ids.append(doc_id) def search(self, query_vec: np.ndarray, top_k: int = 50): # 余弦相似度 sims = [] for i, v in enumerate(self.vectors): score = float(np.dot(query_vec, v)) sims.append((self.doc_ids[i], score)) sims.sort(key=lambda x: x[1], reverse=True) return sims[:top_k]

实际落地时，只需要把SimpleVectorIndex换成 FAISS / Milvus / pgvector 的封装即可，对上层来说接口是一样的。

五、GEO 模块：距离计算与半径过滤

1. Haversine 距离实现

import math class GeoService: EARTH_RADIUS_KM = 6371.0 @staticmethod def _rad(x: float) -> float: return x * math.pi / 180.0 @classmethod def distance_km(cls, lat1: float, lon1: float, lat2: float, lon2: float) -> float: dlat = cls._rad(lat2 - lat1) dlon = cls._rad(lon2 - lon1) rlat1 = cls._rad(lat1) rlat2 = cls._rad(lat2) a = math.sin(dlat / 2) ** 2 + \ math.cos(rlat1) * math.cos(rlat2) * math.sin(dlon / 2) ** 2 c = 2 * math.asin(math.sqrt(a)) return cls.EARTH_RADIUS_KM * c

2. GEO 过滤逻辑

这里假设我们有一个ShopRepository能根据shop_id拿到ShopDoc：

class ShopRepository: def __init__(self): self._store = {} # id -> ShopDoc def add(self, doc: ShopDoc): self._store[doc.id] = doc def get(self, doc_id: str) -> ShopDoc | None: return self._store.get(doc_id)

GEO 过滤函数：

def geo_filter(candidates, user_lat, user_lon, radius_km, repo: ShopRepository): """ candidates: list[{"id": str, "vec_score": float}] """ output = [] for c in candidates: doc = repo.get(c["id"]) if doc is None: continue d = GeoService.distance_km(user_lat, user_lon, doc.lat, doc.lon) if d <= radius_km: c["geo_distance"] = d c["doc"] = doc output.append(c) return output

六、打分与排序：语义 + 距离 + 业务权重

一旦我们有了vec_score和geo_distance，就可以构造一个简单的打分公式：

• 语义相似度越高越好

• 距离越近越好

• 评分/热度越高越好

对应代码：

def compute_final_score(c, alpha=1.0, beta=0.05, gamma=0.2, delta=0.1): doc: ShopDoc = c["doc"] vec_score = c["vec_score"] dist = c["geo_distance"] rating = doc.rating hot = doc.hot return ( alpha * vec_score - beta * dist + gamma * rating + delta * hot ) def rank_candidates(candidates): for c in candidates: c["final_score"] = compute_final_score(c) candidates.sort(key=lambda x: c["final_score"], reverse=True) return candidates

参数只要支持配置/热更新即可，后续可以配 A/B 测试和离线评估去调。

七、混合检索服务：向量 + GEO 的完整链路

现在把上面的模块串起来，给业务提供一个统一入口：

class HybridSearchService: def __init__(self, embedder: EmbeddingService, index: SimpleVectorIndex, repo: ShopRepository): self.embedder = embedder self.index = index self.repo = repo def search(self, query: str, user_lat: float, user_lon: float, radius_km: float = 3.0, top_k: int = 20): # 1) 向量召回 q_vec = self.embedder.embed(query) vec_results = self.index.search(q_vec, top_k=200) # 先取大一点 # 标准化结构 cand = [{"id": doc_id, "vec_score": score} for doc_id, score in vec_results] # 2) GEO 过滤 cand = geo_filter( candidates=cand, user_lat=user_lat, user_lon=user_lon, radius_km=radius_km, repo=self.repo, ) if not cand: return [] # 3) 排序 cand = rank_candidates(cand) # 4) 截断 + 输出 results = [] for c in cand[:top_k]: doc: ShopDoc = c["doc"] results.append({ "id": doc.id, "title": doc.title, "desc": doc.desc, "lat": doc.lat, "lon": doc.lon, "distance_km": c["geo_distance"], "vec_score": c["vec_score"], "final_score": c["final_score"], "rating": doc.rating, "hot": doc.hot, }) return results

八、初始化与完整示例（可直接跑）

最后给一个最小可运行示例，你可以直接放到一个 Python 文件里跑一遍，然后改成你自己的环境：

def build_demo_engine(): # 1) 初始化组件 embedder = EmbeddingService(dim=128) index = SimpleVectorIndex(dim=128) repo = ShopRepository() # 2) 构造假数据 shops = [ ShopDoc( id="shop_001", title="新宿安静咖啡店", desc="适合一个人看书的咖啡店，环境安静，桌椅舒适。", lat=35.6900, lon=139.7000, rating=4.8, hot=0.7, ), ShopDoc( id="shop_002", title="新宿夜景酒吧", desc="适合约会的小酒吧，可以看到新宿夜景。", lat=35.6910, lon=139.7020, rating=4.6, hot=0.9, ), ShopDoc( id="shop_003", title="涩谷连锁咖啡", desc="普通连锁咖啡店，人多嘈杂，适合简单休息。", lat=35.6590, lon=139.7000, rating=4.0, hot=0.5, ), ] # 3) 加入仓库和索引 for s in shops: repo.add(s) vec = embedder.embed(s.title + " " + s.desc) index.add(s.id, vec) # 4) 构建混合搜索服务 service = HybridSearchService(embedder, index, repo) return service if __name__ == "__main__": service = build_demo_engine() # 用户在新宿车站附近 user_lat = 35.6905 user_lon = 139.7005 query = "附近适合约会的咖啡店" results = service.search( query=query, user_lat=user_lat, user_lon=user_lon, radius_km=3.0, top_k=10, ) for r in results: print( r["id"], r["title"], f"{r['distance_km']:.2f}km", f"score={r['final_score']:.3f}", )