# RAG 系统搭建实战
> 从「让 AI 背书」到「让 AI 翻书」—— 给大模型接上私有知识库的完整方案
>
---
## 一、RAG 整体架构
RAG 分两个阶段：**索引阶段**（离线）和 **查询阶段**（在线）。
+ **索引阶段**：跑一次就行，把文档处理好存进向量库
+ **查询阶段**：每次提问都跑，检索 → 拼 Prompt → 生成回答
**关键认知**：检索质量的上限，取决于索引阶段的质量。切分策略、嵌入模型、向量库选型，每一个环节都影响最终效果。其中，**分块策略决定了 RAG 质量的 70%**。
---
## 二、文档加载：把知识「搬进来」
### 2.1 文档加载器选型
LangChain 提供了 100 多种文档加载器，基本覆盖了你能想到的所有格式。常见的：
| 格式 | 推荐加载器 | 特点 |
| :--- | :--- | :--- |
| PDF | `PyMuPDFLoader` | 速度最快，支持元数据 |
| Word | `Docx2txtLoader` | 轻量，无需额外依赖 |
| Markdown | `UnstructuredMarkdownLoader` | 保留标题结构 |
| 网页 | `WebBaseLoader` | 直接抓 URL |
| CSV | `CSVLoader` | 表格数据友好 |
### 2.2 实战：加载 PDF 文件
```python
# 安装依赖：uv pip install pymupdf -i https://pypi.tuna.tsinghua.edu.cn/simple
from langchain_community.document_loaders import PyMuPDFLoader
# 创建加载器实例，传入 PDF 文件路径
pdf_loader = PyMuPDFLoader("./car_info.pdf")
# 调用 load() 方法，返回一个 Document 列表（每页一个 Document）
pdf_pages = pdf_loader.load()
# 看看加载结果
print(f"文档类型：{type(pdf_pages)}")
print(f"PDF 共 {len(pdf_pages)} 页")
```
每个 `Document` 对象有两个核心属性：
+ `page_content`：该页的文本内容
+ `metadata`：描述性数据（页码、文件名等）
```python
# 查看第一页的内容和元数据
first_page = pdf_pages[0]
print(f"元数据：{first_page.metadata}")
print(f"内容预览：{first_page.page_content[:200]}")
```
---
## 三、数据清洗：给知识「去噪」
原始数据往往很脏。PDF 解析出来的文本，经常有莫名其妙的换行符、特殊符号、多余空格。不清洗就直接用，检索效果会大打折扣。
### 3.1 常见问题与清洗策略
```python
import re
# ---- 问题1：PDF 解析产生的多余换行符 ----
# 现象：一句话被拆成多行，中间插了 \n
# 例如："领克汽车\n车顶行李架\n最大载荷"
# 解决：用正则匹配并删除非中文字符之间的换行符
raw_text = "领克汽车\n车顶行李架\n最大载荷。"
pattern = re.compile(r'[^一](\n)[^一]', re.DOTALL)
clean_text = re.sub(pattern, lambda m: m.group(0).replace('\n', ''), raw_text)
# ---- 问题2：特殊符号干扰 ----
# 现象：PDF 中的项目符号 • 、多余空格等
clean_text = clean_text.replace('•', '')
clean_text = clean_text.replace(' ', ' ') # 合并多余空格
# ---- 问题3：页眉页脚噪声 ----
# 现象：每页都有 "第X页"、"公司名称" 等重复内容
# 解决：根据元数据中的页码信息，过滤掉固定位置的噪声文本
```
### 3.2 清洗函数封装
把清洗逻辑封装成函数，方便复用：
```python
def clean_pdf_text(text: str) -> str:
"""清洗 PDF 解析出的文本，去除常见噪声"""
import re
# 删除非中文字符之间的换行符
text = re.sub(r'[^一](\n)[^一]',
lambda m: m.group(0).replace('\n', ''), text)
# 删除项目符号和多余空格
text = text.replace('•', '').replace(' ', ' ')
# 删除连续的换行符（保留一个）
text = re.sub(r'\n{2,}', '\n', text)
return text.strip()
```
---
## 四、文本分块：RAG 质量的命门
**分块做不好，后面的一切都是白搭。**
### 4.1 为什么要分块？
单个文档的长度往往超过模型的上下文窗口。就算没超，把整本书塞进 Prompt 也不是个好主意 —— 模型会找不到关键信息。
分块的逻辑：把长文档切成小段，每段作为一个检索单元。检索时返回 Top-K 个最相关的段落，拼进 Prompt 让模型回答。
### 4.2 两个核心参数
不管用哪种分块器，都要理解这两个参数：
+ `chunk_size`：每个块包含的最大字符数（或 Token 数）
+ `chunk_overlap`：相邻块之间重叠的字符数，用于保持上下文连贯
```plain
chunk_overlap 的作用示意：
块1: [AAAA BBBB CCCC DDDD]
块2: [CCCC DDDD EEEE FFFF]
↑↑↑↑↑↑↑↑
重叠区域，防止语义断裂
```
**调参经验**：
+ `chunk_size` 一般设 500-1000 字符（中文场景）
+ `chunk_overlap` 设为 `chunk_size` 的 10%-20%
+ `chunk_overlap=0` 是新手最常犯的错 —— 一个完整句子被切成两半，检索时两半都不完整
### 4.3 策略一：固定字符分块（最简单）
最直觉的方法：不管内容，按固定字符数硬切。
```python
#安装依赖：uv pip install langchain-text-splitters -i https://mirrors.aliyun.com/pypi/simple
# 手动实现固定字符分块
from langchain_text_splitters import CharacterTextSplitter
# 1. 定义分块器
text_splitter = CharacterTextSplitter(
separator="\n\n", # 指定分隔符，默认为 "\n\n"
chunk_size=500, # 每个块的最大字符数
chunk_overlap=100, # 块与块之间的重叠字符数，建议设置以保持上下文连贯[citation:4]
length_function=len, # 计算长度的方法，默认按字符数
is_separator_regex=False # 是否将分隔符视为正则表达式
)
# 2. 执行分块，返回字符串列表
text = """
LangChain 是一个让你的 LLM 变得更强大的开源框架。
你想开发一个基于 LLM 的应用，需要什么组件它都有，直接使用就行。
甚至针对常规的应用流程，它利用 Chain 这个概念已经内置标准化方案了。
下面我们从新兴的大语言模型技术栈的角度来看看为何它的理念这么受欢迎。
"""
chunks = text_splitter.split_text(text)
for i, chunk in enumerate(chunks):
print(f"--- 块{i+1} ---\n{chunk}\n")
```
**优点**：实现简单，速度最快
**缺点**：粗暴，容易把句子切断，语义碎片化严重
**适用场景**：快速原型验证、格式非常规整的文本
### 4.4 策略二：递归字符分块（推荐首选）
这是 LangChain 的「默认推荐」，也是大多数场景下的最佳起点。
核心逻辑：**按优先级分隔符递归尝试，优先保持语义完整，实在不行再硬截断**。
流程像「剥洋葱」一样层层递进：
```plain
第一步：按 \n\n（段落）拆分
↓ 如果某段超长
第二步：按 。（句号）拆分
↓ 如果某句超长
第三步：按 ，（逗号）拆分
↓ 如果还是超长
第四步：按空格拆分，实在不行就硬截断
```
```python
'''
* RecursiveCharacterTextSplitter 递归字符文本分割
RecursiveCharacterTextSplitter 将按不同的字符递归地分割(按照这个优先级["\n\n", "\n", " ", ""])，
这样就能尽量把所有和语义相关的内容尽可能长时间地保留在同一位置
RecursiveCharacterTextSplitter需要关注的是4个参数：
* separators - 分隔符字符串数组
* chunk_size - 每个文档的字符数量限制
* chunk_overlap - 两份文档重叠区域的长度
* length_function - 长度计算函数
'''
from langchain_text_splitters import RecursiveCharacterTextSplitter
# 创建递归字符分割器
text_splitter = RecursiveCharacterTextSplitter(
# 分隔符优先级：段落 → 换行 → 句号 → 空格 → 硬切
separators=["\n\n", "\n", "。", "！", "？", " ", ""],
# 每个块最大 50 字符
chunk_size=50,
# 相邻块重叠 10 字符（chunk_size 的 20%）
chunk_overlap=10,
# 长度计算函数
length_function=len
)
# 对单段文本进行分割
text = """
LangChain 是一个让你的 LLM 变得更强大的开源框架。
你想开发一个基于 LLM 的应用，需要什么组件它都有，直接使用就行。
甚至针对常规的应用流程，它利用 Chain 这个概念已经内置标准化方案了。
下面我们从新兴的大语言模型技术栈的角度来看看为何它的理念这么受欢迎。
"""
chunks = text_splitter.split_text(text)
for i, chunk in enumerate(chunks):
print(f"--- 块{i+1} ---\n{chunk}\n")
```
```python
#对整个文件进行切割
split_docs = text_splitter.split_documents(pdf_pages)
print(f"切分后的文件数量：{len(split_docs)}")
print(f"切分后的字符数（可以用来大致评估 token 数）：{sum([len(doc.page_content) for doc in split_docs])}")
```
**优点**：保持语义完整性好，适用范围广
**缺点**：需要调参 separators 和 chunk_size
**适用场景**：大多数场景的默认首选
### 4.5 策略三：语义分块（效果最好）
这种方法不按字符数硬切，而是用 Embedding 模型计算相邻句子的语义相似度，当相似度「断崖式下降」时，就在那里切一刀。
```python
from langchain_experimental.text_splitter import SemanticChunker
from langchain_community.embeddings import DashScopeEmbeddings
# 使用通义的 Embedding 模型做语义分块
embedding_model = DashScopeEmbeddings(
model="text-embedding-v3",
dashscope_api_key="your-api-key"
)
# 创建语义分块器
semantic_splitter = SemanticChunker(
embeddings=embedding_model,
breakpoint_threshold_type="percentile", # 使用百分位数作为阈值
breakpoint_threshold_amount=85 # 相似度低于 85% 百分位时切分
)
# 分块
docs = semantic_splitter.create_documents([raw_text])
print(f"语义分块结果：{len(docs)} 个块")
```
**优点**：语义最完整，每个块都是一个「完整话题」
**缺点**：速度慢（每次都要调用 Embedding），适合离线批量处理
**适用场景**：技术文档、法律合同、学术论文等强结构文本
### 4.6 策略四：文档结构分块
利用文档本身的结构（标题、章节、段落）来定义块边界。
```python
from langchain_text_splitters import MarkdownHeaderTextSplitter
# Markdown 按标题分块
headers_to_split_on = [
("#", "一级标题"),
("##", "二级标题"),
("###", "三级标题"),
]
markdown_splitter = MarkdownHeaderTextSplitter(
headers_to_split_on=headers_to_split_on
)
# 分块（会保留标题信息在 metadata 中）
md_header_splits = markdown_splitter.split_text(markdown_content)
```
**优点**：保留文档原有逻辑结构，块的语义边界清晰
**缺点**：依赖文档格式质量，块大小不均匀
**适用场景**：Markdown 文档、有清晰标题结构的 PDF
### 4.7 策略五：LLM 分块（最智能）
直接让大模型来判断哪里该切、哪里该留。
```python
# 原理示意（实际实现需要调用 LLM API）
#
# Prompt: "请将以下文本分成语义完整的段落，每段不超过 500 字：
# {text}"
#
# LLM 会返回切分好的段落列表
```
**优点**：语义理解最深，最智能
**缺点**：成本最高，速度最慢，受 LLM 上下文窗口限制
**适用场景**：高价值文档、小规模数据、对质量要求极高的场景
### 4.8 分块策略选择指南
```plain
你的场景是什么？
│
├─ 需要快速上手 → 固定字符分块
│
├─ 通用场景 → 递归字符分块 ✓（推荐）
│
├─ 追求最佳效果 → 语义分块
│
└─ 预算充足、数据量小 → LLM 分块
```
**新手建议**：从递归字符分块起步。效果不好，再试语义分块。不要一上来就搞最复杂的。
### 4.9 分块后的过滤
切分完记得过滤掉空块和无效块：
```python
# 过滤掉 page_content 为空或仅含空白的文档
valid_docs = [
doc for doc in split_docs
if doc.page_content and doc.page_content.strip()
]
print(f"有效块数量：{len(valid_docs)}")
print(f"总字符数（可大致评估 Token 数）：{sum(len(d.page_content) for d in valid_docs)}")
```
---
## 五、向量嵌入：把文本变成「数字」
### 5.1 Embedding 是什么？
Embedding 是把文本映射成一串数字(向量)的过程。语义相近的文本，向量距离就近；语义无关的文本，向量距离就远。
类比：把每段文字映射到一个高维空间里，「苹果手机」和「iPhone」在这个空间里距离很近，和「橙子」距离就很远。
### 5.2 国内 Embedding 模型选型
| 模型 | 厂商 | 维度 | 特点 |
| :--- | :--- | :--- | :--- |
| `text-embedding-v3` | [阿里（通义）](https://help.aliyun.com/zh/model-studio/text-embedding-synchronous-api?spm=a2c4g.11186623.help-menu-2400256.d_2_7_0_0.2d9b785b0jVsx4) | 1024/768 | 性价比高，中文效果好 |
| `embedding-3` | 百度（文心） | 1024/2048 | 中文语义理解强 |
| `BAAI/bge-m3` | 智源 | 1024 | 开源，可本地部署 |
| `M3E` | MokaAI | 768 | 开源，中文优化 |
### 5.3 实战：使用通义 Embedding
```python
from langchain_community.embeddings import DashScopeEmbeddings
# 初始化 Embedding 模型
# 需要先在 https://dashscope.console.aliyun.com/ 获取 API Key
embedding_model = DashScopeEmbeddings(
model="text-embedding-v3", # 模型名称
dashscope_api_key="your-api-key" # 替换为你的真实 Key
)
# 单条文本向量化
text = "RAG系统搭建实战"
embedding = embedding_model.embed_query(text)
print(f"向量维度：{len(embedding)}")
print(f"前5个值：{embedding[:5]}")
```
```python
# 批量文本向量化
texts = [
"hello world",
"什么是大语言模型",
"RAG的概念和原理",
"牛顿第一定律是什么",
]
embeddings = embedding_model.embed_documents(texts)
print(f"生成了 {len(embeddings)} 个向量")
print(f"每个向量维度：{len(embeddings[0])}")
```
---
## 六、向量数据库：给向量安个「家」
### 6.1 为什么需要向量数据库？
传统数据库（MySQL、PostgreSQL）擅长精确查询，但不擅长「找相似的」。向量数据库专门为向量检索设计，能高效地做近似最近邻搜索（ANN--用于在海量数据里面寻找“足够相似”的数据）。
### 6.2 常见向量数据库对比
| 数据库 | 类型 | 特点 | 适用场景 |
| :--- | :--- | :--- | :--- |
| **ChromaDB** | 嵌入式 | 轻量、易用、Python 原生 | 快速原型、小规模 |
| **FAISS** | 库 | Facebook 开源，速度快 | 大规模向量检索 |
| **Milvus** | 分布式 | 云原生，支持海量数据 | 生产环境 |
| **Weaviate** | 云服务 | 自带向量化，API 友好 | 全托管场景 |
### 6.3 实战：ChromaDB 基础操作
```python
#安装依赖：uv pip install chromadb -i https://pypi.tuna.tsinghua.edu.cn/simple
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import DashScopeEmbeddings
# 初始化 Embedding 模型
embedding_model = DashScopeEmbeddings(
model="text-embedding-v3",
dashscope_api_key="your-api-key"
)
# 准备一些测试数据
datas = [
"小明特别喜欢吃脆甜多汁的苹果",
"小红对榴莲那独特的味道情有独钟",
"小明和小丽是一对甜蜜的情侣",
"王老师教学认真负责，是公认的好老师",
"小李每天都要吃一根香蕉",
"小王的男朋友长得阳光帅气，是大家公认的大帅哥"
]
# 创建向量数据库并持久化(会同时把文本和对应的向量存入数据库)
db = Chroma.from_texts(
texts=datas,
embedding=embedding_model,
collection_metadata={"hnsw:space": "cosine"}, # 关键配置：指定距离算法为余弦相似度
persist_directory="./chroma_db1" # 数据库存储路径
)
```
```python
# 相似度检索
query = "谁是老师"
results = db.similarity_search(query, k=1) # 返回最相似的 1 条
for doc in results:
print(f"匹配内容：{doc.page_content}")
```
```python
# 带分数的相似度检索（分数越低越相似，0 表示完全匹配）
query = "谁是老师"
results = db.similarity_search_with_score(query, k=3)
for doc, score in results:
print(f"内容：{doc.page_content} | 相似度分数：{score:.4f}")
```
---
## 七、检索器：连接「问题」和「答案」的桥梁
### 7.1 检索器的工作流
检索器（Retriever）是 LangChain 中的核心接口，负责根据用户问题从知识库中查询来获取相关文档。
核心数据流：
```plain
用户输入问题
→ 检索器将问题转为向量
→ 在向量库中查找语义最近的文档块
→ 返回 Top-K 相关文档
```
### 7.2 实战：创建检索器
```python
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import DashScopeEmbeddings
# 假设 split_docs 是前面分块后的文档列表
embedding_model = DashScopeEmbeddings(
model="text-embedding-v3",
dashscope_api_key="your-api-key"
)
# 从文档创建向量库(写数据)
vectorstore = Chroma.from_documents(
documents=valid_docs,
embedding=embedding_model,
collection_metadata={"hnsw:space": "cosine"},
persist_directory="./my_knowledge_db"
)
# 创建检索器，设置返回 Top-3 最相关文档
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
```
```python
# 测试检索
query = "在多少速度范围内，HDC才会激活？"
relevant_docs = retriever.invoke(query)
for i, doc in enumerate(relevant_docs):
print(f"--- 结果{i+1} ---")
print(f"内容：{doc.page_content[:100]}...")
print(f"来源：{doc.metadata}")
print()
```
```python
# 直接加载已持久化的数据库（无需再次添加文档或持久化）
vectordb = Chroma(
persist_directory="./my_knowledge_db",
embedding_function=embedding_model
)
```
---
## 八、生成回答：最后一公里
### 8.1 Prompt 工程
检索到相关文档后，需要把它们拼进 Prompt，让大模型基于这些上下文生成回答。
```python
from langchain_core.prompts import ChatPromptTemplate
# 构建 Prompt 模板
prompt = ChatPromptTemplate.from_template("""
你是一个专业的知识库助手。请根据以下检索到的上下文回答用户问题。
**规则：**
- 只基于提供的上下文回答，不要编造
- 如果上下文中没有相关信息，直接说「根据现有资料，我找不到这个问题的答案」
- 回答要简洁直接，引用原文时用引号
**检索到的上下文：**
{context}
**用户问题：**
{question}
""")
```
### 8.2 调用大模型生成回答
```python
from langchain_community.chat_models import ChatTongyi
from langchain_core.output_parsers import StrOutputParser
# 初始化大模型（这里用通义千问，也可以换成 DeepSeek）
llm = ChatTongyi(
model="qwen-plus", # 模型名称
dashscope_api_key="your-api-key" # 替换为你的真实 Key
)
# 拼装上下文
context_text = "\n\n---\n\n".join([doc.page_content for doc in relevant_docs])
# 构建 Chain 并调用
chain = prompt | llm | StrOutputParser()
response = chain.invoke({
"context": context_text,
"question": query
})
print(response)
```
### 8.3 完整 RAG Chain 串联
把前面所有步骤串起来，形成一个完整的 RAG 流程：
```python
from langchain_community.document_loaders import PyMuPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_community.chat_models import ChatTongyi
from langchain_core.output_parsers import StrOutputParser
# ========== 第一步：加载文档 ==========
loader = PyMuPDFLoader("./your_document.pdf")
pages = loader.load()
# ========== 第二步：清洗数据（可选，根据文档质量决定）==========
# clean_pages = [clean_pdf_text(page) for page in pages]
# ========== 第三步：分块 ==========
splitter = RecursiveCharacterTextSplitter(
# 分隔符优先级：段落 → 换行 → 句号 → 空格 → 硬切
separators=["\n\n", "\n", "。", "！", "？", " ", ""],
# 每个块最大 50 字符
chunk_size=50,
# 相邻块重叠 10 字符（chunk_size 的 20%）
chunk_overlap=10,
# 长度计算函数
length_function=len
)
docs = splitter.split_documents(pages)
# ========== 第四步：向量化 + 存入向量库 ==========
embedding_model = DashScopeEmbeddings(
model="text-embedding-v3",
dashscope_api_key="your-api-key"
)
vectorstore = Chroma.from_documents(
documents=docs,
embedding=embedding_model,
persist_directory="./knowledge_db"
)
# ========== 第五步：创建检索器 ==========
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
# ========== 第六步：提问 ==========
query = "你的问题"
relevant_docs = retriever.invoke(query)
# ========== 第七步：生成回答 ==========
context = "\n\n---\n\n".join([d.page_content for d in relevant_docs])
prompt = ChatPromptTemplate.from_template("""
你是一个专业的知识库助手。请根据以下上下文回答问题。
**规则：**
- 只基于提供的上下文回答，不要编造
- 如果上下文中没有相关信息，直接说「根据现有资料，我找不到这个问题的答案」
- 回答要简洁直接，引用原文时用引号
**上下文：**
{context}
**问题：**
{question}
""")
llm = ChatTongyi(model="qwen-plus", dashscope_api_key="your-api-key")
chain = prompt | llm | StrOutputParser()
answer = chain.invoke({"context": context, "question": query})
print(answer)
```
**这就是一次完整的 RAG 流程**：加载 → 清洗 → 分块 → 嵌入 → 存储 → 检索 → 生成。
---
## 九、实战建议与踩坑总结
### 9.1 分块大小怎么定？
+ 中文文档：500-1000 字符
+ 英文文档：1000-2000 字符
+ 代码文档：按函数/类切分，不要按字符切
### 9.2 Top-K 设多少？
+ K=3：适合简单问答
+ K=5：适合需要综合多个信息源的场景
+ K=10：适合需要全面覆盖的场景（但会增加 Token 消耗）
### 9.3 常见踩坑
1. **chunk_overlap=0**：最常见的错误。相邻块没有重叠，一个完整句子被切成两半，检索时两半都不完整。
2. **不做数据清洗**：PDF 解析出来的噪声会严重干扰检索效果。
3. **Embedding 模型和查询语言不匹配**：用英文 Embedding 模型处理中文，效果会很差。
4. **Top-K 设太小**：只返回 1 条结果，很容易漏掉关键信息。
---