3.2 检索算法改进

3.2 检索算法改进 — RAG高级优化核心技巧

本节导读：深入理解RAG检索算法的三种核心范式（关键词、语义、混合），掌握查询改写、重排序、元数据过滤等高级技巧。学完本节，你能设计出一套针对业务场景定制的检索策略，显著提升检索的准确率和召回率。

学习目标

• 理解BM25和向量检索的原理差异和互补关系
• 掌握混合检索（Hybrid Search）的设计思路和权重调优方法
• 学会查询改写（Query Rewriting）的四种实现方式
• 掌握重排序（Reranking）的原理和Cross-Encoder实现
• 能够基于业务场景设计完整的检索管线

核心概念

检索是RAG系统的瓶颈

在RAG系统中，检索环节是决定最终输出质量的关键瓶颈。我用一个数据来说明这个问题：在一个典型的RAG问答系统中，如果检索阶段返回的Top-10结果中没有包含正确答案，那么无论后续的生成模型多么强大，都不可能产生正确的回答。

检索算法的改进方向可以归纳为两个核心指标：

• 召回率（Recall）：确保相关文档不被漏掉
• 精度（Precision）：确保返回的文档确实相关

这两个指标天然矛盾——提高召回率通常会增加无关结果，提高精度又会漏掉相关结果。好的检索算法就是在这两者之间找到最佳平衡点。

上图展示了现代RAG系统的典型检索管线。每一个环节都是优化点，本节将逐一讲解。

环境准备 / 前置知识

本节代码示例基于以下环境：

• Python 3.10+
• sentence-transformers（嵌入模型）
• FlagEmbedding（Cross-Encoder重排序）
• rank_bm25（BM25实现）
• langchain（检索链路框架）

# 安装依赖
# pip install sentence-transformers FlagEmbedding rank_bm25 langchain
​

分步实战

步骤 1：理解BM25——被低估的关键词检索

BM25（Best Matching 25）是基于概率检索模型的经典算法，至今仍是搜索领域的基石。在RAG系统中，很多人认为有了向量检索就不需要BM25了，这是一个常见的误区。

BM25的核心原理

BM25的打分公式为：

Score(D, Q) = Σ IDF(qi) × [f(qi, D) × (k1 + 1)] / [f(qi, D) + k1 × (1 - b + b × |D| / avgdl)]
​

其中：

• f(qi, D) 是词项 qi 在文档 D 中的词频
• |D| 是文档长度，avgdl 是平均文档长度
• k1 控制词频饱和度（通常取 1.2-2.0）
• b 控制文档长度归一化（通常取 0.75）
• IDF(qi) 是词项的逆文档频率

BM25在RAG中的独特优势

BM25有几个向量检索不具备的优势：

1. 精确匹配能力：当用户查询包含专有名词、产品型号、错误代码等精确关键词时，BM25远优于语义检索
2. 零训练成本：不需要预训练模型和GPU推理
3. 可解释性强：可以直接看到哪些关键词匹配上了
4. 补充长尾词：对于低频但高度区分性的词汇，BM25的IDF加权天然有利

实现BM25检索

from rank_bm25 import BM25Okapi
import jieba

def build_bm25_index(documents):
    """构建BM25索引
    
    Args:
        documents: 文档列表，每个元素为字符串
    
    Returns:
        bm25对象和分词后的文档列表
    """
    tokenized_docs = [list(jieba.cut(doc)) for doc in documents]
    bm25 = BM25Okapi(tokenized_docs)
    return bm25, tokenized_docs

def bm25_search(bm25, tokenized_docs, query, top_k=10):
    """BM25检索
    
    Args:
        bm25: BM25索引
        tokenized_docs: 分词后的文档
        query: 查询文本
        top_k: 返回结果数
    
    Returns:
        包含文档索引和分数的结果列表
    """
    tokenized_query = list(jieba.cut(query))
    scores = bm25.get_scores(tokenized_query)
    
    # 获取Top-K结果
    top_indices = sorted(range(len(scores)), key=lambda i: scores[i], reverse=True)[:top_k]
    return [(idx, scores[idx]) for idx in top_indices]

# 使用示例
documents = [
    "RAG系统的检索模块需要支持混合检索策略",
    "BM25算法是传统信息检索的经典方法",
    "向量检索使用嵌入模型将文本映射到高维空间",
    "重排序模块使用交叉编码器对检索结果进行精排",
]

bm25, tokenized_docs = build_bm25_index(documents)
results = bm25_search(bm25, tokenized_docs, "RAG检索策略", top_k=2)
for idx, score in results:
    print(f"文档 {idx} (分数: {score:.4f}): {documents[idx][:50]}")
​

步骤 2：混合检索——BM25与向量的1+1>2

混合检索的核心思想是：BM25擅长精确关键词匹配，向量检索擅长语义理解，两者结合可以覆盖更全面的相关文档。

实现RRF分数融合

Reciprocal Rank Fusion（RRF）是混合检索中最常用的分数融合方法。它的原理很简单：不使用原始分数，而是使用排名的倒数作为融合依据。

def reciprocal_rank_fusion(
    result_lists,
    weights=None,
    k=60
):
    """Reciprocal Rank Fusion 多路检索结果融合
    
    Args:
        result_lists: 多路检索结果列表，每个元素为 [(doc_id, score), ...]
        weights: 各路结果的权重，默认均等
        k: RRF平滑常数，通常取60
    
    Returns:
        融合后的 [(doc_id, fused_score), ...]，按分数降序排列
    """
    if weights is None:
        weights = [1.0] * len(result_lists)
    
    fused_scores = {}
    
    for results, weight in zip(result_lists, weights):
        for rank, (doc_id, score) in enumerate(results):
            if doc_id not in fused_scores:
                fused_scores[doc_id] = 0
            # RRF公式：1 / (k + rank)，rank从0开始
            fused_scores[doc_id] += weight / (k + rank + 1)
    
    # 按融合分数降序排列
    sorted_results = sorted(
        fused_scores.items(),
        key=lambda x: x[1],
        reverse=True
    )
    return sorted_results

# 使用示例
bm25_results = [(0, 5.2), (2, 3.8), (1, 2.1)]  # BM25返回的Top-K
vector_results = [(3, 0.92), (1, 0.85), (0, 0.78)]  # 向量检索返回的Top-K

fused = reciprocal_rank_fusion(
    [bm25_results, vector_results],
    weights=[0.5, 0.5]
)

print("融合结果:")
for doc_id, score in fused:
    print(f"  文档 {doc_id}: RRF分数 {score:.4f}")
​

混合检索的权重调优

BM25和向量检索的权重比例需要根据业务场景调整：

调整建议：从 0.5:0.5 开始，准备 20-50 个标注好的查询-文档对，计算不同权重下的 NDCG@10（归一化折损累计增益），选择 NDCG 最高的权重组合。

步骤 3：查询改写——让查询更好地匹配文档

用户原始查询往往存在表述模糊、缺少关键词、缺少上下文等问题。查询改写的目标是在检索之前对查询进行优化，使其更准确地反映用户的真实信息需求。

策略一：查询扩展（Query Expansion）

通过添加同义词、相关术语来扩展查询，提高召回率。

# 使用LLM进行查询扩展
def query_expansion(query, llm):
    """使用LLM生成查询的同义改写"""
    prompt = f"""请为以下查询生成3个语义相同但表述不同的改写版本，
每个改写应该包含不同的关键词和角度。

原始查询：{query}

请按以下JSON格式返回：
["改写1", "改写2", "改写3"]

只返回JSON数组，不要其他内容。"""
    
    response = llm.invoke(prompt)
    import json
    expansions = json.loads(response.content)
    return [query] + expansions

# 使用示例
# expanded_queries = query_expansion("如何优化RAG的检索效果", llm)
# 然后对每个扩展查询分别检索，合并结果
​

策略二：假设性文档嵌入（HyDE）

HyDE（Hypothetical Document Embedding）的核心思想是：先让LLM根据查询生成一个假设性的回答文档，然后用这个假设文档的向量去检索，而不是用查询本身的向量。

这种方法在以下场景特别有效：

• 查询很短但需要匹配很长的文档段落
• 查询和文档的表述方式差异很大

def hyde_retrieval(query, llm, embedding_model, vector_store, top_k=5):
    """使用HyDE策略进行检索
    
    1. 让LLM生成假设性回答
    2. 用假设回答的向量去检索
    """
    # 第一步：生成假设性回答
    prompt = f"""请根据以下问题，写一段可能包含答案的专业文档片段。
不需要确认信息是否准确，只需要风格和内容范围接近真实文档。

问题：{query}

请直接输出文档片段，不要加前缀说明。"""
    
    hypothetical_doc = llm.invoke(prompt).content
    
    # 第二步：用假设文档的向量检索
    query_embedding = embedding_model.encode(hypothetical_doc)
    results = vector_store.search(query_embedding, top_k=top_k)
    
    return results, hypothetical_doc
​

策略三：多查询检索（Multi-Query Retrieval）

将用户的查询从不同角度拆分成多个子查询，分别检索后合并结果。这尤其适合复杂的复合型查询。

def multi_query_decomposition(query, llm):
    """将复杂查询分解为多个简单子查询"""
    prompt = f"""将以下复杂查询分解为2-4个更具体的子查询，
每个子查询应该覆盖原始查询的一个方面。

原始查询：{query}

请按JSON格式返回子查询列表：
["子查询1", "子查询2", "子查询3"]

只返回JSON数组。"""
    
    response = llm.invoke(prompt)
    import json
    sub_queries = json.loads(response.content)
    return sub_queries
​

查询改写策略选择指南

步骤 4：重排序——检索管线的最后一道质量关卡

初筛（如BM25+向量混合检索）返回的结果可能有几十到几百条，其中包含不少相关性一般的结果。重排序模块使用更精细但更慢的模型，对初筛结果进行二次排序，只保留真正高相关的Top-K。

Cross-Encoder vs Bi-Encoder

理解重排序，首先要理解两种编码器的区别：

subgraph Cross_Encoder["Cross-Encoder（重排序阶段使用）"]
    Q2[查询] --> CE[交叉编码器] --> SC[相关性分数]
    D2[文档] --> CE
end
​

</div>

Bi-Encoder（双编码器）：查询和文档分别编码，计算向量相似度。优点是文档向量可以预计算、检索速度快。缺点是查询和文档没有交互，语义匹配精度有限。

Cross-Encoder（交叉编码器）：将查询和文档拼接后一起编码，直接输出相关性分数。优点是精度高（查询和文档在注意力层充分交互），缺点是每次需要重新计算、速度慢。

这就是为什么我们用 Bi-Encoder 做初筛（快），用 Cross-Encoder 做重排序（准）。

#### 使用 FlagEmbedding 实现重排序

```python
from FlagEmbedding import FlagReranker

# 初始化重排序模型（首次运行会自动下载）
reranker = FlagReranker('BAAI/bge-reranker-v2-m3', use_fp16=True)

def rerank(query, documents, top_k=5):
    """对检索结果进行重排序
    
    Args:
        query: 用户查询
        documents: 候选文档列表
        top_k: 返回Top-K结果
    
    Returns:
        重排序后的 [(index, score, document), ...]
    """
    # 构造查询-文档对
    pairs = [[query, doc] for doc in documents]
    
    # 计算相关性分数
    scores = reranker.compute_score(pairs, normalize=True)
    
    # 按分数降序排列
    scored_docs = list(enumerate(scores))
    scored_docs.sort(key=lambda x: x[1], reverse=True)
    
    return [(idx, score, documents[idx]) for idx, score in scored_docs[:top_k]]

# 使用示例
query = "RAG系统如何提升检索准确率"
candidates = [
    "RAG系统的检索准确率受到多种因素影响，包括分块策略、嵌入模型选择和检索算法等",
    "深度学习在自然语言处理领域的应用正在快速发展",
    "通过混合检索和重排序可以有效提升RAG系统的检索质量",
    "Python是一种流行的编程语言",
]

results = rerank(query, candidates, top_k=2)
for idx, score, doc in results:
    print(f"排名 {idx+1} (分数: {score:.4f}): {doc[:60]}...")
​

# 在LangChain中使用元数据过滤
from langchain_community.vectorstores import FAISS

# 构建带元数据的文档
documents_with_metadata = [
    {"content": "API网关的配置说明...", "metadata": {"chapter": "部署", "type": "技术文档"}},
    {"content": "系统监控指标定义...", "metadata": {"chapter": "运维", "type": "技术文档"}},
    {"content": "产品定价方案...", "metadata": {"chapter": "商务", "type": "产品信息"}},
]

# 检索时指定元数据过滤条件
results = vectorstore.similarity_search(
    "如何配置API网关",
    k=5,
    filter={"chapter": "部署"}  # 只在"部署"章节中搜索
)
​