让 RAG 不再答旧 API：从分块、向量检索到 HNSW 多路召回

2026-07-23 27 预计阅读时间: 1 分钟
来源: my.oschina.net AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间：12 分钟

大模型不知道你的代码库刚刚升级，也不知道内部优惠券退款流程写在哪份文档里。没有外部上下文时，它会用训练阶段见过的模式补全答案，于是可能给出 v1.x 的旧 API，或为内部流程编造看似合理的步骤。RAG（Retrieval-Augmented Generation，检索增强生成）的工作，是在生成前先从可信资料中找出相关证据，再把证据交给模型回答。

但 RAG 的质量并不取决于“有没有接上向量数据库”。分块方式、Embedding、相似度计算、HNSW 索引参数，以及多路召回如何合并，都会直接决定模型最终看到了什么。

Chunking：检索命中的最小证据单元

文档不能原样塞进索引。一个完整的 API 手册、退款制度或 Wiki 页面通常太长，Embedding 会把多个主题压缩成一个向量，查询时很难精确命中。Chunking 的目标是把资料拆成语义尽量完整、主题尽量单一的片段。

对于技术文档，较稳妥的边界通常是：

  • 按标题、函数、类或配置项切分，而不是只按固定字符数硬切。
  • 每个 chunk 保留足够上下文，例如函数签名、参数说明和一个使用示例应尽量在同一块。
  • 在相邻 chunk 之间保留少量 overlap，避免“前提在上一段、结论在下一段”导致证据断裂。
  • 把 文档标题、版本、更新时间、权限范围、URL 或文档 ID 作为 metadata 保存。

例如，v2.3 的迁移说明与 v1.x 的旧示例不应混在一个 chunk 中。版本字段不仅能用于展示来源，也可以在检索阶段过滤：用户明确问 v2.3 时，优先甚至只查 v2.3 的材料。

分块过小会损失上下文，让模型只看到“调用 connect()”；分块过大则降低主题密度，让“退款条件”和“优惠券发放规则”彼此干扰。实际项目应以查询集评估，而不是凭一个固定 token 数量定终身。

Embedding 与相似度：把问题和资料放进同一空间

Embedding 模型会把文本编码为向量。语义相近的句子在向量空间中通常更接近，因此“退款时优惠券是否返还”和“退单后券的回补规则”即使字面词不同，也有机会被一起召回。

常见度量是余弦相似度：

cosine(a, b) = (a · b) / (||a|| * ||b||)

它关注向量方向而非长度，适合比较归一化后的文本向量。不过，“语义相近”不等于“事实正确”。如果资料中同时存在 v1.x 和 v2.3，且没有版本过滤，旧文档仍可能因为措辞更接近而排在前面。

因此，一个可维护的检索请求通常包含两部分：

  1. 用 query 文本做语义检索。
  2. 用 metadata 做结构化约束，例如 product=framework、version=2.3、status=published。

对内部知识库还应在检索前做权限过滤。不要先检索所有文档、再在提示词里要求模型“忽略无权限内容”，因为内容在那之前已经进入了检索结果链路。

HNSW：在大规模向量中快速找近邻

资料增长到数万、数十万甚至更多 chunk 后，逐一计算 query 与所有向量的相似度会越来越慢。HNSW（Hierarchical Navigable Small World）通过多层近邻图近似搜索：高层帮助快速定位区域，低层在附近继续细查。

它的关键取舍是延迟、内存和召回率：

  • M 越大，图中每个节点可连接的邻居更多，通常召回更好，但索引更占内存。
  • ef_construction 越高，建索引时搜索更充分，构建更慢，索引质量通常更好。
  • 查询阶段的 ef 越高，候选更多，召回率通常更高，但查询延迟也会上升。

HNSW 是近似最近邻索引，不保证每次都找到数学意义上的全局 Top K。对“回答是否合规”或“流程是否正确”这类高风险场景，应该用离线标注查询集测量 Recall@K，并为关键文档设置版本、状态和权限过滤，而不是仅依赖向量排名。

多路召回：让精确关键词和语义理解共同工作

只做向量检索会漏掉一些关键字。错误码、函数名、配置键、订单状态值往往要求精确匹配；反过来，只做关键词检索又难以覆盖同义表达。多路召回通常会并行运行：

  • 向量召回：处理自然语言改写和语义相近的问题。
  • 关键词/BM25 召回：保护 API 名、版本号、错误码、制度术语等精确 token。
  • metadata 过滤或业务规则召回：保证版本、租户、权限、文档状态等边界。

合并时可以采用 RRF（Reciprocal Rank Fusion）或归一化加权分数。RRF 对不同检索器的原始分数尺度不敏感，适合作为初始方案；后续再根据离线评估决定是否引入重排序模型。

可以这样搭建一个最小检索实验

下面的示例用多语言 Embedding、HNSW 向量索引和 BM25 关键词检索实现一个小型多路召回器。它是用于理解链路的最小实验：生产环境还需要文档解析、持久化、增量更新、metadata 过滤、权限控制和可观测性。

安装依赖：

python -m pip install sentence-transformers hnswlib rank-bm25 numpy

将以下内容保存为 rag_retrieval_demo.py 后运行 python rag_retrieval_demo.py。首次执行会下载 paraphrase-multilingual-MiniLM-L12-v2 模型；可替换为已在团队评估过、适合业务语种的 Embedding 模型。

import numpy as np
import hnswlib
from rank_bm25 import BM25Okapi
from sentence_transformers import SentenceTransformer

chunks = [
    {
        "id": 0,
        "version": "2.3",
        "text": "v2.3 客户端初始化：client = Client(base_url=URL, token=TOKEN)。",
    },
    {
        "id": 1,
        "version": "1.x",
        "text": "v1.x 客户端初始化：client = Client(API_KEY)。该写法已废弃。",
    },
    {
        "id": 2,
        "version": "2.3",
        "text": "退款完成后，未过期且未使用的优惠券会在 24 小时内返还到账户。",
    },
    {
        "id": 3,
        "version": "2.3",
        "text": "退款申请需要订单状态为 PAID，处理结果由退款服务异步通知。",
    },
]

query = "v2.3 版本的 Client 应该如何初始化？"
model = SentenceTransformer("paraphrase-multilingual-MiniLM-L12-v2")
texts = [item["text"] for item in chunks]
vectors = model.encode(texts, normalize_embeddings=True).astype("float32")nquery_vector = model.encode([query], normalize_embeddings=True).astype("float32")

# HNSW 的 cosine 距离等于 1 - cosine similarity。
index = hnswlib.Index(space="cosine", dim=vectors.shape[1])
index.init_index(max_elements=len(chunks), ef_construction=100, M=16)
index.add_items(vectors, np.arange(len(chunks)))
index.set_ef(50)
vector_ids, _ = index.knn_query(query_vector, k=3)

# 简化分词仅用于演示。中文生产环境应换成与语料匹配的分词或检索服务。
tokenized = [list(text) for text in texts]
bm25 = BM25Okapi(tokenized)
keyword_scores = bm25.get_scores(list(query))
keyword_ids = np.argsort(keyword_scores)[::-1][:3]

# RRF 按排名融合两路结果；版本过滤放在融合后演示。
def rrf_rank(result_lists, k=60):
    scores = {}
    for result_list in result_lists:
        for rank, item_id in enumerate(result_list, start=1):
            scores[int(item_id)] = scores.get(int(item_id), 0) + 1 / (k + rank)
    return sorted(scores, key=scores.get, reverse=True)

merged_ids = rrf_rank([vector_ids[0], keyword_ids])
results = [chunks[item_id] for item_id in merged_ids if chunks[item_id]["version"] == "2.3"]

for item in results[:3]:
    print(f"[v{item['version']}] {item['text']}")

注意代码中 vectors 定义后的 nquery_vector 应改为下一行的 query_vector，完整可运行版本如下：

query_vector = model.encode([query], normalize_embeddings=True).astype("float32")

这段实验的重点不在生成答案，而在于观察每一层结果：向量 Top K 是什么、BM25 Top K 是什么、融合后是否仍保留 v2.3 文档。把这些结果记录下来，才能判断问题出在文档、分块、模型、索引参数还是过滤规则。

上线前应验证什么

RAG 的目标不是让模型“更会说”，而是让它在可信证据不足时少说、在证据充分时引用正确版本的资料。上线前可以用一组真实问题检查：

  • 查询新版本 API 时，Top K 中是否仍混入废弃版本。
  • 同义问题是否能召回同一份制度或规范。
  • 函数名、错误码、订单状态等精确 token 是否能被关键词通道保住。
  • 无权限、草稿、过期文档是否在检索阶段被排除。
  • 没有足够证据时，生成层是否能够明确拒答或要求补充资料。

先建立小而可信的知识集和评估集，再逐步调节 chunk 大小、HNSW 参数与融合策略。只有检索层把正确、最新、可见的证据送进上下文，生成模型才有条件给出可执行的答案。

# AI # LLM # 全栈 # 后端 # 开源 # 开源中国

相关推荐