深入 Self-RAG 评估：动态决策与 _Relevant 分数的力量
在现代大型语言模型（LLM）的应用中，检索增强生成（RAG）已成为提升模型准确性、减少幻觉并引入最新领域知识的关键技术。然而，传统的RAG流程往往是线性的：检索、然后生成。这种模式缺乏对检索结果质量的内在评估和动态调整能力。当检索到的文档与用户查询不相关或信息不足时，传统RAG的生成质量会大打折扣。
为了克服这些局限，Self-RAG应运而生。Self-RAG 的核心思想是让LLM本身参与到检索和生成过程的自我反思与评估中。它引入了一个“评论家”或“评估器”机制，能够根据检索到的信息对自身的状态进行判断，并据此决定下一步行动：是继续深入检索更多信息，还是已经收集到足够的信息可以开始生成答案，亦或是发现当前查询本身存在问题，需要进行重写才能获得更好的检索结果。
今天，我们将深入探讨Self-RAG中的一个关键评估指标：_Relevant 分数。我们将详细阐述如何利用这个动态分数，在Self-RAG的迭代循环中，智能地决定系统的走向——继续检索、开始生成，或是重写查询。
1. Self-RAG 的基石：超越传统 RAG
在深入 _Relevant 分数之前，我们先简要回顾一下RAG和Self-RAG的基本原理。
传统 RAG 的局限性：
传统 RAG 的流程相对简单：
- 查询嵌入： 将用户查询转换为向量。
- 文档检索： 在预构建的向量数据库中搜索与查询向量最相似的文档块。
- 信息合成： 将原始查询和检索到的文档作为上下文，输入给LLM进行答案生成。
这种模式的缺点在于，它对检索阶段的输出是完全信任的。如果检索到的文档质量不高（例如，不相关、信息过时、包含噪声），LLM 即使再强大，也难以生成高质量的答案，甚至可能“一本正经地胡说八道”。它缺乏一个反馈回路来判断检索结果的质量。
Self-RAG 的引入：自省与动态调整
Self-RAG 旨在通过引入一个“自我评估”机制来解决传统 RAG 的不足。它不再是一个单向的流程，而是一个迭代的、带有决策节点的循环。其核心思想在于：
- 批判性检索 (Critique-on-Retrieval): 在检索到文档后，LLM（或另一个专门的LLM模块）会评估这些文档与原始查询的相关性、信息丰富度等。
- 动态决策 (Dynamic Decision-Making): 基于评估结果，系统会做出策略性的选择，例如：
- 继续检索 (Continue Retrieval): 如果现有文档不足或相关性不高。
- 开始生成 (Start Generation): 如果已收集到足够高质量的文档。
- 重写查询 (Rewrite Query): 如果原始查询导致了持续的低质量检索结果。
这种自省和动态调整的能力，是Self-RAG相较于传统RAG的根本优势，也是我们今天讨论的 _Relevant 分数发挥作用的舞台。
2. _Relevant 分数：衡量检索质量的指南针
在Self-RAG框架中，_Relevant 分数是一个至关重要的信号，它由一个专门的“评论家”或“评估器”LLM生成。这个分数不是简单地衡量文档与查询的表面词法相似度，而是深入评估文档与查询之间的语义相关性、信息完整性以及对回答问题的潜在帮助程度。
2.1 _Relevant 分数的来源与生成
_Relevant 分数通常由一个专门的LLM模块（可以是与主生成LLM相同的模型，但经过特定指令微调，或者是一个更小、更快的模型）生成。这个模块接收以下输入：
- 原始用户查询 (Original User Query): 用户的初始问题。
- 当前检索到的文档 (Currently Retrieved Document): 刚刚从向量数据库中获取的文档块。
- （可选）累积的上下文 (Accumulated Context): 之前检索并被认为是相关的文档集合。
通过精心设计的提示词（Prompt），这个LLM评估器被指示输出一个结构化的结果，其中包含一个表示相关性程度的数值，即 _Relevant 分数。这个分数通常介于0到1之间，或者是一个离散的等级（例如，高、中、低），我们将其标准化为0-1的浮点数以便于量化决策。
示例提示词结构 (Concept for Relevance Critic LLM):
你是一个专业的文档相关性评估员。你的任务是判断给定的文档对于原始用户查询的相关性有多高。
请以JSON格式输出你的评估结果，包含以下字段：
- "relevance_score": 一个0到1之间的浮点数，0表示完全不相关，1表示高度相关且信息丰富。
- "reasoning": 简要说明你给出这个分数的原因。
---
用户查询: "{user_query}"
检索文档: "{retrieved_document_content}"
---
JSON输出:LLM会根据上述指令，分析文档内容与查询的匹配程度，并输出类似：{"relevance_score": 0.85, "reasoning": "文档详细解释了用户查询的核心概念，并提供了相关示例。"}。我们提取其中的 relevance_score 作为 _Relevant 分数。
2.2 _Relevant 分数的意义
- 语义深度： 它超越了简单的关键词匹配。即使文档中没有直接出现查询中的所有关键词，只要它能从语义上回答或支撑查询，
_Relevant分数也会很高。 - 信息完整性： 高分不仅仅意味着相关，还意味着文档包含了足够的信息量，能够对查询提供有意义的贡献。
- 决策依据：
_Relevant是Self-RAG进行动态决策的核心依据。它告诉系统，当前获取的信息是否值得进一步处理，或者是否需要改变策略。
3. 基于 _Relevant 分数的动态决策框架
现在，我们进入本文的核心：如何利用 _Relevant 分数来动态决定下一步行动。Self-RAG的决策逻辑通常在一个迭代循环中实现，每次迭代都会检索新的文档（或使用现有文档），然后进行评估。
3.1 核心决策点与策略
我们将 _Relevant 分数与预设的阈值结合，来指导三种主要行为：
3.1.1 继续检索 (Continue Retrieval)
决策条件：
_Relevant分数低于生成阈值 (e.g.,RELEVANCE_THRESHOLD_GENERATE)：这表明当前检索到的文档集合不足以支撑一个高质量的答案。- 尚未达到最大检索尝试次数 (e.g.,
MAX_RETRIEVAL_ATTEMPTS)：系统仍然有预算去探索更多信息。 - （可选）累积的相关文档数量不足：即使单个文档相关，但如果整体覆盖面不够，也需要继续检索。
目的： 寻找更多相关信息，以期累积足够的上下文来生成全面且准确的答案。这通常意味着向向量数据库发起另一次检索请求，可能使用稍微不同的参数（例如，检索更多文档，或者对原始查询进行轻微的扩展）。
3.1.2 开始生成 (Start Generation)
决策条件：
_Relevant分数达到或超过生成阈值 (e.g.,RELEVANCE_THRESHOLD_GENERATE)：这意味着系统已经检索到了一定数量的高质量相关文档。- （可选）累积的相关文档数量达到预期：例如，至少有N个文档被标记为高相关。
- （可选）达到最大检索尝试次数：即使相关性不完美，但已经没有更多检索机会，也可能被迫生成。
目的： 利用当前已收集到的所有相关文档和原始查询，输入给主LLM，生成最终的答案。这是Self-RAG流程的最终目标。
3.1.3 重写查询 (Rewrite Query)
决策条件：
_Relevant分数持续低于一个更低的重写阈值 (e.g.,RELEVANCE_THRESHOLD_REWRITE)：这表示经过多次检索尝试，系统仍然无法找到高质量的相关文档。这通常暗示原始查询可能存在问题，例如：- 过于模糊或宽泛。
- 使用了不常见的术语。
- 包含了错误的假设。
- 在知识库中根本没有相关信息。
- 已达到一定检索尝试次数，但相关性无显著提升：表明继续使用当前查询检索是徒劳的。
目的： 利用LLM的理解和推理能力，对原始用户查询进行语义上的重构或澄清，生成一个全新的、更精确或更具探索性的查询。然后，系统将使用这个新查询重新开始检索过程。
3.2 决策逻辑表格概览
为了更清晰地理解这些决策，我们可以用一个表格来总结：
| 条件类型 | _Relevant 分数 |
已检索文档数 | 检索尝试次数 | 决策 | 解释 |
|---|---|---|---|---|---|
| 继续检索 (低相关) | < RELEVANCE_THRESHOLD_GENERATE |
< MIN_DOCS_FOR_GEN |
< MAX_RETRIEVAL_ATTEMPTS |
继续检索 | 现有文档不足以生成答案，且仍有检索预算。 |
| 重写查询 (持续低相关) | < RELEVANCE_THRESHOLD_REWRITE (多次) |
任何 | >= MIN_ATTEMPTS_FOR_REWRITE |
重写查询并重新检索 | 原始查询可能存在问题，无法获取有效信息。 |
| 开始生成 (高相关) | >= RELEVANCE_THRESHOLD_GENERATE |
>= MIN_DOCS_FOR_GEN |
任何 | 开始生成 | 已收集到足够高质量的文档。 |
| 开始生成 (无更多尝试) | < RELEVANCE_THRESHOLD_GENERATE |
任何 | == MAX_RETRIEVAL_ATTEMPTS |
开始生成 (次优) | 即使相关性不理想，但已无检索机会，需尝试生成最佳答案。 |
注意： 这里的 MIN_DOCS_FOR_GEN 是指被评估为相关的文档数量，而不是所有检索到的文档数量。
4. Self-RAG 评估与决策的架构组件
为了实现上述动态决策，Self-RAG 系统需要以下核心组件：
- 查询嵌入器 (Query Embedder): 将用户查询（或重写后的查询）转换为向量。
- 文档检索器 (Document Retriever): 根据查询向量在向量数据库中检索相关文档块。
- 相关性评论家 (Relevance Critic – LLM): 接收查询和检索到的文档，输出
_Relevant分数及理由。 - 查询重写器 (Query Rewriter – LLM): 在检索效果不佳时，根据原始查询和检索历史，生成一个新的、更优的查询。
- 答案生成器 (Response Generator – LLM): 接收原始查询和累积的相关文档，生成最终的用户答案。
- Self-RAG 协调器/代理 (Orchestrator/Agent): 这是整个系统的“大脑”，负责管理整个流程的循环、状态（累积文档、检索次数）、调用各个组件，并根据
_Relevant分数做出动态决策。
import uuid
import random
from typing import List, Dict, Any, Optional
# 假设我们有一个LLM客户端，用于模拟不同的LLM调用
class MockLLMClient:
def __init__(self, latency_mean=0.1, latency_std=0.05):
self.latency_mean = latency_mean
self.latency_std = latency_std
def _simulate_latency(self):
import time
time.sleep(max(0, random.gauss(self.latency_mean, self.latency_std)))
def generate(self, prompt: str, temperature: float = 0.7, max_tokens: int = 500) -> str:
self._simulate_latency()
# 这是一个非常简化的模拟，实际LLM会根据prompt生成内容
if "relevance_score" in prompt:
# 模拟相关性评分器的输出
# 随机生成一个相关性分数，模拟不同场景
score = random.uniform(0.1, 0.95)
# 模拟偶尔出现低分的情况
if "无法找到" in prompt or "低质量" in prompt:
score = random.uniform(0.05, 0.3)
elif "详细解释" in prompt or "核心概念" in prompt:
score = random.uniform(0.7, 0.99)
return f'{{"relevance_score": {score:.2f}, "reasoning": "模拟评估结果。"}}'
elif "重写查询" in prompt:
# 模拟查询重写器的输出
original_query = prompt.split("原始查询: ")[1].split("n")[0]
if "人工智能的最新进展" in original_query:
return "解释2023-2024年人工智能领域的技术突破和应用趋势"
elif "量子计算" in original_query:
return "详细阐述量子计算的基础原理、当前挑战及未来应用前景"
return f"重写后的查询：{original_query} 的更具体或不同表述。"
else:
# 模拟答案生成器的输出
return f"根据提供的上下文和查询 '{prompt.split('用户查询: ')[1].split('上下文:')[0].strip()}'，我生成了以下答案：这是一个模拟答案，具体内容会根据实际LLM生成。"
llm_client = MockLLMClient()
# 文档数据结构
class Document:
def __init__(self, content: str, doc_id: Optional[str] = None, metadata: Optional[Dict] = None):
self.doc_id = doc_id if doc_id else str(uuid.uuid4())
self.content = content
self.metadata = metadata if metadata else {}
self.relevance_score: Optional[float] = None # 用于存储评估后的相关性分数
def __repr__(self):
return f"Document(id={self.doc_id[:8]}..., score={self.relevance_score:.2f} if self.relevance_score is not None else 'N/A')"
def to_dict(self):
return {
"doc_id": self.doc_id,
"content": self.content,
"metadata": self.metadata,
"relevance_score": self.relevance_score
}
# 模拟向量数据库和检索功能
class MockVectorDatabase:
def __init__(self, documents: List[str]):
self.documents = [Document(content=doc, metadata={"source": f"source_{i+1}.txt"}) for i, doc in enumerate(documents)]
# 模拟嵌入过程，这里只是存储原始文档
print(f"MockVectorDatabase initialized with {len(self.documents)} documents.")
def retrieve_documents(self, query: str, top_k: int = 3) -> List[Document]:
print(f" [DB] Retrieving {top_k} documents for query: '{query}'...")
# 这是一个非常简化的检索模拟
# 实际中会根据query的嵌入向量进行相似度搜索
# 这里为了演示，我们随机返回一些文档，并尝试模拟相关性
retrieved = []
# 尝试返回一些与查询主题相关的文档
if "人工智能" in query or "AI" in query:
relevant_docs = [doc for doc in self.documents if "AI" in doc.content or "人工智能" in doc.content or "模型" in doc.content]
if relevant_docs:
retrieved.extend(random.sample(relevant_docs, min(top_k, len(relevant_docs))))
if "量子计算" in query:
relevant_docs = [doc for doc in self.documents if "量子" in doc.content or "计算" in doc.content or "纠缠" in doc.content]
if relevant_docs:
retrieved.extend(random.sample(relevant_docs, min(top_k, len(relevant_docs))))
if "气候变化" in query:
relevant_docs = [doc for doc in self.documents if "气候" in doc.content or "变暖" in doc.content]
if relevant_docs:
retrieved.extend(random.sample(relevant_docs, min(top_k, len(relevant_docs))))
# 如果不够，或者没有特定主题，就随机补充
if len(retrieved) < top_k:
remaining_needed = top_k - len(retrieved)
all_other_docs = [doc for doc in self.documents if doc not in retrieved]
if all_other_docs:
retrieved.extend(random.sample(all_other_docs, min(remaining_needed, len(all_other_docs))))
# 确保返回的文档数量不超过top_k
return retrieved[:top_k]
# 模拟知识库内容
mock_corpus = [
"人工智能（AI）是计算机科学的一个分支，旨在创建能够像人类一样思考和学习的机器。近年来，深度学习和神经网络推动了AI的巨大发展。",
"大型语言模型（LLM）是AI领域的重要突破，它们能够理解、生成人类语言，并在各种NLP任务中表现出色，如GPT系列和BERT。",
"量子计算利用量子力学原理，如叠加和纠缠，来执行传统计算机无法完成的复杂计算。它在药物发现和材料科学等领域有巨大潜力。",
"气候变化是全球性的挑战，导致海平面上升、极端天气事件增多。减少碳排放和发展可再生能源是应对的关键。",
"机器学习是AI的一个子集，它使系统能够从数据中学习，而无需明确编程。监督学习、无监督学习和强化学习是其主要范式。",
"生成式AI可以创建新的内容，例如图像、文本和音乐。扩散模型和GANs是其代表技术。",
"经典计算机使用二进制位（0或1），而量子计算机使用量子位（qubit），量子位可以同时是0和1。",
"全球变暖是气候变化的主要表现，主要是由于人类活动导致的温室气体排放增加。",
"自然语言处理（NLP）是AI的一个分支，专注于计算机与人类语言之间的交互。LLM是NLP的最新进展。"
]
vector_db = MockVectorDatabase(mock_corpus)
# 4.1 相关性评论家 (Relevance Critic - LLM)
def evaluate_relevance(query: str, document: Document, llm: MockLLMClient) -> float:
prompt = f"""你是一个专业的文档相关性评估员。你的任务是判断给定的文档对于原始用户查询的相关性有多高。
请以JSON格式输出你的评估结果，包含以下字段：
- "relevance_score": 一个0到1之间的浮点数，0表示完全不相关，1表示高度相关且信息丰富。
- "reasoning": 简要说明你给出这个分数的原因。
---
用户查询: "{query}"
检索文档: "{document.content}"
---
JSON输出:
"""
try:
response = llm.generate(prompt, temperature=0.1, max_tokens=100)
import json
result = json.loads(response)
score = result.get("relevance_score", 0.0)
print(f" [Critic] Evaluated Doc '{document.doc_id[:8]}...' for query '{query[:20]}...': Score={score:.2f}")
return float(score)
except Exception as e:
print(f" [Critic Error] Failed to evaluate relevance: {e}")
return 0.0 # 评估失败默认不相关
# 4.2 查询重写器 (Query Rewriter - LLM)
def rewrite_query(original_query: str, past_retrievals: List[Document], llm: MockLLMClient) -> str:
past_docs_content = "n".join([f"- Doc {d.doc_id[:8]}... (Score: {d.relevance_score:.2f}): {d.content[:100]}..." for d in past_retrievals if d.relevance_score is not None and d.relevance_score > 0.3])
prompt = f"""你是一个智能查询优化器。原始查询未能检索到足够的相关信息，或者检索到的信息相关性不高。
请根据原始查询和过去检索到的一些（可能不完全相关或少量相关）文档，重写一个更精确、更具体或从不同角度切入的新查询。
新的查询应该能够帮助检索系统找到更相关的文档。
---
原始查询: "{original_query}"
过去检索到的部分文档（可能相关性不高或不足）：
{past_docs_content if past_docs_content else "无"}
---
请只输出重写后的新查询，不要包含任何额外说明。
"""
print(f" [Rewriter] Rewriting query for '{original_query[:20]}...'...")
rewritten_query = llm.generate(prompt, temperature=0.7, max_tokens=100).strip()
# 模拟LLM输出格式，去除潜在的"重写后的查询："前缀
if rewritten_query.startswith("重写后的查询："):
rewritten_query = rewritten_query[len("重写后的查询："):].strip()
print(f" [Rewriter] New query: '{rewritten_query}'")
return rewritten_query
# 4.3 答案生成器 (Response Generator - LLM)
def generate_response(original_query: str, relevant_documents: List[Document], llm: MockLLMClient) -> str:
context = "nn".join([doc.content for doc in relevant_documents])
prompt = f"""你是一个专业的问答助手。请根据提供的上下文，简洁、准确地回答用户查询。
如果上下文不足以回答问题，请说明。
---
用户查询: "{original_query}"
上下文:
{context}
---
请生成你的答案:
"""
print(f" [Generator] Generating response for query '{original_query[:20]}...' with {len(relevant_documents)} documents...")
response = llm.generate(prompt, temperature=0.5, max_tokens=500)
return response
### 5. Self-RAG 协调器：基于 `_Relevant` 分数的动态循环实现
现在，我们将所有组件整合到一个 `SelfRAGAgent` 类中，它将负责管理整个迭代过程，并根据 `_Relevant` 分数做出决策。
```python
import collections
class SelfRAGAgent:
def __init__(self,
vector_db: MockVectorDatabase,
llm_client: MockLLMClient,
relevance_threshold_generate: float = 0.6,
relevance_threshold_rewrite: float = 0.3,
max_retrieval_attempts: int = 3,
min_docs_for_generation: int = 2,
top_k_retrieval: int = 3):
self.vector_db = vector_db
self.llm_client = llm_client
self.RELEVANCE_THRESHOLD_GENERATE = relevance_threshold_generate
self.RELEVANCE_THRESHOLD_REWRITE = relevance_threshold_rewrite
self.MAX_RETRIEVAL_ATTEMPTS = max_retrieval_attempts
self.MIN_DOCS_FOR_GENERATION = min_docs_for_generation
self.TOP_K_RETRIEVAL = top_k_retrieval
print(f"SelfRAGAgent Initialized with thresholds: Generate={self.RELEVANCE_THRESHOLD_GENERATE}, Rewrite={self.RELEVANCE_THRESHOLD_REWRITE}")
def run(self, user_query: str) -> str:
current_query = user_query
retrieval_attempts = 0
accumulated_relevant_docs: List[Document] = []
all_retrieved_docs_history: List[Document] = [] # 记录所有检索过的文档，无论相关性如何
print(f"n--- Self-RAG Process Start for Query: '{user_query}' ---")
while retrieval_attempts < self.MAX_RETRIEVAL_ATTEMPTS:
retrieval_attempts += 1
print(f"n[Attempt {retrieval_attempts}/{self.MAX_RETRIEVAL_ATTEMPTS}] Current Query: '{current_query}'")
# 1. 检索文档
retrieved_docs = self.vector_db.retrieve_documents(current_query, top_k=self.TOP_K_RETRIEVAL)
if not retrieved_docs:
print(" [Decision] No documents retrieved. Cannot proceed with current query.")
if retrieval_attempts == self.MAX_RETRIEVAL_ATTEMPTS or not accumulated_relevant_docs:
print(" [Decision] Max attempts reached or no relevant docs. Generating with what we have (if any).")
break # 无法检索到任何文档，结束循环
else:
print(" [Decision] No docs, but not max attempts. Will try rewriting if needed or continue.")
# 此时可能需要重写查询，即便没有新文档
pass # 流程会继续到评估阶段，可能会触发重写
# 2. 评估相关性
current_retrieval_scores = []
for doc in retrieved_docs:
doc.relevance_score = evaluate_relevance(user_query, doc, self.llm_client) # 注意这里用原始user_query评估
all_retrieved_docs_history.append(doc) # 记录所有被检索的文档
current_retrieval_scores.append(doc.relevance_score)
if doc.relevance_score >= self.RELEVANCE_THRESHOLD_GENERATE:
accumulated_relevant_docs.append(doc)
avg_current_relevance = sum(current_retrieval_scores) / len(current_retrieval_scores) if current_retrieval_scores else 0
print(f" [Evaluation] Average relevance of current batch: {avg_current_relevance:.2f}")
print(f" [State] Accumulated {len(accumulated_relevant_docs)} relevant documents.")
# 3. 动态决策
# 决策优先级：生成 > 重写 > 继续检索
# 决策 1: 是否有足够的相关文档可以生成？
if len(accumulated_relevant_docs) >= self.MIN_DOCS_FOR_GENERATION and avg_current_relevance >= self.RELEVANCE_THRESHOLD_GENERATE:
print(" [Decision] Enough highly relevant documents accumulated. Proceeding to generation.")
break # 满足生成条件，跳出循环进行生成
# 决策 2: 是否需要重写查询？ (当相关性持续很低时)
# 这里我们检查历史检索结果的平均相关性，或者当前批次是否特别差
if retrieval_attempts >= 2 and avg_current_relevance < self.RELEVANCE_THRESHOLD_REWRITE:
print(" [Decision] Low average relevance after multiple attempts. Considering query rewrite.")
# 再次检查是否有实际的低分文档导致重写
low_score_docs = [d for d in retrieved_docs if d.relevance_score is not None and d.relevance_score < self.RELEVANCE_THRESHOLD_REWRITE]
if low_score_docs or not retrieved_docs: # 如果检索结果很差，或者根本没有检索到
print(" [Decision] Triggering query rewrite.")
current_query = rewrite_query(user_query, all_retrieved_docs_history, self.llm_client)
# 重写后，重置累积的相关文档，但保留历史记录，以便重写器参考
# 也可以选择不重置，这取决于策略
# accumulated_relevant_docs = [] # 谨慎重置，可能丢失有价值信息
# 为了演示，我们在这里不重置，让重写器基于新查询寻找“额外”信息
continue # 使用新查询重新开始下一轮检索
else:
print(" [Decision] Current batch not bad enough to rewrite. Continuing retrieval.")
# 决策 3: 否则，继续检索 (如果还有尝试次数)
if retrieval_attempts < self.MAX_RETRIEVAL_ATTEMPTS:
print(" [Decision] Not enough relevant docs yet, or not bad enough for rewrite. Continuing retrieval.")
continue
else:
print(" [Decision] Max retrieval attempts reached. Proceeding to generation with available documents.")
break # 达到最大尝试次数，跳出循环进行生成
# 4. 生成答案
if not accumulated_relevant_docs:
print(" [Final] No relevant documents found after all attempts.")
return generate_response(user_query, [Document("无可用相关信息。", metadata={"source": "Self-RAG"})], self.llm_client)
else:
print(f"n--- Self-RAG Process Complete for Query: '{user_query}' ---")
print(f"Final accumulated relevant documents ({len(accumulated_relevant_docs)}):")
for doc in accumulated_relevant_docs:
print(f" - Doc ID: {doc.doc_id[:8]}..., Score: {doc.relevance_score:.2f}, Content: {doc.content[:50]}...")
return generate_response(user_query, accumulated_relevant_docs, self.llm_client)
代码解析：
MockLLMClient: 模拟 LLM 的行为，包括生成相关性分数、重写查询和生成答案。这使得我们可以在没有实际 LLM API 调用的情况下测试逻辑。Document类: 封装文档内容、ID和元数据，并添加了relevance_score字段来存储评估结果。MockVectorDatabase: 模拟向量数据库，根据查询返回一些文档。这里的检索逻辑是简化的，但在实际场景中，它会执行复杂的向量相似度搜索。evaluate_relevance: 这是_Relevant分数的核心生成函数。它构建一个提示词，指示 LLM 评估文档与查询的相关性，并解析 JSON 输出。rewrite_query: 当检索效果不佳时，此函数构建提示词，让 LLM 根据原始查询和已检索到的文档（即使质量不高）来生成一个新的、更优的查询。generate_response: 最终的答案生成函数，将累积的相关文档作为上下文提供给 LLM。SelfRAGAgent类:- 初始化: 设置各种阈值和参数。
run方法: 这是整个 Self-RAG 流程的入口。它包含一个while循环，代表了迭代的检索和评估过程。- 状态管理:
current_query(当前用于检索的查询，可能被重写)、retrieval_attempts(计数器)、accumulated_relevant_docs(存储被认为是高度相关的文档)、all_retrieved_docs_history(存储所有检索过的文档，用于重写器参考)。 - 决策逻辑: 在每次迭代结束时，根据
avg_current_relevance和accumulated_relevant_docs的数量，以及retrieval_attempts来动态决定是跳出循环（生成）、重写查询并继续循环，还是直接继续下一轮检索。
6. 运行与案例分析
让我们通过几个具体的例子来演示 SelfRAGAgent 的行为。
案例 1: 初始查询良好，直接找到相关文档并生成
print("n--- 案例 1: 初始查询良好，直接找到相关文档并生成 ---")
agent = SelfRAGAgent(vector_db, llm_client,
relevance_threshold_generate=0.7, # 较高阈值
relevance_threshold_rewrite=0.3,
max_retrieval_attempts=3,
min_docs_for_generation=2,
top_k_retrieval=2)
query_1 = "解释大型语言模型LLM是什么？"
response_1 = agent.run(query_1)
print(f"n用户查询: '{query_1}'")
print(f"最终答案:n{response_1}")预期输出分析 (模拟):
- Agent 第一次检索，由于查询与“大型语言模型LLM”高度相关，
MockVectorDatabase会返回包含“LLM”和“AI”的文档。 evaluate_relevance会为这些文档生成较高的_Relevant分数 (例如 0.85, 0.90)。accumulated_relevant_docs很快达到min_docs_for_generation(2个)，且平均相关性高于RELEVANCE_THRESHOLD_GENERATE(0.7)。- Agent 立即决定跳出循环，进入生成阶段，并使用这些高质量文档生成答案。
案例 2: 初始查询较模糊，需要多次检索或重写
print("n--- 案例 2: 初始查询较模糊，需要多次检索或重写 ---")
agent = SelfRAGAgent(vector_db, llm_client,
relevance_threshold_generate=0.7,
relevance_threshold_rewrite=0.3,
max_retrieval_attempts=4, # 增加尝试次数
min_docs_for_generation=2,
top_k_retrieval=2)
query_2 = "最新的AI技术是什么？" # 这是一个相对模糊的查询，可能导致初期检索结果不理想
response_2 = agent.run(query_2)
print(f"n用户查询: '{query_2}'")
print(f"最终答案:n{response_2}")预期输出分析 (模拟):
- 尝试 1:
MockVectorDatabase可能返回一些泛泛的 AI 文档，evaluate_relevance给出中等或偏低分数 (例如 0.5, 0.6)。accumulated_relevant_docs可能不足，或平均相关性未达生成阈值。Agent 决定继续检索。 - 尝试 2: 再次检索，可能结果类似。如果
avg_current_relevance持续低于RELEVANCE_THRESHOLD_REWRITE(0.3)，Agent 可能会触发rewrite_query。 - 重写:
rewrite_query将“最新的AI技术是什么？”重写为类似“解释2023-2024年人工智能领域的技术突破和应用趋势”。 - 尝试 3 (使用新查询):
MockVectorDatabase使用新的、更精确的查询进行检索，这次可能找到更具体的文档（尽管我们的模拟数据库内容有限）。_Relevant分数可能会提高。 - 如果此时满足生成条件，则生成答案。否则，继续检索，直到达到
MAX_RETRIEVAL_ATTEMPTS。
案例 3: 知识库中缺乏信息
print("n--- 案例 3: 知识库中缺乏信息 ---")
agent = SelfRAGAgent(vector_db, llm_client,
relevance_threshold_generate=0.7,
relevance_threshold_rewrite=0.3,
max_retrieval_attempts=3,
min_docs_for_generation=2,
top_k_retrieval=2)
query_3 = "如何用Python实现一个基于区块链的去中心化投票系统？" # 我们的模拟数据库中没有相关信息
response_3 = agent.run(query_3)
print(f"n用户查询: '{query_3}'")
print(f"最终答案:n{response_3}")预期输出分析 (模拟):
- 尝试 1-3:
MockVectorDatabase几乎找不到任何相关文档，或者返回的文档与区块链、Python编程无关。evaluate_relevance会持续给出非常低的分数 (例如 0.1, 0.2)。 - 由于
_Relevant分数持续低于RELEVANCE_THRESHOLD_REWRITE，且经过多次尝试，Agent 可能会尝试重写查询。但即使重写，例如重写成“区块链技术在投票系统中的应用”，我们的模拟数据库依然缺乏相关信息。 - 最终，
accumulated_relevant_docs将为空或文档数量不足。Agent 会在达到MAX_RETRIEVAL_ATTEMPTS后，带着“无可用相关信息”的文档去生成答案，LLM 会据此说明它无法回答。
7. 高级考量与挑战
虽然基于 _Relevant 分数的动态决策显著提升了 RAG 系统的智能性，但在实际部署中仍需考虑一些高级问题和挑战：
- 阈值校准 (Threshold Tuning):
RELEVANCE_THRESHOLD_GENERATE和RELEVANCE_THRESHOLD_REWRITE的设置至关重要。它们直接影响系统的行为。过高的阈值可能导致过度检索和频繁的查询重写，增加成本和延迟；过低的阈值则可能导致过早生成低质量答案。校准这些阈值通常需要大量的实验和领域知识。 - 多文档相关性聚合： 当检索到多个文档时，如何综合它们的
_Relevant分数来做出整体决策？是取平均值、最小值、最大值，还是加权平均？不同的聚合策略会影响决策的敏感度。 - 上下文窗口管理： 随着
accumulated_relevant_docs数量的增加，传递给生成 LLM 的上下文可能会超出其最大输入长度。需要策略来选择最相关的文档子集，或对文档进行摘要。 - 计算成本与延迟： Self-RAG 引入了额外的 LLM 调用（评估、重写），这会显著增加计算成本和端到端延迟。对于实时性要求高的应用，可能需要使用更小、更快的评估模型，或者优化 LLM 调用的并行性。
- 评估器 LLM 的鲁棒性：
_Relevant分数的质量直接依赖于评估器 LLM 的准确性和稳定性。评估器本身可能存在幻觉或偏见，从而误导整个 Self-RAG 流程。 - 查询重写的有效性： 重写后的查询是否总能带来更好的检索结果？有时重写可能导致查询偏离原始意图，或者陷入局部最优。需要机制来检测和纠正无效的查询重写。
- 迭代次数的平衡：
MAX_RETRIEVAL_ATTEMPTS的设定需要平衡。太少可能遗漏信息，太多则浪费资源。
8. 总结与展望
Self-RAG 及其核心的 _Relevant 分数机制，代表了 RAG 技术向更智能、更自适应方向发展的重要一步。通过赋予系统在检索过程中进行自我评估和动态决策的能力，我们能够显著提升 LLM 在处理复杂、模糊或信息稀缺查询时的性能和鲁棒性。
这种基于反馈循环的智能体设计，不仅优化了信息检索的效率，也使得最终的生成答案更加精准和可靠。随着 LLM 技术和评估方法学的不断演进，Self-RAG 有望在未来成为构建高度智能和自主问答系统的基石。