Advanced RAG 实战指南

“基础 RAG 解决有无问题，高级 RAG 解决体验问题。”

一、基础 RAG 的瓶颈

在《RAG 实战指南》中，我们介绍了基础 RAG 的流程：检索 → 读取 → 生成。但基础 RAG 有几个常见问题：

问题 现象 根因
语义漂移 检索到的文档和问题表面相关，但不包含答案 embedding 模型不够精确
上下文截断 关键信息被切分到不同 chunk，检索只召回了一半 chunk 策略不当
噪音干扰 检索到太多无关文档，反而稀释了正确答案 检索精度不足
查询意图偏差 用户问的是 A，embedding 却匹配到了 B 查询和文档表达方式不同

Advanced RAG 就是针对这些问题的系统性解决方案。

二、Query 改写（Query Rewriting）

2.1 为什么需要 Query 改写？

用户的问题和知识库中的文档，往往使用不同的表达方式：

1
2
3
4
5
6
7
用户问："LLM 训练需要多少显存？"
文档写："以 FP16 为例，175B 参数的模型需要约 350GB 显存"
          ↑ 不同表述，表面相似度低

用户问："帮我查一下 Transformer 是谁发明的"
文档写："《Attention Is All You Need》由 Google 提出..."
          ↑ 隐式相关，直接匹配容易失败

Query 改写的核心思想：让 LLM 先”理解”用户想问什么，再去检索。

2.2 Query Expansion（查询扩展）

用 LLM 将用户的问题扩展成多个相关查询，再并行检索。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
def query_expansion(original_query: str, llm) -> list[str]:
    """将单一查询扩展为多个相关查询"""
    prompt = f"""
用户原始问题：{original_query}

请将这个问题扩展成 3 个相关的搜索查询，覆盖不同的表述角度。
每个查询一行，直接输出查询，不要解释。

示例：
原问题：如何训练大模型
扩展：
1. LLM 训练方法有哪些
2. 大语言模型训练流程
3. 深度学习模型训练步骤
"""
    response = llm.invoke(prompt)
    queries = [original_query] + response.strip().split('\n')
    return queries

# 使用
queries = query_expansion("LLM 训练需要多少显存", gpt4)
# ['LLM 训练需要多少显存', 
#  '大模型显存计算方法', 
#  '不同精度训练的显存占用',
#  '175B 模型需要多少 GPU 显存']

2.3 Query Decomposition（查询分解）

将复杂问题拆解为多个简单子问题，分别检索后再合并。

1
2
3
4
5
6
7
8
用户问题："LLaMA 和 GPT-3 相比，训练数据有什么差异？"

拆解为：
1. LLaMA 的训练数据是什么？
2. GPT-3 的训练数据是什么？
3. 两者的差异对比

分别检索 → 分别生成 → 合并回答

2.4 HyDE（Hypothetical Document Embedding）

让 LLM 先根据问题假设性回答，然后用这个假设性回答去做检索。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
def hyde_retrieval(query: str, vector_store, llm) -> list[Document]:
    """HyDE 检索"""
    # Step 1: 让 LLM 生成一个假设性回答
    hypothetical = llm.invoke(f"""
假设你是知识库中的一篇文章，请针对以下问题写一段假设性的回答。
注意：这段回答可能是错误的，但请尽量符合逻辑。

问题：{query}

假设性回答：
""")
    
    # Step 2: 用假设性回答做 embedding 检索
    # （假设性回答和真实文档往往在向量空间上接近）
    results = vector_store.similarity_search(hypothetical, k=3)
    return results

为什么 HyDE 有效？ LLM 生成的假设性回答在语言风格和内容分布上与知识库文章相似，因此向量空间中更接近真实文档。

三、混合检索（Hybrid Search）

3.1 单一检索的局限

检索方式 优点 缺点
向量检索 语义匹配强，支持同义词 对专有名词、ID 等精确匹配弱
关键词检索（BM25） 精确匹配专有名词、数字、代码 无法处理语义相关性

最佳实践：两者结合，取长补短。

3.2 Hybrid Search 实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
from rank_bm25 import BM25Okapi
import numpy as np

class HybridRetriever:
    def __init__(self, chunks: list[str], embedder):
        self.chunks = chunks
        self.embedder = embedder
        
        # 构建 BM25 索引
        tokenized_chunks = [chunk.split() for chunk in chunks]
        self.bm25 = BM25Okapi(tokenized_chunks)
        
        # 预计算所有 chunk 的向量
        self.embeddings = embedder.embed(chunks)
    
    def search(self, query: str, k: int = 5, alpha: float = 0.5):
        """
        alpha=0.5 表示向量检索和 BM25 各占一半
        alpha=1.0 表示只用向量检索
        alpha=0.0 表示只用 BM25
        """
        # 1. BM25 关键词检索
        bm25_scores = self.bm25.get_scores(query.split())
        bm25_scores = self._normalize(bm25_scores)
        
        # 2. 向量语义检索
        query_embedding = self.embedder.embed([query])[0]
        vector_scores = self._cosine_similarity(query_embedding, self.embeddings)
        
        # 3. 加权融合
        final_scores = alpha * vector_scores + (1 - alpha) * bm25_scores
        
        # 4. 取 top-k
        top_indices = np.argsort(final_scores)[-k:][::-1]
        return [self.chunks[i] for i in top_indices]
    
    def _normalize(self, scores):
        return (scores - scores.min()) / (scores.max() - scores.min() + 1e-8)
    
    def _cosine_similarity(self, a, b_matrix):
        a_norm = np.linalg.norm(a)
        b_norm = np.linalg.norm(b_matrix, axis=1)
        return np.dot(b_matrix, a) / (a_norm * b_norm + 1e-8)

3.3 RRF（Reciprocal Rank Fusion）

另一种融合策略：将两种检索的结果按排名融合。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
def rrf_fusion(results_list: list[list], k: int = 60) -> list:
    """RRF 融合多个检索结果
    
    results_list: 多个检索系统返回的排序结果列表
    k: RRF 参数，通常设为 60
    """
    scores = {}
    
    for results in results_list:
        for rank, doc in enumerate(results):
            # RRF 公式：1 / (k + rank)
            # 排名越靠前，加分越多
            score = 1 / (k + rank + 1)
            scores[doc] = scores.get(doc, 0) + score
    
    # 按融合分数排序
    sorted_docs = sorted(scores.items(), key=lambda x: x[1], reverse=True)
    return [doc for doc, _ in sorted_docs]

四、句子窗口检索（Sentence Window Retrieval）

4.1 问题

简单按固定长度（如 500 字）切分 chunk，往往会把一个连贯的段落切成两半，导致关键信息分散在不同的 chunk 中。

1
2
3
4
5
6
7
8
原文："Transformer 由 Google 在 2017 年提出，它完全基于注意力机制，
      摒弃了传统的 RNN 结构，在 NLP 领域取得了突破性进展。"

错误切分：
Chunk 1: "Transformer 由 Google 在 2017 年提出，它完全基于注意力机制，"
Chunk 2: "摒弃了传统的 RNN 结构，在 NLP 领域取得了突破性进展。"
                                      ↑
                              检索时只能匹配半句，无法完整理解

4.2 句子窗口检索原理

核心思想：检索时用精确的句子级别匹配，但返回时附上周围的上下文窗口。

1
2
3
1. 将文档切分成"句子"级别（每句一个 chunk）
2. 检索时找最相关的那句话
3. 返回时将周围 k 个句子拼接成完整上下文

4.3 实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
class SentenceWindowRetriever:
    def __init__(self, documents: list[str], embedder, window_size: int = 3):
        # 1. 句子级别切分
        self.sentences = self._split_into_sentences(documents)
        self.embedder = embedder
        self.window_size = window_size
        
        # 2. 句子级别索引
        self.embeddings = self.embedder.embed(self.sentences)
    
    def retrieve(self, query: str, k: int = 5) -> list[dict]:
        """返回检索结果，每条包含句子+上下文窗口"""
        # 向量检索找到最相关的句子
        query_emb = self.embedder.embed([query])[0]
        scores = self._cosine_similarity(query_emb, self.embeddings)
        top_indices = np.argsort(scores)[-k:][::-1]
        
        results = []
        for idx in top_indices:
            # 找到窗口内的所有句子
            start = max(0, idx - self.window_size)
            end = min(len(self.sentences), idx + self.window_size + 1)
            
            window_context = ' '.join(self.sentences[start:end])
            results.append({
                "matched_sentence": self.sentences[idx],
                "window_context": window_context,
                "score": scores[idx]
            })
        
        return results
    
    def _split_into_sentences(self, documents: list[str]) -> list[str]:
        """用简单的启发式分句"""
        import re
        sentences = []
        for doc in documents:
            # 按句号、问号、感叹号分句
            parts = re.split(r'[。！？\n]', doc)
            sentences.extend([p.strip() for p in parts if p.strip()])
        return sentences

五、父子文档检索（Parent-Document Retrieval）

5.1 思想

和句子窗口相反：父文档保证完整性，子文档提供精确检索点。

1
2
3
4
5
结构：
├── 父文档（Parent）：整个章节（2000字），保证上下文完整
│   ├── 子文档 A（Child）：第一段（300字），用于精确检索
│   ├── 子文档 B（Child）：第二段（300字），用于精确检索
│   └── 子文档 C（Child）：第三段（300字），用于精确检索

检索流程：

  1. 用子文档做精确检索
  2. 匹配到的子文档对应的父文档作为最终 context

5.2 实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
class ParentDocumentRetriever:
    def __init__(self, documents: list[str], embedder, 
                 parent_chunk_size: int = 2000, child_chunk_size: int = 300):
        self.embedder = embedder
        self.parent_chunk_size = parent_chunk_size
        self.child_chunk_size = child_chunk_size
        
        # 切分父子文档
        self.parent_chunks, self.child_chunks, self.mapping = \
            self._create_parent_child_chunks(documents)
        
        # 索引子文档
        self.child_embeddings = self.embedder.embed(self.child_chunks)
    
    def _create_parent_child_chunks(self, documents):
        parents, children, mapping = [], [], []
        
        for doc in documents:
            # 先切父文档
            parent_chunks = self._chunk_text(doc, self.parent_chunk_size)
            for p_chunk in parent_chunks:
                # 每个父文档再切子文档
                child_chunks = self._chunk_text(p_chunk, self.child_chunk_size)
                for c_chunk in child_chunks:
                    parents.append(p_chunk)
                    children.append(c_chunk)
                    mapping.append(p_chunk)  # 记录父子关系
        
        return parents, children, mapping
    
    def retrieve(self, query: str, k: int = 5) -> list[str]:
        """检索并返回父文档（完整上下文）"""
        # 1. 检索子文档
        query_emb = self.embedder.embed([query])[0]
        scores = self._cosine_similarity(query_emb, self.child_embeddings)
        top_child_indices = np.argsort(scores)[-k:][::-1]
        
        # 2. 通过子文档索引找到对应的父文档
        seen = set()
        parent_docs = []
        for child_idx in top_child_indices:
            parent = self.mapping[child_idx]
            if parent not in seen:
                seen.add(parent)
                parent_docs.append(parent)
        
        return parent_docs
    
    def _chunk_text(self, text: str, chunk_size: int) -> list[str]:
        """固定长度分块"""
        words = text.split()
        chunks = []
        for i in range(0, len(words), chunk_size):
            chunks.append(' '.join(words[i:i+chunk_size]))
        return chunks

六、重排序（Reranking）

6.1 问题

向量检索找到的是”语义上最接近”的文档，但语义接近不等于”对问题最有价值”。Reranker 的作用是根据问题和文档的关联性重新排序。

1
2
3
4
5
6
7
8
9
10
11
向量检索结果（按向量相似度）：
1. "Transformer 是 2017 年提出的..."  score=0.95
2. "BERT 是 Google 提出的模型..."     score=0.93
3. "注意力机制是 Transformer 的核心..." score=0.91

用户问题："Transformer 是谁发明的？"

Reranker 重排后：
1. "Transformer 是 2017 年 Google 团队提出..."  (准确回答问题)
2. "注意力机制是 Transformer 的核心..."      (相关但非答案)
3. "BERT 是 Google 提出的模型..."             (不相关)

6.2 Cross-Encoder Reranker

Reranker 通常用 Cross-Encoder 模型实现——将 query 和 document 一起输入模型，输出一个关联性分数。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from sentence_transformers import CrossEncoder

class Reranker:
    def __init__(self, model_name: str = "cross-encoder/ms-marco-MiniLM-L-12-v2"):
        # MS MARCO 是常用的 Reranker 训练数据集
        self.model = CrossEncoder(model_name)
    
    def rerank(self, query: str, documents: list[str], top_k: int = 3) -> list[str]:
        """对文档列表进行重排序"""
        # 构造 query-document pair
        pairs = [(query, doc) for doc in documents]
        
        # 一次性输入模型，获得所有分数
        scores = self.model.predict(pairs)
        
        # 按分数排序
        ranked_indices = np.argsort(scores)[::-1][:top_k]
        return [documents[i] for i in ranked_indices]

6.3 Cohere Rerank

Cohere 提供商业化的 Rerank API，效果通常比自己训练的 Cross-Encoder 更好。

1
2
3
4
5
6
7
8
9
10
11
12
import cohere

co = cohere.Client("YOUR_API_KEY")

def cohere_rerank(query: str, documents: list[str], top_k: int = 3):
    response = co.rerank(
        model="rerank-multilingual-v2.0",
        query=query,
        documents=documents,
        top_n=top_k
    )
    return [result.document.text for result in results.results]

6.4 在 RAG 中集成 Reranker

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
def advanced_rag(query: str, vector_store, reranker, llm, k: int = 20, rerank_k: int = 5):
    """完整的高级 RAG 流程"""
    # Step 1: 检索（召回更多，k=20）
    retrieved_docs = vector_store.similarity_search(query, k=k)
    
    # Step 2: 重排序（精选 top 5）
    reranked_docs = reranker.rerank(query, retrieved_docs, top_k=rerank_k)
    
    # Step 3: 生成
    context = '\n\n'.join(reranked_docs)
    prompt = f"""
根据以下上下文回答问题。如果上下文中没有答案，请如实说明。

上下文：
{context}

问题：{query}
"""
    response = llm.invoke(prompt)
    return response

七、Self-RAG：让 RAG 自己判断

7.1 什么是 Self-RAG？

Self-RAG 是斯坦福提出的方法，核心思想是让模型自己判断是否需要检索，以及检索到的内容是否有用。

7.2 Self-RAG 的四种特殊 Token

1
2
3
4
5
6
7
8
# Self-RAG 引入了 4 种特殊 token（反思 token）

[Retrieval]   # 模型决定需要检索时输出
[No Retrieval] # 模型决定不需要检索时输出
[Relevant]     # 检索到的文档对问题有帮助
[Irrelevant]   # 检索到的文档对问题没帮助
[Supported]    # 回答被文档支持
[Contradict]   # 回答与文档矛盾

7.3 Self-RAG 的推理流程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
def self_rag_inference(query: str, rag_model, retriever, k: int = 5):
    response = ""
    
    # 让模型判断是否需要检索
    decision = rag_model.classify(query)  # [Retrieval] or [No Retrieval]
    
    if decision == "[Retrieval]":
        # 1. 检索相关文档
        docs = retriever.search(query, k=k)
        
        # 2. 让模型判断每个文档是否相关
        for doc in docs:
            relevance = rag_model.judge(f"{query}", doc)  # [Relevant] or [Irrelevant]
            
            if relevance == "[Relevant]":
                # 3. 让模型基于这个文档生成回答片段
               片段 = rag_model.generate(f"{query} [Doc] {doc}")
                
                # 4. 判断回答是否被文档支持
                support = rag_model.evaluate(f"{片段} [Doc] {doc}")  # [Supported] or [Contradict]
                
                if support == "[Supported]":
                    response += 片段 + "\n"
                else:
                    response += f"[Contradict] 忽略此片段\n"
    else:
        # 直接生成
        response = rag_model.generate(query)
    
    return response

八、总结

Advanced RAG 的核心技术：

  1. Query 改写：用 LLM 理解用户真实意图，扩展/分解查询
  2. 混合检索：结合向量检索和 BM25，兼顾语义和精确匹配
  3. 句子窗口检索：精确句子匹配 + 周围上下文
  4. 父子文档检索：子文档精确检索 + 父文档完整上下文
  5. Reranker：用 Cross-Encoder 做精细化排序
  6. Self-RAG：让模型自己判断检索的必要性

这些技术可以叠加使用，构建出效果远优于基础 RAG 的系统。

相关文章：