4.2 检索策略设计

4.2 检索策略设计 — RAG 知识库实战检索架构

本节导读：RAG 系统的检索质量不只取决于相似度算法本身，更取决于你怎么组织检索流程。本节讲解多路召回、查询改写、元数据过滤等核心检索策略的设计与实现，帮你把检索准确率从"能用"提升到"好用"。

学习目标

• 理解单路检索的局限性，掌握多路召回的设计思路
• 学会实现查询改写（Query Rewriting）提升语义匹配质量
• 掌握元数据过滤与结构化检索结合的方法
• 了解查询扩展、假设文档嵌入（HyDE）等高级策略
• 能够根据业务场景设计完整的检索策略组合

核心概念

在 RAG 系统中，"检索策略"解决的核心问题是：用户的原始查询往往不是最优的检索输入。

用户的问题可能太短、太模糊、表述角度与文档不同，或者需要的答案分散在多个文档中。一个只做"查询 → 向量化 → Top-K"的简单检索器，在真实场景中召回率往往只有 40-60%。好的检索策略通过多路召回、查询变换和智能过滤，能把召回率提升到 80% 以上。

上图展示了一个典型的多路召回架构。它的核心思想是：不要把所有鸡蛋放在一个篮子里，用不同角度的检索路径覆盖更多可能的正确答案。

检索策略的三个层次

本节重点讲解优化层和高级层的策略设计。

环境准备

# 核心依赖
langchain>=0.3.0          # 查询改写工具
langchain-openai>=0.2.0   # OpenAI 集成
numpy>=1.24.0
​

前置知识：本教程 4.1 节相似度计算算法（理解底层计算原理）、基本的向量检索概念。

分步实战

步骤 1：查询改写——让模糊查询变清晰

用户查询常见的三类问题：

1. 指代不清："它的价格是多少"——"它"指什么？
2. 过于简短："RAG"——用户想了解什么方面？
3. 表述偏移：用户问"怎么让 AI 不胡说"，但文档写的是"减少大语言模型幻觉的方法"

查询改写用 LLM 在检索前对查询做一次"翻译"，让查询的语义表达更接近文档的语言风格。

基础查询改写实现：

from openai import OpenAI

client = OpenAI()  # 需配置 OPENAI_API_KEY

REWRITE_PROMPT = """你是一个搜索查询优化专家。请将用户的原始查询改写为更适合语义检索的查询。

规则：
1. 补充上下文和具体细节，但不要改变用户意图
2. 使用与知识库文档风格一致的专业术语
3. 输出 2-3 个改写后的查询，每行一个
4. 如果原始查询已经很清晰，可以保留原始查询作为其中一个

原始查询：{query}

改写后的查询："""


def rewrite_query(query: str, model: str = "gpt-4o-mini") -> list[str]:
    """
    使用 LLM 对查询进行改写，返回多个候选查询。
    
    Args:
        query: 用户原始查询
        model: 使用的 LLM 模型
    
    Returns:
        改写后的查询列表（包含原始查询）
    """
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": REWRITE_PROMPT.format(query=query)}],
        temperature=0.3,
        max_tokens=300,
    )
    
    rewritten = response.choices[0].message.content.strip().split("\n")
    rewritten = [q.strip().lstrip("0123456789.-、 ") for q in rewritten if q.strip()]
    # 确保原始查询也在列表中
    if query not in rewritten:
        rewritten.append(query)
    return rewritten


# 示例
queries = rewrite_query("RAG 怎么优化")
for i, q in enumerate(queries, 1):
    print(f"  查询 {i}: {q}")
# 可能输出：
#   查询 1: 如何优化 RAG 检索增强生成系统的性能和准确率
#   查询 2: RAG 系统性能优化方法与最佳实践
#   查询 3: 提升 RAG 知识库检索质量和生成效果的策略
#   查询 4: RAG 怎么优化
​

关键设计决策：查询改写应该用轻量级模型（如 GPT-4o-mini），因为这是高频调用路径。改写本身不需要很强的推理能力，但需要理解语义。改写 3-5 个查询是一个合理的平衡点——太少覆盖不够，太多会引入噪声且增加延迟。

步骤 2：假设文档嵌入（HyDE）——生成假文档做检索

HyDE（Hypothetical Document Embedding）的核心思想很巧妙：与其改写查询，不如让 LLM 先"假装"回答这个问题，生成一段假设性的答案文档，然后用这段假文档去检索真实文档。因为假文档的语言风格更接近真实答案，所以与真实相关文档的向量距离更近。

HYDE_PROMPT = """请根据以下问题，写一段 100-200 字的专业技术回答。
要求：使用专业术语，像技术文档一样写作，不需要完全准确。

问题：{query}

回答："""


def generate_hypothetical_doc(query: str, model: str = "gpt-4o-mini") -> str:
    """生成假设性文档用于检索。"""
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": HYDE_PROMPT.format(query=query)}],
        temperature=0.7,  # 稍高温度增加多样性
        max_tokens=500,
    )
    return response.choices[0].message.content.strip()


# 使用示例
query = "FAISS 的 HNSW 索引怎么调参"
hypo_doc = generate_hypothetical_doc(query)
print(f"假设文档：{hypo_doc[:150]}...")

# 用假设文档的向量（而非查询向量）去检索
hypo_vector = embedding_model.embed(hypo_doc)
results = retriever.search(hypo_vector, top_k=5)
​

HyDE 的适用与不适用场景：

• 适合：知识密集型查询（"解释 XX 原理"）、需要专业术语的领域查询
• 不适合：事实型查询（"XX 的作者是谁"）、短答案查询——假文档可能引入偏差
• 实践建议：与原始查询检索做融合，而不是完全替代。原始查询捕获精确匹配，HyDE 捕获语义匹配

步骤 3：多路召回——从多个角度捞文档

多路召回的核心是把不同检索方式的结果合并。最常见的是"稠密 + 稀疏"双路召回。

import numpy as np
from typing import List, Tuple


def reciprocal_rank_fusion(
    result_lists: List[List[Tuple[int, float]]],
    k: int = 60,
) -> List[Tuple[int, float]]:
    """
    RRF（Reciprocal Rank Fusion）多路结果融合算法。
    
    这是目前最稳定、最常用的多路召回融合方法。
    核心公式：score(d) = Σ 1/(k + rank(d))
    
    Args:
        result_lists: 多路检索结果列表，每路是 [(doc_id, score), ...]
        k: 平滑常数（默认 60，来自原始论文推荐值）
    
    Returns:
        融合后的 [(doc_id, rrf_score), ...]，按分数降序
    """
    rrf_scores = {}
    
    for results in result_lists:
        for rank, (doc_id, _) in enumerate(results, 1):
            if doc_id not in rrf_scores:
                rrf_scores[doc_id] = 0.0
            rrf_scores[doc_id] += 1.0 / (k + rank)
    
    # 按融合分数降序排列
    sorted_results = sorted(rrf_scores.items(), key=lambda x: -x[1])
    return sorted_results


# 使用示例：稠密检索 + 稀疏检索（BM25）
dense_results = [(101, 0.85), (205, 0.78), (312, 0.72), (418, 0.65), (523, 0.60)]
sparse_results = [(205, 8.5), (418, 7.2), (101, 6.8), (630, 5.5), (744, 4.8)]

fused = reciprocal_rank_fusion([dense_results, sparse_results])
print("RRF 融合结果：")
for rank, (doc_id, score) in enumerate(fused[:5], 1):
    print(f"  #{rank} 文档ID={doc_id}, RRF分数={score:.4f}")
# 输出（示例）：
#   #1 文档ID=205, RRF分数=0.0323   <- 两路都排在前列
#   #2 文档ID=101, RRF分数=0.0315
#   #3 文档ID=418, RRF分数=0.0283
​

RRF 的核心优势：它不依赖各路检索的绝对分数（不同检索方式的分数尺度不同，无法直接比较），只依赖排名。这意味着你可以把余弦相似度的结果和 BM25 的结果直接融合，不需要做分数归一化。这是 RRF 在工业界被广泛采用的原因——简单且鲁棒。

步骤 4：元数据过滤——精确定位文档范围

在知识库中，文档通常携带元数据（日期、部门、类别、语言等）。元数据过滤在向量检索之前缩小候选范围，既提升精度又减少计算量。

from typing import Dict, Any, Optional


class MetadataFilter:
    """
    文档元数据过滤器。
    在向量检索前缩小候选范围。
    """
    
    def __init__(self, documents: List[Dict[str, Any]]):
        """
        Args:
            documents: 文档列表，每个文档包含 'id' 和 'metadata' 字段
        """
        self.documents = documents
        # 构建元数据索引
        self._index = {}
        for doc in documents:
            for key, value in doc.get('metadata', {}).items():
                if key not in self._index:
                    self._index[key] = {}
                if value not in self._index[key]:
                    self._index[key][value] = []
                self._index[key][value].append(doc['id'])
    
    def filter(
        self,
        conditions: Dict[str, Any],
        match_mode: str = 'and'
    ) -> List[int]:
        """
        按元数据条件过滤文档。
        
        Args:
            conditions: 过滤条件，如 {"category": "技术", "year": 2026}
            match_mode: 'and' 所有条件必须满足，'or' 任一条件满足
        
        Returns:
            符合条件的文档 ID 列表
        """
        result_sets = []
        for key, value in conditions.items():
            if key in self._index and value in self._index[key]:
                result_sets.append(set(self._index[key][value]))
            else:
                result_sets.append(set())  # 无匹配
        
        if not result_sets:
            return []
        
        if match_mode == 'and':
            return list(set.intersection(*result_sets))
        else:  # 'or'
            return list(set.union(*result_sets))
    
    def filtered_retrieval(
        self,
        query_vector: np.ndarray,
        retriever,
        conditions: Dict[str, Any],
        top_k: int = 5,
    ) -> List[Tuple[int, float]]:
        """
        先按元数据过滤，再在候选集内做向量检索。
        """
        candidate_ids = self.filter(conditions)
        if not candidate_ids:
            return []
        
        # 获取候选集的向量
        candidate_vectors = np.array([
            retriever._index[i] for i in candidate_ids
        ]) if hasattr(retriever._index, '__getitem__') else None
        
        # 在候选集内计算相似度
        scores = candidate_vectors @ query_vector
        top_indices = np.argsort(-scores)[:top_k]
        
        return [(candidate_ids[i], float(scores[i])) for i in top_indices]


# 使用示例
documents_with_meta = [
    {"id": 0, "metadata": {"category": "技术", "year": 2026, "lang": "zh"}},
    {"id": 1, "metadata": {"category": "产品", "year": 2026, "lang": "zh"}},
    {"id": 2, "metadata": {"category": "技术", "year": 2025, "lang": "en"}},
    {"id": 3, "metadata": {"category": "技术", "year": 2026, "lang": "zh"}},
]

mf = MetadataFilter(documents_with_meta)

# 只检索 2026 年的中文技术文档
candidate_ids = mf.filter({"category": "技术", "year": 2026, "lang": "zh"})
print(f"符合条件文档: {candidate_ids}")  # [0, 3]
​

元数据过滤的最佳实践：

• 过滤条件应尽量精确（避免过滤后候选集为空）
• 当用户查询隐含过滤条件时（如"2026 年最新的 RAG 优化方法"），用 LLM 自动提取元数据条件
• 元数据字段尽量选择低基数的（如部门、类别），高基数字段（如精确日期）用范围查询更合适

步骤 5：查询扩展——补充同义词和相关概念

查询扩展通过补充同义词、上位词和相关概念来扩大检索覆盖面，解决"用户用的词和文档用的词不一样"的问题。

QUERY_EXPAND_PROMPT = """为以下搜索查询生成 3-5 个相关的扩展查询词或短语。
这些扩展应覆盖：
1. 同义词或近义词
2. 相关专业术语
3. 更宽泛的上位概念

原始查询：{query}

扩展查询（每行一个）："""


def expand_query(query: str, model: str = "gpt-4o-mini") -> list[str]:
    """生成查询扩展词。"""
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": QUERY_EXPAND_PROMPT.format(query=query)}],
        temperature=0.3,
        max_tokens=200,
    )
    
    expansions = response.choices[0].message.content.strip().split("\n")
    return [e.strip().lstrip("0123456789.-、 ") for e in expansions if e.strip()]


# 示例
query = "如何减少 LLM 幻觉"
expansions = expand_query(query)
print(f"原始查询: {query}")
for e in expansions:
    print(f"  扩展: {e}")
# 可能输出：
#   扩展: 大语言模型事实性错误
#   扩展: LLM hallucination mitigation
#   扩展: 提高生成内容准确性
#   扩展: 知识增强生成防幻觉
​

完整示例：多策略组合检索器

将上述策略组合成一个完整的检索器：

class MultiStrategyRetriever:
    """
    多策略组合检索器。
    集成查询改写、HyDE、多路召回和元数据过滤。
    """
    
    def __init__(
        self,
        dense_retriever,
        sparse_retriever=None,
        use_rewrite: bool = True,
        use_hyde: bool = False,
        use_metadata_filter: bool = True,
    ):
        self.dense_retriever = dense_retriever
        self.sparse_retriever = sparse_retriever
        self.use_rewrite = use_rewrite
        self.use_hyde = use_hyde
        self.use_metadata_filter = use_metadata_filter
        self.metadata_filter = None
    
    def retrieve(
        self,
        query: str,
        top_k: int = 5,
        metadata_conditions: dict = None,
    ) -> list:
        """执行多策略检索。"""
        all_results = []
        
        # 1. 查询改写（如果启用）
        queries = [query]
        if self.use_rewrite:
            queries = rewrite_query(query)
        
        # 2. 对每个查询做稠密检索
        for q in queries[:3]:  # 最多用 3 个改写查询
            q_vec = self.dense_retriever.embed(q)
            results = self.dense_retriever.search(q_vec, top_k=top_k * 2)
            all_results.append(results)
        
        # 3. HyDE 检索（如果启用）
        if self.use_hyde:
            hypo_doc = generate_hypothetical_doc(query)
            hypo_vec = self.dense_retriever.embed(hypo_doc)
            hypo_results = self.dense_retriever.search(hypo_vec, top_k=top_k)
            all_results.append(hypo_results)
        
        # 4. 稀疏检索（如果可用）
        if self.sparse_retriever:
            sparse_results = self.sparse_retriever.search(query, top_k=top_k * 2)
            all_results.append(sparse_results)
        
        # 5. RRF 融合
        fused = reciprocal_rank_fusion(all_results)
        
        # 6. 元数据过滤（如果启用且有条件）
        if self.use_metadata_filter and metadata_conditions and self.metadata_filter:
            valid_ids = set(self.metadata_filter.filter(metadata_conditions))
            fused = [(did, score) for did, score in fused if did in valid_ids]
        
        return fused[:top_k]
​

常见问题 FAQ

Q1：多路召回一定比单路好吗？什么情况下不需要多路？

A：不一定。如果你的知识库领域非常集中（比如只存了公司产品的 FAQ），文档语言风格统一，单路稠密检索就可能达到 90%+ 的准确率。多路召回的额外开销（延迟、API 调用成本、融合逻辑的复杂度）在简单场景下不值得。建议：先用单路检索跑基准测试（用 50-100 个真实查询评估 Recall@10），如果低于 70% 再引入多路召回。

Q2：查询改写会增加多少延迟？怎么控制？

A：一次 LLM 查询改写通常增加 200-500ms 延迟（取决于模型和 API 响应速度）。控制方法：用缓存（相同或相似查询命中缓存则跳过改写）、用更快的模型（GPT-4o-mini 比 GPT-4 快 5-10 倍）、异步并行执行（改写和第一步检索可以流水线化）。

Q3：HyDE 和查询改写可以同时用吗？

A：可以，但要注意成本。HyDE 本质上也是一种查询变换，与查询改写是互补的：查询改写生成"更好的查询"，HyDE 生成"假设的答案文档"。两者同时使用时，建议在融合阶段降低 HyDE 结果的权重，因为假文档毕竟不是真实文档，可能引入方向偏差。

Q4：RRF 的 k=60 是怎么来的？可以调整吗？

A：k=60 来自 RRF 原始论文（Cormack et al., 2009）的实验推荐值，在多数场景下表现稳定。k 值越大，排名靠后的结果对融合分数的贡献越大（更"民主"）；k 值越小，排名靠前的结果主导（更"精英"）。实践中 k=50-80 都是可以接受的，差异通常在 1-2 个百分点以内。

最佳实践与避坑

• 先单路后多路：不要一上来就上复杂的多路召回，先用最简单的稠密 Top-K 建立基准线，再逐步叠加策略
• 改写结果要去重：多路召回后用文档 ID 去重，同一个文档被多路命中是好事（说明相关性高），但不要重复送入 LLM
• 监控改写质量：定期抽样检查查询改写的输出，如果发现改写偏离了用户意图，调整改写 Prompt 或切换模型
• 元数据过滤放在向量检索之前：先用元数据把候选集从 10 万缩小到 1 千，再做向量检索，速度提升 100 倍
• 成本意识：每增加一个查询改写调用就是一次 LLM API 调用，在高并发场景下成本不可忽视。考虑用本地小模型（如 Qwen2-1.5B）做改写，或用规则模板替代

本节小结

本节围绕"如何让检索更精准"这一核心问题，系统讲解了五种检索策略：查询改写、HyDE 假设文档嵌入、多路召回（RRF 融合）、元数据过滤和查询扩展。这些策略不是互斥的，而是可以灵活组合——一个成熟的 RAG 系统通常会根据查询复杂度动态选择策略组合。

关键认知：检索策略的设计本质是"用计算换精度"。每增加一层策略都会带来额外的延迟和成本，因此要始终以基准测试数据为依据，只引入确实能带来显著提升的策略。下一节 4.3 将深入讲解重排机制——在检索之后、送入 LLM 之前，如何用重排模型对候选文档做精细排序，进一步提升最终的相关性。

延伸阅读

• 官方文档：LangChain 官方文档 v0.3 版本（MultiQueryRetriever 和 ContextualCompressionRetriever 模块）
• 相关论文：Gao 等人 2022 年发表的 HyDE 论文，提出假设文档嵌入的检索增强方法
• 相关章节：本教程 4.1 节相似度计算算法（底层度量基础），4.4 节混合检索方法（稠密与稀疏检索的深度融合方案）

关键词：RAG 知识库实战, 检索策略, 查询改写, HyDE, 多路召回, RRF, 元数据过滤, 检索优化
难度：进阶
预计阅读：18 分钟