RAG 架构设计模式
RAG（Retrieval-Augmented Generation）是将检索与生成结合的主流架构，用于构建知识库问答系统。
一、RAG 核心概念
1.1 什么是 RAG
RAG = 检索（Retrieval）+ 生成（Generation）
检索：从知识库中查找相关信息
生成：基于检索结果生成回答
优势：
- 减少幻觉（基于事实）
- 支持私有知识
- 可追溯来源
- 成本可控1.2 基本架构
用户问题
↓
[查询处理]
↓
[向量检索] → [向量数据库]
↓
[结果重排序]
↓
[构建 Prompt] → [大模型]
↓
生成回答二、设计模式
2.1 Naive RAG（基础 RAG）
# 最简单的 RAG 实现
def naive_rag(query, documents, top_k=3):
# 1. 向量化查询
query_embedding = embed(query)
# 2. 检索相似文档
results = vector_db.search(query_embedding, top_k)
# 3. 构建上下文
context = "\n\n".join([doc.text for doc in results])
# 4. 生成回答
prompt = f"""
基于以下信息回答问题：
{context}
问题：{query}
"""
answer = llm.generate(prompt)
return answer优点：实现简单、快速上线 缺点：检索质量不稳定、容易遗漏信息
2.2 Advanced RAG（高级 RAG）
# 增加优化的 RAG
def advanced_rag(query, documents, top_k=5):
# 查询优化
query_expanded = expand_query(query) # 查询扩展
# 多路检索
results_bm25 = bm25_search(query_expanded, top_k=10)
results_vector = vector_search(query_expanded, top_k=10)
# 结果融合
merged_results = reciprocal_rank_fusion(
results_bm25,
results_vector
)
# 重排序
reranked = cross_encoder_rerank(
query,
merged_results,
top_k=5
)
# 生成回答
answer = generate_answer(query, reranked)
return answer优化点：
- 查询扩展（Query Expansion）
- 混合检索（BM25 + 向量）
- 重排序（Cross-Encoder）
2.3 Modular RAG（模块化 RAG）
# 模块化设计
class ModularRAG:
def __init__(self):
self.query_processor = QueryProcessor()
self.retriever = HybridRetriever()
self.reranker = CrossEncoderReranker()
self.generator = LLMGenerator()
def query(self, user_query):
# 1. 查询处理
processed_query = self.query_processor.process(
user_query,
history=self.conversation_history
)
# 2. 检索
documents = self.retriever.retrieve(
processed_query,
filters=self.get_filters()
)
# 3. 重排序
ranked_docs = self.reranker.rerank(
processed_query,
documents
)
# 4. 生成
response = self.generator.generate(
query=processed_query,
context=ranked_docs,
style="professional"
)
return response优势：
- 各模块独立优化
- 易于扩展和维护
- 支持 A/B 测试
三、核心组件
3.1 文档处理流水线
class DocumentProcessor:
def process(self, file_path):
# 1. 加载文档
doc = self.load_document(file_path)
# 2. 清洗
doc = self.clean(doc)
# 3. 分块
chunks = self.chunk(doc)
# 4. 向量化
embeddings = self.embed(chunks)
# 5. 存储
self.store(chunks, embeddings)
return chunks
def chunk(self, doc, chunk_size=500, overlap=50):
"""
分块策略：
- 固定大小分块
- 语义分块
- 递归分块
"""
chunks = []
for i in range(0, len(doc), chunk_size - overlap):
chunk = doc[i:i + chunk_size]
chunks.append(chunk)
return chunks3.2 检索优化
# 混合检索
class HybridRetriever:
def retrieve(self, query, top_k=10):
# 关键词检索
bm25_results = self.bm25.search(query, top_k=top_k*2)
# 向量检索
vector_results = self.vector_db.search(
self.embed(query),
top_k=top_k*2
)
# 融合结果
fused = self.fuse(bm25_results, vector_results)
return fused[:top_k]
def fuse(self, results1, results2):
# 倒数排名融合（RRF）
scores = {}
for i, doc in enumerate(results1):
scores[doc.id] = scores.get(doc.id, 0) + 1/(i+1)
for i, doc in enumerate(results2):
scores[doc.id] = scores.get(doc.id, 0) + 1/(i+1)
return sorted(scores.items(), key=lambda x: x[1], reverse=True)3.3 重排序策略
# Cross-Encoder 重排序
class CrossEncoderReranker:
def rerank(self, query, documents, top_k=5):
# 计算 query-document 相关性分数
pairs = [[query, doc.text] for doc in documents]
scores = self.model.predict(pairs)
# 按分数排序
ranked = sorted(
zip(documents, scores),
key=lambda x: x[1],
reverse=True
)
return [doc for doc, score in ranked[:top_k]]四、评估体系
4.1 评估指标
# RAGAS 评估框架
from ragas import evaluate
from ragas.metrics import (
faithfulness, # 忠实度
answer_relevancy, # 答案相关性
context_precision, # 上下文精确度
context_recall # 上下文召回率
)
# 评估
results = evaluate(
dataset=test_dataset,
metrics=[
faithfulness,
answer_relevancy,
context_precision,
context_recall
]
)
print(results)4.2 A/B 测试
# 对比不同 RAG 配置
def ab_test_rag(query_set):
config_a = {"top_k": 3, "rerank": False}
config_b = {"top_k": 5, "rerank": True}
results_a = []
results_b = []
for query in query_set:
answer_a = rag_system.query(query, **config_a)
answer_b = rag_system.query(query, **config_b)
results_a.append(evaluate_answer(answer_a))
results_b.append(evaluate_answer(answer_b))
# 统计显著性检验
from scipy import stats
t_stat, p_value = stats.ttest_ind(results_a, results_b)
if p_value < 0.05:
print(f"配置 B 显著优于配置 A (p={p_value})")五、最佳实践
5.1 分块策略
# 推荐配置
chunk_config = {
"chunk_size": 500, # 500 tokens
"chunk_overlap": 50, # 10% 重叠
"separator": "\n\n", # 段落分隔
"length_function": len # token 计数
}
# 根据文档类型调整
if doc_type == "code":
chunk_size = 200 # 代码块较小
elif doc_type == "legal":
chunk_size = 1000 # 法律文档较长5.2 检索配置
# 生产环境推荐配置
retrieval_config = {
"top_k": 10, # 初检数量
"rerank_top_k": 5, # 重排序后数量
"score_threshold": 0.7, # 最低相关性阈值
"use_hybrid": True, # 混合检索
"rerank": True # 启用重排序
}5.3 性能优化
# 缓存策略
from functools import lru_cache
class CachedRAG:
@lru_cache(maxsize=1000)
def embed(self, text):
return self.model.encode(text)
@lru_cache(maxsize=100)
def retrieve(self, query_hash):
return self.db.search(query_hash)六、总结
RAG 架构核心要点：
| 模式 | 复杂度 | 效果 | 适用场景 |
|---|---|---|---|
| Naive RAG | 低 | 中 | MVP、简单场景 |
| Advanced RAG | 中 | 高 | 生产环境 |
| Modular RAG | 高 | 最高 | 企业级应用 |
RAG 是构建知识库问答系统的主流架构，关键是平衡检索质量和系统复杂度。