朴素RAG的致命缺陷
朴素RAG对每个问题都执行检索,这带来两个问题:简单问题浪费检索资源(“1+1等于几"不需要查文档),复杂问题单次检索不够(“对比三家公司五年财报"需要多次检索)。自适应RAG的核心思想是让模型自己决定:是否需要检索、检索几次、检索什么。
Self-RAG:自我反思的检索
核心机制
Self-RAG训练模型输出特殊的反思token来控制检索行为:
[Retrieve]:是否需要检索[Relevant]:检索结果是否相关[Supported]:生成内容是否被检索结果支持[Useful]:检索结果是否有用
工作流程
输入问题 → 模型判断[Retrieve] →
是 → 检索Top-K文档 → 模型判断[Relevant] →
过滤不相关文档 → 逐段生成 → [Supported]标注 → 输出
否 → 直接生成 → 输出
训练方法
Self-RAG的训练分为两阶段:
阶段1:训练 critic 模型
收集标注数据训练一个判断模型:
- 何时需要检索(基于问题类型)
- 文档是否相关(人工标注)
- 生成是否被支持(对照检查)
class CriticModel:
def __init__(self, base_model):
self.model = base_model
def should_retrieve(self, question):
# 简单事实问题不需要检索
# 需要最新信息的问题需要检索
# 需要引用来源的问题需要检索
prompt = f"判断以下问题是否需要检索外部信息:\n{question}"
return self.model.generate(prompt) == "yes"
def is_relevant(self, question, document):
prompt = f"判断文档是否与问题相关:\n问题:{question}\n文档:{document}"
return self.model.generate(prompt) == "relevant"
阶段2:训练生成模型
在生成模型中注入反思token的训练:
def self_rag_generate(question, retriever, model):
# Step 1: 判断是否检索
if model.should_retrieve(question):
docs = retriever.search(question, top_k=5)
# 过滤不相关文档
relevant_docs = [d for d in docs if model.is_relevant(question, d)]
else:
relevant_docs = []
# Step 2: 逐段生成
response = ""
for segment in model.generate_segments(question, relevant_docs):
# 检查是否被来源支持
support_level = model.check_support(segment, relevant_docs)
if support_level == "fully_supported":
response += segment + " [Supported]"
elif support_level == "partially_supported":
response += segment + " [Partially Supported]"
else:
# 无支持的内容触发重新检索
new_docs = retriever.search(segment, top_k=3)
response += segment + " [No Support - Retrieved]"
return response
效果对比
在多个基准上,Self-RAG相比朴素RAG:
| 指标 | 朴素RAG | Self-RAG | 提升 |
|---|---|---|---|
| 事实准确率 | 72% | 85% | +13% |
| 幻觉率 | 18% | 7% | -11% |
| 引用准确率 | 65% | 89% | +24% |
| 简单问题延迟 | 500ms | 100ms | -80% |
Adaptive-RAG:问题难度感知
自适应检索策略
Adaptive-RAG根据问题复杂度选择不同的检索策略:
class AdaptiveRAG:
def __init__(self, classifier, strategies):
self.classifier = classifier # 问题复杂度分类器
self.strategies = strategies # 检索策略集合
def answer(self, question):
# 1. 分类问题复杂度
complexity = self.classifier.classify(question)
# Level A: 简单事实查询
# Level B: 多跳推理
# Level C: 开放分析
# 2. 选择检索策略
if complexity == "A":
return self._no_retrieve(question)
elif complexity == "B":
return self._single_retrieve(question)
elif complexity == "C":
return self._multi_retrieve(question)
def _no_retrieve(self, question):
# 直接用模型参数化知识回答
return self.model.generate(question)
def _single_retrieve(self, question):
docs = self.retriever.search(question, top_k=3)
return self.model.generate(question, context=docs)
def _multi_retrieve(self, question):
# 分解为子问题
sub_questions = self.model.decompose(question)
all_context = []
for sub_q in sub_questions:
docs = self.retriever.search(sub_q, top_k=3)
all_context.extend(docs)
# 综合所有上下文生成答案
return self.model.generate(
question,
context=deduplicate(all_context)
)
问题分类器训练
def train_classifier():
"""训练问题复杂度分类器"""
training_data = [
("法国首都是哪里?", "A"),
("中国的GDP什么时候超过日本?", "B"),
("分析AI对就业市场的影响", "C"),
("1+1等于几?", "A"),
("比较Llama和Qwen在代码生成上的差异", "B"),
("如何构建一个可扩展的AI系统?", "C"),
]
# 使用LLM做few-shot分类
classifier = LLMClassifier(
examples=training_data,
model="gpt-4o"
)
return classifier
检索时机判断的细化
基于置信度的检索
模型可以先尝试回答,根据自身置信度决定是否需要检索:
def confidence_based_retrieve(question, model, retriever):
# 1. 先尝试直接回答
answer, confidence = model.generate_with_confidence(question)
if confidence > 0.9:
return answer # 高置信度,无需检索
# 2. 中等置信度,检索验证
if confidence > 0.5:
docs = retriever.search(question, top_k=3)
verified_answer = model.verify_and_revise(answer, docs)
return verified_answer
# 3. 低置信度,必须检索
docs = retriever.search(question, top_k=5)
return model.generate(question, context=docs)
基于生成过程的检索
在生成过程中,当模型遇到不确定的部分时触发检索:
def inline_retrieve_generate(question, model, retriever):
tokens = []
for token in model.stream_generate(question):
tokens.append(token)
# 检测不确定性信号
current_text = "".join(tokens)
if detect_uncertainty(current_text):
# 触发检索
docs = retriever.search(current_text, top_k=3)
# 注入检索结果
model.inject_context(docs)
if token == "<END>":
break
return "".join(tokens)
def detect_uncertainty(text):
"""检测模型不确定的信号"""
signals = [
"我不确定", "可能是", "据我所知",
"大约", "似乎", "可能是"
]
return any(signal in text[-50:] for signal in signals)
迭代检索
IRCoT:交错推理与检索
def ircot(question, model, retriever, max_steps=5):
"""Interleaving Retrieval with Chain-of-Thought"""
reasoning = ""
all_docs = []
for step in range(max_steps):
# 1. 基于当前推理生成下一步
next_thought = model.generate(
f"问题:{question}\n已有推理:{reasoning}\n下一步推理:"
)
# 2. 基于推理生成检索查询
search_query = model.generate(
f"基于以下推理,生成搜索查询:{next_thought}"
)
# 3. 检索
new_docs = retriever.search(search_query, top_k=3)
all_docs.extend(new_docs)
# 4. 更新推理
reasoning += next_thought
# 5. 判断是否可以回答
if model.can_answer(question, reasoning, all_docs):
break
return model.generate_answer(question, reasoning, all_docs)
评估自适应RAG
评估维度
- 检索效率:简单问题是否避免了不必要的检索
- 答案质量:复杂问题是否通过多次检索提升了质量
- 延迟分布:不同复杂度问题的延迟差异
- 成本节约:相比always-retrieve减少了多少API调用
基准测试
在Natural Questions + HotpotQA混合基准上:
| 方法 | 准确率 | 平均检索次数 | 平均延迟 |
|---|---|---|---|
| No-RAG | 45% | 0 | 200ms |
| Always-RAG | 68% | 1.0 | 600ms |
| Self-RAG | 72% | 0.4 | 350ms |
| Adaptive-RAG | 74% | 0.5 | 380ms |
| IRCoT | 78% | 2.3 | 1200ms |
结语
自适应检索代表了RAG技术从"一刀切"到"智能化"的演进。让模型自己决定何时检索、检索什么、检索几次,不仅提升了答案质量,还显著降低了推理成本。随着模型元认知能力的提升,自适应RAG将成为生产系统的标配——不是所有问题都需要查资料,聪明的模型知道什么时候该"翻书”。