第3章搜索策略¶

学习目标¶

区分稠密检索（Dense）与稀疏检索（Sparse）的原理与适用场景
理解 BM25 算法及其在关键词检索中的角色
掌握混合检索（Hybrid Search）的实现与融合方法（RRF）
理解重排序（Reranker）的作用与 Cross-Encoder 原理
能够用 Python 实现多路召回 + Reranker 的两阶段检索

问题导入¶

某医疗知识库 RAG 上线后遇到一个怪现象：用户问"阿司匹林的禁忌症"，稠密向量检索返回的是"布洛芬的副作用"——语义相近但答非所问；而医生搜索"ICD-10 编码 I21.9"这种精确编码时，向量检索完全失效，因为编码本身没有语义。

这是单路检索的典型短板：稠密检索擅长语义匹配但弱于精确关键词；稀疏检索（BM25）擅长关键词匹配但无法理解语义。生产级 RAG 必须用混合检索（Hybrid Search）+ 重排序（Reranker）的两阶段架构，才能兼顾召回率与精度。本章系统讲解搜索策略。

核心概念¶

1. 稠密检索 vs 稀疏检索¶

维度	稠密检索（Dense）	稀疏检索（Sparse/BM25）
表示	1536 维稠密向量	高维稀疏词袋（词→权重）
匹配方式	语义相似度	关键词精确匹配
优势	理解同义词、近义、跨语言	精确词、专有名词、编码
劣势	弱于精确关键词	无法理解语义
模型依赖	需 Embedding 模型	无需模型（基于统计）
典型实现	OpenAI Embedding + Qdrant	Elasticsearch / BM25

2. BM25 算法¶

BM25（Best Matching 25）是经典的关键词检索算法，基于词频（TF）与逆文档频率（IDF）：

词频饱和：词频贡献随出现次数递减（避免堆词刷分）
文档长度归一：长文档词频天然偏高，需归一惩罚
IDF 权重：罕见词权重高，常见词（"的"、"是"）权重低

公式简化：score(D, Q) = Σ IDF(qi) · (f(qi,D) · (k1+1)) / (f(qi,D) + k1·(1 - b + b·|D|/avgdl))

典型参数 k1=1.2, b=0.75。BM25 在 Elasticsearch、Lucene 中是默认打分算法。

3. 混合检索（Hybrid Search）¶

同时执行稠密与稀疏检索，再融合结果。融合方法：

加权融合：final_score = α · dense_score + (1-α) · bm25_score，需先归一化两路分数。简单但对分数尺度敏感。

RRF（Reciprocal Rank Fusion）：score(d) = Σ 1 / (k + rank_i(d))，只看排名不看分数，鲁棒性强，是生产环境首选。典型 k=60。

4. 重排序（Reranker）¶

两阶段检索架构：

第一阶段（召回）：用高效的向量检索/BM25 召回 Top-20-100
第二阶段（精排）：用 Cross-Encoder 对 query 与每个候选拼接后联合编码打分，精选 Top-3-5

Bi-Encoder vs Cross-Encoder：

维度	Bi-Encoder（双塔）	Cross-Encoder（交叉）
编码方式	query 与 doc 独立编码	query+doc 拼接联合编码
速度	快（doc 可预编码）	慢（每对都要前向）
精度	中（交互浅）	高（深层交互）
角色	召回阶段	精排阶段

常见 Reranker：Cohere Rerank（API）、bge-reranker-large（开源）、jina-reranker。

5. 多路召回¶

除稠密+稀疏外，可再加更多召回通道：

元数据过滤召回：先按属性筛再向量检索
查询改写召回：用 LLM 生成多个子查询并行检索
HyDE 召回：让 LLM 先生成假设答案，用答案向量去检索

多路结果统一用 RRF 融合。

原理图¶

flowchart TD
    Query[用户查询] --> Dense[稠密检索<br/>Embedding + HNSW]
    Query --> Sparse[稀疏检索<br/>BM25]
    Query --> Rewritten[查询改写<br/>多子查询]

    Dense --> TopA[Top-50 候选]
    Sparse --> TopB[Top-50 候选]
    Rewritten --> TopC[Top-50 候选]

    TopA --> RRF[RRF 融合<br/>score=Σ 1/(k+rank)]
    TopB --> RRF
    TopC --> RRF

    RRF --> Merged[Top-20 合并候选]
    Merged --> Rerank[Cross-Encoder Reranker<br/>query+doc 联合打分]
    Rerank --> TopN[Top-5 精选]
    TopN --> Prompt[拼装 Prompt]

图：多路召回 + RRF 融合 + Reranker 两阶段检索

示例代码¶

实现稠密 + BM25 混合检索与 Reranker 精排：

import os
import math
from collections import defaultdict
from qdrant_client import QdrantClient
from openai import OpenAI

os.environ.setdefault("OPENAI_API_KEY", "sk-xxx")
oai = OpenAI()
qdrant = QdrantClient(":memory:")

COLLECTION = "kb"


def embed(text: str) -> list[float]:
    return oai.embeddings.create(
        model="text-embedding-3-small", input=text
    ).data[0].embedding


def dense_search(query: str, top_k: int = 20) -> list[dict]:
    """稠密向量检索。"""
    results = qdrant.search(
        collection_name=COLLECTION,
        query_vector=embed(query),
        limit=top_k,
        with_payload=True,
    )
    return [{"text": r.payload["text"], "score": r.score} for r in results]


def bm25_search(corpus: list[str], query: str, top_k: int = 20) -> list[dict]:
    """简易 BM25 实现（生产环境用 Elasticsearch/rank_bm25 库）。"""
    k1, b = 1.2, 0.75
    tokenized = [doc.split() for doc in corpus]
    avgdl = sum(len(d) for d in tokenized) / len(tokenized)
    df = defaultdict(int)
    for doc in tokenized:
        for w in set(doc):
            df[w] += 1
    idf = {w: math.log((len(corpus) - df[w] + 0.5) / (df[w] + 0.5) + 1) for w in df}
    q_tokens = query.split()
    scored = []
    for i, doc in enumerate(tokenized):
        score = 0.0
        for q in q_tokens:
            f = doc.count(q)
            if f == 0:
                continue
            score += idf.get(q, 0) * (f * (k1 + 1)) / (f + k1 * (1 - b + b * len(doc) / avgdl))
        scored.append({"text": corpus[i], "score": score})
    scored.sort(key=lambda x: x["score"], reverse=True)
    return scored[:top_k]


def rrf_fuse(rank_lists: list[list[dict]], k: int = 60, top_n: int = 20) -> list[dict]:
    """Reciprocal Rank Fusion 融合多路结果。"""
    scores = defaultdict(float)
    text_map = {}
    for ranks in rank_lists:
        for rank, item in enumerate(ranks):
            scores[item["text"]] += 1.0 / (k + rank + 1)
            text_map[item["text"]] = item
    fused = sorted(scores.items(), key=lambda x: x[1], reverse=True)[:top_n]
    return [{"text": t, "score": s} for t, s in fused]


def rerank(query: str, candidates: list[dict], top_n: int = 5) -> list[dict]:
    """用 Cohere Rerank API 精排（示例用，需 cohere 包）。"""
    # from cohere import Client
    # co = Client(os.environ["COHERE_API_KEY"])
    # docs = [c["text"] for c in candidates]
    # resp = co.rerank(model="rerank-multilingual-v3.0", query=query, documents=docs, top_n=top_n)
    # return [{"text": docs[r.index], "score": r.relevance_score} for r in resp.results]

    # 无 API 时用 Cross-Encoder 本地模型（需 sentence-transformers）
    from sentence_transformers import CrossEncoder
    model = CrossEncoder("BAAI/bge-reranker-base")
    pairs = [(query, c["text"]) for c in candidates]
    scores = model.predict(pairs)
    ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
    return [{"text": c["text"], "score": float(s)} for c, s in ranked[:top_n]]


def hybrid_search(query: str, corpus: list[str]) -> list[dict]:
    """两阶段混合检索：稠密 + BM25 → RRF → Reranker。"""
    dense = dense_search(query, top_k=20)
    sparse = bm25_search(corpus, query, top_k=20)
    fused = rrf_fuse([dense, sparse], top_n=20)
    return rerank(query, fused, top_n=5)


if __name__ == "__main__":
    corpus = [
        "阿司匹林用于预防心血管疾病。",
        "布洛芬是非甾体抗炎药，用于止痛。",
        "阿司匹林禁用于活动性溃疡患者。",
        "ICD-10 编码 I21.9 表示急性心肌梗死。",
    ]
    print(hybrid_search("阿司匹林禁忌症", corpus))

运行结果说明：纯稠密检索可能把"布洛芬止痛"排前面（语义接近"药物"），但加入 BM25 后"阿司匹林禁用于活动性溃疡"因关键词命中加权提升，再经 Reranker 精排后稳居 Top-1，体现了混合检索的价值。

企业案例¶

案例 1：客服中心高召回搜索¶

某运营商客服 RAG 从纯向量检索升级为 Hybrid + Cohere Rerank： - 召回率（Recall@5）从 78% → 92% - Top-1 命中率从 61% → 84% - 端到端延迟从 80ms → 220ms（可接受）客户满意度提升 18%，工单一次解决率提升 22%。

案例 2：法律条款精确检索¶

某法律平台对"民法典第1064条"这类精确编号检索，纯向量失效（编号无语义）。加入 BM25 通道后精确命中，再经 Reranker 把相关司法解释排前，律师检索效率提升 4 倍。

案例 3：多路召回调优¶

某电商搜索用"稠密 + BM25 + 查询改写"三路召回 + RRF 融合 + Reranker。查询改写让"夏天穿的薄外套"改写为"夏季薄款外套透气"，召回率比单路稠密提升 25%。

常见问题¶

Q1：Reranker 必须用吗？ A：不是必须，但强烈推荐。Reranker 通常能提升 Top-3 命中率 10-20 个百分点，成本是增加 50-150ms 延迟。对答案精度敏感的场景（医疗、法律）必上。

Q2：RRF 的 k 值怎么选？ A：k=60 是经验默认值，含义是排名第 1 与第 61 的权重比为 61:1。k 越大对低排名项越宽容，k 越小越强调头部。多数场景用 60 即可。

Q3：稠密和稀疏的权重怎么分配？ A：用 RRF 时无需调权重（只看排名）。用加权融合时，关键词型查询（编码、型号）给 BM25 更高权重，语义型查询给稠密更高权重。可用 LLM 判断查询类型动态调权。

Q4：Reranker 太慢怎么办？ A：减少 Reranker 输入候选数（从 50 降到 20）；用轻量模型（bge-reranker-base 而非 large）；批量并行调用；或用 Cohere API（GPU 加速）。

面试题¶

1. 为什么混合检索优于单一检索？ 参考答案：稠密检索擅长语义匹配但弱于精确关键词（如产品型号、ICD 编码）；BM25 擅长关键词但无法理解同义。两者互补，融合后召回率与精度同时提升，是生产 RAG 主流方案。

2. Reranker 与向量检索的区别？ 答题要点：向量检索用 Bi-Encoder，query 与 doc 独立编码，doc 可预编码入库，速度快但交互浅；Reranker 用 Cross-Encoder，query+doc 拼接联合编码，能捕获深层交互，精度高但慢，用于精排少量候选。

3. RRF 相比加权融合的优势？ 参考答案：RRF 只用排名不用分数，对不同检索器的分数尺度不敏感，无需归一化，对异常分鲁棒。加权融合需要两路分数尺度一致，调参成本高。生产环境首选 RRF。

4. 如何评估检索质量？ 答题要点：用标注的查询-相关文档对评测集，计算 Recall@K（召回率）、MRR（平均倒数排名）、NDCG（归一化折损累计增益）。检索与生成分开评估，避免混淆。

练习题¶

BM25 实现：用 rank_bm25 库替换本章手写 BM25，对比检索结果差异。
融合方法对比：对同一组多路召回结果，分别用 RRF（k=60）和加权融合（α=0.5）融合，对比 Top-5 排序差异。
Reranker 实验：用 bge-reranker-base 与 bge-reranker-large 在同一评测集上对比 Top-3 命中率与延迟。

小结¶

搜索策略是 RAG 的核心环节。本章介绍了稠密与稀疏两类检索、BM25 算法、混合检索与 RRF 融合、Cross-Encoder Reranker 两阶段架构。生产环境推荐"稠密 + BM25 双路召回 + RRF 融合 + Reranker 精排"作为默认方案，兼顾召回率与精度。下一章将讲解元数据管理与过滤查询，让检索支持更精细的业务控制。

第3章 搜索策略¶