Appearance
章节3：RAG检索增强生成
学习目标
- 理解 RAG 架构原理与"检索+生成"协同优势
- 掌握文档切片与分块策略（语义分块 / 重叠分块）
- 了解 Embedding 模型与向量数据库（text2vec / BGE / ChromaDB）
- 理解召回排序与重排序 Reranker（粗排 + 精排）
- 掌握引用溯源与答案质量优化方法
核心知识点
3.1 RAG 架构原理与优势
定义
RAG（Retrieval-Augmented Generation，检索增强生成）：在 LLM 生成回答前，先从外部知识库中检索相关文档片段，将其作为上下文注入提示词，让模型基于检索到的信息进行生成。
用户问题
│
▼
┌─────────────┐ ┌──────────────┐
│ 向量化检索 │───▶│ 关联上下文 │
│ (Embedding) │ │ (Top-K文档) │
└─────────────┘ └──────┬───────┘
│
▼
┌────────────────────┐
│ LLM 生成回答 │
│ (问题 + 上下文) │
└────────────────────┘RAG 的核心优势：
| 对比维度 | 传统 LLM | RAG 方案 |
|---|---|---|
| 知识时效性 | 训练数据截止日期 | 实时更新知识库 |
| 幻觉问题 | 可能编造事实 | 检索结果约束，降低幻觉 |
| 可解释性 | 黑盒输出 | 可溯源至原文 |
| 领域适配 | 需全量微调 | 只需更新知识库 |
代码示例：RAG 流程示意
python
# RAG 简化流程
def rag_query(query, knowledge_base, embedder, llm, top_k=3):
"""
RAG 查询流程
"""
# 1. 将查询向量化
query_vector = embedder.encode(query)
# 2. 在知识库中检索最相关片段
results = knowledge_base.search(query_vector, top_k=top_k)
# 3. 构建上下文
context = "\n\n".join([r["content"] for r in results])
# 4. 构造增强提示
augmented_prompt = f"""
请基于以下参考信息回答问题。如果参考信息不足以回答，请诚实地说"信息不足"。
参考信息：
{context}
问题：{query}
回答：
"""
# 5. 调用 LLM 生成
answer = llm.generate(augmented_prompt)
return {
"answer": answer,
"sources": [r["source"] for r in results]
}3.2 文档切片与分块策略
定义
文档分块（Chunking）是 RAG 的基础环节，直接影响检索质量：
| 策略 | 方法 | 适用场景 |
|---|---|---|
| 固定长度分块 | 按字符/Token 数切分 | 通用场景，实现简单 |
| 重叠分块 | 相邻块保留部分重叠内容 | 防止关键信息被截断 |
| 语义分块 | 按自然段落/章节切分 | 长文档、教材、论文 |
| 递归分块 | 先按段落切，再调整大小 | LangChain 推荐策略 |
代码示例：不同分块策略实现
python
# 固定长度分块
def fixed_size_chunk(text, chunk_size=500, overlap=50):
"""固定长度分块，带重叠"""
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
chunk = text[start:end]
chunks.append(chunk)
start += chunk_size - overlap
return chunks
# 语义分块（按段落）
def semantic_chunk(text):
"""按段落切分，保留语义完整性"""
paragraphs = text.split("\n\n")
chunks = []
for para in paragraphs:
para = para.strip()
if len(para) > 50: # 过滤过短段落
chunks.append(para)
return chunks
# LangChain 递归分块
# pip install langchain-text-splitters
from langchain_text_splitters import RecursiveCharacterTextSplitter
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50,
separators=["\n\n", "\n", "。", ".", " ", ""],
length_function=len,
)
# document_chunks = text_splitter.split_text(long_document_text)
# print(f"文档被切分为 {len(document_chunks)} 个块")3.3 Embedding 模型与向量数据库
定义
Embedding 模型：将文本转换为固定维度的高维向量，语义相近的文本在向量空间中距离更近。
| 模型 | 维度 | 特点 | 推荐场景 |
|---|---|---|---|
| text2vec-base-chinese | 768 | 中文语义理解好 | 中文 RAG |
| BGE-large-zh | 1024 | BAAI出品，中文最佳 | 企业级中文场景 |
| text-embedding-3-small | 1536 | OpenAI 通用模型 | 多语言、英文场景 |
向量数据库：专为向量相似度搜索优化（余弦相似度、内积、欧氏距离）。
代码示例：ChromaDB 完整使用
python
# pip install chromadb sentence-transformers
import chromadb
from sentence_transformers import SentenceTransformer
# 1. 初始化 Embedding 模型
encoder = SentenceTransformer("BAAI/bge-large-zh-v1.5")
print(f"向量维度: {encoder.get_sentence_embedding_dimension()}")
# 2. 创建 ChromaDB 客户端
client = chromadb.PersistentClient(path="./chroma_db")
# 3. 创建或获取集合
collection = client.get_or_create_collection(
name="course_materials",
metadata={"hnsw:space": "cosine"} # 余弦相似度
)
# 4. 添加文档
documents = [
{"id": "doc1", "text": "深度学习是机器学习的一个分支...", "source": "intro.pdf"},
{"id": "doc2", "text": "Transformer架构由编码器和解码器组成...", "source": "transformer.pdf"},
]
for doc in documents:
embedding = encoder.encode(doc["text"]).tolist()
collection.add(
embeddings=[embedding],
documents=[doc["text"]],
metadatas=[{"source": doc["source"]}],
ids=[doc["id"]]
)
# 5. 检索
query = "什么是Transformer"
query_embedding = encoder.encode(query).tolist()
results = collection.query(
query_embeddings=[query_embedding],
n_results=2,
include=["documents", "metadatas", "distances"]
)
print("检索结果：")
for i, (doc, meta, dist) in enumerate(zip(
results["documents"][0],
results["metadatas"][0],
results["distances"][0]
)):
print(f"{i+1}. [距离={dist:.4f}] {doc[:100]}...\n 来源: {meta['source']}")3.4 召回排序与重排序 Reranker
定义
RAG 典型采用两阶段检索：
| 阶段 | 方法 | 目的 | 特点 |
|---|---|---|---|
| 粗排（Retrieval） | Embedding 向量检索 | 快速召回 Top-50（海选） | 速度快，精度一般 |
| 精排（Rerank） | Cross-Encoder 重排序 | 对 Top-50 精确排序 | 精度高，速度慢 |
Reranker 模型：将查询和候选文档拼接输入，计算精确的相关性分数。
代码示例：Reranker 使用
python
# pip install sentence-transformers
from sentence_transformers import CrossEncoder
# 加载 Reranker 模型
reranker = CrossEncoder("BAAI/bge-reranker-v2-m3")
query = "Python中的装饰器是什么"
# 假设这是粗排召回的结果
candidates = [
"装饰器是Python中一个非常强大的功能...",
"Python中的列表推导式是一种简洁语法...",
"Python装饰器本质上是一个函数，它接受一个函数作为参数...",
"Python中的异常处理使用try-except语句..."
]
# 计算相关性分数
pairs = [[query, doc] for doc in candidates]
scores = reranker.predict(pairs)
# 按分数排序
ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
print("重排序结果：")
for doc, score in ranked:
print(f"分数: {score:.4f} | {doc[:60]}...")3.5 引用溯源与答案质量优化
定义
RAG 的核心可信度保障——每个回答都必须能追溯到原始文档。
质量优化策略：
| 问题 | 解决方案 |
|---|---|
| 检索不相关 | 调整分块大小、增加 Top-K、使用 Reranker |
| 上下文过长 | 压缩 Prompt、关键信息提取 |
| 答案无引用 | 要求 LLM 输出引用源（Source Citation） |
| 多跳问题 | 拆解为子问题，多轮检索（Multi-Hop RAG） |
代码示例：带引用的 RAG 回答
python
def rag_with_citation(query, collection, encoder, llm, top_k=5):
"""
带引用溯源的 RAG 查询
"""
# 检索
q_vec = encoder.encode(query).tolist()
results = collection.query(
query_embeddings=[q_vec],
n_results=top_k,
include=["documents", "metadatas", "distances"]
)
docs = results["documents"][0]
metas = results["metadatas"][0]
# 构建带编号的上下文
context_parts = []
for i, (doc, meta) in enumerate(zip(docs, metas)):
context_parts.append(f"[文档{i+1}] (来源: {meta['source']})\n{doc}")
context = "\n\n".join(context_parts)
# 要求 LLM 引用来源
prompt = f"""
请基于以下参考文档回答问题。在回答中，请在每个观点后用 [文档编号] 标注来源。
参考文档：
{context}
问题：{query}
回答（请标注引用来源）：
"""
answer = llm.generate(prompt)
return answer
# 输出示例：
# "Python装饰器本质上是一个函数[文档1]，它允许我们在不修改原函数代码的情况下扩展功能[文档3]。"小结与练习
小结
- RAG 通过"检索+生成"协同，解决了 LLM 知识时效性、幻觉、可解释性问题
- 文档分块策略直接影响检索质量，重叠分块和语义分块是主流方案
- Embedding 模型（BGE/text2vec）+ 向量数据库（ChromaDB）构成了 RAG 的检索基础
- 两阶段检索（粗排+精排/重排序）能显著提升召回的精确度
- 引用溯源是 RAG 可信度的关键保障
练习题
- 基础题：简述 RAG 架构的核心流程（至少 4 步），并说明每步的作用。
- 实践题：使用 ChromaDB + BGE Embedding 实现一个本地文件检索系统，支持文档添加和语义搜索。
- 实践题：在基础 RAG 上增加 Reranker 重排序，对比有无重排序的回答质量差异。
- 分析题：你的 RAG 系统回答总是不准确，请列出至少 3 种可能的故障原因和对应的排查方向。