06 · RAG 进阶与生产实践：把检索做扎实

前置：第 04 章讲了 Embedding、向量检索与分块的原理；第 05 章用纯 Python 手写了一个最小 RAG。本章在第 05 章的基础上，解决"检索质量不好、进不了生产"的问题：查询改写、混合检索、重排序、评估，以及生产落地注意点。

本章你将学到

• 用一张「症状 -> 根因 -> 对策」表快速定位检索质量问题的来源；
• 查询改写（Rewrite-Retrieve-Read）：把口语化问题翻译成文档里"长那样"的查询；
• 混合检索（Hybrid Search）：BM25 词法检索 + 向量语义检索，用 RRF 公式融合两路结果；
• 重排序（Rerank）：用 bge-reranker 做精排，救回 Top-K 里被噪音淹没的相关段落；
• HyDE 与 Multi-Query 两个低成本高收益的检索技巧；
• 父子分块（ParentDocumentRetriever）：小块检索定位、父块喂给 LLM；
• 用 Ragas 离线评估 faithfulness / answer_relevancy / context_precision / context_recall，并根据指标定位问题；
• 生产注意点：增量更新、权限过滤、Embedding 缓存、检索质量监控。

先立一个心智模型：RAG 的质量 = 检索的"召回" × 检索的"精排" × 生成的忠实度。这与你们做过的搜索引擎、推荐系统完全同构——召回层求全（Recall），精排层求准（Precision），生成层求不瞎编（Faithfulness）。本章的每一节，都是在给这三个环节加一个可落地的旋钮。

6.1 检索质量诊断：症状与对策映射

生产里 RAG 答得不好，90% 的锅在检索，而不是模型。先别急着换大模型，先用下表对号入座。

先分清楚是哪一环的问题，再动手。一个实用的归因口诀：

1. 先看检索：把 RAG 的检索结果单独打印出来（doc_id + 片段 + 分数），人眼判断 Top-3 里有没有相关段落；
2. 有相关段落但答案还是错 -> 问题在生成侧，改 Prompt、加"只依据上下文回答"的约束（第 03 章）；
3. 没有相关段落 -> 问题在检索侧，往下看 6.2 ~ 6.6；
4. 检索结果里有相关段落但排得靠后 -> 上 Rerank（6.4），这是性价比最高的一步。

本章所有示例都用同一个小型语料：机票/航班客服政策。代码全部完整可运行，Python 3.10+，统一走 OPENAI_API_KEY / OPENAI_BASE_URL 环境变量（DeepSeek / Qwen / GLM 等国产兼容端点直接可用）。

6.2 查询改写 Query Rewrite（Rewrite-Retrieve-Read）

原理

用户的提问是口语，文档是书面语。"退票扣多少钱？"在文档里长这样："退票手续费按退票时间分档收取……"。两者词面差异很大，Embedding 对同义改写有一定容忍度，但远不如"把查询先翻译成文档的语言"可靠。

Rewrite-Retrieve-Read 三步流水线，对应搜索引擎里的 Query 归一化 / Query 理解：

用户问题 ──▶ LLM改写 ──▶ 检索友好查询 ──▶ 检索 ──▶ 拼接上下文 ──▶ LLM作答
 (口语)     (一次小LLM调用)   (术语化)        (召回)      (Read 阶段)

工程类比：这就像你给数据库查询做查询优化器改写（谓词下推、子查询上提），或者给用户搜索词做同义词扩展——不改变问题本身，只是让检索系统更容易命中。

完整代码

# -*- coding: utf-8 -*-
"""6.2 查询改写（Rewrite-Retrieve-Read）完整示例。
用法：先 export OPENAI_API_KEY（可选 export OPENAI_BASE_URL 指向国产端点），再运行：
    python query_rewrite.py
依赖：pip install langchain langchain-openai langchain-community rank-bm25 jieba
"""
import os

import jieba
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from rank_bm25 import BM25Okapi

# 小型语料：只保留 4 条，够演示即可
CORPUS = {
    "d1": "国内航班退票手续费按退票时间和舱位计算：起飞前24小时以上退票，收取票面价5%的手续费；"
          "起飞前4至24小时退票，收取票面价10%；起飞前4小时以内退票，收取票面价30%。特价机票（X舱）不可退票，仅退还机建燃油费。",
    "d2": "改签分为免费改签与收费改签：同一舱位且无票价差额时，起飞前2小时以上可免费改签一次；"
          "存在票价差额时，需补足差价并支付50元改签手续费。改签后航班票价低于原票价的，差额不退还。",
    "d5": "因承运人原因导致航班延误超过4小时，航司提供补偿：延误4至8小时补偿200元，"
          "延误8小时以上补偿400元，或免费改签、全额退票。因天气等不可抗力原因延误的，不提供经济补偿，但提供餐食与住宿安排。",
    "d6": "已满2周岁不满12周岁的儿童票价为成人票面价的50%，单独占座；不满2周岁的婴儿票价为成人票面价的10%，"
          "不单独占座，一名成人限带一名婴儿。儿童票退改签规则与同舱位成人票一致。",
}


def tokenize(text: str) -> list[str]:
    """BM25 检索前需要对中文切词；jieba 搜索模式类似搜索引擎的 Query 切词。"""
    return list(jieba.cut_for_search(text))


def build_bm25_index():
    """rank_bm25 是 Elasticsearch 里 BM25 打分算法的纯 Python 实现。"""
    return BM25Okapi([tokenize(t) for t in CORPUS.values()])


def search_topk(bm25, query: str, top_k: int = 2) -> list[tuple[str, float]]:
    scores = bm25.get_scores(tokenize(query))
    ranked = sorted(range(len(scores)), key=lambda i: scores[i], reverse=True)[:top_k]
    return [(list(CORPUS.keys())[i], float(scores[i])) for i in ranked]


def main():
    # base_url 不传：openai 客户端会自动读取 OPENAI_BASE_URL，国产兼容端点通用
    llm = ChatOpenAI(
        model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"),
        temperature=0,
    )

    # 1) Rewrite：口语 -> 检索友好的术语化查询
    rewrite_prompt = ChatPromptTemplate.from_messages([
        ("system",
         "你是一名机票客服系统的搜索查询改写器。用户问题往往口语化，而文档里是书面术语。"
         "请把问题改写为适合文档检索的查询：提取关键实体与术语，去掉语气词，用空格分隔，不超过8个词。"
         "只输出改写结果，不要解释。"),
        ("human", "{question}"),
    ])
    rewrite_chain = rewrite_prompt | llm | StrOutputParser()

    # 3) Read：基于检索到的上下文作答
    answer_prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一名机票客服。只依据【上下文】回答，不要编造上下文没有的信息。"),
        ("human", "【上下文】\n{context}\n\n【问题】{question}"),
    ])
    answer_chain = answer_prompt | llm | StrOutputParser()

    bm25 = build_bm25_index()
    questions = [
        "退票扣多少钱？",
        "机票不要了，能退吗？",
        "带小朋友坐飞机，票怎么算？",
    ]
    for q in questions:
        rewritten = rewrite_chain.invoke({"question": q}).strip()
        # 2) Retrieve：用改写后的查询去检索（这里先用 BM25 演示，混合检索见 6.3）
        hits = search_topk(bm25, rewritten)
        context = "\n".join(f"[{doc_id}] {CORPUS[doc_id]}" for doc_id, _ in hits)
        final = answer_chain.invoke({"question": q, "context": context})
        print(f"原始问题: {q}")
        print(f"改写查询: {rewritten}")
        print(f"命中文档: {[doc_id for doc_id, _ in hits]}")
        print(f"最终回答: {final}")
        print("-" * 60)


if __name__ == "__main__":
    main()

运行后你会看到类似这样的改写效果：

原始问题: 退票扣多少钱？
改写查询: 退票手续费 收取规则

关键点：

• 改写的目标是"让检索命中"，不是"回答问题"。所以 Prompt 里要求输出术语化查询而不是答案，检索友好。
• 改写是一个小 LLM 调用，在链路里成本极低（一次对话），收益是检索命中率的大幅提升。
• 改写结果最好缓存：同一用户口语问题经常反复出现（线上客服日志里 Top 问题高度集中），按原问题哈希缓存改写结果，可以省掉大量重复 LLM 调用（见 6.8 的缓存思路）。
• 改写也可以不做 LLM：词典同义扩展、基于日志的 Query 归一化，都是工程上常见的降级方案。

6.3 混合检索 Hybrid Search（核心）

为什么纯向量不够

Embedding 向量检索擅长"语义相似"，但有两个致命盲区：

1. 词法盲区：专有名词、术语、编号（舱位号 "X舱"、航线号、"航班取消" 这种冷门词），如果文档里没出现过高度相似的表达，向量可能完全召不回；
2. 短查询退化：用户问得很短（"退票费"），向量几乎拿不到有效信号，结果近似随机。

而 BM25 恰好相反：它按词频 + 逆文档频率（TF-IDF）给词法匹配打分，对精确术语、编号极其敏锐，但对同义改写（"不要了" vs "退票"）完全失灵。混合检索 = 两路召回 + 融合，本质是把"词法信号"和"语义信号"都留下来。

这与 ES 的做法一致：ES 里做搜索本来就有"BM25 文本分 + dense_vector 向量分"融合打分（RRF 或 script_score）的思路，只是这里我们用 Python 在 RAG 链路里自己实现一遍。

RRF（Reciprocal Rank Fusion）公式

多路召回结果的分数量纲不可比（BM25 是 0~N 的词频分，向量是余弦相似度），不能直接相加。RRF 的思路是：不看分数，只看排名：

RRFscore(d) = Σ( 1 / (k + rank_r(d)) )    对所有召回路 r

其中 rank_r(d) 是文档 d 在第 r 路结果中的排名（从 1 开始），k 是平滑常数，一般取 60。

• 两路都排在前面 => 分数叠加，稳稳靠前（双路共识）；
• 只在某一路靠前 => 也能进 Top，保住单路的强项。

工程类比：这就是你们熟悉的多路召回归并（推荐系统把 tag 召回、协同过滤召回、向量召回 merge 成一个候选池），RRF 相当于一个不挑量纲的 merge 规则。

完整代码：对比纯向量 vs 混合检索

# -*- coding: utf-8 -*-
"""6.3 混合检索（Hybrid Search）：BM25 词法检索 + 向量语义检索 + RRF 融合。
对比同一个查询在「纯向量」「纯BM25」「混合」下的结果差异。
依赖：pip install langchain langchain-openai langchain-community chromadb rank-bm25 jieba
"""
import os

import jieba
from langchain_community.vectorstores import Chroma
from langchain_core.documents import Document
from langchain_openai import OpenAIEmbeddings
from rank_bm25 import BM25Okapi

CORPUS = [
    ("d1", "国内航班退票手续费规则",
     "国内航班退票手续费按退票时间和舱位计算：起飞前24小时以上退票，收取票面价5%的手续费；"
     "起飞前4至24小时退票，收取票面价10%；起飞前4小时以内退票，收取票面价30%。"
     "特价机票（X舱）不可退票，仅退还机建燃油费。退票必须于航班起飞前在APP或官网办理。"),
    ("d2", "航班改签与变更规则",
     "改签分为免费改签与收费改签：同一舱位且无票价差额时，起飞前2小时以上可免费改签一次；"
     "存在票价差额时，需补足差价并支付50元改签手续费。改签后航班票价低于原票价的，差额不退还。"
     "改签需在原航班起飞前办理。"),
    ("d3", "免费托运行李额度",
     "国内航班经济舱免费托运行李额度为20公斤，公务舱30公斤，头等舱40公斤。单件托运行李重量不得超过23公斤，"
     "三边之和不得超过158厘米。超重行李按每公斤票面价1.5%收取逾重费。充电宝不得托运，必须随身携带。"),
    ("d4", "值机时间与方式",
     "国内航班起飞前30分钟停止值机，起飞前45分钟停止办理行李托运。旅客可通过APP在线值机，"
     "在线值机开放时间为起飞前48小时至起飞前60分钟。选座免费，紧急出口座位需到机场柜台办理。"),
    ("d5", "航班延误与取消赔偿",
     "因承运人原因导致航班延误超过4小时，航司提供补偿：延误4至8小时补偿200元，延误8小时以上补偿400元，"
     "或免费改签、全额退票。因天气等不可抗力原因延误的，不提供经济补偿，但提供餐食与住宿安排。"),
    ("d6", "儿童与婴儿票购买规则",
     "已满2周岁不满12周岁的儿童票价为成人票面价的50%，单独占座；不满2周岁的婴儿票价为成人票面价的10%，"
     "不单独占座，一名成人限带一名婴儿。儿童票退改签规则与同舱位成人票一致。"),
]


def tokenize(text: str) -> list[str]:
    """中文切词；rank_bm25 的 BM25Okapi 就是 BM25 算法本身，不依赖 langchain。"""
    return list(jieba.cut_for_search(text))


def build_bm25(docs):
    return BM25Okapi([tokenize(d.page_content) for d in docs])


def bm25_search(bm25, docs, query: str, top_k: int = 5) -> list[tuple[str, float]]:
    scores = bm25.get_scores(tokenize(query))
    ranked = sorted(range(len(docs)), key=lambda i: scores[i], reverse=True)[:top_k]
    return [(docs[i].metadata["doc_id"], float(scores[i])) for i in ranked]


def vector_search(vectorstore, query: str, top_k: int = 5) -> list[tuple[str, float]]:
    # relevance score 归一化到 [0,1]，越大越相似
    hits = vectorstore.similarity_search_with_relevance_scores(query, k=top_k)
    return [(d.metadata["doc_id"], float(score)) for d, score in hits]


def rrf_fusion(ranked_lists: list[list[str]], k: int = 60) -> list[tuple[str, float]]:
    """Reciprocal Rank Fusion：只看排名不看分数量纲。
    公式：score(d) = Σ 1 / (k + rank_r(d))，rank 从 1 开始，k 一般取 60。
    """
    agg: dict[str, float] = {}
    for ranked in ranked_lists:
        for rank, doc_id in enumerate(ranked, start=1):
            agg[doc_id] = agg.get(doc_id, 0.0) + 1.0 / (k + rank)
    return sorted(agg.items(), key=lambda kv: kv[1], reverse=True)


def main():
    docs = [Document(page_content=t, metadata={"doc_id": i, "title": title})
            for i, title, t in CORPUS]

    # 注意：langchain-community 0.3.x 的 Chroma 包装类仍可用但已标记废弃，
    # 官方新家在 langchain_chroma 包，升级后改 import 即可（以官方文档为准）。
    embeddings = OpenAIEmbeddings(model=os.getenv("EMBEDDING_MODEL", "text-embedding-3-small"))
    vectorstore = Chroma.from_documents(
        docs, embeddings, collection_name="flight_policy", persist_directory="./chroma_db",
    )
    bm25 = build_bm25(docs)

    queries = [
        "退票手续费",             # 词法强：文档里的精确术语
        "机票不要了能退多少钱",     # 语义强：口语，词法上几乎没有命中
        "儿童票怎么买",            # 术语 + 语义混合
    ]
    for q in queries:
        v_rank = [doc_id for doc_id, _ in vector_search(vectorstore, q, top_k=5)]
        b_rank = [doc_id for doc_id, _ in bm25_search(bm25, docs, q, top_k=5)]
        h_rank = [doc_id for doc_id, _ in rrf_fusion([v_rank, b_rank])]
        print(f"查询: {q}")
        print(f"  纯向量 Top5: {v_rank}")
        print(f"  纯BM25 Top5: {b_rank}")
        print(f"  混合RRF Top5: {h_rank}")
        print()


if __name__ == "__main__":
    main()

典型输出（示意）：

查询: 机票不要了能退多少钱
  纯向量 Top5: ['d1', 'd5', 'd2', 'd6', 'd3']
  纯BM25 Top5: ['d5', 'd1', 'd6', 'd2', 'd3']
  混合RRF Top5: ['d1', 'd5', 'd2', 'd6', 'd3']

观察结论：

• 对 "退票手续费" 这类精确术语，BM25 一骑绝尘，纯向量反而不稳（尤其小语料、短查询时向量分全部趋同）；
• 对 "机票不要了" 这类口语，纯向量赢，BM25 只能靠碰运气命中 "退" 字；
• 混合后：语义负责召回，词法负责保底，任何一路丢掉的都不至于彻底消失。把上面的 top_k 从 5 调大，对比会更明显（比如只让 BM25 召回 d1，而向量没召回到 d1 时）。

框架版捷径：EnsembleRetriever

上面是手写的 RRF，理解了公式之后，LangChain 的 EnsembleRetriever 就是同样逻辑的现成封装（内部就是 RRF 融合，fusion_mode="rrf"）：

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever

# BM25Retriever 内部用的就是 rank_bm25，preprocess_func 传 jieba 分词
bm25_retriever = BM25Retriever.from_texts(
    texts=[d.page_content for d in docs],
    metadatas=[d.metadata for d in docs],
    preprocess_func=tokenize,
    k=5,
)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
ensemble = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.5, 0.5],          # 两路权重，默认等权
)
for d in ensemble.invoke("机票不要了能退多少钱"):
    print(d.metadata["doc_id"], d.page_content[:20])

工程建议：混合检索应当是生产 RAG 的默认配置，纯向量适合原型、小型 demo。切换成本很低（换一个 retriever），但"查不到"类工单能明显下降。

6.4 重排序 Rerank（核心）

为什么 Top-K 的排序不可靠

向量检索的分数本质是余弦相似度，它衡量"语义接近"，不衡量"这条上下文能不能回答这个问题"。两条与问题都"相关"的段落，哪条更有用，向量分几乎无法区分；更糟的是，一些看似相关实则误导的段落（比如同时提到"退票"和"改签"的段落）会排得很靠前。

另外，Top-K 一刀切：K 取小了丢召回，K 取大了混进噪音。Rerank 解决的就是这个问题：召回阶段放大 K（比如 10~20 条），再用一个更强的打分器做精排，只留 3~5 条给 LLM。

这个"召回（粗排）-> 精排"的两段式架构，就是你们在搜索/推荐里天天见的分层打分结构。区别只在精排器：搜索用 LR/GBDT 特征模型，这里用交叉编码器（cross-encoder）。

Rerank 模型：cross-encoder vs bi-encoder

cross-encoder 精度高但贵（每个 pair 都要过一遍模型），所以只能用在召回后的 Top-K 上——这正是"精排只精排候选集"的老规矩。

完整代码：ContextualCompressionRetriever + bge-reranker

LangChain 把"对召回结果做二次处理"抽象成 ContextualCompressionRetriever（上下文压缩检索器），它由两部分组成：base_retriever（先召回）和 base_compressor（对召回结果压缩/重排）。我们自定义一个 compressor，内部调用 bge-reranker：

# -*- coding: utf-8 -*-
"""6.4 重排序（Rerank）：召回 Top-K 后，用交叉编码器 bge-reranker 精排。
演示 LangChain ContextualCompressionRetriever + 自定义 DocumentCompressor。
依赖：pip install FlagEmbedding torch
      bge-reranker-base 首次运行会从 HuggingFace 下载模型（国内可先 export HF_ENDPOINT=https://hf-mirror.com）
"""
import os

from FlagEmbedding import FlagReranker
from langchain.retrievers import ContextualCompressionRetriever
from langchain_community.vectorstores import Chroma
from langchain_core.documents import Document
from langchain_core.retrievers import BaseDocumentCompressor
from langchain_openai import OpenAIEmbeddings

# 复用 6.3 的语料（此处省略，直接粘贴 6.3 的 CORPUS 定义）
CORPUS = [
    ("d1", "国内航班退票手续费规则",
     "国内航班退票手续费按退票时间和舱位计算：起飞前24小时以上退票，收取票面价5%的手续费；"
     "起飞前4至24小时退票，收取票面价10%；起飞前4小时以内退票，收取票面价30%。"
     "特价机票（X舱）不可退票，仅退还机建燃油费。"),
    ("d2", "航班改签与变更规则",
     "改签分为免费改签与收费改签：同一舱位且无票价差额时，起飞前2小时以上可免费改签一次；"
     "存在票价差额时，需补足差价并支付50元改签手续费。改签后航班票价低于原票价的，差额不退还。"),
    ("d3", "免费托运行李额度",
     "国内航班经济舱免费托运行李额度为20公斤，公务舱30公斤，头等舱40公斤。单件托运行李重量不得超过23公斤。"),
    ("d4", "值机时间与方式",
     "国内航班起飞前30分钟停止值机，起飞前45分钟停止办理行李托运。旅客可通过APP在线值机，"
     "在线值机开放时间为起飞前48小时至起飞前60分钟。"),
    ("d5", "航班延误与取消赔偿",
     "因承运人原因导致航班延误超过4小时，航司提供补偿：延误4至8小时补偿200元，延误8小时以上补偿400元，"
     "或免费改签、全额退票。因天气等不可抗力原因延误的，不提供经济补偿。"),
    ("d6", "儿童与婴儿票购买规则",
     "已满2周岁不满12周岁的儿童票价为成人票面价的50%，单独占座；不满2周岁的婴儿票价为成人票面价的10%，"
     "不单独占座，一名成人限带一名婴儿。"),
]


class BGERerankCompressor(BaseDocumentCompressor):
    """自定义 compressor：把 BaseDocumentCompressor 接到 ContextualCompressionRetriever 上。
    语义就是搜索里的"精排层"：对召回结果重新打分，只保留 top_n。
    """

    def __init__(self, model_name: str = "BAAI/bge-reranker-base", top_n: int = 3):
        # 本机无 GPU 时去掉 use_fp16
        self.reranker = FlagReranker(model_name, use_fp16=True)
        self.top_n = top_n

    def compress_documents(self, documents, query):
        # cross-encoder 有长度上限：bge-reranker-base 是 512 token，先截断
        pairs = [[query, d.page_content[:500]] for d in documents]
        # 新版本 FlagEmbedding 支持 normalize=True；低版本去掉该参数即可（以官方文档为准）
        scores = self.reranker.compute_score(pairs, normalize=True)
        if isinstance(scores, float):  # 单条输入时可能直接返回 float
            scores = [scores]
        ranked = sorted(zip(documents, scores), key=lambda x: x[1], reverse=True)
        return [d for d, _ in ranked[: self.top_n]]

    def acompress_documents(self, documents, query):
        return self.compress_documents(documents, query)


def main():
    docs = [Document(page_content=t, metadata={"doc_id": i, "title": title})
            for i, title, t in CORPUS]
    embeddings = OpenAIEmbeddings(model=os.getenv("EMBEDDING_MODEL", "text-embedding-3-small"))
    vectorstore = Chroma.from_documents(
        docs, embeddings, collection_name="flight_policy_rerank", persist_directory="./chroma_db_rerank",
    )

    query = "航班取消可以赔多少钱？"

    # 召回阶段：粗排，故意放大 K=10，让噪音进来
    base_retriever = vectorstore.as_retriever(search_kwargs={"k": 10})
    print("召回 Top-10（纯向量，含噪音）：")
    for d in base_retriever.invoke(query):
        print(f"  {d.metadata['doc_id']}: {d.page_content[:24]}...")

    # 精排阶段：bge-reranker 对 10 条重新打分，只留 3 条
    compressor = BGERerankCompressor(top_n=3)
    compression_retriever = ContextualCompressionRetriever(
        base_compressor=compressor, base_retriever=base_retriever,
    )
    print("\nRerank 后 Top-3：")
    for d in compression_retriever.invoke(query):
        print(f"  {d.metadata['doc_id']}: {d.page_content[:40]}...")


if __name__ == "__main__":
    main()

运行结果会直观地展示：向量 Top-10 里 "d5（延误赔偿）" 未必排第一（可能被 d1/d2 挤下去），Rerank 之后 d5 稳定进入 Top-3。问题越"绕"（比如同时涉及两个政策）、语料越杂，Rerank 的收益越明显。

关于模型选择的说明：

• BAAI/bge-reranker-base：开源、中英文、CPU 也能跑（慢一点），生产首选；更大的是 bge-reranker-large。
• Cohere Rerank：闭源 API，效果强，但需要 Cohere 的 Key 和 langchain-cohere 包。集成方式与本例完全同构——写一个 compressor 调 cohere.rerank()，或者用 langchain_cohere.CohereRerank 直接作为 base_compressor 传入。具体参数以官方文档为准。
• 国内云厂商一般也有"文本精排/重排序"API，接入方式同 Cohere。

Rerank 的工程要点：它按请求计费/计耗时，所以永远放在召回之后、喂给 LLM 之前；K 一般取 10~20，精排后取 3~5。

6.5 HyDE 与 Multi-Query Retriever：两个低成本技巧

HyDE（Hypothetical Document Embeddings）

思路一句话：与其拿用户的问题去检索，不如先让 LLM 写一篇"假设的标准答案"，再用这篇假答案的向量去检索。

为什么有用：问题短、口语化，向量信号弱；而"假设答案"是用文档的语言写出来的完整段落，与库里的段落天然更接近。等于是用一次小 LLM 调用，把问题翻译成文档侧的表达——和 6.2 查询改写是同一思想的两面（一个改查询，一个生成文档）。

注意它的风险：假答案如果编错方向，会把检索带偏。所以 HyDE 适合"问题在文档里有明确答案"的场景，不适合开放性问题。

Multi-Query Retriever

思路一句话：一个问题生成多个角度的子问题，分别检索，结果合并去重。

"退票扣多少钱？"
  ├─ "退票手续费怎么算？"
  ├─ "退票要付多少违约金？"
  └─ "机票退款比例是多少？"

多路子查询各自命中文档的不同表述，合并后覆盖面更大——专门治"同一问题换个问法结果就飘"。

完整代码

# -*- coding: utf-8 -*-
"""6.5 HyDE 与 Multi-Query Retriever。
依赖：pip install langchain langchain-openai langchain-community chromadb
"""
import os

from langchain.retrievers import MultiQueryRetriever
from langchain_community.vectorstores import Chroma
from langchain_core.documents import Document
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

# 复用 6.3 的 CORPUS（此处省略，直接粘贴）
CORPUS = [
    ("d1", "国内航班退票手续费规则",
     "国内航班退票手续费按退票时间和舱位计算：起飞前24小时以上退票，收取票面价5%的手续费；"
     "起飞前4至24小时退票，收取票面价10%；起飞前4小时以内退票，收取票面价30%。特价机票（X舱）不可退票。"),
    ("d2", "航班改签与变更规则",
     "改签分为免费改签与收费改签：同一舱位且无票价差额时，起飞前2小时以上可免费改签一次；"
     "存在票价差额时，需补足差价并支付50元改签手续费。"),
    ("d5", "航班延误与取消赔偿",
     "因承运人原因导致航班延误超过4小时，航司提供补偿：延误4至8小时补偿200元，延误8小时以上补偿400元，"
     "或免费改签、全额退票。因天气等不可抗力原因延误的，不提供经济补偿。"),
    ("d6", "儿童与婴儿票购买规则",
     "已满2周岁不满12周岁的儿童票价为成人票面价的50%，单独占座；不满2周岁的婴儿票价为成人票面价的10%，"
     "不单独占座，一名成人限带一名婴儿。"),
]


def main():
    docs = [Document(page_content=t, metadata={"doc_id": i, "title": title})
            for i, title, t in CORPUS]
    embeddings = OpenAIEmbeddings(model=os.getenv("EMBEDDING_MODEL", "text-embedding-3-small"))
    vectorstore = Chroma.from_documents(
        docs, embeddings, collection_name="flight_policy_hyde", persist_directory="./chroma_db_hyde",
    )
    llm = ChatOpenAI(model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"), temperature=0)

    query = "带小朋友坐飞机，票怎么算？"

    # ---- 方式一：HyDE ----
    hyde_prompt = ChatPromptTemplate.from_messages([
        ("system",
         "你是一个机票客服文档的写手。给定用户问题，写一段风格与文档一致的假设性答案，"
         "尽量使用文档里可能出现的关键词（如'儿童票''票价''退改'）。只输出答案，不要解释。"),
        ("human", "{question}"),
    ])
    hypo_doc = (hyde_prompt | llm | StrOutputParser()).invoke({"question": query})
    print("假设性文档：", hypo_doc[:60], "...")
    hits = vectorstore.similarity_search_with_relevance_scores(hypo_doc, k=3)
    print("HyDE 检索结果：", [d.metadata["doc_id"] for d, _ in hits])

    # ---- 方式二：Multi-Query Retriever（框架自带，零 Prompt 成本）----
    mq = MultiQueryRetriever.from_llm(
        retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),
        llm=llm,
    )
    docs_mq = mq.invoke(query)
    print("Multi-Query 检索结果：", [d.metadata["doc_id"] for d in docs_mq])


if __name__ == "__main__":
    main()

工程建议：两个技巧都很便宜，优先上 Multi-Query（框架自带、无 Prompt 维护成本、还顺带拉高召回）；HyDE 在"问题短、答案在文档里是完整段落"的问答场景里收益更大，但需要人工观察是否被假答案带偏。

6.6 父子分块 ParentDocumentRetriever

原理：小块检索定位、父块喂给 LLM

上一章（第 05 章）讲过：分块太大，检索不准、浪费 token；分块太小，上下文碎片化、答案缺上下文。这是一个矛盾：检索想要小块，生成想要大块。

父子分块同时满足两者：

• 子块（child）：切得很小（如 100~200 字），进向量库做检索——定位精准；
• 父块（parent）：包含子块的完整章节（如 500~1000 字），存 docstore——检索命中子块后，回表把整块父文档取出来喂给 LLM。

                  ┌── 向量库（索引小块，检索用） ── 命中 [child-3] ──┐
  长文档 ──拆父块──┤                                                   ├── 把父块喂给 LLM
                  └── docstore（按 doc_id 存父块） ── 由子块ID回表 ──┘

工程类比：这就像 MySQL 里明细表（子块）定位 + 主表（父块）取数，用小的索引结构精确定位，再回表拿全量行。代价是 docstore 要多存一份父块，但换来的是"检索准 + 上下文完整"。

完整代码

# -*- coding: utf-8 -*-
"""6.6 父子分块（ParentDocumentRetriever）：小块检索、父块喂给 LLM。
依赖：pip install langchain langchain-openai langchain-community langchain-text-splitters chromadb
"""
import os

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_core.documents import Document
from langchain_openai import OpenAIEmbeddings

# 两个较长的政策文档，一个 parent 会被切成多个 child
LONG_DOCS = [
    Document(
        page_content=(
            "国内航班退改签总则：退票手续费按退票时间与舱位分档收取。起飞前24小时以上退票，经济舱收取票面价5%作为手续费；"
            "起飞前4至24小时退票，收取票面价10%；起飞前4小时以内退票，收取票面价30%。特价机票（X舱）不可退票，仅退还机建燃油费。"
            "改签分为免费改签与收费改签：同一舱位且无票价差额时，起飞前2小时以上可免费改签一次；存在票价差额时，"
            "需补足差价并支付50元改签手续费。改签后航班票价低于原票价的，差额不退还。"
            "因承运人原因导致航班延误超过4小时的，延误4至8小时补偿200元，8小时以上补偿400元，或免费改签、全额退票。"
            "因天气等不可抗力原因延误的，不提供经济补偿，但提供餐食与住宿安排。所有退改签业务需在航班起飞前办理。"
        ),
        metadata={"doc_id": "p1", "title": "退改签总则"},
    ),
    Document(
        page_content=(
            "乘机服务须知：国内航班经济舱免费托运行李额度为20公斤，公务舱30公斤，头等舱40公斤。"
            "单件托运行李重量不得超过23公斤，三边之和不得超过158厘米，超重行李按每公斤票面价1.5%收取逾重费。"
            "充电宝不得托运，必须随身携带。值机方面，起飞前30分钟停止值机，起飞前45分钟停止办理行李托运；"
            "在线值机开放时间为起飞前48小时至起飞前60分钟，选座免费。"
            "儿童乘机需由成人陪伴，已满2周岁不满12周岁的儿童票价为成人票面价的50%。"
        ),
        metadata={"doc_id": "p2", "title": "乘机服务须知"},
    ),
]


def main():
    embeddings = OpenAIEmbeddings(model=os.getenv("EMBEDDING_MODEL", "text-embedding-3-small"))
    parent_splitter = RecursiveCharacterTextSplitter(chunk_size=600, chunk_overlap=50)
    child_splitter = RecursiveCharacterTextSplitter(chunk_size=150, chunk_overlap=20)

    docstore = InMemoryStore()  # 父文档按 doc_id 存在这里
    vectorstore = Chroma(
        collection_name="parent_child",
        embedding_function=embeddings,
        persist_directory="./chroma_db_parent",
    )

    retriever = ParentDocumentRetriever(
        vectorstore=vectorstore,
        docstore=docstore,
        child_splitter=child_splitter,   # 子块：进向量库，用于检索
        parent_splitter=parent_splitter, # 父块：进 docstore，用于喂 LLM
        search_kwargs={"k": 2},          # 命中 2 个子块，回表取 2 个父块
    )
    retriever.add_documents(LONG_DOCS)
    print("向量库里的子块数量：", len(vectorstore.get()["ids"]))

    query = "改签要收多少钱？"
    parents = retriever.invoke(query)
    print(f"\n命中父文档 {len(parents)} 条（喂给 LLM 的是父块，包含完整上下文）：")
    for d in parents:
        print(f"  [{d.metadata['doc_id']}] {d.page_content[:40]}...")


if __name__ == "__main__":
    main()

说明与坑：

• 向量库里存的是子块（数量比父块多很多），invoke 返回的是父块。中间发生了什么：子块命中 -> 从子块的 doc_id 元数据定位父块 -> 从 docstore 取回父块。
• 生产环境用 InMemoryStore 只适合单机、小规模；大规模用 RedisStore（langchain_community.storage）或 PostgreSQL/对象存储，以官方文档为准。
• 子块粒度决定检索精度，父块粒度决定上下文完整度，两个参数分开调，别混为一谈。

6.7 RAG 评估（核心）

为什么要评估：给 RAG 写"回归测试"

你在 Java 工程里写单元测试、CI 回归，天然接受"没有测试就谈不上上线"。RAG 也一样：没有离线评估集，你无法知道一次改动（换分块、上 rerank）到底是变好还是变坏。"感觉效果好多了"不是证据。

评估的最小闭环：

1. 构建离线测试集：每一条 = 问题 + 标准答案 + 期望命中的段落 id（几十到几百条，从线上日志抽）；
2. 跑评估脚本：把线上 RAG 管线在测试集上跑一遍，得到 4 个指标；
3. 定位问题：根据哪个指标低，去找对应的环节（见下方映射表）；
4. 优化后重跑，指标回升即验证通过——这就是检索侧的"回归测试"。

四个指标

指标范围 [0, 1]，越高越好。其中 faithfulness 是最接近"有没有瞎编"的指标；context_recall 最接近"查不查得到"。

说明：Ragas 这几个指标的实现细节在不同小版本有差异（例如 context_precision 的排序加权方式），具体以 Ragas 官方文档为准。本节给出的 API 是 0.1.x ~ 0.2.x 通用的经典写法。

完整评估脚本

# -*- coding: utf-8 -*-
"""6.7 RAG 离线评估：用 Ragas 计算 faithfulness / answer_relevancy /
context_precision / context_recall。
依赖：pip install ragas datasets
注意：评估本身也需要 LLM 与 Embedding 模型，默认读取 OPENAI_API_KEY / OPENAI_BASE_URL。
"""
from datasets import Dataset
from ragas import evaluate
from ragas.metrics import (
    answer_relevancy,
    context_precision,
    context_recall,
    faithfulness,
)

# 离线测试集：question + contexts（线上 RAG 实际检索到的段落）+
# answer（线上 RAG 实际生成的答案）+ ground_truth（期望命中段落对应的标准答案）
TEST_CASES = [
    {  # 理想样本：上下文准确，答案有依据
        "question": "退票要扣多少钱？",
        "contexts": [
            "国内航班退票手续费按退票时间和舱位计算：起飞前24小时以上退票，收取票面价5%的手续费；"
            "起飞前4至24小时退票，收取票面价10%；起飞前4小时以内退票，收取票面价30%。",
            "改签分为免费改签与收费改签：同一舱位且无票价差额时，起飞前2小时以上可免费改签一次。",
        ],
        "answer": "退票手续费按退票时间分档：起飞前24小时以上收票面价5%，4至24小时收10%，4小时以内收30%。",
        "ground_truth": "国内航班退票按退票时间分档收取手续费：24小时以上5%，4至24小时10%，4小时内30%。",
    },
    {  # 问题样本1：上下文里有噪音段落（行李额），答案被带偏
        "question": "航班延误多久可以赔偿？",
        "contexts": [
            "国内航班经济舱免费托运行李额度为20公斤，超重行李按每公斤票面价1.5%收取逾重费。",
            "因承运人原因导致航班延误超过4小时，航司提供补偿：延误4至8小时补偿200元，延误8小时以上补偿400元。",
        ],
        "answer": "经济舱免费托运行李额度为20公斤，超重按每公斤票面价1.5%收取逾重费。",
        "ground_truth": "因承运人原因延误超过4小时的，延误4至8小时补偿200元，延误8小时以上补偿400元。",
    },
    {  # 问题样本2：上下文没召回核心信息（漏了儿童票段落）
        "question": "带两岁半的小孩坐飞机怎么买票？",
        "contexts": [
            "国内航班起飞前30分钟停止值机，起飞前45分钟停止办理行李托运。",
            "儿童乘机需由成人陪伴，已满2周岁不满12周岁的儿童票价为成人票面价的50%。",
        ],
        "answer": "起飞前30分钟停止值机，请提前到机场办理手续；儿童乘机需由成人陪伴。",
        "ground_truth": "已满2周岁不满12周岁的儿童票价为成人票面价的50%，单独占座。",
    },
    {  # 完美对照：上下文完整、答案忠实
        "question": "航班延误多久可以赔偿？",
        "contexts": [
            "因承运人原因导致航班延误超过4小时，航司提供补偿：延误4至8小时补偿200元，延误8小时以上补偿400元，"
            "或免费改签、全额退票。因天气等不可抗力原因延误的，不提供经济补偿。",
        ],
        "answer": "因承运人原因延误超过4小时才启动补偿：4至8小时补200元，8小时以上补400元，或免费改签、全额退票；"
                  "天气等不可抗力造成的延误不补偿。",
        "ground_truth": "因承运人原因延误超过4小时的，延误4至8小时补偿200元，8小时以上补偿400元；不可抗力不补偿。",
    },
]


def main():
    dataset = Dataset.from_dict({
        "question": [t["question"] for t in TEST_CASES],
        "answer": [t["answer"] for t in TEST_CASES],
        "contexts": [t["contexts"] for t in TEST_CASES],
        "ground_truth": [t["ground_truth"] for t in TEST_CASES],
    })

    result = evaluate(
        dataset,
        metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
    )
    print(result)          # 聚合分数
    print()
    print(result.to_pandas())  # 每个样本的分项分数

    # 若 Ragas 0.2.x 提示需要显式指定评估模型，可改为显式传入（以官方文档为准）：
    # from ragas.llms import LangchainLLMWrapper
    # from ragas.embeddings import LangchainEmbeddingsWrapper
    # from langchain_openai import ChatOpenAI, OpenAIEmbeddings
    # result = evaluate(
    #     dataset,
    #     metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
    #     llm=LangchainLLMWrapper(ChatOpenAI(model="gpt-4o-mini")),
    #     embeddings=LangchainEmbeddingsWrapper(OpenAIEmbeddings(model="text-embedding-3-small")),
    # )


if __name__ == "__main__":
    main()

把上面的测试集跑完，你会看到第 2、3 条样本的指标明显低于第 1、4 条——这就证明指标能区分好坏的检索/生成。

如何根据指标定位问题

工程建议：

• 测试集优先从线上日志抽样（真实用户问题），人工补标准答案与期望段落，50~100 条就能支撑一轮回归。
• Ragas 评估消耗 token 较多（faithfulness 要逐条抽"声明"再核对），按批次小样本跑，别拿全量日志跑。
• 把 4 个指标当 CI 门槛：每次改分块/改检索/换模型，重跑一遍，任何指标回落后退，这就是 RAG 的"回归测试"。

6.8 生产注意点

1) 增量更新：重建索引 vs 增量 upsert

增量更新的核心纪律：doc_id 必须稳定（等于数据库主键），才能做到"同 id 覆盖、不重复堆积"：

# 片段：以 doc_id 为幂等键做增量更新
vectorstore.delete(ids=["d1"])                      # 先删旧版本（确保 embedding 一起换掉）
vectorstore.add_documents([new_d1_doc])             # 再写新版本
# 部分库也提供 update_documents(ids, docs)，但历史上对 embedding 的更新有坑，
# "delete + add" 语义最稳（以官方文档为准）

三个容易踩的坑：

• 换 Embedding 模型必须全量重建：不同模型产出的向量空间不兼容，新旧向量混在一个集合里等于让检索失效；
• 分块参数改了必须全量重建：同一文本在不同分块下是不同向量，老向量残留会导致重复/漏检；
• 删除走 upsert 不管：内容下架只更新了文本没删向量，就会检索到已下架内容。

2) 权限过滤（metadata filter）

多租户、内部知识库都要做行级权限：过滤必须发生在召回阶段（下推到向量库），而不是全量取回后再过滤——全量再过滤等于没有索引。

# 片段：Chroma 的 where 过滤下推，检索时只在一个租户的文档里召回
tenant_retriever = vectorstore.as_retriever(
    search_kwargs={
        "k": 5,
        "filter": {
            "tenant_id": "tenant_001",                          # 等值
            "dept": {"$in": ["运营", "客服"]},                  # 集合
            "is_archived": {"$ne": True},                       # 取反
        },
    },
)
docs = tenant_retriever.invoke("退票手续费")

写入文档时，把 tenant_id、dept、is_archived 这些字段放进 metadata（即每块的 metadata），权限判断的输入（当前用户的租户/角色）来自你现有的认证体系，检索时动态拼到 filter 里。

3) Embedding 缓存

Embedding 按 token 计费、且推理有延迟。线上问题高度重复（客服场景 Top 问题集中），缓存收益很大。缓存键用文本的哈希，命中即免调模型：

# 片段：函数级缓存，生产可换 Redis（key = md5(文本)，value = 向量 JSON）
from functools import lru_cache

import hashlib

_emb = OpenAIEmbeddings(model="text-embedding-3-small")


@lru_cache(maxsize=4096)
def cached_embed(text: str) -> list[float]:
    return _emb.embed_query(text)


def embed_for_store(text: str) -> list[float]:
    return cached_embed(text)

注意：换 embedding 模型或加 temperature 类参数前，先清缓存，否则会混入不同模型产出的向量。

4) 检索质量监控

生产环境要能看到"检索干得好不好"。最小做法：全链路结构化日志 + 命中率统计 + 周期性离线复评。

# 片段：每次请求记录检索与生成的快照，供线上监控与离线抽样评估
import json
import logging
import time

logger = logging.getLogger("rag")

def log_rag_call(question: str, query: str, hit_doc_ids: list[str],
                 scores: list[float], answer: str, latency_ms: float) -> None:
    logger.info(json.dumps({
        "event": "rag_call",
        "ts": int(time.time()),
        "question": question,
        "query_after_rewrite": query,
        "hit_doc_ids": hit_doc_ids,      # 命中了哪些块（查得到吗）
        "top_scores": scores,            # 检索分数（分布漂移了吗）
        "answer": answer,
        "latency_ms": latency_ms,
    }, ensure_ascii=False))

从这些日志里可以统计：

• 命中率：Top-3 里"人眼判断相关"的比例，抽样人工标注；
• 零命中率：检索一个相关块都没有的请求占比，这是"查不到"工单的源头；
• 分数漂移：检索分数整体下降往往意味着知识库更新出了问题（如 6.8-1 的换模型没重建）；
• 定期把线上样本抽回 6.7 的 Ragas 评估脚本，做回归对比。

监控与可观测性的整体清单（日志、链路追踪、成本、限流降级）在第 12 章展开。

6.9 演进方向

GraphRAG（一句话简介）

传统 RAG 在"文本块"上检索，GraphRAG 先把文档抽成实体与关系建知识图谱，回答需要跨文档推理的"多跳"问题时，沿图谱关系路径检索。适用场景：公司制度、研报、多文档对比这类"知识点分散在多处、需要串起来"的问答。它不是银弹——图谱构建成本高、维护重，先用好本章的检索优化再说。

Agentic RAG（预告第 11 章）

本章的检索是一次性的：查一次、拼上下文、生成答案。Agentic RAG 把"检索"封装成工具，由 Agent 自主决定：要不要查、先查什么、结果不够再查一次、多轮追问里逐步缩小范围（比如先查"退改签政策"，再查"退票手续费"）。这就是"把 RAG 变成 Agent 的一项技能"，编排细节在第 11 章《Agent 工程原理与设计模式》展开。

小结

• 检索质量问题先归因：召回不足、噪音过多、还是生成不忠实，分别对症下药，别一上来就换模型。
• 查询改写（Rewrite-Retrieve-Read）把口语翻译成文档术语，成本一次小 LLM 调用，收益是命中率。
• 混合检索 = BM25（jieba 分词）+ 向量，RRF 按排名融合，生产默认配置。
• Rerank 是"精排层"：召回 Top-10 到 20，cross-encoder 重排取 3~5，用 ContextualCompressionRetriever + bge-reranker 落地。
• HyDE 与 Multi-Query 是廉价提召回的技巧，前者可能被假答案带偏，后者零成本。
• 父子分块同时满足"检索要小块、生成要大块"，子块索引 + 父块回表。
• 评估是 RAG 的回归测试：离线测试集 + Ragas 四指标（faithfulness / answer_relevancy / context_precision / context_recall），根据低指标定位环节。
• 生产四件事：doc_id 稳定的增量更新（换模型/换分块必须全量重建）、召回阶段做权限过滤、Embedding 缓存、全链路日志 + 命中率监控。

练习

1. 在 6.3 的脚本里把 top_k 从 5 改成 20，重新对比纯向量与混合检索，观察 RRF 对"单路弱命中"的保底作用。
2. 用你自己项目里的文档跑 6.4，对比 Rerank 前后喂给 LLM 的答案，记录 10 个问题的前后差异。
3. 从线上日志抽 20 条真实用户问题，构造 6.7 的测试集（含期望命中的 doc_id），跑出四个指标，并针对最低的一项做一次优化再复测。
4. 给知识库加一个 tenant_id 元数据，写一个带权限过滤的检索函数，验证不同租户查不到彼此的文档。
5. 把 6.2、6.3、6.4 串成一条完整流水线：改写 -> 混合检索 -> Rerank -> 生成，作为第 14 章综合实战的前置练习。

参考答案

<details>
<summary>练习1（top_k=20 观察 RRF 保底）</summary>

把 top_k 从 5 改成 20，重跑纯向量与混合检索。预期现象：

• 纯向量：Top-20 里会混入更多噪音，但更关键的是——如果某条相关文档语义 Embedding 得不好（分词/写法差异），它在向量路排名可能跌出前 5，此时 BM25 路可能正好把它排在前面；
• 混合检索 + RRF：单路弱命中的文档因为"在另一路排进前列"而获得较高融合分，保住排名——这就是 RRF 对单路弱命中的保底作用；
• 观察点：对比两条路各自的 Top-5 命中集合与融合后 Top-5 的命中集合，找出"单路漏了、融合后救回"的样本，数一数这类样本有多少。若你的数据上这类样本很少，说明数据用纯向量就够了，混合检索的收益有限——用数据说话，别为用而用。
  </details>

<details>
<summary>练习2（Rerank 前后对比）</summary>

跑 6.4 的 ContextualCompressionRetriever + bge-reranker，对 10 个问题记录三组结果：

# 记录：纯向量 Top-5 → Rerank 后 Top-3 → 最终回答
for q in questions:
    before = vector_retriever.invoke(q)              # 召回 Top-5
    after  = rerank_retriever.invoke(q)              # 压缩后 Top-3
    # 分别喂给 LLM，记录回答质量（完整/正确/多余噪音/引用是否可信）

典型差异：

• 命中率不变或略升（Rerank 不负责召回，只负责精排——它救不回本来就没召回的内容）；
• 回答"纯度高"：Rerank 前 Top-3 常夹带无关块，Rerank 后喂进去的几乎全有用，回答更干净、幻觉率更低；
• 速度与成本：多了一次 bge-reranker 推理（本地模型，免费但耗 CPU/GPU），但最终喂给 LLM 的 token 更少。

如果 10 个问题上 Rerank 前后回答没差异，说明你的检索已经足够好，Rerank 在此场景是冗余优化——它主要对"召回宽、Top-K 噪音大"的场景收益明显。
</details>

<details>
<summary>练习3（20 条测试集 + 四指标优化）</summary>

步骤参考：

1. 构造测试集：从线上日志抽 20 条真实问题，每条标注 expected_doc_id（人工看日志确认答对时命中的文档）；
2. 跑 Ragas 四指标（faithfulness / answer_relevancy / context_precision / context_recall），记录基线；

3. 定位最低指标，对症下药：

   • context_recall 低 → 召回不足 → 上混合检索 / HyDE / Multi-Query；
   • context_precision 低 → Top-K 里无关多 → 上 Rerank / 收紧切分；
   • faithfulness 低 → 生成不忠实 → 收紧 System 约束（"只依据资料，没有就明说"）、检查是否信息被截断；
   • answer_relevancy 低 → 答非所问 → 改 Prompt、检查 top_k 是否给了过多无关上下文干扰；
4. 优化一次再复测，记录指标变化。一次只改一个变量，否则无法归因。

这套流程就是 RAG 的"回归测试"：之后任何改动（换模型、换分块参数）都先跑它，指标不降再上线（结合第 12 章挂 CI）。
</details>

<details>
<summary>练习4（tenant_id 权限过滤）</summary>

入库时为每个 chunk 打上元数据，检索时按租户过滤：

# 入库：chunk 关联 tenant_id
vectorstore.add_texts(
    texts,
    metadatas=[{"doc_id": d, "tenant_id": tenant} for d in docs],
)

# 检索：filter 只返回本租户的 chunk
docs = vectorstore.similarity_search(
    question,
    k=3,
    filter={"tenant_id": current_tenant},     # 关键：检索阶段就隔离
)

验证：用租户 A 的会话问"退改签政策"，检索结果里只应有 tenant_id=A 的文档；换租户 B 问同一句，返回 B 的文档。若检索层没有 filter，即使生成时没泄露，日志里也会暴露出跨租户的"查询痕迹"。

安全红线：权限过滤必须做在检索阶段（数据访问边界），不能依赖 Prompt 里的"只回答本租户内容"——Prompt 是软的，filter 是硬的。RAG 的越权是最容易被忽略的漏洞，OWASP LLM Top 10 中注入与数据泄露的常见载体就是它。

</details>

<details>
<summary>练习5（完整流水线：改写→混合检索→Rerank→生成）</summary>

把 6.2/6.3/6.4 串成一条 LCEL 链：

def rewrite(query: str) -> str:
    return rewrite_chain.invoke(query)          # 6.2：口语→文档术语

def hybrid_retrieve(query: str) -> list:
    return ensemble_retriever.invoke(query)[:20]   # 6.3：BM25+向量，RRF 融合

def rerank_docs(docs: list) -> list:
    compressed = compression_retriever.compress_docs(docs, query)   # 6.4
    return compressed[:3]

def format_docs(docs: list) -> str:
    return "\n".join(f"[{i}] {d.page_content}" for i, d in enumerate(docs, 1))

pipeline = (
    RunnablePassthrough.assign(rewritten=lambda x: rewrite(x["question"]))
    | RunnableLambda(lambda x: {**x, "docs": hybrid_retrieve(x["rewritten"])})
    | RunnableLambda(lambda x: {**x, "top": rerank_docs(x["docs"])})
    | RunnablePassthrough.assign(context=lambda x: format_docs(x["top"]))
    | rag_prompt | llm | StrOutputParser()
)

注意：rewrite 的结果同时给检索用；检索 Query 走改写后的版本，Rerank 时也用同一 Query。这条链就是第 14 章综合实战"政策问答"分支的核心骨架，先在这里跑通，后面直接复用。
</details>

延伸阅读

• LangChain 官方文档 Retrievers 章节：BM25Retriever、EnsembleRetriever、MultiQueryRetriever、ParentDocumentRetriever、ContextualCompressionRetriever（以官方文档为准，注意 0.3.x 与 1.x 的差异）。
• Ragas 官方文档与 Metrics 页面：faithfulness、answer_relevancy、context_precision、context_recall 的定义与参数。
• FlagEmbedding 仓库（BAAI）：bge-reranker-base 的模型说明、下载与推理参数。
• RRF 原文：Cormack et al., "Reciprocal Rank Fusion outperforms Condorcet and individual Rank Learning Methods"（SIGIR 2009）。
• GraphRAG：Microsoft GraphRAG 论文与仓库。
• 关联章节：检索与生成的分工详见第 04 章、第 05 章；框架抽象（LCEL、Retriever 体系）在第 07 章；监控与可观测性清单在第 12 章；Agentic RAG 与工具编排在第 11 章；端到端系统在第 14 章。