RAG 检索增强生成知识系统
本文档系统阐述 RAG（Retrieval-Augmented Generation，检索增强生成）的完整技术链路，涵盖从文档解析到评测的十个核心环节
适合作为构建企业知识库和 RAG 系统的核心技术参考资料。
1. 文档解析
1.1 定义
文档解析是 RAG 系统的第一步，将 PDF、Word、Markdown、HTML 等不同格式的文档统一提取为纯文本加结构信息，为后续的清洗、分块和向量化做准备。解析质量直接决定整个 RAG 系统的知识上限。
1.2 核心原理
+----------------------------------------------------------+
| 文档解析流程 |
+----------------------------------------------------------+
| |
| 原始文档 |
| +----------+ +----------+ +----------+ +----------+ |
| | Markdown | | HTML | | PDF | | Word | |
| +----+-----+ +----+-----+ +----+-----+ +----+-----+ |
| | | | | |
| v v v v |
| +-----------------------------------------------+ |
| | 格式特定的解析器 | |
| | (标题层级/段落/列表/表格/图片描述) | |
| +----------------------+------------------------+ |
| | |
| v |
| +-----------------------------------------------+ |
| | 统一输出: sections = [ | |
| | {heading, level, content, metadata}, | |
| | ... | |
| | ] | |
| +-----------------------------------------------+ |
| |
+----------------------------------------------------------+1.3 Markdown 解析
def parse_markdown(text: str) -> dict:
"""解析 Markdown 文本，提取标题层级和正文内容"""
lines = text.strip().split("\n")
sections = []
current_heading = ""
current_level = 0
current_content = []
for line in lines:
if line.startswith("#"):
# 保存上一个段落
if current_content:
sections.append({
"heading": current_heading,
"level": current_level,
"content": "\n".join(current_content).strip(),
})
current_content = []
current_level = len(line) - len(line.lstrip("#"))
current_heading = line.lstrip("# ").strip()
else:
current_content.append(line)
if current_content:
sections.append({
"heading": current_heading,
"level": current_level,
"content": "\n".join(current_content).strip(),
})
return {"format": "markdown", "sections": sections}1.4 HTML 解析
def parse_html(text: str) -> dict:
"""简易 HTML 解析：提取标题和正文，去除标签"""
import re
# 提取 <title>
title_match = re.search(r"<title>(.*?)</title>", text, re.DOTALL)
title = title_match.group(1).strip() if title_match else ""
# 移除 script 和 style 标签
text = re.sub(r"<script[^>]*>.*?</script>", "", text, flags=re.DOTALL)
text = re.sub(r"<style[^>]*>.*?</style>", "", text, flags=re.DOTALL)
# 提取 <h1>~<h3> 和 <p> 标签内容
headings = re.findall(r"<h[1-3][^>]*>(.*?)</h[1-3]>", text, re.DOTALL)
paragraphs = re.findall(r"<p[^>]*>(.*?)</p>", text, re.DOTALL)
# 清除剩余 HTML 标签
clean_headings = [re.sub(r"<[^>]+>", "", h).strip() for h in headings]
clean_paragraphs = [re.sub(r"<[^>]+>", "", p).strip() for p in paragraphs]
return {"format": "html", "title": title, "headings": clean_headings, "paragraphs": clean_paragraphs}1.5 不同格式解析对比
1.6 解析中的关键决策
解析策略选择:
1. 保留结构 vs 纯文本
- 保留结构: 标题层级用于分块边界
- 纯文本: 简单但丢失文档结构
2. 表格处理
- 转为文本描述: "表格：列A=值1, 列B=值2"
- 保留 Markdown 表格: 保留结构
- 跳过: 信息损失
3. 图片处理
- OCR 提取文字
- 图片描述（VLM 生成）
- 跳过
4. 元数据保留
- 来源文件名、页码、章节
- 用于检索结果溯源1.7 与其他概念的关联
-> 文本清洗与分块：解析输出是分块的输入
-> 向量化与存储：解析后的文本将被向量化
<- 无前置依赖：文档解析是 RAG 的入口
2. 文本清洗与分块策略
2.1 定义
文本清洗去除噪声（多余空白、特殊字符、页眉页脚），分块（Chunking）把长文档切成适合检索的证据片段。分块策略直接决定 RAG 的召回质量——块太大稀释语义、太小丢失上下文。
2.2 文本清洗
def clean_text(text: str) -> str:
"""清洗文本：去除多余空白、特殊字符、统一格式"""
lines = text.split("\n")
cleaned_lines = []
for line in lines:
line = line.strip()
if not line:
continue
# 去除纯数字行（页码）
if re.match(r"^\d{1,3}$", line):
continue
# 去除 HTML 残留标签
line = re.sub(r"<[^>]+>", "", line)
# 合并多个空格
line = re.sub(r"[ \t]+", " ", line)
cleaned_lines.append(line)
return "\n".join(cleaned_lines)
def normalize_text(text: str) -> str:
"""归一化文本：全角转半角，统一标点"""
text = text.replace("\u3000", " ") # 全角空格
text = re.sub(r"[\uff10-\uff19]", lambda m: chr(ord(m.group()) - 0xFFEE), text)
text = re.sub(r"[\uff21-\uff3a]", lambda m: chr(ord(m.group()) - 0xFF00 + 0x41), text)
text = re.sub(r"[\uff41-\uff5a]", lambda m: chr(ord(m.group()) - 0xFF00 + 0x61), text)
return text2.3 分块策略
2.3.1 固定长度分块
def chunk_by_fixed_size(text: str, chunk_size: int = 200, overlap: int = 50) -> list[str]:
"""固定长度分块：按字符数切割，带重叠窗口"""
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
chunk = text[start:end]
chunks.append(chunk)
if end >= len(text):
break
start = end - overlap # 重叠部分
return chunks2.3.2 按句子分块
def chunk_by_sentence(text: str, sentences_per_chunk: int = 3) -> list[str]:
"""按句子分块：以句末标点为分割点，每 N 句一个块"""
sentences = re.split(r"(?<=[。！？.!?\n])", text)
sentences = [s.strip() for s in sentences if s.strip()]
chunks = []
for i in range(0, len(sentences), sentences_per_chunk):
chunk = "".join(sentences[i : i + sentences_per_chunk])
chunks.append(chunk)
return chunks2.3.3 按 Token 分块
def chunk_by_tokens(text: str, max_tokens: int = 256, overlap: int = 50) -> list[str]:
"""按 Token 数分块：使用 tiktoken 精确控制"""
enc = tiktoken.get_encoding("cl100k_base")
tokens = enc.encode(text)
chunks = []
start = 0
while start < len(tokens):
end = start + max_tokens
chunk_tokens = tokens[start:end]
chunks.append(enc.decode(chunk_tokens))
if end >= len(tokens):
break
start = end - overlap
return chunks2.3.4 按分隔符分块
def chunk_by_separator(text: str, separator: str = "\n\n") -> list[str]:
"""按分隔符分块：保留文档自然段落结构"""
chunks = text.split(separator)
return [c.strip() for c in chunks if c.strip()]2.4 分块策略对比
2.5 重叠窗口的作用
无重叠 (可能丢失上下文):
块1: "RAG 是一种检索增强生成技术。它的核心思想是"
块2: "通过检索外部知识库来增强模型的回答能力。"
-> 检索到块2时，缺少"RAG"的上下文
有重叠 (overlap=10):
块1: "RAG 是一种检索增强生成技术。它的核心思想是"
块2: "检索增强生成技术。它的核心思想是通过检索外部知识库..."
-> 块2包含前文上下文，语义更完整2.6 分块大小对 RAG 的影响
块大小的影响:
太小 (50 tokens):
- 语义不完整，检索准确率低
- 需要检索更多块才能覆盖答案
- 优点：定位精确
适中 (200-500 tokens):
- 语义完整，检索准确率高
- 上下文充分，适合多数场景
- 推荐默认值
太大 (1000+ tokens):
- 一个块包含多个主题，稀释语义
- 检索准确率下降
- 占用过多上下文窗口2.7 与其他概念的关联
<- 文档解析：解析结果作为分块输入
-> 向量化与存储：分块结果将被向量化存入数据库
-> 向量检索：分块质量直接影响检索效果
3. 向量化与向量数据库
3.1 定义
向量化（Embedding）把文本变成高维向量，向量数据库存储这些向量并支持高效相似度搜索。这是 RAG 检索的基础设施——没有向量数据库，就无法实现毫秒级的语义检索。
对应 Demo: demos/02_RAG/03_向量化与存储.py
3.2 向量化流程
文本分块 -> Embedding 模型 -> 向量 -> 向量数据库
"RAG 是检索增强生成" -> model -> [0.12, -0.34, ...] -> Qdrant
"它通过检索知识库回答" -> model -> [0.15, -0.30, ...] -> Qdrant
向量维度: 64 ~ 3072 (取决于模型)
常用模型: text-embedding-ada-002 (1536维)
text-embedding-3-small (1536维)
text-embedding-3-large (3072维)
BGE-zh (1024维)3.3 Qdrant 向量数据库操作
def create_collection(client: QdrantClient, collection_name: str, vector_size: int) -> None:
"""创建向量集合（相当于数据库中的表）"""
if client.collection_exists(collection_name):
client.delete_collection(collection_name)
client.create_collection(
collection_name=collection_name,
vectors_config=VectorParams(size=vector_size, distance=Distance.COSINE),
)
def upsert_points(client: QdrantClient, collection_name: str, chunks: list[dict], dim: int) -> None:
"""将文档分块向量化并写入向量数据库"""
points = []
for chunk in chunks:
vector = mock_embedding(chunk["content"], dim)
points.append(
PointStruct(
id=chunk["id"],
vector=vector,
payload={ # payload 存储原始文本和元数据
"content": chunk["content"],
"source": chunk["metadata"]["source"],
"chunk_index": chunk["metadata"]["chunk_index"],
},
)
)
client.upsert(collection_name=collection_name, points=points)3.4 向量数据库核心概念
+----------------------------------------------------------+
| Qdrant 数据模型 |
+----------------------------------------------------------+
| |
| Collection (集合) = 一类文档的向量表 |
| | |
| +-- Point (点) = 一个文档块 |
| | |
| +-- Vector (向量): [0.12, -0.34, ...] |
| | |
| +-- Payload (元数据): { |
| "content": "文档块文本", |
| "source": "文件名", |
| "chunk_index": 3, |
| "category": "技术" |
| } |
| |
| 距离度量: |
| - Cosine (余弦): 适合文本语义匹配 |
| - Euclid (欧氏): 适合图像 |
| - Dot (点积): 适合已归一化向量 |
| |
+----------------------------------------------------------+3.5 向量数据库对比
3.6 向量索引算法
暴力搜索 (Flat):
遍历所有向量计算相似度 -> O(n)
精确但慢，适合小数据集 (<10万)
HNSW (Hierarchical Navigable Small World):
构建层次图结构，近似最近邻搜索
速度快，精度高，适合中等数据集 (10万~1000万)
IVF (Inverted File Index):
聚类后只搜索最近的几个簇
速度极快，精度略低，适合超大数据集
PQ (Product Quantization):
向量压缩，减少内存占用
牺牲精度换取存储效率3.7 与其他概念的关联
<- 文本清洗与分块：分块结果是向量化的输入
-> 向量检索：向量数据库支持相似度搜索
-> 混合检索：向量检索是混合检索的组件之一
4. 向量检索
4.1 定义
向量检索（语义检索）是 RAG 的核心：把用户问题向量化，在向量库中找到最相似的文档块。关键在于相似度计算、Top-K 召回和元数据过滤。
对应 Demo: demos/02_RAG/04_向量检索.py
4.2 检索流程
用户问题: "如何训练大语言模型"
|
v
Embedding 模型 -> query_vector [0.12, -0.34, ...]
|
v
向量数据库 (Qdrant) 搜索
|
v
计算 query_vector 与所有文档块向量的余弦相似度
|
v
返回 Top-K 最相似的文档块
|
v
结果:
1. [score=0.92] "大语言模型的训练包括预训练和微调..."
2. [score=0.87] "预训练使用大规模无标注文本..."
3. [score=0.85] "微调阶段使用监督学习..."
4. [score=0.78] "模型评估指标包括..."
5. [score=0.71] "训练成本与参数量..."4.3 基本检索实现
def vector_search(client: QdrantClient, collection: str, query: str, dim: int, top_k: int = 5) -> list[dict]:
"""向量语义检索：返回 Top-K 结果"""
query_vec = mock_embedding(query, dim)
results = client.search(
collection_name=collection,
query_vector=query_vec,
limit=top_k,
)
return [{"doc_id": int(r.id), "score": r.score, "content": r.payload["content"]} for r in results]4.4 元数据过滤
def filtered_search(
client: QdrantClient, collection: str, query: str, dim: int, source_filter: str = None, category_filter: str = None
) -> list[dict]:
"""带元数据过滤的向量检索"""
query_vec = mock_embedding(query, dim)
# 构建过滤条件
conditions = []
if source_filter:
conditions.append(FieldCondition(key="source", match=MatchValue(value=source_filter)))
if category_filter:
conditions.append(FieldCondition(key="category", match=MatchValue(value=category_filter)))
search_filter = Filter(must=conditions) if conditions else None
results = client.search(
collection_name=collection,
query_vector=query_vec,
query_filter=search_filter,
limit=5,
)
return results4.5 阈值过滤
def search_with_threshold(results: list[dict], min_score: float = 0.7) -> list[dict]:
"""阈值过滤：只保留相似度高于阈值的结果"""
return [r for r in results if r["score"] >= min_score]Top-K vs 阈值过滤:
Top-K=5: 固定返回 5 条结果
-> 可能包含低相关性的结果
阈值=0.7: 只返回相似度 >= 0.7 的结果
-> 结果数量不确定，但质量有保证
组合: Top-K=10 + 阈值=0.7
-> 最多 10 条，且都高于 0.74.6 检索质量影响因素
4.7 与其他概念的关联
<- 向量化与存储：检索基于已存储的向量
-> 混合检索：向量检索是混合检索的语义组件
-> 重排序：向量检索结果是重排序的输入
5. BM25 关键词检索
5.1 定义
BM25（Best Matching 25）是经典的关键词检索算法，基于词频（TF）和逆文档频率（IDF）评分。它擅长精确匹配关键词，与向量检索互补——向量理解语义，BM25 精确匹配字面，两者结合大幅提升召回率。
对应 Demo: demos/02_RAG/05_关键词检索_BM25.py
5.2 BM25 评分公式
5.3 倒排索引
倒排索引结构:
文档1: "Python 是编程语言"
文档2: "Python 可以做数据分析"
文档3: "编程语言有很多种"
分词后:
文档1: [python, 编程, 语言]
文档2: [python, 数据, 分析]
文档3: [编程, 语言, 很多, 种]
倒排索引:
python -> [文档1, 文档2]
编程 -> [文档1, 文档3]
语言 -> [文档1, 文档3]
数据 -> [文档2]
分析 -> [文档2]
很多 -> [文档3]
种 -> [文档3]
查询 "python 编程":
python -> [文档1, 文档2]
编程 -> [文档1, 文档3]
合并 -> 文档1(2次命中), 文档2(1次), 文档3(1次)
BM25评分 -> 文档1 > 文档2 > 文档35.4 BM25 实现
class BM25Index:
"""BM25 倒排索引"""
def __init__(self, k1: float = 1.5, b: float = 0.75):
"""
:param k1: 词频饱和参数（控制词频增长的影响）
:param b: 长度归一化参数（控制文档长度的影响）
"""
self.k1 = k1
self.b = b
self.documents: list[list[str]] = []
self.doc_len: list[int] = []
self.avg_doc_len: float = 0.0
self.inverted_index: dict[str, list[int]] = {}
self.idf: dict[str, float] = {}
self.doc_count: int = 0
def add_documents(self, docs: list[str]) -> None:
"""构建索引：分词 -> 统计词频 -> 构建倒排索引 -> 计算 IDF"""
self.doc_count = len(docs)
for doc_id, doc in enumerate(docs):
tokens = tokenize(doc)
self.documents.append(tokens)
self.doc_len.append(len(tokens))
tf = Counter(tokens)
for term in tf:
if term not in self.inverted_index:
self.inverted_index[term] = []
self.inverted_index[term].append(doc_id)
self.avg_doc_len = sum(self.doc_len) / self.doc_count if self.doc_count > 0 else 0
# 计算 IDF
for term, doc_ids in self.inverted_index.items():
n = len(doc_ids)
self.idf[term] = math.log((self.doc_count - n + 0.5) / (n + 0.5) + 1)
def search(self, query: str, top_k: int = 5) -> list[tuple[int, float]]:
"""检索：计算查询与每个文档的 BM25 得分"""
query_tokens = tokenize(query)
scores = [0.0] * self.doc_count
for term in query_tokens:
if term not in self.inverted_index:
continue
for doc_id in self.inverted_index[term]:
tf = self.documents[doc_id].count(term)
# BM25 评分
numerator = tf * (self.k1 + 1)
denominator = tf + self.k1 * (1 - self.b + self.b * self.doc_len[doc_id] / self.avg_doc_len)
scores[doc_id] += self.idf[term] * (numerator / denominator)
ranked = sorted(enumerate(scores), key=lambda x: x[1], reverse=True)
return ranked[:top_k]5.5 参数调优
k1 的影响:
k1=0: 完全忽略词频（退化为布尔检索）
k1=1.5: 标准值，词频有适度影响
k1=10: 词频影响很大（几乎线性）
b 的影响:
b=0: 不考虑文档长度
b=0.75: 标准值，适度归一化
b=1: 完全归一化（长文档严重被惩罚）5.6 BM25 vs 向量检索
5.7 与其他概念的关联
-> 混合检索：BM25 与向量检索融合是标准做法
<- 文本清洗与分块：分块结果构建 BM25 索引
-> RAG 评测：BM25 检索质量需要评测
6. 混合检索
6.1 定义
混合检索（Hybrid Search）融合语义检索（向量）和关键词检索（BM25）的结果。语义检索理解意思，关键词检索精确匹配，两者互补提升召回率。核心挑战是不同检索系统的分数不可直接比较，需要融合策略。
对应 Demo: demos/02_RAG/06_混合检索.py
6.2 为什么需要混合检索
场景1: 查询 "RAG 的 hallucination 问题"
向量检索结果:
1. "检索增强生成可以减少模型编造信息" (score=0.89)
2. "RAG 系统的架构设计" (score=0.82)
-> 找到了语义相关的内容，但可能遗漏包含 "hallucination" 关键词的文档
BM25 检索结果:
1. "Hallucination 是大模型的常见问题" (score=15.2)
2. "减少 hallucination 的方法" (score=12.8)
-> 精确匹配了关键词，但可能遗漏语义相关但用词不同的文档
混合检索结果:
1. "检索增强生成可以减少模型编造信息" (语义匹配 + 部分词匹配)
2. "Hallucination 是大模型的常见问题" (词匹配)
3. "减少 hallucination 的方法" (词匹配)
-> 综合两者优势，召回率最高6.3 分数融合策略
6.3.1 RRF（Reciprocal Rank Fusion，倒数排名融合）
其中
是常数（通常 60），
是文档
在检索系统
中的排名。
def rrf_fusion(vector_results: list[dict], bm25_results: list[dict], k: int = 60, top_n: int = 5) -> list[dict]:
"""RRF 融合：基于排名的融合，不依赖原始分数"""
scores = {}
# 向量检索结果按排名融合
for rank, result in enumerate(vector_results):
doc_id = result["doc_id"]
scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1)
# BM25 检索结果按排名融合
for rank, result in enumerate(bm25_results):
doc_id = result["doc_id"]
scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1)
# 排序并返回 Top-N
ranked = sorted(scores.items(), key=lambda x: x[1], reverse=True)
return [{"doc_id": doc_id, "rrf_score": score} for doc_id, score in ranked[:top_n]]6.3.2 加权融合
def weighted_fusion(
vector_results: list[dict], bm25_results: list[dict], vector_weight: float = 0.7, bm25_weight: float = 0.3
) -> list[dict]:
"""加权融合：将分数归一化后按权重相加"""
# 归一化向量检索分数到 [0, 1]
max_vec_score = max(r["score"] for r in vector_results) if vector_results else 1
# 归一化 BM25 分数到 [0, 1]
max_bm25_score = max(r["score"] for r in bm25_results) if bm25_results else 1
scores = {}
for r in vector_results:
doc_id = r["doc_id"]
normalized = r["score"] / max_vec_score if max_vec_score > 0 else 0
scores[doc_id] = scores.get(doc_id, 0) + vector_weight * normalized
for r in bm25_results:
doc_id = r["doc_id"]
normalized = r["score"] / max_bm25_score if max_bm25_score > 0 else 0
scores[doc_id] = scores.get(doc_id, 0) + bm25_weight * normalized
ranked = sorted(scores.items(), key=lambda x: x[1], reverse=True)
return [{"doc_id": doc_id, "fused_score": score} for doc_id, score in ranked]6.4 融合策略对比
6.5 权重选择
权重选择指南:
语义检索为主 (vector=0.7, bm25=0.3):
-> 适合自然语言问答、概念查询
关键词检索为主 (vector=0.3, bm25=0.7):
-> 适合精确查找、专有名词、代码搜索
均衡 (vector=0.5, bm25=0.5):
-> 通用场景的起点
动态权重:
-> 查询长度短: 提高 BM25 权重
-> 查询长度长: 提高向量权重
-> 包含专有名词: 提高 BM25 权重6.6 与其他概念的关联
<- 向量检索 + BM25：混合检索融合两者结果
-> 重排序：混合检索结果是重排序的输入
-> 端到端 RAG：混合检索是完整 RAG 流程的核心组件
7. 重排序
7.1 定义
重排序（Reranking）对初步检索结果做精细化二次排序。向量检索是"双塔"模型（快但粗），Reranker 是"交叉"模型（慢但准）。先粗筛再精排是 RAG 检索的标准模式——用快速检索召回较多候选，再用精准模型选出最优。
对应 Demo: demos/02_RAG/07_重排序.py
7.2 两阶段检索架构
+----------------------------------------------------------+
| 两阶段检索架构 |
+----------------------------------------------------------+
| |
| 用户查询 |
| | |
| v |
| +---------------+ |
| | 第一阶段: 粗排 | 向量检索/BM25/混合检索 |
| | (召回 Top-K) | K=20~50 (召回率高，速度快) |
| +-------+-------+ |
| | |
| v |
| +---------------+ |
| | 第二阶段: 精排 | Cross-Encoder Reranker |
| | (精排 Top-N) | N=3~5 (精准度高，速度慢) |
| +-------+-------+ |
| | |
| v |
| 最终结果 (Top-N) |
| |
+----------------------------------------------------------+
双塔模型 (Bi-Encoder) vs 交叉模型 (Cross-Encoder):
Bi-Encoder (向量检索):
Query -> Encoder -> Vector
Doc -> Encoder -> Vector
相似度 = cos(Query_Vector, Doc_Vector)
-> 快 (可预计算), 但粗
Cross-Encoder (Reranker):
[Query, Doc] -> Encoder -> Score
-> 慢 (不可预计算), 但准7.3 重排序实现
def mock_reranker_score(query: str, document: str) -> float:
"""
模拟 Cross-Encoder Reranker 评分
真实场景应调用 Cohere Rerank / BGE-Reranker
"""
q_tokens = set(re.findall(r"[\u4e00-\u9fff]|[a-zA-Z]+", query.lower()))
d_tokens = re.findall(r"[\u4e00-\u9fff]|[a-zA-Z]+", document.lower())
if not d_tokens:
return 0.0
# 关键词命中数
hits = sum(1 for t in d_tokens if t in q_tokens)
# 命中率
hit_rate = hits / len(q_tokens) if q_tokens else 0
# 文档中查询词的密度
density = hits / len(d_tokens)
# 综合评分
return 0.6 * hit_rate + 0.4 * density
def first_stage_retrieve(client, collection, query, dim, top_k=10) -> list[dict]:
"""第一阶段：向量检索粗排，召回较多候选"""
query_vec = mock_embedding(query, dim)
results = client.search(collection_name=collection, query_vector=query_vec, limit=top_k)
return [{"doc_id": int(r.id), "score": r.score, "content": r.payload["content"]} for r in results]
def second_stage_rerank(candidates: list[dict], query: str, top_n: int = 3) -> list[dict]:
"""第二阶段：Reranker 精排，对候选重新打分"""
reranked = []
for c in candidates:
rerank_score = mock_reranker_score(query, c["content"])
reranked.append({
"doc_id": c["doc_id"],
"vector_score": c["score"],
"rerank_score": rerank_score,
"content": c["content"],
})
# 按 Reranker 分数排序
reranked.sort(key=lambda x: x["rerank_score"], reverse=True)
return reranked[:top_n]7.4 常用 Reranker 模型
7.5 两阶段检索参数
参数选择:
第一阶段 (粗排):
top_k = 20 ~ 50 (召回足够多候选)
使用快速检索 (向量/BM25/混合)
第二阶段 (精排):
top_n = 3 ~ 5 (只保留最相关的)
使用 Cross-Encoder (慢但准)
权衡:
top_k 太大 -> 精排计算量大
top_k 太小 -> 可能遗漏正确答案
top_n 太大 -> 上下文窗口占用多
top_n 太小 -> 可能缺少必要信息7.6 与其他概念的关联
<- 混合检索：重排序对混合检索结果做精排
-> 端到端 RAG：重排序是完整 RAG 流程的关键环节
-> RAG 评测：重排序提升检索质量，需评测验证
8. 查询改写
8.1 定义
查询改写（Query Rewrite）是 RAG 检索的前置优化。用户问题可能太短、口语化或缺少关键词，通过 LLM 改写为更适合检索的表述，提升召回率。主要策略包括查询扩展、查询分解和 HyDE。
对应 Demo: demos/02_RAG/08_查询改写.py
8.2 查询扩展
def query_expansion(query: str) -> list[str]:
"""
查询扩展：为原始查询添加同义词和相关词
生成多个改写查询，分别检索后合并结果
"""
queries = [query] # 原始查询始终保留
for term, synonyms in SYNONYM_MAP.items():
if term in query:
for syn in synonyms:
expanded = query.replace(term, syn)
queries.append(expanded)
return queries查询扩展示例:
原始查询: "如何解决幻觉问题"
扩展后:
1. "如何解决幻觉问题" (原始)
2. "如何解决 hallucination 问题" (英文同义词)
3. "如何解决虚假信息问题" (中文同义词)
4. "如何解决编造内容问题" (中文同义词)
对每个查询分别检索，合并去重后排序8.3 查询分解
def query_decomposition(query: str) -> list[str]:
"""
查询分解：将复杂问题拆分为多个子问题
每个子问题独立检索，最后综合答案
"""
# 模拟 LLM 分解复杂查询
if "和" in query or "与" in query:
parts = re.split(r"[和与]", query)
return [p.strip() for p in parts if p.strip()]
return [query]查询分解示例:
原始查询: "比较 RAG 和微调的优缺点"
分解后:
1. "RAG 的优点是什么"
2. "RAG 的缺点是什么"
3. "微调的优点是什么"
4. "微调的缺点是什么"
分别检索 -> 分别生成答案 -> 综合为比较分析8.4 HyDE（Hypothetical Document Embeddings）
def hyde(query: str, llm_handler) -> str:
"""
HyDE: 先让 LLM 生成一个假设性答案文档，
再用该文档做向量检索（而非用原始查询）
"""
prompt = f"请为以下问题生成一段可能的答案（无需准确）：\n{query}"
hypothetical_doc = llm_handler(prompt)
# 用假设文档而非原始查询做向量检索
return hypothetical_docHyDE 原理:
传统流程:
查询 -> Embedding -> 检索文档
问题: 查询（短问题）与文档（长答案）的向量分布不匹配
HyDE 流程:
查询 -> LLM 生成假设答案 -> 假设答案 Embedding -> 检索文档
优势: 假设答案与文档长度和风格更接近，向量匹配更准确
示例:
查询: "什么是 RAG"
假设答案: "RAG 是检索增强生成技术，通过检索外部知识库
来增强大语言模型的回答能力，减少幻觉..."
-> 用假设答案的向量去检索，比用短查询更准确8.5 查询改写策略对比
8.6 与其他概念的关联
-> 向量检索 / 混合检索：改写后的查询用于检索
<- LLM API 调用：查询改写需要调用 LLM
-> 端到端 RAG：查询改写是 RAG 流程的可选前置步骤
9. 端到端 RAG 流程
9.1 定义
端到端 RAG 流程将文档解析、清洗分块、向量化、混合检索、重排序、上下文构建和生成回答串联成完整 Pipeline。这是前面所有环节的集大成，展示各组件如何协同工作构成可运行的 RAG 系统。
对应 Demo: demos/02_RAG/09_完整RAG流程.py
9.2 完整流程架构
+------------------------------------------------------------------+
| 端到端 RAG Pipeline |
+------------------------------------------------------------------+
| |
| 离线索引阶段: |
| |
| 文档库 -> [解析] -> [清洗] -> [分块] -> [向量化] -> [存储] |
| | | | | | |
| Markdown 去噪 固定长度 Embedding Qdrant |
| HTML 归一化 按句子 模型 向量库 |
| PDF 去重 按 Token |
| |
| 在线查询阶段: |
| |
| 用户问题 |
| | |
| v |
| [查询改写] --(可选)--> 扩展/分解/HyDE |
| | |
| v |
| +----------------+ +----------------+ |
| | 向量检索 (Top-K)| | BM25 检索(Top-K)| |
| +-------+--------+ +-------+--------+ |
| | | |
| +--------+-------+-------+ |
| | |
| v |
| [混合融合 (RRF/加权)] |
| | |
| v |
| [重排序 (Reranker)] |
| | |
| v |
| [上下文构建] |
| System: "基于以下信息回答..." |
| Context: [Top-N 检索结果] |
| User: 原始问题 |
| | |
| v |
| [LLM 生成回答] |
| | |
| v |
| 最终回答 + 引用来源 |
| |
+------------------------------------------------------------------+9.3 RAG 类实现
class RAGPipeline:
"""端到端 RAG 流程"""
def __init__(self, vector_dim: int = 64):
"""初始化各组件"""
self.vector_dim = vector_dim
self.client = QdrantClient(":memory:")
self.collection = "rag_docs"
self.bm25 = BM25Index()
self.documents: list[Chunk] = []
def index(self, raw_docs: list[str], source: str = "default") -> None:
"""索引阶段：解析 -> 清洗 -> 分块 -> 向量化 -> 存储"""
all_chunks = []
for i, doc in enumerate(raw_docs):
# 1. 清洗
cleaned = clean_text(doc)
# 2. 分块
chunks = chunk_by_fixed_size(cleaned, chunk_size=200, overlap=50)
for j, chunk_text in enumerate(chunks):
chunk = Chunk(
id=f"{source}_{i}_{j}",
content=chunk_text,
source=source,
metadata={"chunk_index": j},
)
all_chunks.append(chunk)
self.documents = all_chunks
# 3. 向量化 + 存储
self._setup_vector_store(all_chunks)
# 4. BM25 索引
self.bm25.add_documents([c.content for c in all_chunks])
def query(self, question: str, top_k: int = 5, top_n: int = 3) -> dict:
"""查询阶段：检索 -> 融合 -> 重排 -> 生成"""
# 1. 向量检索
vec_results = self._vector_search(question, top_k)
# 2. BM25 检索
bm25_results = self._bm25_search(question, top_k)
# 3. RRF 融合
fused = self._fuse(vec_results, bm25_results, top_k)
# 4. 重排序
reranked = self._rerank(fused, question, top_n)
# 5. 构建上下文 + 生成
context = self._build_context(reranked)
answer = self._generate(question, context)
return {"answer": answer, "sources": reranked}9.4 上下文构建
def _build_context(self, results: list[RetrievalResult]) -> str:
"""构建 LLM 上下文"""
context_parts = []
for i, r in enumerate(results, 1):
context_parts.append(f"[{i}] (来源: {r.chunk.source})\n{r.chunk.content}")
return "\n\n".join(context_parts)
def _generate(self, question: str, context: str) -> str:
"""生成回答（模拟 LLM 调用）"""
prompt = f"""请基于以下检索到的信息回答问题。如果信息不足以回答，请说明。
检索到的信息：
{context}
问题：{question}
回答："""
# 真实场景: return llm_client.chat(prompt)
return f"[模拟回答] 基于 {len(context)} 字符的上下文回答了问题。"9.5 RAG 的关键设计决策
9.6 常见问题与解决
问题1: 检索结果不相关
原因: 分块太大/太小, Embedding 模型差, 查询表述差
解决: 调整分块策略, 更换 Embedding, 查询改写
问题2: 回答不忠于检索结果（幻觉）
原因: LLM 忽略上下文, 上下文不足
解决: 强化 System Prompt, 增加上下文量, 忠实度评测
问题3: 检索速度慢
原因: 向量库数据量大, 检索方式复杂
解决: 使用 HNSW 索引, 减小 Top-K, 缓存常见查询
问题4: 知识更新不及时
原因: 索引未更新
解决: 增量索引, 定期重建, 文档变更监听9.7 与其他概念的关联
<- 所有前置环节：RAG 流程整合所有组件
-> RAG 评测：完整流程需要系统性评测
-> Agent 核心能力：RAG 可作为 Agent 的工具
10. RAG 评测
10.1 定义
RAG 评测衡量检索系统和生成系统的质量。检索评测核心指标包括召回率（Recall）、精确率（Precision）、MRR、NDCG；生成质量评估包括忠实度（Faithfulness）和相关度（Relevance）。没有评测的 RAG 系统是盲目的——无法知道改进方向。
对应 Demo: demos/02_RAG/10_RAG评测.py
10.2 评测体系架构
+------------------------------------------------------------------+
| RAG 评测体系 |
+------------------------------------------------------------------+
| |
| 检索评测 (Retrieval Metrics) |
| +-------------------+ +-------------------+ |
| | Recall@K | | Precision@K | |
| | 召回率: 相关文档 | | 精确率: 结果中 | |
| | 被检索到的比例 | | 相关文档的占比 | |
| +-------------------+ +-------------------+ |
| +-------------------+ +-------------------+ |
| | MRR | | NDCG | |
| | 第一个相关文档的 | | 考虑排名位置的 | |
| | 排名倒数 | | 加权累积增益 | |
| +-------------------+ +-------------------+ |
| |
| 生成评测 (Generation Metrics) |
| +-------------------+ +-------------------+ |
| | Faithfulness | | Relevance | |
| | 忠实度: 回答是否 | | 相关度: 回答是否 | |
| | 基于检索到的事实 | | 切实回答了问题 | |
| +-------------------+ +-------------------+ |
| +-------------------+ +-------------------+ |
| | Answer Correctness| | Context Precision | |
| | 答案正确性 | | 上下文精确率 | |
| +-------------------+ +-------------------+ |
| |
+------------------------------------------------------------------+10.3 检索评测指标
10.3.1 Recall@K（召回率）
def recall_at_k(retrieved_ids: list[str], relevant_ids: set[str], k: int) -> float:
"""计算 Recall@K：Top-K 中命中相关文档的比例"""
top_k = retrieved_ids[:k]
hits = len(set(top_k) & relevant_ids)
return hits / len(relevant_ids) if relevant_ids else 0.010.3.2 Precision@K（精确率）
10.3.3 MRR（Mean Reciprocal Rank）
其中
是第
个查询中第一个相关文档的排名。
10.3.4 NDCG（Normalized Discounted Cumulative Gain）
def evaluate_retrieval(eval_cases: list[EvalCase], search_results: list[list[str]], k: int = 5) -> RetrievalMetrics:
"""计算检索评测指标"""
total_recall = 0.0
total_precision = 0.0
total_mrr = 0.0
total_ndcg = 0.0
hit_count = 0
for case, results in zip(eval_cases, search_results):
relevant = case.relevant_doc_ids
top_k = results[:k]
# Recall@K
hits = len(set(top_k) & relevant)
recall = hits / len(relevant) if relevant else 0
total_recall += recall
# Precision@K
precision = hits / k if k > 0 else 0
total_precision += precision
# MRR
for rank, doc_id in enumerate(top_k, 1):
if doc_id in relevant:
total_mrr += 1 / rank
break
# NDCG
dcg = sum(1 / math.log2(i + 2) for i, d in enumerate(top_k) if d in relevant)
idcg = sum(1 / math.log2(i + 2) for i in range(min(len(relevant), k)))
total_ndcg += dcg / idcg if idcg > 0 else 0
# Hit Rate
if hits > 0:
hit_count += 1
n = len(eval_cases)
return RetrievalMetrics(
recall_at_k=total_recall / n,
precision_at_k=total_precision / n,
mrr=total_mrr / n,
ndcg=total_ndcg / n,
hit_rate=hit_count / n,
)10.4 指标对比
10.5 生成评测
生成评测 (LLM-as-Judge):
忠实度 (Faithfulness):
问题: 回答是否完全基于检索到的上下文？
评分: 0~1 (1=完全忠实, 0=全部编造)
方法: LLM 检查回答中每个事实是否可在上下文中找到
相关度 (Relevance):
问题: 回答是否切实回答了用户的问题？
评分: 0~1 (1=完全相关, 0=答非所问)
方法: LLM 判断回答与问题的相关程度
答案正确性 (Answer Correctness):
问题: 回答是否与标准答案一致？
评分: 0~1
方法: 与人工标注的参考答案比较def evaluate_generation(answer: str, context: str, question: str, expected: str = "") -> dict:
"""评估生成质量（模拟 LLM-as-Judge）"""
# 忠实度: 回答中的事实是否在上下文中
answer_facts = extract_facts(answer)
context_facts = extract_facts(context)
faithful = sum(1 for f in answer_facts if f in context_facts)
faithfulness = faithful / len(answer_facts) if answer_facts else 0
# 相关度: 回答是否涉及问题关键词
q_tokens = set(tokenize(question))
a_tokens = set(tokenize(answer))
relevance = len(q_tokens & a_tokens) / len(q_tokens) if q_tokens else 0
return {
"faithfulness": faithfulness,
"relevance": min(relevance, 1.0),
}10.6 评测数据集构建
评测数据集结构:
EvalCase:
query: "什么是 RAG?" # 用户问题
relevant_doc_ids: {"doc_1", "doc_5"} # 人工标注的相关文档
expected_answer: "RAG 是检索增强..." # 参考答案（可选）
构建方法:
1. 人工标注: 领域专家标注相关文档和参考答案
2. LLM 辅助: LLM 生成问题和候选答案，人工审核
3. 真实查询: 从生产环境收集真实用户查询
4. 合成数据: 基于文档自动生成问答对
评测集要求:
- 覆盖不同类型的问题（事实型/分析型/比较型）
- 包含简单和困难查询
- 定期更新以反映知识库变化10.7 RAG 评测框架对比
10.8 与其他概念的关联
<- 端到端 RAG：评测衡量完整 RAG 流程的质量
-> 持续优化：评测结果指导各环节优化方向
-> Agent 评测：RAG 评测是 Agent 评测的子集
概念关系总览
+--------------------------------------------------+
| RAG 技术体系 |
+--------------------------------------------------+
数据处理层 检索层 生成层 评测层
| | | |
v v v v
+-----------+ +---------------+ +---------------+ +-----------+
|文档解析 | |向量检索 | |上下文构建 | |检索评测 |
|(MD/HTML/PDF)| |(语义匹配) | |(Prompt 组装) | |(Recall等) |
+-----+-----+ +-------+-------+ +-------+-------+ +-----------+
| | |
v | |
+-----------+ | |
|文本清洗与 |---> +---------------+ |
|分块策略 | |BM25 检索 | |
|(分块/重叠) | |(关键词匹配) | |
+-----+-----+ +-------+-------+ |
| | |
v v |
+-----------+ +---------------+ |
|向量化与 | |混合检索 | |
|向量数据库 | |(RRF/加权融合) | |
|(Qdrant) | +-------+-------+ |
+-----------+ | |
v |
+---------------+ |
|重排序 | |
|(粗排+精排) | |
+-------+-------+ |
| |
v |
+---------------+ +------v-------+
|查询改写 | |LLM 生成回答 |
|(扩展/分解/HyDE)| |(基于上下文) |
+-------+-------+ +------+-------+
| |
v v
+---------------+ +---------------+
|端到端 RAG 流程 |-->| RAG 评测 |
|(完整 Pipeline) | |(检索+生成) |
+---------------+ +---------------+概念间的依赖关系
文档解析 -> 文本清洗与分块 -> 向量化与存储 -> 向量检索
|
BM25 检索
|
混合检索 (融合两者)
|
重排序
|
查询改写 ----> (向量检索 / 混合检索) |
|
端到端 RAG 流程
|
RAG 评测
评论区