# RAG 检索增强生成实战指南
什么是 RAG?
RAG(Retrieval-Augmented Generation,检索增强生成)是一种将大规模语言模型(LLM)与外部知识库相结合的技术架构。其核心思想是:当用户提出问题时,系统先从知识库中检索相关文档,再将检索结果作为上下文提供给 LLM,由 LLM 基于这些真实信息生成回答。
RAG 的主要价值在于:
RAG 工作流程
完整的 RAG 流程包含以下核心环节:
用户问题 → 向量化查询 → 向量数据库检索 → 结果重排 → 上下文组装 → LLM 生成回答
下面我们逐一深入讲解各环节的实现要点。
文档处理与切分
切分策略的重要性
文档切分是 RAG 系统的第一道关口。切分粒度直接影响检索效果:
| 切分方式 | 适用场景 | 优缺点 |
|---------|---------|--------|
| 固定长度切分 | 通用场景 | 简单但可能破坏语义完整性 |
| 基于语义切分 | 知识库结构清晰 | 更精准但实现复杂 |
| 递归字符切分 | 混合内容 | 兼顾效率与效果 |
切分实现示例
from langchain.text_splitter import RecursiveCharacterTextSplitter
def split_documents(documents, chunk_size=500, chunk_overlap=50):
"""
递归字符切分器,支持多种分隔符
"""
splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size, # 每个文本块的目标大小
chunk_overlap=chunk_overlap, # 块之间的重叠字符数
separators=["\n\n", "\n", "。", "!", "?", " "] # 分隔优先级
)
return splitter.split_documents(documents)
**实操要点**:
向量化与 Embedding
Embedding 模型选择
Embedding 模型将文本转换为高维向量,直接决定语义检索的质量。主流选择包括:
向量化代码示例
from sentence_transformers import SentenceTransformer
import numpy as np
class EmbeddingModel:
def __init__(self, model_name="BAAI/bge-large-zh"):
self.model = SentenceTransformer(model_name)
self.dimension = self.model.get_sentence_embedding_dimension()
def encode(self, texts, normalize=True):
"""将文本列表转为向量"""
embeddings = self.model.encode(
texts,
normalize_embeddings=normalize # L2 归一化,便于余弦相似度计算
)
return embeddings.tolist()
def get_dimension(self):
return self.dimension
# 使用示例
embedder = EmbeddingModel()
query_embedding = embedder.encode(["RAG 的工作原理是什么?"])
向量检索
主流向量数据库对比
| 数据库 | 特点 | 适用规模 | 部署方式 |
|-------|------|---------|---------|
| Milvus | 功能丰富,国产 | 亿级 | 云/私有 |
| Chroma | 轻量易用 | 百万级 | 本地 |
| FAISS | 高性能 | 千万级 | 嵌入应用 |
| Qdrant | 支持混合检索 | 百万级 | 云/私有 |
检索实现
import faiss
import numpy as np
class VectorStore:
def __init__(self, dimension):
self.dimension = dimension
self.index = faiss.IndexFlatIP(dimension) # 内积索引(余弦相似度需归一化)
self.id_to_doc = {}
self._doc_id = 0
def add_vectors(self, vectors, documents):
"""批量添加向量和对应文档"""
vectors = np.array(vectors).astype('float32')
self.index.add(vectors)
for doc in documents:
self.id_to_doc[self._doc_id] = doc
self._doc_id += 1
def search(self, query_vector, top_k=5):
"""检索最相似的 top_k 个文档"""
query = np.array([query_vector]).astype('float32')
distances, indices = self.index.search(query, top_k)
results = []
for dist, idx in zip(distances[0], indices[0]):
if idx != -1: # -1 表示无结果
results.append({
"score": float(dist),
"document": self.id_to_doc[idx]
})
return results
**关键参数说明**:
结果重排(Reranking)
初筛结果往往存在相关性排序不理想的问题。重排模型(如 `bge-reranker-large`)会综合语义匹配度与文本特征,对候选结果重新排序。
from sentence_transformers import CrossEncoder
class Reranker:
def __init__(self, model_name="BAAI/bge-reranker-large"):
self.model = CrossEncoder(model_name)
def rerank(self, query, candidates, top_k=3):
"""
对候选文档重排,返回 top_k 个最终结果
"""
pairs = [[query, doc["document"]] for doc in candidates]
scores = self.model.predict(pairs