RAG 进阶实战：从基础到生产级应用的完整指南
你搭过 RAG 系统吗？
简单版本：把文档切块，存入向量数据库，用 embedding 模型做相似度检索，然后把检索结果和用户的 prompt 一起喂给大模型。
听起来很简单，对吧？但当你真正把它用到生产环境，问题就来了：
- 检索结果相关性差，答非所问
- 响应速度慢，用户体验糟糕
- 长文档处理效果差，关键信息遗漏
- 不知道哪里出了问题，调试困难
今天这篇，帮你从"能跑"进阶到"好用"。
一、检索质量提升：不只是余弦相似度
问题：余弦相似度的局限
传统的向量检索使用余弦相似度，计算的是向量之间的夹角。但这有个问题：
余弦相似度只关注方向，不关注绝对距离。两个向量即使夹角很小，内容可能并不相关。
方案一：混合检索
结合多种检索方式：
from langchain.retrievers import EnsembleRetriever
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import OpenAIEmbeddings
from langchain.retrievers import BM25Retriever
# 向量检索
vector_retriever = Chroma(
embedding_function=OpenAIEmbeddings(),
collection_name="docs"
).as_retriever(search_kwargs={"k": 5})
# 关键词检索（BM25）
bm25_retriever = BM25Retriever.from_documents(documents)
# 混合检索
ensemble_retriever = EnsembleRetriever(
retrievers=[vector_retriever, bm25_retriever],
weights=[0.7, 0.3]
)
混合检索结合了语义理解和关键词匹配的优势。对于技术文档、代码搜索等场景，效果明显更好。
方案二：元数据过滤
在检索时加入元数据过滤，可以大幅提高准确性：
# 检索时指定元数据过滤
retriever = vector_store.as_retriever(
search_type="similarity",
search_kwargs={
"k": 5,
"filter": {
"source": "user_manual",
"language": "zh",
"date_range": {"$gte": "2025-01-01"}
}
}
)
比如用户问"如何重置密码"，你可以先过滤出"用户手册"类型的文档，排除掉"开发文档"。
方案三：重排序（Rerank）
检索返回的结果需要进一步排序。可以用重排序模型来提升准确性：
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CrossEncoderReranker
from langchain_community.cross_encoders import HuggingFaceCrossEncoder
# 加载重排序模型
cross_encoder = HuggingFaceCrossEncoder(
model_name="cross-encoder/ms-marco-MiniLM-L-6-v2"
)
# 创建压缩检索器
compressor = CrossEncoderReranker(encoder=cross_encoder, top_n=3)
compression_retriever = ContextualCompressionRetriever(
base_retriever=vector_retriever,
document_compressors=[compressor]
)
重排序模型会重新评估检索结果的排序，把最相关的文档排在前面。
二、文档处理优化：切块的艺术
问题：切块方式影响检索效果
切块太粗，检索结果不够精确。切块太细，上下文信息丢失。
方案一：按语义切块
不是简单地按字符数切分，而是按语义边界切分：
from langchain.text_splitter import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(
chunk_size=800, # 每个块的大小
chunk_overlap=100, # 块之间的重叠
separators=["\n\n", "\n", "。", " ", ""], # 按这些分隔符切分
length_function=len
)
RecursiveCharacterTextSplitter 会尝试按你指定的分隔符依次切分，优先按段落切，其次按句子切。
方案二：父子索引（Parent-Child Indexing）
这是生产环境常用的技巧：
- 把文档切分成小块（子文档）
- 检索时返回小块
- 但实际传给大模型的是小块的父文档（更大的上下文）
from langchain.text_splitter import RecursiveCharacterTextSplitter
# 小粒度切分用于检索
child_splitter = RecursiveCharacterTextSplitter(
chunk_size=400,
chunk_overlap=50
)
# 大粒度切分用于返回上下文
parent_splitter = RecursiveCharacterTextSplitter(
chunk_size=1200,
chunk_overlap=200
)
# 先用子文档切分检索，再映射回父文档
这样既能精确检索，又能提供足够的上下文。
方案三：元数据增强
给每个文本块添加元数据，帮助检索器理解内容：
from langchain.docstore.document import Document
doc = Document(
page_content="如何重置密码...",
metadata={
"section": "用户帮助",
"topic": "账户安全",
"priority": "high",
"author": "support_team"
}
)
检索时可以用这些元数据做加权。
三、索引构建最佳实践
批量处理大文档
处理大文档时，不要一次性加载：
from langchain_community.document_loaders import DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
# 分批次加载和处理
loader = DirectoryLoader(
"./documents",
glob="**/*.md",
show_progress=True
)
all_documents = loader.load()
# 分批处理
batch_size = 100
for i in range(0, len(all_documents), batch_size):
batch = all_documents[i:i + batch_size]
chunks = splitter.split_documents(batch)
# 分批添加到向量数据库
vector_store.add_documents(chunks)
print(f"Processed {i + len(batch)}/{len(all_documents)} documents")
增量更新
不要每次都重新索引全部文档：
import hashlib
from datetime import datetime
def compute_doc_hash(content):
return hashlib.md5(content.encode()).hexdigest()
# 检查文档是否变更
if doc_hash not in existing_hashes:
# 只更新变更的文档
update_document(doc)
索引监控
定期检查索引质量：
# 检查向量数据库状态
stats = vector_store.get_stats()
print(f"总文档数: {stats['total_documents']}")
print(f"总向量数: {stats['total_vectors']}")
# 检查最近检索的准确率（需要人工标注）
四、性能优化
检索加速
- 使用本地向量数据库（如 Chroma、Qdrant）而非在线服务
- 启用 HNSW 索引（近似最近邻搜索）
- 缓存检索结果
from functools import lru_cache
import hashlib
@lru_cache(maxsize=1000)
def cached_retrieve(query: str, top_k: int = 5):
# 查询缓存
query_hash = hashlib.md5(query.encode()).hexdigest()
return vector_store.similarity_search_with_score(query, k=top_k)
流式输出
不要等所有检索结果都返回再生成回答：
from langchain.callbacks import AsyncIteratorCallbackHandler
async def stream_response(query):
callback = AsyncIteratorCallbackHandler()
# 异步检索和生成
task = asyncio.create_task(
chain.ainvoke({"question": query}, config={"callbacks": [callback]})
)
# 流式输出
async for token in callback.aiter():
yield token
批量 Embedding
# 批量生成 embedding，减少 API 调用次数
embeddings = embedding_model.embed_documents(
[doc.page_content for doc in documents]
)
五、常见问题排查
问题一：检索结果不相关
排查步骤：
- 检查 embedding 模型是否适合你的领域
- 检查切块大小是否合理
- 检查是否有足够的训练数据来微调模型
解决方案：
# 尝试不同的切块策略
splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50,
separators=["\n\n", "\n", "。", " ", ""]
)
# 尝试不同的 embedding 模型
from langchain.embeddings import HuggingFaceEmbeddings
embedding = HuggingFaceEmbeddings(
model_name="thenlper/gte-large"
)
问题二：回答太长或不简洁
解决方案：
- 调整 prompt，要求简洁回答
- 限制检索的文档数量
- 使用摘要策略
prompt = """请根据以下上下文，简洁回答用户问题。
回答不超过3句话。
上下文：
{context}
用户问题：{question}
回答："""
问题三：系统响应慢
排查步骤：
- 检查向量数据库查询性能
- 检查 embedding 模型推理时间
- 检查 LLM API 响应时间
解决方案：
- 使用更快的 embedding 模型（如 text-embedding-3-small）
- 启用向量数据库的缓存
- 并行检索多个索引源
六、完整生产级 RAG 架构
用户请求
↓
API Gateway（限流、认证）
↓
Query Preprocessor（查询预处理）
├─ 查询改写
├─ 实体识别
└─ 意图分类
↓
Retrieval Layer（检索层）
├─ 向量检索
├─ 关键词检索
└─ 元数据过滤
↓
Reranker（重排序）
↓
Context Builder（上下文构建）
├─ 父子索引映射
└─ 冗余去重
↓
LLM Generator（生成层）
├─ Prompt 组装
└─ 流式生成
↓
Output Validator（输出验证）
├─ 内容过滤
└─ 格式校验
↓
响应返回
总结
从基础 RAG 到生产级应用，关键在于：
- 检索质量：混合检索 + 重排序
- 文档处理：语义切块 + 父子索引
- 索引管理：增量更新 + 监控
- 性能优化：缓存 + 批量处理 + 流式输出
你的 RAG 系统现在卡在哪个环节？是哪个问题困扰你最多？
AI Token 计算器 可以帮你估算 RAG 系统的 token 消耗，优化成本。