大模型的知识有截止日期，且无法直接获取你的私有数据。RAG（Retrieval-Augmented Generation，检索增强生成）解决的就是这个问题：让大模型基于你提供的知识库来回答问题，而不是凭"记忆"。
这篇文章从工程实践角度，完整拆解RAG系统的每一个环节：文档处理、向量化、检索、重排序、生成。附核心代码和踩坑经验。
一、RAG系统架构全景
一个完整的RAG系统包含两条链路：
【离线索引链路】
原始文档 → 文档解析 → 文本分块 → Embedding向量化 → 存入向量数据库
【在线查询链路】
用户提问 → 问题Embedding → 向量相似度检索 → Rerank重排序 → 构造Prompt → 大模型生成回答二、文档解析与预处理
企业知识库的文档格式五花八门：PDF、Word、Excel、PPT、HTML、Markdown。第一步是把它们统一转为纯文本。
常用解析工具
| 格式 | 推荐工具 | 注意事项 |
|---|---|---|
| PyMuPDF / pdfplumber | 扫描件PDF需要OCR预处理 | |
| Word | python-docx | 注意表格和嵌入图片的处理 |
| HTML | BeautifulSoup / trafilatura | 去除导航、页脚等噪音 |
| Markdown | 直接处理 | 保留标题层级作为元数据 |
import fitz # PyMuPDF
def parse_pdf(file_path):
"""解析PDF文件，提取文本和元数据"""
doc = fitz.open(file_path)
pages = []
for i, page in enumerate(doc):
text = page.get_text("text")
if text.strip():
pages.append({
"content": text.strip(),
"metadata": {
"source": file_path,
"page": i + 1,
"total_pages": len(doc)
}
})
return pages三、文本分块（Chunking）
分块是RAG效果的关键影响因素。太大的块包含太多无关信息，太小的块丢失上下文。
分块策略对比
| 策略 | 原理 | 适用场景 |
|---|---|---|
| 固定长度 | 按字符/Token数切割 | 结构不明确的长文本 |
| 递归分割 | 按段落→句子→字符逐级切割 | 通用场景（推荐默认） |
| 语义分块 | 用Embedding检测语义断点 | 对精度要求高的场景 |
| 按文档结构 | 按标题/章节切割 | 结构化文档（手册、规范） |
推荐参数
from langchain.text_splitter import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(
chunk_size=500, # 每块约500字符（中文约250字）
chunk_overlap=80, # 块间重叠80字符，保留上下文连续性
separators=["\n\n", "\n", "。", "；", "，", " "], # 中文优先分割符
length_function=len
)
chunks = splitter.split_documents(documents)chunk_overlap非常重要。如果没有重叠，一句话被切成两半，两个块各拿到半句，检索时都不完整。80-100字符的overlap是一个比较好的平衡点。
四、Embedding向量化
将文本转为高维向量，使语义相近的文本在向量空间中距离更近。
Embedding模型选型
| 模型 | 维度 | 中文效果 | 特点 |
|---|---|---|---|
| BGE-large-zh | 1024 | 优秀 | 开源、中文专优化、MTEB排行榜前列 |
| BGE-M3 | 1024 | 优秀 | 多语言、支持稀疏+稠密混合检索 |
| 通义千问Embedding | 1536 | 优秀 | API调用、无需本地部署 |
| GTE-large-zh | 1024 | 良好 | 阿里开源、推理速度快 |
from sentence_transformers import SentenceTransformer
# 加载BGE模型
model = SentenceTransformer("BAAI/bge-large-zh-v1.5")
# 对查询需要加前缀（BGE特有）
query_embedding = model.encode("为用户解答：什么是RAG？")
doc_embeddings = model.encode(["文档块1的内容", "文档块2的内容"])BGE模型在编码查询时需要添加
"为用户解答："前缀，而编码文档时不需要。这是训练时的设计，不加前缀会降低检索效果。不同模型的前缀要求不同，使用前一定要查文档。五、向量数据库选型与使用
主流向量数据库对比
| 数据库 | 类型 | 适用规模 | 特点 |
|---|---|---|---|
| Milvus | 独立部署 | 百万-十亿级 | 高性能、支持多种索引、国产 |
| Chroma | 嵌入式 | 万-百万级 | 极简API、开发调试方便 |
| FAISS | 库 | 百万级 | Meta开源、纯内存、速度极快 |
| Qdrant | 独立部署 | 百万级 | Rust编写、过滤能力强 |
Milvus实战
from pymilvus import Collection, CollectionSchema, FieldSchema, DataType, connections
# 连接Milvus
connections.connect(host="localhost", port="19530")
# 定义Schema
fields = [
FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
FieldSchema(name="content", dtype=DataType.VARCHAR, max_length=2000),
FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=1024),
FieldSchema(name="source", dtype=DataType.VARCHAR, max_length=256),
]
schema = CollectionSchema(fields, description="知识库文档集合")
collection = Collection("knowledge_base", schema)
# 创建索引（IVF_FLAT适合百万级数据）
index_params = {
"index_type": "IVF_FLAT",
"metric_type": "COSINE",
"params": {"nlist": 128}
}
collection.create_index("embedding", index_params)
# 插入数据
collection.insert([contents, embeddings, sources])
# 检索
collection.load()
results = collection.search(
data=[query_embedding],
anns_field="embedding",
param={"metric_type": "COSINE", "params": {"nprobe": 16}},
limit=10,
output_fields=["content", "source"]
)六、Rerank重排序
向量检索返回的Top-K结果，排序并不总是准确。Rerank用一个交叉编码器对"查询-文档"对进行精排，显著提升结果质量。
from sentence_transformers import CrossEncoder
reranker = CrossEncoder("BAAI/bge-reranker-large")
# 对检索结果重新打分
pairs = [(query, doc["content"]) for doc in search_results]
scores = reranker.predict(pairs)
# 按分数重排
reranked = sorted(zip(search_results, scores), key=lambda x: x[1], reverse=True)
top_docs = [doc for doc, score in reranked[:5]]Rerank的计算成本比Embedding高很多（交叉编码器需要对每个Query-Doc对做完整推理），所以通常是先用向量检索召回Top-20，再用Rerank精排到Top-5。不要对全库做Rerank。
七、Prompt构造与生成
最终把检索到的文档作为上下文，构造Prompt让大模型生成回答：
def build_rag_prompt(query, docs):
context = "\n\n---\n\n".join([
f"【来源：{doc['source']}】\n{doc['content']}"
for doc in docs
])
return f"""基于以下参考资料回答用户的问题。如果参考资料中没有相关信息，请明确说明"根据现有资料无法回答"，不要编造。
## 参考资料
{context}
## 用户问题
{query}
## 回答要求
- 基于参考资料回答，引用具体来源
- 如果多个资料有矛盾，指出差异
- 语言简洁，重点突出"""调用大模型
import dashscope
from dashscope import Generation
response = Generation.call(
model="qwen-plus",
messages=[
{"role": "system", "content": "你是一个企业知识库助手。"},
{"role": "user", "content": build_rag_prompt(query, top_docs)}
],
temperature=0.1, # RAG场景用低温度，减少发散
max_tokens=2000
)
answer = response.output.text八、效果优化的关键指标
- 召回率（Recall）：正确答案是否在检索结果中？→ 调整chunk_size、增加检索数量
- 精确率（Precision）：检索结果中有多少是相关的？→ Rerank、优化Embedding模型
- 答案准确性：生成的回答是否忠实于文档？→ 调整Prompt、降低temperature
- 答案完整性：是否覆盖了所有相关信息？→ 增加Top-K数量、优化分块策略
九、常见踩坑与优化
- 中文分块不要用英文分隔符：默认的RecursiveCharacterTextSplitter的分隔符是英文标点，中文文本要自定义
- PDF表格数据会丢失结构：表格内容被解析为平铺文本后语义丢失，建议用专门的表格解析工具
- 重复文档要去重：同一份文档被多次导入会导致检索结果重复，浪费上下文窗口
- Embedding模型和Reranker模型建议配套：BGE-large-zh + BGE-reranker-large是验证过的组合
- 生产环境Milvus建议用IVF_PQ索引：比IVF_FLAT省内存，大规模数据下性价比更高