本文以通俗易懂、带教学风格的方式，详细讲解 RAG（检索增强生成）的优化技术，包括数据清洗、查询扩展、自查询、提示压缩、效果评估、分块、嵌入模型、文档解析、提示工程，以及 Advanced RAG 和 Modular RAG 的概念。内容结合 Python 代码示例和旅游问答场景，适合初学者和开发者！
一、在 RAG 中如何进行数据清洗和预处理？
数据清洗和预处理是优化 RAG 检索精度的基础，确保知识库文档质量高、格式统一。以下是具体步骤：
1.1 数据清洗和预处理的步骤
-
去除噪声数据
删除广告、导航栏等无关内容。
例子：从旅游攻略网页中去除“订阅 newsletter”文本。
方法：用 BeautifulSoup 过滤 HTML 标签。
-
文本规范化
统一编码（如 UTF-8），去除多余空格。
例子：将“免簽”统一为“免签”。
方法：用 unicodedata 规范化。
-
分段与结构化
按语义分段，添加元数据。
例子：将 PDF 拆为“签证政策”“入境要求”。
方法：用 spaCy 识别段落。
-
去除冗余
删除重复或低价值内容。
例子：去除重复的“瑞士免签”条款。
方法：用余弦相似度检测。
-
语义增强
提取关键词，添加元数据。
例子：为文档添加“2025年”标签。
方法：用 TextRank 提取关键词。
1.2 Python 代码示例：数据清洗
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
|
import re
from bs4 import BeautifulSoup
import unicodedata
def clean_text(html_content: str) -> str:
soup = BeautifulSoup(html_content, 'html.parser')
for tag in soup(['script', 'style', 'nav', 'footer']):
tag.decompose()
text = soup.get_text(separator=' ')
text = unicodedata.normalize('NFKC', text)
text = re.sub(r'\s+', ' ', text.strip())
text = re.sub(r'[\r\n]+', ' ', text)
text = re.sub(r'(点击这里|联系我们|订阅.*?\.)', '', text, flags=re.IGNORECASE)
return text
def segment_text(text: str, max_length: int = 200) -> list:
sentences = text.split('。')
segments = []
current_segment = ""
for sentence in sentences:
sentence = sentence.strip()
if not sentence:
continue
if len(current_segment) + len(sentence) <= max_length:
current_segment += sentence + '。'
else:
if current_segment:
segments.append(current_segment)
current_segment = sentence + '。'
if current_segment:
segments.append(current_segment)
return segments
def main():
html_content = """
<html>
<body>
<h1>2025年瑞士旅游政策</h1>
<p>免签国家增至70个。入境需提供健康码。点击这里订阅我们的 newsletter！</p>
<nav>首页 | 关于我们</nav>
</body>
</html>
"""
cleaned_text = clean_text(html_content)
print("清洗后的文本：", cleaned_text)
segments = segment_text(cleaned_text)
print("分段结果：")
for i, segment in enumerate(segments, 1):
print(f"段落 {i}: {segment}")
if __name__ == "__main__":
main()
|
1.3 优化技巧
- 定期更新知识库，删除过时文档。
- 人工检查清洗结果，确保无噪声。
- 确保分段不破坏语义。
二、查询扩展：定义与必要性
2.1 什么是查询扩展？
查询扩展（Query Expansion）是指自动扩充用户查询，添加同义词或相关术语，提升检索精度。
比喻：像图书管理员帮你把“瑞士”扩展为“瑞士旅游”“瑞士政策”，确保找到更多相关书籍。
2.2 为什么需要查询扩展？
- 模糊查询：用户输入简短（如“瑞士政策”），需补充上下文。
- 语义多样性：覆盖不同术语（如“免签” vs. “visa-free”）。
- 提升召回率：检索更多相关文档。
- 多语言支持：扩展为其他语言的同义词。
2.3 实现方法
- 基于词典：用 WordNet 添加同义词。
- 基于嵌入：用
sentence-transformers 找语义相近词。
- 基于大模型：用 Grok 生成改写查询。
2.4 Python 代码示例：查询扩展
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
|
from sentence_transformers import SentenceTransformer
import numpy as np
def query_expansion(query: str, knowledge_base: list, model, top_k: int = 3) -> list:
embedder = model
query_embedding = embedder.encode([query])[0]
kb_embeddings = embedder.encode(knowledge_base)
similarities = np.dot(kb_embeddings, query_embedding) / (
np.linalg.norm(kb_embeddings, axis=1) * np.linalg.norm(query_embedding)
)
top_indices = np.argsort(similarities)[::-1][:top_k]
return [knowledge_base[i] for i in top_indices]
def main():
knowledge_base = [
"瑞士旅游政策",
"瑞士签证要求",
"瑞士入境健康码",
"东京旅游攻略",
"瑞士文化介绍"
]
model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')
query = "瑞士政策"
expanded_terms = query_expansion(query, knowledge_base, model)
print(f"原始查询：{query}")
print("扩展后的相关词：")
for term in expanded_terms:
print(f"- {term}")
if __name__ == "__main__":
main()
|
三、自查询：定义与必要性
3.1 什么是自查询？
自查询（Self-Querying）是指系统根据用户查询，自动生成子查询或过滤条件，精确检索结构化数据。
比喻：像图书管理员自动加上“出版年=2025”的过滤条件。
3.2 为什么需要自查询？
- 复杂查询：处理带条件的查询（如“2025年瑞士政策”）。
- 提升精度：减少无关文档。
- 结构化数据：利用元数据（如日期）。
- 用户体验：减少手动指定条件。
3.3 实现方法
- 基于规则：用正则表达式解析条件。
- 基于 NLP：用 NER 提取实体。
- 基于大模型：生成结构化查询。
3.4 举例
查询：“2025年瑞士旅游政策”
- 解析：提取“瑞士”“2025年”。
- 子查询：语义查询“瑞士旅游政策” + 过滤“year=2025”。
四、提示压缩：定义与必要性
4.1 什么是提示压缩？
提示压缩（Prompt Compression）是指精简输入大模型的提示，减少 token 数量，提高效率和质量。
比喻：像把长邮件浓缩成重点。
4.2 为什么需要提示压缩？
- 降低成本：减少计算资源。
- 避免过载：去除冗余信息。
- 模型限制：不超过 token 限制。
- 提升体验：加快响应速度。
4.3 实现方法
- 提取关键句：用 TextRank 摘要。
- 去除冗余：删除重复内容。
- 基于模型：用大模型精简。
4.4 Python 代码示例：提示压缩
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
|
from summa import summarizer
def compress_context(context: str, ratio: float = 0.3) -> str:
summary = summarizer.summarize(context, ratio=ratio, language='chinese')
return summary
def main():
context = """
2025年瑞士旅游政策更新：免签国家增至70个，涵盖大部分欧洲国家和部分亚洲国家。
入境瑞士的游客需通过官方APP提前申请健康码。健康码需包含疫苗接种证明或核酸检测阴性结果。
瑞士政府还推出了新的旅游补贴计划，鼓励游客前往乡村地区。
此外，东京和大阪的酒店价格在2025年预计上涨10%。
"""
compressed_context = compress_context(context)
print("原始上下文：")
print(context)
print("\n压缩后的上下文：")
print(compressed_context)
if __name__ == "__main__":
main()
|
五、RAG 调优后的效果评估
5.1 评估的重要性
评估确保 RAG 系统满足用户需求，覆盖检索和生成质量。
5.2 评估标准
- 检索质量
- 生成质量
- 用户体验
5.3 评估方法
- 人工评估：专家打分。
- 自动化评估：BLEU、ROUGE、语义相似度。
- A/B 测试：比较调优前后。
- 日志分析：识别失败案例。
5.4 应用场景：旅游问答
- 测试集：100 个查询和参考回答。
- 检索：计算精确率（0.8）、召回率（0.75）。
- 生成：用户评分（4.2/5）、ROUGE-L（0.65）。
- 体验：响应时间 2 秒，满意度 4.3/5。
5.5 Python 代码示例：检索评估
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
|
from typing import List, Set
def evaluate_retrieval(retrieved_docs: List[int], relevant_docs: Set[int]) -> dict:
retrieved_set = set(retrieved_docs)
true_positives = len(retrieved_set & relevant_docs)
precision = true_positives / len(retrieved_set) if retrieved_set else 0.0
recall = true_positives / len(relevant_docs) if relevant_docs else 0.0
f1 = 2 * (precision * recall) / (precision + recall) if (precision + recall) > 0 else 0.0
return {"precision": precision, "recall": recall, "f1": f1}
def main():
retrieved_docs = [1, 2, 3, 4]
relevant_docs = {1, 2, 5}
metrics = evaluate_retrieval(retrieved_docs, relevant_docs)
print("检索评估结果：")
print(f"精确率: {metrics['precision']:.2f}")
print(f"召回率: {metrics['recall']:.2f}")
print(f"F1 分数: {metrics['f1']:.2f}")
if __name__ == "__main__":
main()
|
六、RAG 中的分块：定义、必要性与策略
6.1 什么是分块？
分块（Chunking）是将长文档拆为短文本片段（chunk），便于嵌入、检索和生成。
比喻：像把书分成章节，方便查找。
6.2 为什么需要分块？
- 嵌入质量：短文本嵌入更精准。
- 检索效率：提升速度和精度。
- 上下文限制：不超过 token 限制。
- 语义完整：减少无关信息。
6.3 常见分块策略
- 固定长度：按 200 词分割，简单但可能破坏语义。
- 句子/段落：按句子分割，语义完整但大小不均。
- 语义分块：基于主题分割，精度高但计算复杂。
- 滑动窗口：200 词窗口，50 词重叠，兼顾语义和效率。
6.4 举例
文档：1000 词文章。
- 固定长度：5 个 200 词 chunk。
- 句子分块：8 个大小不等 chunk。
- 语义分块：4 个主题 chunk。
- 滑动窗口：6 个重叠 chunk。
6.5 Python 代码示例：句子分块
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
|
import spacy
def sentence_chunking(text: str, max_length: int = 200) -> list:
nlp = spacy.load("zh_core_web_sm")
doc = nlp(text)
chunks = []
current_chunk = ""
for sent in doc.sents:
sent_text = sent.text.strip()
if not sent_text:
continue
if len(current_chunk) + len(sent_text) <= max_length:
current_chunk += sent_text + " "
else:
if current_chunk:
chunks.append(current_chunk.strip())
current_chunk = sent_text + " "
if current_chunk:
chunks.append(current_chunk.strip())
return chunks
def main():
text = """
2025年瑞士旅游政策更新。免签国家增至70个。入境需健康码。
健康码需包含疫苗证明。瑞士还推出旅游补贴计划。
"""
chunks = sentence_chunking(text)
print("分块结果：")
for i, chunk in enumerate(chunks, 1):
print(f"Chunk {i}: {chunk}")
if __name__ == "__main__":
main()
|
七、RAG 中的 Embedding 嵌入
7.1 什么是 Embedding 嵌入？
嵌入是将文本转为向量，捕捉语义信息，用于语义检索。
比喻：像把句子翻译成“坐标”，相近含义的句子坐标靠近。
7.2 常见 Embedding Model
- Sentence-BERT：擅长句子嵌入，多语言支持。
- multilingual-e5：优化长文本和多语言。
- BGE：中文优化，效率高。
- OpenAI text-embedding-ada-002：高质量，闭源。
- GTE：支持长文本，性能均衡。
7.3 如何选择？
- 语种：多语言选 e5，中文选 BGE。
- 文本长度：短文本选 Sentence-BERT，长文本选 e5。
- 资源：小型选 MiniLM，大型选 e5-large。
- 开源：预算有限选 BGE。
- 测试：比较检索性能。
7.4 举例
场景：中文旅游知识库。
- 选择：
BAAI/bge-large-zh。
- 理由：中文优化，开源，精确率 0.85。
八、文档解析与提示工程
8.1 文档解析
文档解析是将原始文档转为可检索的结构化数据。
步骤：
- 格式转换：PDF 用
pdfplumber，网页用 BeautifulSoup。
- 文本提取：去除噪声。
- 结构化：识别标题、表格。
- 分块：按句子分割。
- 嵌入生成：生成向量。
- 元数据存储：保存日期、来源。
8.2 Python 代码示例：PDF 解析
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
|
import pdfplumber
import spacy
import re
def parse_pdf(pdf_path: str) -> list:
nlp = spacy.load("zh_core_web_sm")
chunks = []
with pdfplumber.open(pdf_path) as pdf:
for page in pdf.pages:
text = page.extract_text()
if not text:
continue
text = re.sub(r'页眉.*?\n|页脚.*?\n', '', text)
doc = nlp(text)
current_chunk = ""
max_length = 200
for sent in doc.sents:
sent_text = sent.text.strip()
if not sent_text:
continue
if len(current_chunk) + len(sent_text) <= max_length:
current_chunk += sent_text + " "
else:
if current_chunk:
chunks.append(current_chunk.strip())
current_chunk = sent_text + " "
if current_chunk:
chunks.append(current_chunk.strip())
return chunks
def main():
pdf_path = "japan_policy_2025.pdf"
chunks = parse_pdf(pdf_path)
print("解析结果：")
for i, chunk in enumerate(chunks, 1):
print(f"Chunk {i}: {chunk}")
if __name__ == "__main__":
main()
|
8.3 提示工程心得
- 明确指令：如“基于文档，简洁回答”。
- 提供上下文：包含 2-3 个 chunk。
- 指定风格：如“友好，适合旅游爱好者”。
- 示例驱动：提供 1-2 个示例。
- 迭代优化：测试不同提示。
提示示例：
基于以下文档，以友好、简洁的语气回答用户问题，适合旅游爱好者。如果信息不足，请说明：
文档：2025年瑞士免签国家增至70个。入境需健康码。
用户问题：2025年去瑞士旅游需要签证吗？
九、Advanced RAG 与 Modular RAG
9.1 Advanced RAG
Advanced RAG 引入高级功能，提升性能：
- 多模态：支持图像、表格。
- 上下文感知：利用对话历史。
- 自适应检索：动态调整策略。
- 后处理：事实检查、润色。
- 微调：领域特定优化。
比喻：像全能选手，能看图、记对话。
9.2 Modular RAG
Modular RAG 将 RAG 组件解耦为模块：
- 数据预处理、检索、生成、优化、评估。
- 特点：灵活、可扩展、易维护。
比喻：像乐高积木，自由拼装。
9.3 举例
- Advanced RAG：支持照片输入、多轮对话。
- Modular RAG：模块化设计，可替换嵌入模型。
十、总结与实践建议
本文全面剖析了 RAG 的优化技术和高级形式。以下是实践建议：
欢迎在评论区分享经验！
评论 0