Chapter 06

检索策略优化

超越基础向量检索——混合检索、Reranker 重排序、MMR 多样性,打造高精度的 Advanced RAG 检索层

为什么纯向量检索不够

稠密向量检索(Dense Retrieval)捕捉语义,但有两个明显局限:

稀疏 vs 稠密:互补而非替代

稀疏检索(BM25)擅长精确关键词匹配,对专有名词、编号、代码片段效果好。稠密检索(向量)擅长语义理解,能处理同义词、释义表达。混合检索融合两者优势,是目前 RAG 最佳实践。

稀疏检索:BM25 算法

BM25
Best Match 25,基于词频(TF)和逆文档频率(IDF)的关键词检索算法。是传统搜索引擎(Elasticsearch)的核心算法,对精确关键词匹配效果优秀。
TF-IDF
词频-逆文档频率。TF 衡量词在当前文档中出现的频率,IDF 衡量词在所有文档中的稀有程度。BM25 是 TF-IDF 的改进版,加入了文档长度归一化。
稀疏向量
大多数维度为零的高维向量。BM25 可以表示为稀疏向量:每个维度对应词汇表中的一个词,值为该词的 BM25 分数。Qdrant 1.7+ 原生支持稀疏向量存储和检索。
RRF(Reciprocal Rank Fusion)
倒数排名融合,混合检索的主流融合算法。将多个排名列表合并:每个文档的得分 = Σ(1 / (k + rank_i)),k 通常取 60。简单有效,不需要对齐不同检索器的分数量纲。

BM25 独立使用

# pip install rank-bm25
from rank_bm25 import BM25Okapi
import jieba  # 中文分词

# 文档集合
corpus = [
    "RAG 检索增强生成由 Meta AI 提出,解决 LLM 幻觉问题",
    "Qdrant 向量数据库使用 HNSW 索引实现高性能检索",
    "Embedding 模型将文本转换为高维向量,捕捉语义信息",
    "BGE-M3 支持稠密向量、稀疏向量和多向量混合检索",
]

# 中文分词(BM25 需要分词后的词列表)
tokenized_corpus = [list(jieba.cut(doc)) for doc in corpus]

# 构建 BM25 索引
bm25 = BM25Okapi(
    tokenized_corpus,
    k1=1.5,  # 词频饱和参数(默认 1.5)
    b=0.75   # 文档长度归一化系数(默认 0.75)
)

# 查询
query = "HNSW 索引如何工作?"
tokenized_query = list(jieba.cut(query))
scores = bm25.get_scores(tokenized_query)

# 输出排名
ranked_idx = sorted(range(len(scores)), key=lambda i: -scores[i])
for i in ranked_idx:
    print(f"分数 {scores[i]:.3f}: {corpus[i][:40]}")

混合检索(Hybrid Search)

方案一:RRF 融合(推荐)

from typing import TypeVar
import numpy as np

def reciprocal_rank_fusion(
    results_list: list[list[str]],
    k: int = 60
) -> list[tuple[str, float]]:
    """
    RRF 排名融合
    results_list: 多个检索器返回的文档 ID 列表(已排序)
    k: 平滑参数,默认 60
    """
    scores: dict[str, float] = {}

    for results in results_list:
        for rank, doc_id in enumerate(results, start=1):
            if doc_id not in scores:
                scores[doc_id] = 0.0
            scores[doc_id] += 1.0 / (k + rank)

    # 按融合分数降序排列
    return sorted(scores.items(), key=lambda x: -x[1])

# 示例:融合向量检索和 BM25 检索结果
dense_ids = ["doc_3", "doc_1", "doc_4", "doc_2"]   # 向量检索排名
sparse_ids = ["doc_1", "doc_5", "doc_3", "doc_2"]   # BM25 排名

fused = reciprocal_rank_fusion([dense_ids, sparse_ids])
print(fused)
# [('doc_1', 0.0306), ('doc_3', 0.0304), ('doc_2', 0.0157), ...]

方案二:Qdrant 原生混合检索

from qdrant_client import QdrantClient, models

client = QdrantClient(host="localhost", port=6333)

# Qdrant 1.7+ 支持稀疏向量 + 稠密向量混合
# 创建支持混合检索的 Collection
client.create_collection(
    collection_name="hybrid_collection",
    vectors_config={
        "dense": models.VectorParams(size=1536, distance=models.Distance.COSINE),
    },
    sparse_vectors_config={
        "sparse": models.SparseVectorParams(
            index=models.SparseIndexParams(on_disk=False)
        )
    }
)

# 写入稀疏+稠密向量(使用 BGE-M3 同时生成两种向量)
from FlagEmbedding import BGEM3FlagModel

m3_model = BGEM3FlagModel("BAAI/bge-m3", use_fp16=True)
output = m3_model.encode(["RAG 检索增强生成"], return_sparse=True)

dense_vec = output["dense_vecs"][0].tolist()
sparse_weights = output["lexical_weights"][0]  # {token_id: weight}

# 转换稀疏向量格式
sparse_vec = models.SparseVector(
    indices=list(sparse_weights.keys()),
    values=list(sparse_weights.values())
)

# 查询:混合检索
results = client.query_points(
    collection_name="hybrid_collection",
    prefetch=[
        models.Prefetch(
            query=dense_vec,     # 稠密向量检索
            using="dense",
            limit=20,
        ),
        models.Prefetch(
            query=sparse_vec,    # 稀疏向量检索
            using="sparse",
            limit=20,
        ),
    ],
    query=models.FusionQuery(fusion=models.Fusion.RRF),  # RRF 融合
    limit=5,
)

Reranker 重排序

向量检索的第一阶段(粗排)用双编码器快速召回 20–50 个候选,第二阶段(精排)用交叉编码器对候选重新精确排序。两阶段组合可以在保持速度的同时大幅提升精度。

# pip install sentence-transformers
from sentence_transformers import CrossEncoder
from typing import NamedTuple

# 加载交叉编码器(中英文通用)
reranker = CrossEncoder(
    "BAAI/bge-reranker-v2-m3",  # 支持中文,效果优秀
    max_length=512,
)

def rerank(
    query: str,
    candidates: list[str],
    top_k: int = 5,
) -> list[tuple[str, float]]:
    """
    交叉编码器重排序
    返回:[(文档内容, 相关性分数), ...],按分数降序
    """
    # 构建 (query, passage) 对
    pairs = [(query, doc) for doc in candidates]

    # 批量预测相关性分数
    scores = reranker.predict(
        pairs,
        batch_size=16,
        show_progress_bar=False,
    )

    # 排序并返回 Top-K
    ranked = sorted(zip(candidates, scores), key=lambda x: -x[1])
    return ranked[:top_k]

# 两阶段检索示例
query = "HNSW 索引的工作原理是什么?"

# 第一阶段:向量检索召回 20 个候选
candidates = vector_search(query, top_k=20)

# 第二阶段:Reranker 精排,取 Top-5
final_results = rerank(query, candidates, top_k=5)

for doc, score in final_results:
    print(f"[{score:.3f}] {doc[:60]}")
Reranker 的性价比

Reranker 通常只需 50–100ms 处理 20 个候选,而精度提升通常在 5–15 个百分点。对于生产 RAG 系统,Reranker 是投入产出比最高的优化手段之一,强烈推荐。

MMR:最大边际相关性

MMR(Maximal Marginal Relevance)算法解决 Top-K 结果高度重复的问题,在相关性和多样性之间取得平衡。

import numpy as np

def mmr(
    query_vec: np.ndarray,
    doc_vecs: np.ndarray,
    doc_texts: list[str],
    lambda_param: float = 0.5,  # 0=最多样,1=最相关
    top_k: int = 5,
) -> list[str]:
    """
    MMR 算法:平衡相关性与多样性
    lambda=1.0: 等价于纯相关性排序
    lambda=0.0: 等价于最大化多样性
    """
    # 计算所有文档与查询的相似度
    query_sims = doc_vecs @ query_vec

    selected = []
    remaining = list(range(len(doc_texts)))

    while len(selected) < top_k and remaining:
        if not selected:
            # 第一个选最相关的
            best = max(remaining, key=lambda i: query_sims[i])
        else:
            # 后续选"相关性高 AND 与已选文档相似度低"的
            best_score = -np.inf
            best = remaining[0]
            selected_vecs = doc_vecs[selected]

            for i in remaining:
                # 相关性:与查询的相似度
                relevance = query_sims[i]
                # 冗余度:与已选文档的最大相似度
                redundancy = (selected_vecs @ doc_vecs[i]).max()
                # MMR 分数
                mmr_score = lambda_param * relevance - (1 - lambda_param) * redundancy

                if mmr_score > best_score:
                    best_score = mmr_score
                    best = i

        selected.append(best)
        remaining.remove(best)

    return [doc_texts[i] for i in selected]

查询改写(Query Rewriting)

用 LLM 将用户的原始问题改写为更适合检索的多个查询,可以大幅提升召回率。

from openai import OpenAI

client = OpenAI()

def multi_query_rewrite(question: str, n: int = 3) -> list[str]:
    """生成 N 个不同角度的查询变体"""
    prompt = f"""你是信息检索专家。为以下问题生成 {n} 个不同角度的查询变体,
用于在知识库中检索相关文档。每行一个查询,不要编号。

原问题:{question}

查询变体:"""

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.7,
    )

    variants = response.choices[0].message.content.strip().split("\n")
    return [question] + [v.strip() for v in variants if v.strip()]

# 使用多查询检索
question = "RAG 系统的性能瓶颈在哪里?"
queries = multi_query_rewrite(question)
# ["RAG 系统的性能瓶颈在哪里?",
#  "RAG 响应速度慢的原因",
#  "如何优化 RAG 系统的检索延迟",
#  "RAG 吞吐量限制因素分析"]

# 对每个查询检索,用 RRF 融合结果
all_results = [vector_search(q, top_k=10) for q in queries]
fused = reciprocal_rank_fusion(all_results)

本章总结