向量检索引擎实战：手写 HNSW 近似最近邻索引与召回率调优

为什么 RAG 把向量检索推上了 C 位

2023 年以来，检索增强生成（RAG）成了大模型落地的标配。但一个被忽视的事实是：RAG 的瓶颈从来不是生成，而是检索。向量数据库要在百万、亿级向量里，毫秒级返回与 query 最相似的 k 条，靠的是近似最近邻（ANN）索引。

暴力扫描的复杂度是 O(N·d)，当 N 达到千万级、d=768 时，单次查询要算数亿次乘加，完全不可接受。于是业界演化出三条主线：

• IVF（倒排文件）：用 k-means 把空间切成簇，只搜最近的几个簇。召回随簇数波动大，且高维下"簇"本身就很模糊。
• PQ（乘积量化）：把向量压缩到几个字节，用近似距离换内存。召回损失明显，但能扛十亿级。
• HNSW（分层可导航小世界图）：用图结构做导航，召回和延迟同时做到极致，代价是内存。

工程上，凡是"要召回、要低延迟、单机扛得住"的场景，HNSW 基本是默认答案。本文不堆概念，直接手写一个可运行的 HNSW，讲清它的实现陷阱与调优旋钮。

HNSW 的两个核心思想

HNSW 站在 NSW（Navigable Small World）图和跳表（skip-list）两个肩膀上。

第一，NSW 图保证"可导航"。 把每个向量当图节点，连若干近邻边。从任意入口出发，沿"离目标更近"的边贪婪前进，几跳就能逼近最近邻——这叫"小世界"特性。

第二，分层把"长途"和"短途"分离。 像跳表一样，少数节点出现在高层（长边，快速跨越大范围），所有节点出现在第 0 层（短边，做精细搜索）。查询时从最高层进入，逐层下降，高层负责"粗定位"，低层负责"精修"。层数是随机的：节点以概率 1/2、1/4、1/8……出现在第 1、2、3 层，本质是 l = -ln(uniform(0,1)) / ln(m_L)。

从零实现一个 HNSW

下面给出约 80 行、可直接跑的 Python 实现。为聚焦算法，距离用"1 − 余弦"（假设向量已 L2 归一化）。

import heapq, random, math

def dist(a, b):
    # 假设 a, b 已 L2 归一化，余弦相似度即点积，距离取 1 - 余弦
    return 1.0 - sum(x * y for x, y in zip(a, b))

class HNSW:
    def __init__(self, M=16, Mmax0=32, ef_construction=200, mL=1 / math.log(16)):
        self.M = M                 # 每层基础邻居数（建图时用）
        self.Mmax0 = Mmax0         # 第 0 层允许的最大邻居数
        self.ef = ef_construction  # 建图时每层搜索的候选宽度
        self.mL = mL               # 层数衰减系数
        self.nodes = {}            # id -> {"vec": [...], "nb": {layer: set}}
        self.entry = None          # 当前最高层入口节点
        self.max_layer = -1
        self._id = 0

    def _random_level(self):
        return int(-math.log(random.random()) * self.mL)

    def _search_layer(self, q, entry_points, ef, layer):
        # 返回该层里离 q 最近的 ef 个 (距离, id)，最近在前
        visited, candidates, found = set(), [], []
        for ep in entry_points:
            d = dist(q, self.nodes[ep]["vec"])
            heapq.heappush(candidates, (d, ep))
            heapq.heappush(found, (-d, ep))   # 最大堆，堆顶是最远
            visited.add(ep)
        while candidates:
            d, c = heapq.heappop(candidates)
            if d > -found[0][0] and len(found) >= ef:
                break
            for nb in self.nodes[c]["nb"].get(layer, ()):
                if nb not in visited:
                    visited.add(nb)
                    nd = dist(q, self.nodes[nb]["vec"])
                    if len(found) < ef or nd < -found[0][0]:
                        heapq.heappush(candidates, (nd, nb))
                        heapq.heappush(found, (-nd, nb))
                        if len(found) > ef:
                            heapq.heappop(found)
        return [(-d, n) for d, n in found]

    def insert(self, vec):
        nid = self._id; self._id += 1
        self.nodes[nid] = {"vec": vec, "nb": {}}
        level = self._random_level()
        if self.entry is None:
            self.entry, self.max_layer = nid, level
            return
        ep = [self.entry]
        for lc in range(self.max_layer, level, -1):          # 高层只找入口
            ep = [n for _, n in self._search_layer(vec, ep, 1, lc)]
        for lc in range(min(level, self.max_layer), -1, -1):  # 逐层连边
            cand = self._search_layer(vec, ep, self.ef, lc)
            neighbors = [n for _, n in cand[:self.M]]
            self.nodes[nid]["nb"][lc] = set(neighbors)
            mmax = self.Mmax0 if lc == 0 else self.M
            for nb in neighbors:                              # 双向连接 + 裁剪
                self.nodes[nb]["nb"].setdefault(lc, set()).add(nid)
                if len(self.nodes[nb]["nb"][lc]) > mmax:
                    lst = list(self.nodes[nb]["nb"][lc])
                    lst.sort(key=lambda x: dist(self.nodes[nb]["vec"], self.nodes[x]["vec"]))
                    self.nodes[nb]["nb"][lc] = set(lst[:mmax])
            ep = neighbors
        if level > self.max_layer:
            self.max_layer, self.entry = level, nid

    def search(self, q, k=10, ef=None):
        ef = ef or max(k, self.ef // 4)
        ep = [self.entry]
        for lc in range(self.max_layer, 0, -1):
            ep = [n for _, n in self._search_layer(q, ep, 1, lc)]
        res = self._search_layer(q, ep, ef, 0)
        res.sort(key=lambda x: x[0])
        return [(n, -d) for d, n in res[:k]]

_search_layer 是整个算法的心脏：它维护一个"候选最小堆"和一个"已找到最大堆"，每扩展一个节点就把它邻居中足够近的纳入候选，直到候选里最远的距离超过已找到集合里最远的距离——这就是贪婪搜索的终止条件。insert 的关键是双向连边后做邻居裁剪，否则热门节点度数爆炸会把图变成星形、破坏可导航性。

四个旋钮决定生死

1. M（每层基础邻居数）：召回、内存、插入速度三者此消彼长。太小图太稀、容易迷路；太大内存线性增长、插入时裁剪开销上升。默认 16 是甜点，召回优先可上 32。

2. efConstruction（建图候选宽度）：只影响建图质量。调大图更连通、召回更高，但建库更慢。索引只建一次，建议给到 200–400。

3. efSearch（查询候选宽度）：运行期可调，是召回/延迟的直接旋钮。它不重构图，只是让搜索看得更广。线上先定延迟预算，再往上推 efSearch 直至满足召回。

4. mL：层间衰减，反比于 M。它决定层数分布——mL 越小层越多，高层长边越多、跨域更快，但内存略增。

与 IVF-PQ 的实战取舍

HNSW 的弱点是内存：每个节点存 M 条边，边是指针 + 原始向量副本，十亿级轻松吃掉上百 GB。而 IVF-PQ 把向量压到几字节，能在同内存下扛更大规模。

所以选型很清晰：

• 单机 / 千万级 / 要低延迟高召回 → HNSW（如 Milvus、Qdrant、Weaviate 默认引擎）。
• 十亿级以上 / 内存敏感 / 可接受略低召回 → IVF-PQ 或 DiskANN。

三个没人告诉你的坑

• 删除是硬伤：原生 HNSW 不支持删除，删边会导致图不连通。生产做法是用 tombstone 标记 + 定期离线重建，或上支持标记删除的变体（如 HNSWlib 的 mark_deleted）。
• 归一化必须一致：混合用余弦和欧氏，索引会返回"近"但语义远的结果。建库和查询用同一度量，最好入库前统一 L2 归一化。
• 维度灾难在 1000 维以上加剧：高维下距离区分度坍塌，HNSW 召回也会下滑。此时先降维（PCA）或用量化补偿，比无脑堆 efSearch 更划算。

结语

HNSW 不是魔法，它只是把"跳表的层 + 小世界图的边 + 贪婪搜索的终止条件"组合到了一个工程上极稳的结构里。理解 _search_layer 的堆逻辑和邻居裁剪，你就能看懂任何一款向量数据库的源码，也能在召回不达标时知道该拧哪个旋钮——而不是盲调参数。

真正拉开差距的，从来不是会不会调 efSearch，而是知不知道它为什么有效。