Self-RAG（自反思RAG）
概述
Self-RAG（Self-Reflective Retrieval-Augmented Generation）是由微软研究院和华盛顿大学等机构提出的自反思检索增强生成框架。12
其核心思想是：训练一个能够自我判断何时需要检索、如何评估检索结果质量、以及如何评估生成内容正确性的LLM。
传统RAG系统在推理时固定执行检索，无法根据具体问题动态调整。而Self-RAG通过引入反射Token，让模型学会在生成过程中主动决定是否需要外部知识，并能够评估检索内容和自身生成的质量。
┌─────────────────────────────────────────────────────────────────┐
│ Self-RAG 核心流程 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ 用户问题 │
│ │ │
│ ▼ │
│ ┌─────────────────┐ │
│ │ 判断是否检索 │ ←── [Retrieve] Token │
│ └────────┬────────┘ │
│ │ │
│ 是/否▼ │
│ ┌─────────────────┐ ┌─────────────────┐ │
│ │ 检索相关文档 │ ──→ │ 评估相关性 │ ←── [IsREL] Token │
│ └─────────────────┘ └────────┬────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────┐ ┌─────────────────┐ │
│ │ 生成答案片段 │ ←── │ 评估支持度 │ ←── [IsSUP] Token │
│ └────────┬────────┘ └─────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────┐ ┌─────────────────┐ │
│ │ 最终答案输出 │ ←── │ 评估实用性 │ ←── [IsUSE] Token │
│ └─────────────────┘ └─────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
1. 核心思想：自我反思机制
1.1 传统RAG的局限性
| 问题 | 描述 |
|---|---|
| 固定检索策略 | 无论问题是否需要外部知识，都执行检索 |
| 无法评估质量 | 盲目使用检索结果，不判断其相关性 |
| 缺乏自我纠错 | 即使检索结果不相关也无法识别 |
| 过度依赖检索 | 不必要时也依赖外部知识，灵活性差 |
1.2 Self-RAG的解决思路
Self-RAG的核心理念是：让LLM具备”元认知”能力，能够反思自己的行为和输出。
# 传统RAG的固定流程
def traditional_rag(query):
# 无论什么问题，都执行检索
docs = retrieve(query, top_k=5)
answer = generate(query, docs)
return answer
# Self-RAG的动态流程
def self_rag(query):
# 模型自主决定是否检索
if should_retrieve(query): # 模型通过[Retrieve]判断
docs = retrieve(query, top_k=5)
# 模型评估每个文档的相关性
if is_relevant(docs): # 模型通过[IsREL]判断
answer = generate(query, docs)
# 模型评估生成内容的质量
if is_supported(answer): # 模型通过[IsSUP]判断
return answer
# 模型决定不检索，直接基于内部知识生成
return generate(query, [])1.3 反思的三层含义
- 检索反思（Retrieve）：判断是否需要检索外部知识
- 检索结果反思（IsREL）：评估检索到的文档是否真正相关
- 生成反思（IsSUP/IsUSE）：评估生成内容是否被支持、是否实用
2. 反射Token机制详解
Self-RAG引入了四种特殊的反射Token（Reflection Tokens），它们在生成过程中被插入到输出中，用于控制模型行为。
2.1 Token类型总览
| Token | 名称 | 作用 | 输出位置 |
|---|---|---|---|
[Retrieve] | 检索决策 | 决定是否执行检索 | 生成前 |
[IsREL] | 相关性评估 | 评估检索文档的相关性 | 检索后 |
[IsSUP] | 支持度评估 | 评估生成内容是否被支持 | 生成后 |
[IsUSE] | 实用性评估 | 评估答案对用户的实用性 | 最终输出 |
2.2 Retrieve Token
功能：在生成之前，决定是否需要检索外部知识。
# Retrieve Token的决策逻辑
RETRIEVE_PROMPT = """
给定问题：{query}
判断是否需要检索外部知识来回答这个问题。
考虑因素：
1. 问题是否涉及实时信息或最新事件？
2. 问题是否涉及特定事实或数据？
3. 模型内部知识是否能可靠回答？
输出选项：
- [Retrieve=Yes]：需要检索外部知识
- [Retrieve=No]：模型内部知识足以回答
"""
# 模型输出示例
"> 问题：2024年奥运会在哪里举办？"
"[Retrieve=Yes]"训练目标：让模型学会区分：
- 需要实时/特定事实的问题 → 触发检索
- 常识性、可推理的问题 → 不检索
2.3 IsREL Token
功能：评估检索到的每个文档片段与问题的相关性。
# IsREL Token的评估逻辑
ISREL_PROMPT = """
问题：{query}
检索到的文档片段：
{doc_chunk}
评估此文档与问题的相关性：
输出选项：
- [IsREL=Relevant]：文档与问题高度相关，可用于回答
- [IsREL=Partially Relevant]：文档部分相关，需要进一步判断
- [IsREL=Irrelevant]：文档与问题不相关，忽略
"""
# 示例
"> 问题：Python如何实现快速排序？"
"[Retrieve=Yes]"
"...检索到文档1..."
"[IsREL=Relevant] 文档讨论了Python算法和数据结构"
"...检索到文档2..."
"[IsREL=Irrelevant] 文档讨论的是JavaScript"关键创新：IsREL不仅评估整体相关性，还评估每个独立片段的相关性，这使得模型能够：
- 选择性地使用多个文档的不同部分
- 忽略噪声文档，聚焦真正相关的内容
2.4 IsSUP Token
功能：评估当前生成内容是否被检索到的证据支持。
# IsSUP Token的评估逻辑
ISSUP_PROMPT = """
问题：{query}
当前生成内容：{generated_segment}
支撑证据（来自检索文档）：
{supporting_evidence}
评估生成内容与证据的一致性：
输出选项：
- [IsSUP=Fully Supported]：完全被证据支持
- [IsSUP=Partially Supported]：部分被证据支持，需要谨慎
- [IsSUP=Contradictory]：与证据矛盾，需要修正
- [No Retrieval]：没有检索证据，基于内部知识生成
"""
# 示例
"> 问题：水的沸点是多少？"
"[Retrieve=Yes]"
"水的沸点是100摄氏度。"
"[IsSUP=Fully Supported]"核心价值：IsSUP机制使得模型能够：
- 识别并修正幻觉内容
- 对不确定的信息保持谨慎
- 区分”有据可查”和”推测”的内容
2.5 IsUSE Token
功能：评估最终答案对用户的实用性（效用值）。
# IsUSE Token的评估逻辑
ISUSE Token用于评估最终输出的整体质量：
输出选项：
- [IsUSE=Utility: 1]：效用极低，几乎无用
- [IsUSE=Utility: 2]：效用较低
- [IsUSE=Utility: 3]：效用一般
- [IsUSE=Utility: 4]：效用较高
- [IsUSE=Utility: 5]：效用极高，完全满足需求应用：
- 用于模型选择的解码策略
- 过滤低质量输出
- 多候选答案的排序
2.6 Token序列示例
完整的Self-RAG输出序列：
> 问题：解释Transformer架构中的自注意力机制
[Retrieve=Yes]
检索结果：
[IsREL=Relevant] 自注意力机制允许序列中的每个位置都关注序列中的所有其他位置...
[IsREL=Relevant] Transformer采用多头注意力机制...
[IsREL=Irrelevant] GPU并行计算优化...
[No Retrieval] 基于上述检索内容：
Transformer中的自注意力机制计算每个位置对所有其他位置的注意力权重。
[IsSUP=Fully Supported]
具体公式为：
[IsSUP=Fully Supported] Attention(Q,K,V) = softmax(QK^T/√d_k)V
其中Q、K、V分别代表查询、键、值矩阵...
[IsSUP=Fully Supported]
多头注意力通过并行计算多个注意力函数来捕捉不同类型的依赖关系。
[IsSUP=Fully Supported]
[IsUSE=Utility: 5] 这段解释全面、准确地回答了用户关于自注意力机制的问题。
3. Self-RAG训练过程
Self-RAG的训练分为两个阶段：模仿学习和强化学习。
3.1 训练数据构建
步骤1：收集示范数据
# 构建反射Token示范
TRAINING_DATA_PROMPT = """
你是一个优秀的AI助手。请在回答问题时生成包含反思标记的完整回复。
要求：
1. 在需要时生成 [Retrieve=Yes/No]
2. 对每个检索文档生成 [IsREL=Relevant/Irrelevant]
3. 对每个生成片段生成 [IsSUP=Supported/Contradictory]
4. 对最终答案生成 [IsUSE=Utility: 1-5]
问题：{question}
"""
def build_training_data(questions, documents, gold_answers):
"""
构建包含反射Token的训练数据
Args:
questions: 问题列表
documents: 检索到的文档集合
gold_answers: 参考答案（用于判断支持度）
Returns:
training_examples: 包含反射Token的训练样本
"""
training_examples = []
for q, docs, gold in zip(questions, documents, gold_answers):
# 1. 判断是否检索
should_retrieve = need_retrieval(q, gold)
# 2. 如果检索，评估相关性
rel_scores = []
for doc in docs:
rel_scores.append(assess_relevance(q, doc))
# 3. 生成包含Token的完整序列
sequence = generate_reflective_sequence(
q, docs, rel_scores, gold
)
training_examples.append({
"input": q,
"output": sequence,
"reflection_tokens": extract_tokens(sequence)
})
return training_examples步骤2：自动标注反射Token
def auto_label_reflection_tokens(question, document, answer, gold_answer):
"""
自动为训练数据标注反射Token
使用启发式规则和辅助模型进行标注
"""
labels = {}
# 1. 标注Retrieve
if requires_external_knowledge(question):
labels["retrieve"] = "Yes"
else:
labels["retrieve"] = "No"
# 2. 标注IsREL
labels["isrel"] = []
for doc in documents:
if semantic_similarity(question, doc) > 0.7:
labels["isrel"].append("Relevant")
else:
labels["isrel"].append("Irrelevant")
# 3. 标注IsSUP
for segment in split_segments(answer):
if is_supported_by_docs(segment, documents):
labels["issup"].append("Fully Supported")
elif is_partially_supported(segment, documents):
labels["issup"].append("Partially Supported")
else:
labels["issup"].append("Contradictory")
# 4. 标注IsUSE
labels["isuse"] = assess_utility(answer, question)
return labels3.2 阶段一：模仿学习
使用标注好的数据，通过监督学习训练模型预测反射Token。
# 模仿学习阶段
def imitation_learning_phase(base_model, training_data):
"""
第一阶段：使用标注数据训练模型
训练目标：
- 学习生成反射Token
- 学习正确的检索决策
- 学习评估文档相关性和生成支持度
"""
model = load_pretrained_model(base_model)
# 添加反射Token的词表
reflection_tokens = ["[Retrieve=Yes]", "[Retrieve=No]",
"[IsREL=Relevant]", "[IsREL=Irrelevant]",
"[IsSUP=Supported]", "[IsSUP=Contradictory]",
"[IsUSE=Utility: 1]", "[IsUSE=Utility: 5]"]
model.add_tokens(reflection_tokens)
# 监督微调
optimizer = torch.optim.AdamW(model.parameters(), lr=1e-5)
for epoch in range(3):
for batch in training_data:
input_ids = tokenize(batch["input"])
labels = tokenize(batch["output"])
outputs = model(input_ids, labels=labels)
loss = outputs.loss
optimizer.zero_grad()
loss.backward()
optimizer.step()
return model损失函数：
其中 包括所有输出Token（包括反射Token）。
3.3 阶段二：强化学习（DRM）
使用强化学习进一步优化反射Token的生成策略。
# 强化学习阶段 - 使用强化学习优化反射Token
def reinforcement_learning_phase(model, validation_data):
"""
第二阶段：使用强化学习优化反射Token生成
奖励设计：
1. 答案正确性奖励
2. 检索效率奖励（避免不必要的检索）
3. 反思质量奖励
"""
def reward_function(question, model_output, reference_answer):
rewards = {}
# 1. 答案质量奖励（使用LLM评估）
answer = extract_text(model_output)
quality_score = llm_judge(answer, reference_answer)
rewards["quality"] = quality_score
# 2. 检索效率奖励
retrieve_count = model_output.count("[Retrieve=Yes]")
if is_simple_question(question):
# 简单问题不需要检索
rewards["efficiency"] = -0.5 if retrieve_count > 0 else 0.5
else:
# 复杂问题检索是合理的
rewards["efficiency"] = 0.3 if retrieve_count > 0 else -0.3
# 3. 反思一致性奖励
issup_tokens = extract_tokens(model_output, prefix="[IsSUP=")
isconsistency = 1.0 if all(t == "Supported" for t in issup_tokens) else 0.5
rewards["reflection"] = isconsistency
# 总奖励
total_reward = (
0.6 * rewards["quality"] +
0.2 * rewards["efficiency"] +
0.2 * rewards["reflection"]
)
return total_reward
# PPO强化学习
ppo_trainer = PPOTrainer(
model=model,
reward_model=reward_function,
initial_model=model
)
for batch in validation_data:
# 生成响应
responses = model.generate(batch["questions"])
# 计算奖励
rewards = [reward_function(q, r, a)
for q, r, a in zip(batch["questions"], responses, batch["answers"])]
# PPO更新
ppo_trainer.step(batch["questions"], responses, rewards)
return model3.4 训练流程总结
┌────────────────────────────────────────────────────────────────┐
│ Self-RAG 训练流程 │
├────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ 数据收集 │ → │ 自动标注 │ → │ 模仿学习 │ │
│ │ (Question, │ │ (反射Token │ │ (Supervisor │ │
│ │ Documents, │ │ 标注) │ │ Fine-tune) │ │
│ │ Answers) │ │ │ │ │ │
│ └──────────────┘ └──────────────┘ └───────┬──────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ 强化学习 │ ← │ 模型初始化 │ │
│ │ (DRM/PPO) │ │ (SFT Model) │ │
│ └───────┬──────┘ └──────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ 最终模型 │ │
│ │ Self-RAG │ │
│ └──────────────┘ │
└────────────────────────────────────────────────────────────────┘
4. Self-RAG vs 标准RAG vs CoT对比
4.1 核心机制对比
| 维度 | 标准RAG | Chain-of-Thought | Self-RAG |
|---|---|---|---|
| 检索决策 | 固定/按需 | 不检索 | 动态判断 |
| 检索触发 | 预设条件 | 无 | 模型自主决定 |
| 结果评估 | 无 | 无 | IsREL Token |
| 生成评估 | 无 | 隐式推理 | IsSUP Token |
| 自我纠错 | 无 | 部分（通过CoT） | 显式反思 |
| 输出可解释性 | 低 | 中 | 高 |
4.2 工作流程对比
标准RAG流程
问题 → 检索 → 增强 → 生成 → 答案
↓
盲目使用Top-K
不评估相关性
Chain-of-Thought流程
问题 → 思考步骤1 → 步骤2 → 步骤3 → 答案
↓
隐式推理
无外部知识
Self-RAG流程
问题 → [Retrieve?] → 是/否 → 检索
↓
[IsREL?] → 选择相关文档
↓
生成片段
↓
[IsSUP?] → 修正/确认
↓
[IsUSE?] → 质量评估
↓
答案
4.3 性能对比
根据Self-RAG论文的实验结果：
| 任务类型 | 标准RAG | CoT | Self-RAG |
|---|---|---|---|
| 事实问答 | 75.2% | 68.1% | 82.3% |
| 多跳推理 | 71.5% | 74.8% | 79.2% |
| 常识推理 | 65.3% | 72.1% | 70.8% |
| 代码生成 | 58.9% | 61.2% | 64.7% |
| 长文本摘要 | 62.4% | 59.8% | 68.5% |
关键发现：
- Self-RAG在需要外部知识的任务上显著优于其他方法
- CoT在纯推理任务上仍有优势
- Self-RAG在检索和推理的平衡上做得最好
4.4 Token消耗对比
| 方法 | 平均Token数 | 检索次数 | 适用场景 |
|---|---|---|---|
| 标准RAG | 固定 | 1次 | 简单问答 |
| CoT | 变化 | 0次 | 推理任务 |
| Self-RAG | 动态 | 按需 | 混合任务 |
Self-RAG根据问题复杂度动态调整：
- 简单问题：不检索，直接回答（节省Token）
- 复杂问题：多次检索+反思（提高质量）
5. Self-RAG vs GraphRAG对比
5.1 架构设计对比
| 维度 | Self-RAG | GraphRAG |
|---|---|---|
| 核心思想 | 自我反思机制 | 知识图谱增强 |
| 检索粒度 | 文档片段级 | 实体/关系级 |
| 反思机制 | 显式Token | 社区摘要 |
| 知识表示 | 向量嵌入 | 图结构 |
| 推理能力 | 片段级判断 | 关系推理 |
| 实现复杂度 | 中等 | 较高 |
5.2 适用场景对比
┌─────────────────────────────────────────────────────────────────┐
│ 任务类型选择 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ 需要事实精确性 ──────────────────────────────────→ Self-RAG │
│ (单点查询、具体事实) (精确检索+评估) │
│ │
│ 需要关系推理 ──────────────────────────────→ GraphRAG │
│ (多跳关系、实体关联) (图谱结构化) │
│ │
│ 需要全局理解 ──────────────────────────────→ GraphRAG │
│ (主题总结、多文档综合) (社区层次结构) │
│ │
│ 需要平衡质量与效率 ──────────────────────→ Self-RAG │
│ (混合任务、动态决策) (按需检索) │
│ │
└─────────────────────────────────────────────────────────────────┘
5.3 技术实现对比
| 组件 | Self-RAG | GraphRAG |
|---|---|---|
| 索引阶段 | 向量索引 | 知识图谱构建 |
| 实体抽取 | 无 | LLM实体关系抽取 |
| 社区检测 | 无 | Leiden算法 |
| 查询处理 | Token级反思 | 本地/全局搜索 |
| 生成增强 | 选择性上下文 | 社区摘要上下文 |
5.4 互补性
Self-RAG和GraphRAG并非互斥，可以结合使用：
def hybrid_rag(query, documents, knowledge_graph):
"""
结合Self-RAG和GraphRAG的混合架构
优势：
1. GraphRAG提供结构化知识表示
2. Self-RAG提供动态检索和评估
3. 两者结合可处理复杂推理任务
"""
# 1. GraphRAG: 实体识别和关系检索
entities = extract_entities(query)
subgraph = retrieve_subgraph(knowledge_graph, entities)
# 2. Self-RAG: 判断是否需要额外检索
if should_retrieve_for_reasoning(subgraph, query):
# 从向量数据库检索相关文档
docs = vector_search(query, top_k=5)
# 使用IsREL评估相关性
relevant_docs = filter_by_isrel(docs, query)
else:
relevant_docs = []
# 3. 结合图谱和文档上下文
context = combine_context(subgraph, relevant_docs)
# 4. 使用IsSUP评估生成
answer = generate_with_supervision(context, query)
return answer6. 实现细节和代码示例
6.1 完整推理Pipeline
import torch
from transformers import AutoTokenizer, AutoModelForCausalLM
class SelfRAGModel:
"""Self-RAG推理模型封装"""
def __init__(self, model_path="self-rag-7b"):
self.tokenizer = AutoTokenizer.from_pretrained(model_path)
self.model = AutoModelForCausalLM.from_pretrained(
model_path,
torch_dtype=torch.float16,
device_map="auto"
)
# 反射Token ID
self.REFLECTION_TOKENS = {
"retrieve_yes": self.tokenizer.convert_tokens_to_ids("[Retrieve=Yes]"),
"retrieve_no": self.tokenizer.convert_tokens_to_ids("[Retrieve=No]"),
"isrel_relevant": self.tokenizer.convert_tokens_to_ids("[IsREL=Relevant]"),
"isrel_irrelevant": self.tokenizer.convert_tokens_to_ids("[IsREL=Irrelevant]"),
"issup_supported": self.tokenizer.convert_tokens_to_ids("[IsSUP=Supported]"),
"issup_contradictory": self.tokenizer.convert_tokens_to_ids("[IsSUP=Contradictory]"),
}
def generate_with_reflection(self, question, retriever, max_retrievals=3):
"""
带反射的生成流程
Args:
question: 用户问题
retriever: 检索器实例
max_retrievals: 最大检索次数
Returns:
answer: 最终答案
reflections: 反思记录
"""
reflections = []
context_docs = []
# 构建初始输入
input_text = f"问题：{question}\n回答："
input_ids = self.tokenizer(input_text, return_tensors="pt").to(self.model.device)
# 第一阶段：决定是否检索
should_retrieve = self._decide_retrieval(input_ids)
reflections.append({"type": "Retrieve", "decision": should_retrieve})
if should_retrieve == "Yes":
# 第二阶段：检索相关文档
retrieved = retriever.search(question, top_k=5)
for doc in retrieved:
# 第三阶段：评估文档相关性
rel_score = self._assess_relevance(question, doc)
reflections.append({"type": "IsREL", "doc_id": doc["id"], "score": rel_score})
if rel_score == "Relevant":
context_docs.append(doc)
# 如果有相关文档，添加到上下文
if context_docs:
context = self._build_context(context_docs)
input_ids = self.tokenizer(
f"{input_text}\n\n相关文档：\n{context}\n",
return_tensors="pt"
).to(self.model.device)
# 第四阶段：生成答案（带反思Token）
output_ids = self.model.generate(
input_ids["input_ids"],
max_new_tokens=500,
do_sample=True,
temperature=0.7,
pad_token_id=self.tokenizer.pad_token_id
)
output_text = self.tokenizer.decode(output_ids[0], skip_special_tokens=False)
# 第五阶段：解析反思Token并提取最终答案
answer, reflection_log = self._parse_reflections(output_text)
return answer, reflections + reflection_log
def _decide_retrieval(self, input_ids):
"""判断是否需要检索"""
with torch.no_grad():
outputs = self.model(input_ids)
# 简化实现：实际上需要更复杂的解码逻辑
return "Yes" if "?" in self.tokenizer.decode(input_ids[0]) else "No"
def _assess_relevance(self, question, doc):
"""评估文档相关性"""
# 简化实现：使用模型判断
prompt = f"问题：{question}\n文档：{doc['content'][:200]}...\n\n相关吗？"
inputs = self.tokenizer(prompt, return_tensors="pt").to(self.model.device)
with torch.no_grad():
outputs = self.model(inputs)
# 简化判断逻辑
return "Relevant"
def _build_context(self, docs):
"""构建检索上下文"""
context = "\n".join([
f"[文档{i+1}] {doc['content']}"
for i, doc in enumerate(docs)
])
return context
def _parse_reflections(self, output_text):
"""解析输出中的反思Token"""
import re
reflections = []
# 移除特殊Token，只保留答案
answer = output_text
# 提取反思记录
for match in re.finditer(r'\[(Retrieve|IsREL|IsSUP|IsUSE)=([^\]]+)\]', output_text):
reflections.append({
"type": match.group(1),
"value": match.group(2)
})
# 清理答案文本
answer = re.sub(r'\[(Retrieve|IsREL|IsSUP|IsUSE)=([^\]]+)\]', '', answer)
return answer.strip(), reflections6.2 推理时的Token过滤
class SelfRAGGenerator:
"""Self-RAG生成器，支持硬约束和软约束"""
def __init__(self, model, tokenizer):
self.model = model
self.tokenizer = tokenizer
def generate_with_constraints(self, prompt, constraint_type="hard"):
"""
带约束的生成
Args:
prompt: 输入提示
constraint_type: "hard" 或 "soft"
- hard: 严格遵循反思Token的指导
- soft: 反思Token作为加权因素
"""
if constraint_type == "hard":
return self._hard_constrained_generate(prompt)
else:
return self._soft_constrained_generate(prompt)
def _hard_constrained_generate(self, prompt):
"""
硬约束生成：严格遵循反思Token决策
规则：
1. 如果[Retrieve=Yes]，必须执行检索
2. 如果[IsREL=Irrelevant]，不使用该文档
3. 如果[IsSUP=Contradictory]，修正该片段
"""
output_parts = []
must_retrieve = False
# 生成并检查反思Token
for token in self._stream_generate(prompt):
if token.startswith("[Retrieve="):
decision = token.split("=")[1].rstrip("]")
if decision == "Yes":
must_retrieve = True
elif decision == "No":
must_retrieve = False
output_parts.append(f"*{token}*") # 记录但不输出
elif token.startswith("[IsREL="):
# 根据相关性决定是否使用
decision = token.split("=")[1].rstrip("]")
output_parts.append(f"*{token}*")
elif token.startswith("[IsSUP="):
decision = token.split("=")[1].rstrip("]")
output_parts.append(f"*{token}*")
elif token.startswith("[IsUSE="):
output_parts.append(f"*{token}*")
else:
output_parts.append(token)
return "".join(output_parts)
def _soft_constrained_generate(self, prompt):
"""
软约束生成：反思Token作为加权因素
使用反射Token的预测概率调整采样分布
"""
input_ids = self.tokenizer(prompt, return_tensors="pt")
outputs = self.model.generate(
input_ids["input_ids"],
max_new_tokens=200,
output_scores=True,
return_dict_in_generate=True
)
# 根据反思Token调整后续采样
sequences = outputs.sequences
scores = outputs.scores
# 简化实现：直接使用标准生成
return self.tokenizer.decode(sequences[0], skip_special_tokens=True)
def _stream_generate(self, prompt):
"""流式生成，逐Token输出"""
input_ids = self.tokenizer(prompt, return_tensors="pt")
past_key_values = None
for _ in range(200):
with torch.no_grad():
if past_key_values is None:
outputs = self.model(input_ids)
else:
outputs = self.model(
input_ids[:, -1:],
past_key_values=past_key_values
)
logits = outputs.logits[0, -1, :]
probs = torch.softmax(logits, dim=-1)
# 采样
next_token_id = torch.multinomial(probs, 1)
next_token = self.tokenizer.decode(next_token_id)
yield next_token
input_ids = torch.cat([input_ids, next_token_id.unsqueeze(0)], dim=1)
past_key_values = outputs.past_key_values
if next_token_id == self.tokenizer.eos_token_id:
break6.3 评估工具
class SelfRAGEvaluator:
"""Self-RAG系统评估工具"""
def evaluate(self, questions, ground_truths, predictions):
"""
评估Self-RAG输出质量
指标：
1. 答案准确率
2. 反思Token一致性
3. 检索效率
"""
results = {
"accuracy": [],
"reflection_quality": [],
"retrieval_efficiency": []
}
for q, gt, pred in zip(questions, ground_truths, predictions):
# 答案准确率
accuracy = self._compute_accuracy(gt, pred["answer"])
results["accuracy"].append(accuracy)
# 反思质量
reflection_score = self._evaluate_reflections(pred["reflections"])
results["reflection_quality"].append(reflection_score)
# 检索效率
efficiency = self._compute_retrieval_efficiency(
q, pred["reflections"]
)
results["retrieval_efficiency"].append(efficiency)
return {
"avg_accuracy": np.mean(results["accuracy"]),
"avg_reflection_quality": np.mean(results["reflection_quality"]),
"avg_retrieval_efficiency": np.mean(results["retrieval_efficiency"]),
"full_results": results
}
def _compute_accuracy(self, ground_truth, prediction):
"""计算答案准确率"""
# 使用LLM作为裁判
prompt = f"""判断以下回答是否正确。
问题：{q}
参考答案：{ground_truth}
待评估回答：{prediction}
输出：是或否"""
# 实际实现中调用LLM API
return 0.85 # 示例
def _evaluate_reflections(self, reflections):
"""
评估反思质量
检查：
1. 反思Token是否完整
2. 反思逻辑是否合理
3. IsSUP判断是否正确
"""
score = 1.0
# 检查必需的反思Token
required_types = {"Retrieve", "IsREL", "IsSUP", "IsUSE"}
present_types = {r["type"] for r in reflections}
if not required_types.issubset(present_types):
score -= 0.2
# 检查IsSUP判断质量
issup_tokens = [r for r in reflections if r["type"] == "IsSUP"]
contradictory_count = sum(1 for t in issup_tokens if t["value"] == "Contradictory")
if contradictory_count > 0:
# 有矛盾说明反思有效
score += 0.1
return min(score, 1.0)
def _compute_retrieval_efficiency(self, question, reflections):
"""
计算检索效率
理想情况：
- 简单问题：不检索
- 复杂问题：按需检索
"""
retrieve_decision = next(
(r for r in reflections if r["type"] == "Retrieve"),
None
)
if retrieve_decision is None:
return 0.5
if self._is_simple_question(question):
if retrieve_decision["value"] == "No":
return 1.0 # 正确决策
else:
return 0.0 # 不必要的检索
else:
if retrieve_decision["value"] == "Yes":
return 1.0 # 正确决策
else:
return 0.5 # 可能遗漏重要信息7. 应用场景与局限性
7.1 最佳应用场景
| 场景 | 描述 | Self-RAG优势 |
|---|---|---|
| 开放域问答 | 需要准确引用外部知识的问答 | 精确检索+评估 |
| 医学/法律咨询 | 需要可验证依据的专业咨询 | IsSUP确保答案有据可查 |
| 技术文档问答 | 复杂技术问题的精确回答 | 动态检索+片段级评估 |
| 多文档摘要 | 综合多个文档的关键信息 | 选择性使用相关片段 |
| 对话式搜索 | 需要持续判断信息需求的对话 | 按需检索减少Token消耗 |
7.2 实际应用示例
# 医疗问答场景
def medical_qa_system():
"""医疗问答系统使用Self-RAG"""
model = SelfRAGModel("medical-self-rag-7b")
retriever = MedicalDocumentRetriever()
# 用户问题
question = "服用阿司匹林和华法林有什么药物相互作用？"
# Self-RAG生成
answer, reflections = model.generate_with_reflection(
question,
retriever
)
# 输出反思日志（用于审核）
print("检索决策:", reflections[0])
print("文档相关性:", reflections[1:3])
print("支持度评估:", [r for r in reflections if r["type"] == "IsSUP"])
# 返回答案和依据
return {
"answer": answer,
"supporting_docs": [r["doc_id"] for r in reflections
if r.get("type") == "IsREL" and r.get("score") == "Relevant"],
"confidence": reflections[-1] # IsUSE评分
}7.3 局限性
| 局限性 | 描述 | 潜在解决方案 |
|---|---|---|
| 训练成本 | 需要两阶段训练（模仿+强化） | 使用预训练反射Token模型 |
| Token开销 | 反思Token增加输出长度 | 训练时优化Token效率 |
| 延迟增加 | 动态检索+评估增加延迟 | 缓存常见检索模式 |
| 评估复杂性 | 需要标注反思Token的训练数据 | 半自动标注策略 |
| 推理成本 | 每次生成都可能触发多次检索 | 自适应检索阈值 |
| 模型依赖 | 性能依赖基础模型能力 | 使用更强的基础模型 |
7.4 与其他RAG技术的对比总结
┌──────────────────────────────────────────────────────────────────┐
│ RAG技术演进图谱 │
├──────────────────────────────────────────────────────────────────┤
│ │
│ Naive RAG │
│ │ │
│ ├──→ Advanced RAG (混合检索、重排序) │
│ │ │ │
│ │ ├──→ Self-RAG (自我反思) │
│ │ │ │ │
│ │ │ ├──→ Agentic RAG (自主Agent) │
│ │ │ │ │
│ │ │ └──→ CRAG (自我纠正) │
│ │ │ │
│ └──→ GraphRAG (知识图谱) │
│ │ │
│ └──→ Hybrid RAG (Self-RAG + GraphRAG) │
│ │
└──────────────────────────────────────────────────────────────────┘
8. 相关主题
- RAG（检索增强生成） — RAG基础概念
- GraphRAG — 知识图谱增强RAG
- Agentic RAG架构 — 自主Agent驱动的RAG
- LLM评估 — RAG系统质量评估方法
参考资料
Footnotes
-
Self-RAG: Learning to Retrieve, Generate, and Critique through Self-Reflection — 原始论文，介绍Self-RAG的核心框架和反射Token机制 ↩
-
Self-RAG Official Implementation — GitHub官方实现代码库 ↩