进阶 RAG:分块、重排与混合检索

进阶 RAG:分块、重排与混合检索

本节摘要:朴素 RAG 检索 top-k 个最相似块,简单问题够用,但在多跳推理、模糊查询、超大语料上分崩离析。进阶 RAG,是「能跑通 10 篇文档的 demo」与「能服务 1000 万文档的系统」之间的分水岭。本节带你吃透把朴素管线打磨成生产级的四件套:混合检索(稠密向量 + BM25 关键词,用倒数排名融合 RRF 合并)、重排序(双编码器检索快但粗,交叉编码器重排慢但准,先 top-50 再精排 top-5)、查询变换(HyDE 用假设答案检索、多查询、step-back 退一步)、父子分块(小块精准命中、大块提供上下文)。最后给出 RAG 评估的三把尺子——召回率 Recall@k、忠实度、答案正确度,让你能客观判断改一处参数到底是变好还是变坏。

对应原课程:Phase 11 · Lesson 07 · advanced-rag(原英文 phases/11-llm-engineering/07-advanced-rag/docs/en.md)。

学习目标

阅读完本节,你应当能够:

1. 实现进阶分块策略(语义、递归、父子),保留文档结构与上下文。
2. 搭建一条混合检索管线:BM25 关键词匹配 + 语义向量检索 + 交叉编码器重排。
3. 运用查询变换(HyDE、多查询、step-back)改善模糊或复杂问题的检索。
4. 诊断并修复常见 RAG 失败:检错块、答案不在上下文里、多跳推理断裂。

一、问题与直觉

你在第 06 节搭了条朴素 RAG 管线,小语料上的简单问题能答。现在试试这几类:

模糊查询:「上季度营收是多少?」语义检索返回的是营收策略、营收预测、CFO 对营收增长的看法——这些都与「营收」语义相近,却都不含那个具体数字。正确的块写着「2025 年 Q3 的 4720 万美元」,却用了「earnings(盈利)」这个词而非「revenue(营收)」。嵌入模型觉得「营收策略」比「Q3 盈利 4720 万」离查询更近。

多跳问题:「哪个团队客户满意度提升最大?」要先找各团队满意度分数、比较、再取最大。没有任何单块包含答案,信息散落在各团队报告里。

大语料问题:你有 200 万块,正确答案在第 1847293 块。你的 top-5 拉回的是第 14、89201、1200000、44、901333 块——嵌入空间里都近,却都不含答案。这个规模下,近似最近邻搜索引入的误差足以把相关结果挤出 top-k。

朴素 RAG 失败的根因:向量相似 ≠ 相关。一块可以与查询语义相近,却对答题毫无帮助。进阶 RAG 用四类技术应对:混合检索(加关键词匹配)、重排序(更仔细地给候选打分)、查询变换(检索前先修查询)、更好的分块(在正确的粒度上检索)。

混合检索:语义 + 关键词

语义检索(向量相似)擅长理解含义。「如何取消订阅?」能匹配到「终止你套餐的步骤」,即使两句话没一个共同词。但它漏掉精确匹配:「错误码 E-4021」可能匹不到含「E-4021」的块,因为嵌入模型把它当噪声。

关键词检索(BM25)正相反。它精通精确匹配,「E-4021」完美命中;但「取消订阅」如果文档写「终止套餐」,就零结果。

混合检索:两边都跑,再合并结果。

BM25(Best Matching 25) 是关键词检索的标准算法,自 1990 年代起就是搜索引擎的脊梁。公式:

BM25(q, d) = 对查询中每个词 t 求和:
    IDF(t) * (tf(t,d) * (k1 + 1)) / (tf(t,d) + k1 * (1 - b + b * |d| / avgdl))
​

其中 tf(t,d) 是词 t 在文档 d 中的频次,IDF(t) 是逆文档频率,|d| 是文档长度,avgdl 是平均文档长度,k1 控制词频饱和(默认 1.2),b 控制长度归一(默认 0.75)。

通俗讲:BM25 给含查询词(尤其罕见词)的文档打高分,但重复出现收益递减——一个文档出现 50 次「revenue」,并不比出现 1 次的相关 50 倍。

倒数排名融合(RRF)

你有两份排名表:一份来自向量检索,一份来自 BM25。怎么合?倒数排名融合(Reciprocal Rank Fusion)是标准做法。

RRF_score(d) = 对每份排名 R 求和:
    1 / (k + rank_R(d))
​

其中 k 是常数(典型 60),防止榜首结果过度主导。

一份在向量检索排第 1、BM25 排第 5 的文档得:1/(60+1) + 1/(60+5) = 0.0164 + 0.0154 = 0.0318。

一份在向量排第 3、BM25 排第 2 的文档得:1/(60+3) + 1/(60+2) = 0.0159 + 0.0161 = 0.0320。

RRF 自然地平衡两个信号:两份榜都靠前的文档拿最高分;只在一个榜排第 1、另一榜缺席的拿中等分。它之所以稳健,是因为用的是排名而非原始分数——两个系统分数分布的差异不影响结果。

重排序

检索(无论向量、关键词还是混合)快但粗。它用双编码器(bi-encoder):查询和每个文档各自独立嵌入再比较,嵌入算一次可缓存,能扩展到百万文档。

重排序用交叉编码器(cross-encoder):查询和候选文档一起喂进模型,输出一个相关性分数。模型同时看到两段文本,能捕捉它们之间的细粒度交互。交叉编码器能理解「Q3 盈利多少?」与含「Q3 的 4720 万」的块高度相关,即使双编码器漏掉了这层联系。

代价是:交叉编码器比双编码器慢 100~1000 倍,因为它要把「查询-文档」对联合处理,无法对百万文档预计算。解法:检索一个大候选集(混合检索 top-50),再用交叉编码器精排出最终 top-5。

主流重排模型(2026 阵容):Cohere Rerank 3.5(托管、多语言、混合语料召回提升最大)、Voyage rerank-2.5(托管、延迟最低)、Jina-Reranker-v2 多语种(开源、100+ 语言)、bge-reranker-v2-m3(开源、强基线)、cross-encoder/ms-marco-MiniLM-L-6-v2(开源、CPU 可跑用于原型)、ColBERTv2/Jina-ColBERT-v2(晚交互多向量重排,打分是 O(token) 而非 O(doc))。

查询变换

有时候问题不在检索,而在查询本身。「那个新政策变动的事是啥来着?」是个糟糕的检索查询:不含具体词,嵌入模糊,任何检索系统都找不到对的文档。

查询改写:把用户查询改写成更好的检索查询,LLM 能做。

用户:"那个新政策变动的事是啥来着?"
改写:"近期的政策变更与更新"
​

HyDE(假设文档嵌入):不用原查询去搜,而是先生成一个假设答案,嵌入它,再搜与它相似的真实文档。

查询:"企业版退款政策是什么?"
假设答案:"企业版客户在购买后 60 天内可获全额退款。
退款按剩余订阅期按比例计算,5~7 个工作日内处理。"
​

嵌入这个假设答案,搜与它相似的真实文档。直觉是:假设答案在嵌入空间里离真实答案更近,比原问题更近。问题和答案是不同的语言结构,生成假设答案就在嵌入空间里搭了一座桥,从「问题空间」跨到「答案空间」。

HyDE 在检索前多一次 LLM 调用,增加 500~2000ms 延迟。当原查询检索质量差时才值。

多查询(multi-query):让 LLM 把一个查询拆成 3~5 个不同表述,各自检索后合并去重——覆盖不同措辞。

Step-back(退一步):对细节问题,先让 LLM 抽出一个更宽泛的概念问题去检索背景,再答具体问题。

父子分块

标准分块逼你做权衡:小块检索精准但缺上下文,大块上下文足但稀释相关性。父子分块消除这个权衡。

索引用小块(128 token)做检索;小块被命中时,返回它的父块(512 token)进提示。小块精准匹配查询,父块提供足够上下文让 LLM 生成好答案。

查询「企业版退款?」精准命中子块 C2,但提示收到的是完整父块 P,含处理时间和提交流程的周边上下文。

元数据过滤

跑向量检索前,先按元数据过滤语料:日期、来源、类别、作者、语言。这缩小搜索空间,防止无关结果。

「上个月安全政策有什么变更?」应只搜最近 30 天、安全类别的文档。不做元数据过滤,你会搜遍整个语料,可能捞回一份两年前的安全文档——只因它语义相近。

生产 RAG 给每块存元数据:源文档、创建日期、类别、作者、版本。向量库支持按元数据预过滤再做相似度搜索,这对大规模性能至关重要。

评估

你搭了 RAG 系统,怎么知道它行不行?三把尺子:

检索相关性(Recall@k):对一组带已知相关文档的测试问题,top-k 结果里出现了百分之多少的相关文档?某题答案在第 47 块,那第 47 块在不在 top-5 里?

忠实度(Faithfulness):生成的答案是否锚定于检索文档?检索块说「60 天退款窗口」,模型却说「90 天」,这就是忠实度失败——明明有正确上下文,模型还幻觉了。

答案正确度:生成的答案是否匹配期望答案?这是端到端指标,综合了检索质量与生成质量。

一个简单的忠实度检查:把生成答案里的每条断言拿出来,验证它(实质上)出现在检索块里。答案里有任何检索块都不存在的事实,大概率是幻觉。

二、从零实现

完整代码见原课程 code/advanced_rag.py,这里给出关键骨架。

步骤 1:BM25 实现

import math
from collections import Counter

class BM25:
    def __init__(self, k1=1.2, b=0.75):
        self.k1, self.b = k1, b
        self.docs, self.doc_lengths, self.avg_dl = [], [], 0
        self.doc_freqs, self.n_docs = {}, 0

    def index(self, documents):
        self.docs = documents
        self.n_docs = len(documents)
        self.doc_lengths, self.doc_freqs = [], {}
        for doc in documents:
            words = doc.lower().split()
            self.doc_lengths.append(len(words))
            for w in set(words):
                self.doc_freqs[w] = self.doc_freqs.get(w, 0) + 1
        self.avg_dl = sum(self.doc_lengths) / self.n_docs if self.n_docs else 1

    def score(self, query, doc_idx):
        q_words = query.lower().split()
        d_words = self.docs[doc_idx].lower().split()
        d_len = self.doc_lengths[doc_idx]
        counts = Counter(d_words)
        s = 0.0
        for term in q_words:
            if term not in counts:
                continue
            tf = counts[term]
            df = self.doc_freqs.get(term, 0)
            idf = math.log((self.n_docs - df + 0.5) / (df + 0.5) + 1)
            num = tf * (self.k1 + 1)
            den = tf + self.k1 * (1 - self.b + self.b * d_len / self.avg_dl)
            s += idf * num / den
        return s

    def search(self, query, top_k=10):
        scores = [(i, self.score(query, i)) for i in range(self.n_docs)]
        scores.sort(key=lambda x: x[1], reverse=True)
        return scores[:top_k]
​

步骤 2:倒数排名融合

def reciprocal_rank_fusion(ranked_lists, k=60):
    scores = {}
    for ranked in ranked_lists:
        for rank, (doc_id, _) in enumerate(ranked):
            scores[doc_id] = scores.get(doc_id, 0.0) + 1.0 / (k + rank + 1)
    return sorted(scores.items(), key=lambda x: x[1], reverse=True)
​

步骤 3:混合检索管线

def hybrid_search(query, chunks, vec_embs, vocab, idf, bm25,
                  top_k=5, fusion_k=60):
    q_emb = tfidf_embed(query, vocab, idf)
    vec_res = search(q_emb, vec_embs, top_k=top_k * 3)      # 多取些候选
    bm25_res = bm25.search(query, top_k=top_k * 3)
    fused = reciprocal_rank_fusion([vec_res, bm25_res], k=fusion_k)
    return fused[:top_k]
​

两路各取 top_k*3,留足候选给融合后再裁剪——这是「检索宽、精排严」的体现。

步骤 4:简易重排器

生产里用交叉编码器。这里做一个用词重叠、词重要性、短语匹配来打分的重排器。

def rerank(query, candidates, chunks):
    stop = {"the","a","an","is","are","was","were","what","how","why",
            "when","where","do","does","for","of","in","to","and","or",
            "on","at","by","it","its","this","that","with","from","be"}
    q_words = set(query.lower().split())
    q_terms = q_words - stop
    q_list = [w for w in query.lower().split() if w not in stop]
    q_bigrams = {q_list[i] + " " + q_list[i+1] for i in range(len(q_list)-1)}

    scored = []
    for doc_id, init_score in candidates:
        chunk = chunks[doc_id].lower()
        c_words = set(chunk.split())
        term_overlap = len(q_terms & c_words)
        bigram_match = sum(1 for bg in q_bigrams if bg in chunk)
        pos_boost = 0
        for t in q_terms:
            p = chunk.find(t)
            if p != -1 and p < len(chunk) // 3:    # 出现在前 1/3 加分
                pos_boost += 0.5
        score = term_overlap * 1.0 + bigram_match * 2.0 + pos_boost + init_score * 5.0
        scored.append((doc_id, score))
    scored.sort(key=lambda x: x[1], reverse=True)
    return scored
​

步骤 5:HyDE(假设文档嵌入)

def hyde_generate_hypothesis(query):
    templates = {
        "what": "「{query}」的答案如下:根据我们的文档,{topic} 涉及若干定义流程如何运作的具体政策与程序。",
        "how":  "针对「{query}」:流程分几步。首先发起请求,然后系统按既定规则处理。",
        "default": "关于「{query}」:我们的记录显示与该主题相关的具体细节和政策,能提供完整答案。",
    }
    ql = query.lower()
    key = "what" if ql.startswith("what") else "how" if ql.startswith("how") else "default"
    stop = {"what","is","the","how","do","does","a","an","for","of","to","in","on","at","by","and","or"}
    topic_words = [w for w in ql.split() if w not in stop]
    topic = " ".join(topic_words) if topic_words else "这个主题"
    return templates[key].format(query=query, topic=topic)

def hyde_search(query, chunks, vec_embs, vocab, idf, top_k=5):
    hyp = hyde_generate_hypothesis(query)
    hyp_emb = tfidf_embed(hyp, vocab, idf)
    return search(hyp_emb, vec_embs, top_k), hyp
​

生产里把 hyde_generate_hypothesis 换成一次真正的 LLM 调用,让模型写一段「假如答案长这样」的段落。

步骤 6:父子分块

def create_parent_child_chunks(text, parent_size=200, child_size=50):
    words = text.split()
    parents, children, child_to_parent = [], [], {}
    parent_idx, start = 0, 0
    while start < len(words):
        parent_end = min(start + parent_size, len(words))
        parents.append(" ".join(words[start:parent_end]))
        child_start = start
        while child_start < parent_end:
            child_end = min(child_start + child_size, parent_end)
            child_idx = len(children)
            children.append(" ".join(words[child_start:child_end]))
            child_to_parent[child_idx] = parent_idx
            child_start += child_size
        parent_idx += 1
        start += parent_size
    return parents, children, child_to_parent
​

检索时用 children 索引、用 child_to_parent 映射回父块、把父块放进提示。

步骤 7:忠实度评估

def evaluate_faithfulness(answer, retrieved_chunks):
    sents = [s.strip() for s in answer.split("。") if len(s.strip()) > 10]
    if not sents:
        return 1.0, []
    stop = {"的","是","在","和","与","了","为","对","由","从"}
    context = " ".join(retrieved_chunks).lower()
    grounded, ungrounded = 0, []
    for s in sents:
        words = set(s.lower().split()) - stop
        if not words:
            grounded += 1
            continue
        matched = sum(1 for w in words if w in context)
        if matched / len(words) >= 0.5:
            grounded += 1
        else:
            ungrounded.append(s)
    return grounded / len(sents), ungrounded

def evaluate_retrieval_recall(qa_pairs, retrieval_fn, k=5):
    total, results = 0.0, []
    for query, relevant in qa_pairs:
        retrieved = set(idx for idx, _ in retrieval_fn(query, k))
        rel = set(relevant)
        hits = len(retrieved & rel)
        recall = hits / len(rel) if rel else 1.0
        total += recall
        results.append({"query": query, "recall": recall, "hits": hits})
    return total / len(qa_pairs) if qa_pairs else 0, results
​

💡 评估要分层做:Recall@k 测检索、faithfulness 测接地、correctness 测端到端。改了分块大小只 Recall 动了,说明问题在检索;Recall 没动 correctness 掉了,说明问题在生成。分层让你定位而不是瞎猜。

三、框架对比

用真实交叉编码器重排

# from sentence_transformers import CrossEncoder
# reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")
# def rerank_with_ce(query, candidates, chunks, top_k=5):
#     pairs = [(query, chunks[doc_id]) for doc_id, _ in candidates]
#     scores = reranker.predict(pairs)
#     scored = sorted(zip([d for d,_ in candidates], scores), key=lambda x:x[1], reverse=True)
#     return scored[:top_k]
​

Cohere 托管重排:

# import cohere
# co = cohere.Client()
# def rerank_with_cohere(query, candidates, chunks, top_k=5):
#     docs = [chunks[doc_id] for doc_id, _ in candidates]
#     resp = co.rerank(model="rerank-english-v3.0", query=query,
#                      documents=docs, top_n=top_k)
#     return [(candidates[r.index][0], r.relevance_score) for r in resp.results]
​

真实 LLM 做 HyDE:

# import anthropic
# client = anthropic.Anthropic()
# def hyde_with_llm(query):
#     r = client.messages.create(model="claude-sonnet-5", max_tokens=256,
#         messages=[{"role":"user","content":
#             f"写一小段文字,作为下面这个问题的理想答案。不要说不知道,就写答案该有的样子。\n\n问题:{query}"}])
#     return r.content[0].text
​

生产级混合检索用 Weaviate:

# import weaviate
# client = weaviate.connect_to_local()
# collection = client.collections.get("Documents")
# resp = collection.query.hybrid(query="企业版退款政策", alpha=0.5, limit=10)
​

alpha 控制平衡:0.0 纯关键词(BM25),1.0 纯向量,0.5 等权。多数生产系统 alpha 在 0.3~0.7 之间。

LangChain 与 LlamaIndex 都内置了 ContextualCompressionRetriever、MultiQueryRetriever、ParentDocumentRetriever,把本节的手写逻辑封装成开箱即用组件。优点是省胶水代码,代价是默认参数不一定适配你的语料——本节的从零实现让你看懂每个旋钮在做什么。

四、可复用产物

本节产出两个可复用文件(位于原课程 outputs/):

• prompt-advanced-rag-debugger.md:一个元提示,诊断和修复 RAG 质量问题。喂给它失败案例(查询、检索到的块、生成的答案、期望答案),它定位是检索失败、接地失败还是生成失败,并给修复建议。
• skill-advanced-rag.md:一个技能文件,教 Agent 如何搭建带混合检索与重排的生产级 RAG。

Python 代码(code/advanced_rag.py)是一条独立进阶管线,把模拟函数换成真实嵌入/重排/LLM API 即可投产,BM25、RRF、混合检索、父子分块、评估逻辑均无需修改。

五、练习

1. 三路对比:在样例文档上对比 BM25、向量检索、混合检索。对 5 条测试查询,记录哪种方法把最相关块放在第 1 位。混合检索应至少在 3 条上胜出。

2. 元数据过滤:给每份文档加「类别」字段(安全、计费、API、产品)。向量检索前先按类别过滤。用「用了什么加密?」测试,验证它只搜安全类别的块。

3. 完整 HyDE 管线:用第 06 节的简易生成函数搭一条完整 HyDE 管线。在 5 条测试查询上对比直接检索与 HyDE 检索的 top-3 相关性。HyDE 应在模糊查询上提升明显。

4. 父子分块对比:在样例文档上实现父子分块,子块 30、父块 100。用子块检索、返回父块进提示,与标准分块(chunk_size=50)对比生成答案质量。

5. 评估数据集:造 10 个带已知答案块的问题,测量(纯向量、纯 BM25、混合、混合+重排)四套方案在 Recall@3、Recall@5、Recall@10 上的表现,画图找出重排帮助最大的区间。

本节要点回顾

1. 向量相似 ≠ 相关:语义相近未必对答题有用,这是朴素 RAG 在模糊、多跳、大语料上崩盘的根因。
2. 混合检索互补:语义擅长含义、BM25 擅长精确匹配,两路并用比单路强。
3. BM25 是关键词检索脊梁:含查询词(尤其罕见词)打高分,重复收益递减,k1=1.2、b=0.75 是默认。
4. RRF 用排名而非分数融合:1/(k+rank) 求和,稳健地平衡两路信号,不受分数分布差异影响。
5. 重排是「检索宽、精排严」:双编码器粗排 top-50,交叉编码器精排 top-5,慢 100~1000 倍但只在候选集上跑。
6. 查询变换修查询本身:HyDE 用假设答案检索(答案空间离真答案更近)、多查询覆盖措辞、step-back 抽背景。
7. 父子分块消除粒度矛盾:小块精准命中、大块提供上下文,两全其美。
8. 元数据过滤缩小搜索空间:按日期/来源/类别预过滤,防止捞回陈旧或无关文档,大规模性能关键。
9. 评估分三层:Recall@k 测检索、faithfulness 测接地、correctness 测端到端,分层让你定位而非瞎猜。
10. 进阶 RAG 是 demo 与生产的分水岭:把朴素 top-k 升级为「查询变换 → 混合检索 → 重排 → 父子分块 → 三层评估」,才能从 10 篇文档跑通到服务 1000 万文档。

下一节,我们离开检索、转向微调——当 RAG 不够、你需要模型习得特定风格或领域推理模式时,LoRA 与 QLoRA 如何用极小显存把大模型调教成你的专属模型。