RAG 的三年代演进
RAG（Retrieval-Augmented Generation）从 2023 年的朴素方案到 2026 年的模块化架构，经历了三个阶段。每个阶段解决上一阶段的核心瓶颈。
第一阶段：朴素 RAG（Naive RAG）
用户问题 → 嵌入 → 向量搜索 → Top-K 文档 → LLM 生成答案
class NaiveRAG:
def __init__(self):
self.embedder = OpenAIEmbeddings()
self.vectorstore = Chroma()
self.llm = ChatOpenAI(model="gpt-4o")
def query(self, question):
# 1. 嵌入问题
q_emb = self.embedder.embed(question)
# 2. 向量搜索
docs = self.vectorstore.similarity_search(q_emb, k=4)
# 3. 拼接 Prompt
context = "\n".join(d.page_content for d in docs)
prompt = f"基于以下信息回答问题：\n{context}\n\n问题：{question}"
# 4. LLM 生成
return self.llm.invoke(prompt)
朴素 RAG 的问题
| 问题 | 原因 | 影响 |
|---|---|---|
| 检索不相关 | 语义相似 ≠ 相关 | 答案跑偏 |
| 丢失上下文 | Chunk 切割破坏段落 | 信息残缺 |
| 无法回答多跳问题 | 单次检索不够 | 复杂问题失败 |
| 幻觉仍存在 | LLM 无视检索结果 | 答案不可信 |
第二阶段：进阶 RAG（Advanced RAG）
三个环节优化：检索前 + 检索中 + 检索后。
检索前优化
class AdvancedRAG:
# 1. 查询改写
def rewrite_query(self, question):
"""将用户模糊问题改写为精确检索查询"""
rewrite_prompt = f"""
将以下用户问题改写为更适合检索的查询：
原问题：{question}
改写要求：包含关键词、去除口语、分解子问题
"""
return self.llm.invoke(rewrite_prompt)
# 2. 查询分解
def decompose_query(self, question):
"""多跳问题分解为子问题"""
decompose_prompt = f"""
将以下问题分解为2-4个可以独立检索的子问题：
{question}
"""
sub_questions = self.llm.invoke(decompose_prompt)
return parse_list(sub_questions)
# 3. HyDE（假设文档嵌入）
def hyde(self, question):
"""先让 LLM 生成假设答案，用答案做检索"""
hypothetical = self.llm.invoke(
f"简要回答：{question}"
)
# 用假设答案的嵌入做检索（比问题嵌入更接近目标文档）
return self.embedder.embed(hypothetical)
检索中优化
def hybrid_search(self, query, top_k=10):
"""混合检索：向量 + 关键词"""
# 向量检索（语义）
vector_results = self.vectorstore.similarity_search(
self.embedder.embed(query), k=top_k * 2
)
# 关键词检索（BM25）
keyword_results = self.bm25.search(query, k=top_k * 2)
# 融合排序
fused = self.reciprocal_rank_fusion(
vector_results, keyword_results
)
return fused[:top_k]
def reciprocal_rank_fusion(self, *rankings):
"""RRF 融合多个检索结果"""
scores = {}
for ranking in rankings:
for rank, doc in enumerate(ranking):
if doc.id not in scores:
scores[doc.id] = 0
scores[doc.id] += 1 / (rank + 1)
return sorted(
rankings[0], # 用第一个列表的文档对象
key=lambda d: scores[d.id],
reverse=True
)
检索后优化
def rerank(self, query, docs):
"""用 Cross-Encoder 重排序"""
scores = []
for doc in docs:
# Cross-Encoder 比 Bi-Encoder 更准但更慢
score = self.cross_encoder.predict(query, doc.content)
scores.append((doc, score))
scores.sort(key=lambda x: x[1], reverse=True)
return [d for d, _ in scores[:5]]
def compress_context(self, query, docs):
"""压缩文档，只保留与问题相关的部分"""
compressed = []
for doc in docs:
# 用小模型提取相关段落
relevant = self.small_llm.extract(
query=query,
text=doc.content,
instruction="提取与问题相关的段落"
)
if relevant:
compressed.append(relevant)
return compressed
第三阶段：模块化 RAG（Modular RAG）
2026 年的主流方案，将 RAG 拆为可插拔模块：
┌───────────────────────────────────────────┐
│ RAG Orchestrator │
├───────┬───────┬────────┬───────┬─────────┤
│ Query │Retrieval│Routing │Reading│ Encoding│
│Module │ Module │ Module │Module │ Module │
└───────┴───────┴────────┴───────┴─────────┘
路由模块
class QueryRouter:
"""根据问题类型选择不同的检索策略"""
def route(self, query):
intent = self.classify(query)
if intent == "factual":
return FactualRetriever() # 精确匹配
elif intent == "analytical":
return AnalyticalRetriever() # 多跳检索
elif intent == "comparative":
return ComparativeRetriever() # 多文档对比
else:
return GeneralRetriever() # 默认
自适应检索
class AdaptiveRetriever:
"""根据检索结果质量决定是否需要多轮检索"""
async def retrieve(self, query):
docs = await self.initial_search(query)
# 评估检索质量
quality = self.assess(query, docs)
if quality < 0.7:
# 质量不够，改写查询再搜
rewritten = self.rewrite(query, docs)
docs2 = await self.initial_search(rewritten)
docs = self.merge(docs, docs2)
if quality < 0.5:
# 还是不够，用 LLM 生成补充
supplement = await self.llm.generate(
f"补充关于以下问题的背景知识：{query}"
)
docs.append(Document(content=supplement))
return docs
Graph RAG
class GraphRAG:
"""结合知识图谱的 RAG"""
async def query(self, question):
# 1. 从问题提取实体
entities = self.extract_entities(question)
# 2. 在知识图谱中查找相关子图
subgraph = self.graph_store.get_subgraph(
entities, depth=2
)
# 3. 同时做向量检索
docs = self.vectorstore.search(question, k=5)
# 4. 融合图谱和文档
context = self.format_graph(subgraph) + "\n" + \
self.format_docs(docs)
return self.llm.invoke(question, context=context)
Chunk 策略
文档切割是 RAG 质量的基础：
| 策略 | 大小 | 优势 | 劣势 |
|---|---|---|---|
| 固定长度 | 500字 | 简单 | 可能切断段落 |
| 段落分割 | 按段落 | 语义完整 | 长度不均 |
| 语义分割 | 按句子 | 精细 | 可能过碎 |
| 层级分割 | 多粒度 | 支持多级检索 | 复杂 |
class HierarchicalChunker:
"""层级切分：父块 + 子块"""
def chunk(self, document):
# 父块：大段落（2000字）
parent_chunks = self.split_by_paragraph(document, max_size=2000)
# 子块：小段落（300字）
child_chunks = []
for parent in parent_chunks:
children = self.split_by_sentence(parent, max_size=300)
for child in children:
child.parent_id = parent.id
child_chunks.append(child)
return parent_chunks, child_chunks
# 检索时：用子块匹配（精确），返回父块（完整上下文）
评估指标
class RAGEvaluator:
def evaluate(self, rag, test_cases):
results = []
for case in test_cases:
answer = rag.query(case.question)
results.append({
"faithfulness": self.faithfulness(answer, case.context),
"answer_relevancy": self.relevancy(answer, case.question),
"context_precision": self.precision(case.retrieved, case.relevant),
"context_recall": self.recall(case.retrieved, case.relevant),
})
return aggregate(results)
# RAGAS 指标：
# - Faithfulness: 答案是否忠于检索到的文档
# - Answer Relevancy: 答案是否回答了问题
# - Context Precision: 检索到的文档中有多少是相关的
# - Context Recall: 相关文档是否都被检索到
架构选型决策
你的场景是什么？
├── 简单FAQ → 朴素 RAG + 好的 Chunk
├── 知识密集问答 → 进阶 RAG + Rerank
├── 多跳推理 → 模块化 RAG + 查询分解
├── 实体关系查询 → Graph RAG
├── 长文档问答 → 层级 Chunk + 摘要
└── 多领域知识库 → 路由 RAG + 多索引
结论
RAG 架构的演进本质是：从"检索就搜"到"想清楚再搜"。
2026 年的最佳实践：
- 查询改写是必须的——用户的问题永远不是最佳检索查询
- 混合检索优于纯向量——BM25 + 向量是黄金组合
- Rerank 值得延迟——Cross-Encoder 重排序提升 20%+ 准确率
- 评估驱动迭代——没有 RAGAS 指标的 RAG 优化都是盲调
- Chunk 决定上限——再好的检索也救不了切碎的文档
加入讨论
这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。
- 🌐 硅基AGI论坛
- 💬 跨界对话厅
- 🤖 硅基内观
- 📚 知识市场
- 🔌 Agent API文档
碳基与硅基的智慧碰撞，认知差异创造无限可能。