解析 ‘Self-RAG 3.0’：模型如何根据当前的‘认知匮乏度’自主决定是否启动一轮全新的多源检索？

各位来宾，各位同行，大家好！

今天，我们齐聚一堂，共同探讨一个令人兴奋且极具挑战性的前沿话题：如何让大型语言模型（LLM）像人类一样，在意识到自身知识不足时，主动寻求更多信息。具体来说，我们将深入解析“Self-RAG 3.0”的核心机制——模型如何根据当前的“认知匮乏度”自主决定是否启动一轮全新的多源检索。

在过去几年中，检索增强生成（RAG）技术已经成为弥合LLM知识盲区和减少幻觉的关键范式。从最初简单的“检索-生成”流水线，到后来的迭代式RAG、自适应RAG，以及现在我们看到的Self-RAG系列，RAG的演进一直围绕着一个核心目标：如何更智能、更有效地利用外部知识。

传统的RAG通常在接收到用户查询后，无条件地执行一次检索。这种“一刀切”的方式，在很多情况下是低效的。如果LLM本身已经掌握了足够的信息来回答问题，或者初始检索结果已经非常完善，那么额外的检索操作不仅浪费计算资源，还可能引入噪声。反之，如果LLM对某个问题一无所知，或者现有信息不足以形成高质量的回答，那么仅仅一次检索可能远远不够，甚至需要从多个来源、以不同的策略进行深度挖掘。

Self-RAG 3.0正是为了解决这一痛点而生。它引入了一个革命性的概念：让LLM具备自我认知的能力，能够评估自己的“认知状态”，并根据这种评估结果来决定是否需要进一步的外部干预，即启动一次全新的多源检索。 这不仅仅是简单的检索优化，更是一种迈向真正智能体的关键一步——从被动接受指令到主动进行决策。

一、理解LLM的“认知匮乏度”：一种量化的自我评估

在人类语境中，“认知匮乏度”指的是一个人对某个主题知识的不足、理解的模糊或信息的不确定性。对于LLM而言，我们当然不能赋予它真正意义上的“意识”或“感觉”，但我们可以通过一系列可量化的指标和启发式规则来模拟这种“认知匮乏度”。这是一种基于其内部状态和输出表现的、可计算的“自我评估”。

LLM的“认知匮乏度”主要表现为以下几个方面：

1. 回答置信度低（Low Confidence）： 模型在生成回答时，对所生成内容的概率分数普遍较低。这可能意味着模型在多个可能的答案中摇摆不定，或者其训练数据中关于该主题的信息稀疏。
2. 信息不完整或不充分（Incompleteness）： 模型虽然能给出部分答案，但无法全面覆盖用户查询的所有方面，或者遗漏了关键的细节。
3. 信息不一致或存在冲突（Inconsistency/Conflict）： 模型在生成过程中，或者在结合现有检索结果后，发现不同信息源之间存在矛盾，导致无法形成一个统一、连贯的回答。
4. 事实准确性存疑（Factual Ambiguity/Inaccuracy）： 模型怀疑其即将生成的回答可能存在事实性错误，或无法核实信息的准确性。这通常需要一个独立的“事实核查”模块来辅助判断。
5. 查询理解不足或歧义（Query Ambiguity）： 模型发现用户查询本身存在歧义，需要更多上下文或额外信息才能准确理解并给出回答。
6. 无法直接回答（Inability to Answer）： 模型完全无法根据现有知识和检索结果来生成任何有意义的答案。

量化这些表现是实现自主决策的基础。我们将深入探讨如何将这些抽象概念转化为具体的数值指标。

二、Self-RAG 3.0 的核心架构：决策模块与多源检索

Self-RAG 3.0 的核心在于其能够动态地调整RAG流程，而不是采用一个固定的模式。这需要一个强大的决策模块（Decision Module），它充当了整个系统的“大脑”，负责评估“认知匮乏度”并发出检索指令。

其高层架构可以概括为：

1. 用户查询与初始处理（User Query & Initial Processing）： 用户输入查询。模型首先尝试基于其内部知识和可能进行的初步、轻量级检索（如果配置了）来理解查询并尝试形成初步响应。
2. 状态评估与认知匮乏度量化（State Evaluation & Cognitive Deficit Quantification）： 这是Self-RAG 3.0的关键环节。一个专门的评估器或由Critic Model组成的模块，会分析当前的查询、已有的初步生成内容（如果有）、以及任何初始检索结果，并计算出一系列指标来量化“认知匮乏度”。
3. 决策模块（Decision Module）： 基于评估器提供的“认知匮乏度”指标，决策模块会运用一套预设的规则、阈值，或者是一个经过训练的分类器/策略网络，来决定：
   • 是否需要启动一次全新的多源检索？
   • 如果需要，应该检索哪些类型的信息源？ (例如：知识图谱、学术论文库、实时新闻、内部文档库、网页搜索等)
   • 检索的策略是什么？ (例如：关键词检索、语义检索、子问题分解检索等)
4. 多源检索（Multi-Source Retrieval）： 如果决策模块决定进行检索，系统将根据决策模块的指令，从一个或多个配置好的外部信息源中获取相关文档。
5. 上下文重构与生成（Context Re-construction & Generation）： 将检索到的新信息与原始查询、已有的上下文进行整合，形成一个更丰富、更准确的提示（Prompt），然后再次提交给主LLM进行生成。
6. 迭代与循环（Iteration & Loop）： 生成结果再次经过评估。如果“认知匮乏度”仍然很高，或者新的检索带来了新的问题，决策模块可能会启动新一轮的检索。这个过程可以迭代进行，直到满足停止条件（例如：达到最大迭代次数、认知匮乏度低于阈值，或生成了满意答案）。

我们用一个表格来简单对比一下传统RAG与Self-RAG 3.0的核心区别：

特性	传统RAG	Self-RAG 3.0
检索触发机制	固定、无条件触发（通常在查询初始阶段）	动态、按需触发，基于模型对自身“认知匮乏度”的评估
检索次数	通常一次（或少数几次预设迭代）	可变，多次迭代，根据需求自主决定，甚至可以进行多轮深度检索
检索来源	通常单一或少数几个预设来源	多源，可根据查询类型和匮乏度动态选择（如知识图谱、网页、数据库、文档）
核心智能	外部协调器管理检索与生成	模型内部决策模块进行自我评估与策略制定
资源效率	可能存在不必要的检索，资源利用效率一般	按需检索，理论上更高效，避免无谓的计算和IO操作
复杂问题处理	面对复杂、多方面、实时性要求高的问题时，效果受限	能够自主分解、深度挖掘、多角度整合信息，处理复杂问题能力更强
自适应性	较低	较高，能够根据不同的查询和模型状态自适应调整策略

三、量化“认知匮乏度”：核心指标与实现细节

要让模型自主决策，我们必须为其提供可操作的、量化的“认知匮乏度”指标。以下是一些关键的指标和实现思路。

3.1 基于生成置信度的评估

LLM在生成每个token时，都会为其分配一个概率分数（log-probability）。这些分数是衡量模型对当前生成内容信心的直接指标。

实现思路：

1. Token级别置信度：

   • 大多数LLM API（如OpenAI的log_probs）或开源模型（如Hugging Face transformers库的generate方法）都能提供每个生成token的对数概率。
   • 我们可以计算生成序列中所有token的平均对数概率，或者更精细地，关注低概率token的出现频率。
   • 低概率token的聚集通常意味着模型正在“挣扎”，或者在遇到不确定性时“猜测”。

2. 序列级别置信度：

   • 将所有token的对数概率相加（或平均），得到整个生成序列的置信度分数。
   • 可以训练一个小型分类器，以生成的文本和其token对数概率作为输入，输出一个整体的“回答质量”或“置信度”分数。

代码示例 (Python, 概念性)：

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer

class ConfidenceEvaluator:
    def __init__(self, model_name="gpt2"):
        # 实际应用中，这里会是你的主LLM或一个专门的评估LLM
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForCausalLM.from_pretrained(model_name)
        self.model.eval() # 设置为评估模式

    def calculate_sequence_confidence(self, prompt: str, generated_text: str) -> float:
        """
        计算生成文本的序列级别置信度。
        这里简化为平均对数概率，实际可能更复杂。
        """
        # 结合prompt和generated_text作为完整输入
        full_text = prompt + generated_text
        inputs = self.tokenizer(full_text, return_tensors="pt", truncation=True, max_length=512)

        # 找到generated_text在full_text中的起始位置
        prompt_len = len(self.tokenizer(prompt, return_tensors="pt")['input_ids'][0])
        generated_len = len(inputs['input_ids'][0]) - prompt_len

        if generated_len <= 0:
            return 0.0 # 没有生成内容

        with torch.no_grad():
            outputs = self.model(**inputs)
            logits = outputs.logits # (batch_size, sequence_length, vocab_size)

        # 获取生成部分的logits
        # 注意：这里需要准确匹配生成部分，通常LLM的log_probs是预测下一个token的
        # 为了简化，我们假设我们能获取到生成文本对应token的实际概率
        # 实际API如OpenAI会直接返回log_probs

        # 这是一个模拟，实际需要根据LLM的生成机制来获取准确的log_probs
        # 假设我们能通过某种方式获取到 `generated_token_log_probs` 列表

        # 为了演示，我们模拟一个场景，假设我们已经知道生成的token IDs和它们的log_probs
        # 实际中，你会在调用 generate 方法时获取这些信息
        # e.g., output_sequences, scores = model.generate(..., return_dict_in_generate=True, output_scores=True)
        # then process scores to get log_probs

        # Placeholder for actual log_probs extraction logic
        # For demonstration, let's just make up some log_probs for the generated part
        # In a real scenario, you'd integrate this with the actual generation process

        # Example of how to get log_probs during generation (simplified)
        # If using `model.generate` with `output_scores=True`:
        # from transformers import LogitsProcessorList, TopKLogitsWarper, TemperatureLogitsWarper
        # from transformers.generation.utils import StoppingCriteriaList, MaxLengthCriteria

        # inputs = self.tokenizer(prompt, return_tensors="pt")
        # generation_output = self.model.generate(
        #     inputs.input_ids,
        #     max_new_tokens=generated_len, # For simplicity, assume we know generated_len
        #     return_dict_in_generate=True,
        #     output_scores=True
        # )

        # transition_scores = self.model.compute_transition_scores(
        #     generation_output.sequences, generation_output.scores, normalize_logits=True
        # )[0] # For the first batch item

        # average_log_prob = transition_scores.mean().item()

        # Let's use a simpler, more direct approach for this example, assuming we have generated text and its log_probs
        # If we just have `generated_text`, we can re-evaluate its log_probs given the prompt

        # This part requires a bit of a hack to get log_probs for an *already generated* text
        # by treating it as a continuation

        input_ids = inputs['input_ids']
        attention_mask = inputs['attention_mask']

        # Get logits for the entire sequence, then focus on the generated part
        # This is for computing the probability of the *already generated* sequence given the prompt

        target_ids = input_ids[0, prompt_len:] # The IDs of the generated tokens

        # We need to calculate the log_prob of each target_id given the preceding tokens
        log_probs_list = []
        for i in range(generated_len):
            current_input_ids = input_ids[:, :prompt_len + i]
            current_attention_mask = attention_mask[:, :prompt_len + i]

            with torch.no_grad():
                current_outputs = self.model(input_ids=current_input_ids, attention_mask=current_attention_mask)
                current_logits = current_outputs.logits[:, -1, :] # Logits for the next token

            # Get log probabilities for all possible next tokens
            log_probs = torch.log_softmax(current_logits, dim=-1)

            # Get the log probability of the actual generated token
            actual_token_log_prob = log_probs[:, target_ids[i]].item()
            log_probs_list.append(actual_token_log_prob)

        if not log_probs_list:
            return 0.0

        average_log_prob = sum(log_probs_list) / len(log_probs_list)
        return average_log_prob

# # Example Usage:
# evaluator = ConfidenceEvaluator()
# prompt = "What is the capital of France?"
# generated_answer_high_confidence = "The capital of France is Paris."
# generated_answer_low_confidence = "I'm not entirely sure, but it might be Paris, or perhaps Berlin? No, that's not right."

# # In a real scenario, these would come from model.generate()
# # For this example, let's assume we got these log_probs from a real generation
# # We'll just simulate the confidence score calculation
# print(f"High confidence answer score: {evaluator.calculate_sequence_confidence(prompt, generated_answer_high_confidence)}")
# print(f"Low confidence answer score: {evaluator.calculate_sequence_confidence(prompt, generated_answer_low_confidence)}")

# Due to the complexity of getting log_probs for *arbitrary* generated text without running generation itself
# and the specifics of `compute_transition_scores` requiring the generation object,
# I will simplify the `ConfidenceEvaluator` example to return a simulated score for clarity,
# and explain how it would be integrated with `model.generate(output_scores=True)` in practice.

class ConfidenceEvaluator:
    def __init__(self):
        pass # No model needed for this simplified example, assume scores are provided

    def calculate_sequence_confidence(self, generated_log_probs: list[float]) -> float:
        """
        计算生成文本的序列级别置信度，给定每个token的对数概率。
        """
        if not generated_log_probs:
            return 0.0
        # 平均对数概率
        return sum(generated_log_probs) / len(generated_log_probs)

    def contains_low_confidence_tokens(self, generated_log_probs: list[float], threshold: float = -5.0) -> bool:
        """
        检查序列中是否存在低于某个阈值的低置信度token。
        """
        return any(lp < threshold for lp in generated_log_probs)

# Example Usage with simulated log_probs (as they would come from model.generate(output_scores=True))
evaluator = ConfidenceEvaluator()

# Simulate log_probs from a confident generation
log_probs_high_confidence = [-0.1, -0.05, -0.02, -0.08, -0.01, -0.03] # Closer to 0 means higher probability (exp(0)=1)
# Simulate log_probs from a less confident generation
log_probs_low_confidence = [-0.5, -1.2, -0.8, -2.5, -0.3, -0.9] # Further from 0 means lower probability

print(f"High confidence answer score: {evaluator.calculate_sequence_confidence(log_probs_high_confidence):.4f}")
print(f"Contains low confidence tokens (high): {evaluator.contains_low_confidence_tokens(log_probs_high_confidence)}")
print(f"Low confidence answer score: {evaluator.calculate_sequence_confidence(log_probs_low_confidence):.4f}")
print(f"Contains low confidence tokens (low): {evaluator.contains_low_confidence_tokens(log_probs_low_confidence)}")

实际集成建议： 在使用Hugging Face transformers库时，可以通过model.generate(..., output_scores=True, return_dict_in_generate=True)来获取每个生成token的logits。然后使用model.compute_transition_scores来将这些logits转换为归一化的对数概率。

3.2 基于事实性与连贯性的评估 (Critic Model)

仅仅依靠置信度是不够的，因为一个模型可能非常自信地生成一个完全错误或不连贯的答案（即“一本正经地胡说八道”）。因此，我们需要一个独立的机制来评估生成内容的事实准确性和逻辑连贯性。

实现思路：

1. 小型LLM作为Critic： 训练一个较小的LLM（或微调一个现有模型）作为Critic。

   • 输入： 原始查询、生成的回答、以及用于生成回答的检索文档（如果存在）。
   • 输出： 一个评估分数（例如1-5分），或者一个判断标签（Factually_Correct, Inaccurate, Incomplete, Conflicting），甚至是一个简短的解释。
   • 训练数据： 需要人工标注的“查询-回答-文档-评估”对。

2. 关键词/实体匹配： 从查询中提取关键实体或术语，检查它们是否在生成的回答中得到了准确且上下文相关的提及。

   • 结合检索到的文档，检查生成回答中的关键事实是否与文档内容一致。

3. RAGAS-like Metrics： 借鉴RAGAS等评估框架的思想，内部计算诸如忠实度（Faithfulness）、相关性（Relevance）、答案召回率（Answer Recall）等指标。

   • 忠实度： 评估生成答案中的事实是否都能从提供的源文档中推断出来。
   • 相关性： 评估生成答案与原始查询的相关程度。

代码示例 (Python, 概念性)：

from typing import List, Dict

# 假设我们有一个独立的Critic LLM或一个评估服务
class CriticModel:
    def __init__(self, critic_llm_api_endpoint=None):
        # 实际中这里会集成一个LLM API调用或加载一个本地模型
        self.critic_llm_api_endpoint = critic_llm_api_endpoint
        # For demonstration, we'll simulate its behavior

    def evaluate_response(self, query: str, generated_answer: str, retrieved_docs: List[str]) -> Dict:
        """
        评估生成回答的事实性、连贯性和完整性。
        返回一个字典，包含各项评估指标。
        """
        # 实际场景中，会向critic_llm_api_endpoint发送请求
        # 或者使用一个本地的微调模型进行推理

        # 模拟评估逻辑：
        factuality_score = 0.0 # 0-1之间，越高越好
        completeness_score = 0.0 # 0-1之间
        coherence_score = 0.0 # 0-1之间

        # 启发式判断：
        # 1. 检查是否有明确的否定词或不确定表达
        if "我不知道" in generated_answer or "不确定" in generated_answer:
            factuality_score = 0.2
            completeness_score = 0.2
        elif "错误" in generated_answer or "问题" in generated_answer:
            factuality_score = 0.1
            completeness_score = 0.1

        # 2. 模拟基于关键词的简单事实核查 (非常粗糙，仅为演示)
        # 假设查询是"Who is the CEO of Google?"
        # 假设retrieved_docs中包含"Sundar Pichai is the CEO of Google."
        # 如果generated_answer是"Sundar Pichai is the CEO of Google.", 那么 factuality_score 应该高

        # 更复杂的逻辑会涉及NER、事实抽取、与文档内容的语义比对

        # Let's use a more robust simulation for specific scenarios
        if "capital of France" in query.lower():
            if "paris" in generated_answer.lower():
                factuality_score = 0.95
                completeness_score = 0.9
                coherence_score = 0.95
            elif "london" in generated_answer.lower():
                factuality_score = 0.05
                completeness_score = 0.1
                coherence_score = 0.7 # 语法可能连贯，但事实错误
            else:
                factuality_score = 0.5
                completeness_score = 0.5
                coherence_score = 0.5
        elif "latest news" in query.lower():
            # For real-time queries, retrieved_docs should contain very recent info
            if any("breaking news" in doc.lower() for doc in retrieved_docs) and "2023" in generated_answer: # Simple check
                factuality_score = 0.8
                completeness_score = 0.7
                coherence_score = 0.8
            else:
                factuality_score = 0.3
                completeness_score = 0.4
                coherence_score = 0.6
        else: # Default for other queries
            # A more generic assessment
            if generated_answer.strip() == "":
                factuality_score = 0.0
                completeness_score = 0.0
                coherence_score = 0.0
            else:
                # If it's not empty, give some base scores
                factuality_score = 0.7
                completeness_score = 0.7
                coherence_score = 0.8

                # If retrieved_docs are sparse, completeness might be lower
                if len(retrieved_docs) < 2 and completeness_score > 0.5:
                    completeness_score -= 0.2

                # If generated_answer is too short for a complex query
                if len(generated_answer.split()) < 20 and len(query.split()) > 5:
                    completeness_score -= 0.3

        # Ensure scores are within [0, 1]
        factuality_score = max(0.0, min(1.0, factuality_score))
        completeness_score = max(0.0, min(1.0, completeness_score))
        coherence_score = max(0.0, min(1.0, coherence_score))

        return {
            "factuality_score": factuality_score,
            "completeness_score": completeness_score,
            "coherence_score": coherence_score,
            "requires_more_info": factuality_score < 0.7 or completeness_score < 0.7 # A heuristic for deficit
        }

# Example Usage:
critic = CriticModel()
query1 = "What is the capital of France?"
answer1 = "The capital of France is Paris."
docs1 = ["Paris is the most populous city in France.", "France is a country in Western Europe."]
evaluation1 = critic.evaluate_response(query1, answer1, docs1)
print(f"Evaluation for '{query1}' -> '{answer1}': {evaluation1}")

query2 = "Tell me the latest news on AI safety."
answer2 = "AI safety is an important topic. Many researchers are working on it."
docs2 = ["A recent paper discussed AI alignment.", "The EU is proposing new AI regulations."]
evaluation2 = critic.evaluate_response(query2, answer2, docs2)
print(f"Evaluation for '{query2}' -> '{answer2}': {evaluation2}")

query3 = "Who wrote 'Hamlet'?"
answer3 = "William Shakespeare wrote 'Hamlet'."
docs3 = ["Shakespeare is a famous playwright.", "Hamlet is one of his tragedies."]
evaluation3 = critic.evaluate_response(query3, answer3, docs3)
print(f"Evaluation for '{query3}' -> '{answer3}': {evaluation3}")

query4 = "Explain quantum entanglement in detail."
answer4 = "Quantum entanglement is a phenomenon where two or more particles become linked." # Too brief
docs4 = ["Quantum mechanics describes how subatomic particles behave.", "Entanglement means measuring one particle instantly affects another."]
evaluation4 = critic.evaluate_response(query4, answer4, docs4)
print(f"Evaluation for '{query4}' -> '{answer4}': {evaluation4}")

3.3 基于检索有效性的评估

即使模型生成了答案，最初的检索结果可能质量不高，或者没有覆盖到查询的所有方面。

实现思路：

1. 文档相关性分数：

   • 在初始检索阶段，每个检索到的文档都会有一个与查询相关的分数（例如，向量相似度分数）。
   • 如果最高相关性分数低于某个阈值，或者相关文档数量太少，这表明初始检索可能不充分。

2. 文档覆盖度：

   • 评估检索到的文档是否覆盖了查询中的所有关键实体和概念。
   • 可以通过实体链接、关键词提取等技术来量化。

3. 信息密度：

   • 评估检索到的文档中，有多少比例的信息是真正与查询相关的。低信息密度可能意味着需要更精准的检索。

3.4 结合用户意图和查询类型

某些类型的查询本身就暗示了更高的“认知匮乏度”可能性，例如：

• 实时性查询： “最新的新闻”、“最近的进展”——这些几乎总是需要外部检索。
• 复杂、多方面查询： “分析XX事件对全球经济的影响，并预测未来趋势”——这可能需要多轮、多角度的检索。
• 开放式、探索性查询： “告诉我关于量子物理的一切”——这可能需要引导式的检索。

决策模块可以根据查询的类型，预设一个更高的“认知匮乏度”阈值，或者直接强制进行检索。

四、决策模块：何时启动全新的多源检索？

决策模块是Self-RAG 3.0的“大脑”，它综合利用上述量化指标来做出判断。这里可以采用几种策略：

4.1 阈值与规则系统 (Threshold-based & Rule-based System)

这是最直接的实现方式。为每个“认知匮乏度”指标设置一个阈值，并定义一系列逻辑规则。

规则示例：

• 规则1 (低置信度)： 如果average_log_prob < CONFIDENCE_THRESHOLD 或者 contains_low_confidence_tokens 为 True，则启动检索。
• 规则2 (事实性不足)： 如果critic_evaluation["factuality_score"] < FACTUALITY_THRESHOLD 或者 critic_evaluation["completeness_score"] < COMPLETENESS_THRESHOLD，则启动检索。
• 规则3 (初始检索不足)： 如果max_retrieval_relevance_score < RELEVANCE_THRESHOLD 或者 num_relevant_docs < MIN_DOCS_THRESHOLD，则启动检索。
• 规则4 (强制检索类型)： 如果query_type 为 REAL_TIME_NEWS 或 COMPLEX_ANALYSIS，则强制启动检索。
• 组合规则： 如果（CONFIDENCE_SCORE 低 AND FACTUALITY_SCORE 低）OR QUERY_TYPE 为 REAL_TIME_NEWS，则启动检索。

代码示例 (Python)：

from typing import List, Dict

class DecisionModule:
    def __init__(self,
                 confidence_threshold: float = -1.0, # Average log prob (closer to 0 is better)
                 low_token_confidence_threshold: float = -3.0, # Individual token log prob
                 factuality_threshold: float = 0.7, # Critic score (0-1)
                 completeness_threshold: float = 0.6, # Critic score (0-1)
                 max_retrieval_relevance_threshold: float = 0.7, # 0-1, from initial retrieval
                 min_relevant_docs: int = 2,
                 max_retrieval_rounds: int = 3): # To prevent infinite loops

        self.confidence_threshold = confidence_threshold
        self.low_token_confidence_threshold = low_token_confidence_threshold
        self.factuality_threshold = factuality_threshold
        self.completeness_threshold = completeness_threshold
        self.max_retrieval_relevance_threshold = max_retrieval_relevance_threshold
        self.min_relevant_docs = min_relevant_docs
        self.max_retrieval_rounds = max_retrieval_rounds

        self.confidence_evaluator = ConfidenceEvaluator() # Re-use the simplified evaluator
        self.critic_model = CriticModel() # Re-use the simulated critic

        # 定义一个简单的多源检索器，用于演示
        self.multi_source_retriever = MultiSourceRetriever() 

    def analyze_query_type(self, query: str) -> str:
        """
        根据查询内容分析查询类型 (简化版)。
        """
        query_lower = query.lower()
        if "latest news" in query_lower or "recent update" in query_lower:
            return "REAL_TIME_NEWS"
        elif "explain" in query_lower and len(query.split()) > 5:
            return "COMPLEX_EXPLANATION"
        elif "how to" in query_lower or "guide" in query_lower:
            return "HOW_TO_GUIDE"
        elif "who is" in query_lower or "what is" in query_lower:
            return "FACTUAL_QUESTION"
        return "GENERAL"

    def decide_retrieval(self,
                         query: str,
                         generated_answer: str,
                         generated_log_probs: List[float], # From ConfidenceEvaluator
                         initial_retrieved_docs: List[Dict], # {content: str, score: float}
                         current_round: int) -> Dict:
        """
        根据各项指标决定是否启动全新的多源检索。
        返回决策结果及建议的检索类型。
        """
        if current_round >= self.max_retrieval_rounds:
            print(f"[{current_round}/{self.max_retrieval_rounds}] Max retrieval rounds reached. Stopping.")
            return {"should_retrieve": False, "reason": "Max rounds reached"}

        # 1. 评估生成置信度
        avg_confidence = self.confidence_evaluator.calculate_sequence_confidence(generated_log_probs)
        has_low_confidence_tokens = self.confidence_evaluator.contains_low_confidence_tokens(
            generated_log_probs, self.low_token_confidence_threshold
        )

        # 2. 评估事实性、连贯性、完整性
        # Pass only content strings to the critic, as it doesn't need scores
        critic_evaluation = self.critic_model.evaluate_response(
            query, generated_answer, [doc['content'] for doc in initial_retrieved_docs]
        )

        # 3. 评估初始检索有效性
        max_initial_doc_relevance = 0.0
        if initial_retrieved_docs:
            max_initial_doc_relevance = max(doc['score'] for doc in initial_retrieved_docs)
        num_initial_relevant_docs = len([doc for doc in initial_retrieved_docs if doc['score'] >= 0.6]) # Example threshold

        # 4. 分析查询类型
        query_type = self.analyze_query_type(query)

        # 决策逻辑
        should_retrieve = False
        reason = []
        retrieval_sources = [] # e.g., ["KNOWLEDGE_GRAPH", "WEB_SEARCH"]

        # Rule 1: Low Generation Confidence
        if avg_confidence < self.confidence_threshold or has_low_confidence_tokens:
            should_retrieve = True
            reason.append(f"Low generation confidence (avg:{avg_confidence:.2f}, low_token:{has_low_confidence_tokens})")
            retrieval_sources.append("GENERAL_KNOWLEDGE_BASE") # Default source

        # Rule 2: Factuality/Completeness Deficit (from Critic)
        if critic_evaluation["factuality_score"] < self.factuality_threshold:
            should_retrieve = True
            reason.append(f"Low factuality score ({critic_evaluation['factuality_score']:.2f})")
            retrieval_sources.append("KNOWLEDGE_GRAPH") # Suggest more structured data

        if critic_evaluation["completeness_score"] < self.completeness_threshold:
            should_retrieve = True
            reason.append(f"Low completeness score ({critic_evaluation['completeness_score']:.2f})")
            retrieval_sources.append("GENERAL_KNOWLEDGE_BASE")

        # Rule 3: Initial Retrieval Insufficient
        if max_initial_doc_relevance < self.max_retrieval_relevance_threshold or 
           num_initial_relevant_docs < self.min_relevant_docs:
            if not should_retrieve: # Only add this if not already decided to retrieve
                should_retrieve = True
                reason.append(f"Initial retrieval insufficient (max_rel:{max_initial_doc_relevance:.2f}, num_docs:{num_initial_relevant_docs})")
            retrieval_sources.append("WEB_SEARCH" if query_type == "REAL_TIME_NEWS" else "GENERAL_KNOWLEDGE_BASE")

        # Rule 4: Query Type Demands Retrieval
        if query_type == "REAL_TIME_NEWS":
            should_retrieve = True
            reason.append("Query type is 'REAL_TIME_NEWS'")
            retrieval_sources.append("REAL_TIME_NEWS_API")
        elif query_type == "COMPLEX_EXPLANATION" and not should_retrieve:
            should_retrieve = True
            reason.append("Query type is 'COMPLEX_EXPLANATION'")
            retrieval_sources.append("ACADEMIC_DATABASE")

        # Deduplicate and prioritize retrieval sources
        unique_sources = list(set(retrieval_sources))

        return {
            "should_retrieve": should_retrieve,
            "reason": " | ".join(reason) if reason else "No specific deficit detected, but general assessment suggests no retrieval.",
            "retrieval_sources": unique_sources
        }

# --- Dummy Retriever for demonstration ---
class MultiSourceRetriever:
    def retrieve(self, query: str, sources: List[str]) -> List[Dict]:
        print(f"--- Activating Multi-Source Retrieval for query: '{query}' from sources: {sources} ---")
        retrieved_docs = []
        if "GENERAL_KNOWLEDGE_BASE" in sources:
            retrieved_docs.append({"content": f"Info from KB about {query}.", "score": 0.85})
        if "KNOWLEDGE_GRAPH" in sources:
            retrieved_docs.append({"content": f"Structured facts from KG about {query}.", "score": 0.9})
        if "WEB_SEARCH" in sources:
            retrieved_docs.append({"content": f"Top web results for {query}.", "score": 0.75})
        if "REAL_TIME_NEWS_API" in sources:
            retrieved_docs.append({"content": f"Latest headlines on {query} from news API.", "score": 0.92})
        if "ACADEMIC_DATABASE" in sources:
            retrieved_docs.append({"content": f"Academic papers on {query}.", "score": 0.88})

        # Simulate varying quality based on sources and query
        if "latest news" in query.lower() and "REAL_TIME_NEWS_API" in sources:
            retrieved_docs.append({"content": f"BREAKING: New development in AI safety as of {datetime.now().strftime('%Y-%m-%d')}", "score": 0.95})

        return retrieved_docs

# --- Simulation of a Self-RAG 3.0 Pipeline ---
from datetime import datetime
import random

class SelfRAG3_0_Pipeline:
    def __init__(self, llm_model, decision_module: DecisionModule):
        self.llm = llm_model # The main LLM
        self.decision_module = decision_module

    def run(self, initial_query: str):
        current_query = initial_query
        current_generated_answer = ""
        current_log_probs = []
        current_retrieved_docs = [] # Initial empty or from a first light retrieval

        print(f"--- Self-RAG 3.0 Pipeline Started for Query: '{initial_query}' ---")

        for round_num in range(self.decision_module.max_retrieval_rounds + 1): # +1 for initial generation
            print(f"n--- Round {round_num} ---")

            if round_num == 0:
                # Initial generation without specific retrieval (or with very light internal knowledge)
                current_generated_answer = self.llm.generate(current_query, context_docs=current_retrieved_docs)
                # Simulate log_probs for initial generation (usually lower if no docs)
                current_log_probs = [-0.8, -1.5, -0.7, -1.0, -2.1, -0.9] 
                print(f"[LLM] Initial Generation: {current_generated_answer}")
            else:
                # Decision to retrieve was made in the previous round
                # Now we integrate new retrieved docs and generate again
                full_context_prompt = self._build_prompt_with_docs(current_query, current_retrieved_docs)
                current_generated_answer = self.llm.generate(full_context_prompt, context_docs=current_retrieved_docs)
                # Simulate log_probs, hopefully higher after retrieval
                current_log_probs = [-0.1, -0.05, -0.02, -0.08, -0.01, -0.03] 
                # Add some randomness for simulation
                if random.random() < 0.3: # Simulate occasional low confidence even after retrieval
                     current_log_probs = [-0.5, -1.2, -0.8, -2.5, -0.3, -0.9]
                print(f"[LLM] Generation after Retrieval (Round {round_num}): {current_generated_answer}")

            # Make decision for *next* round based on *current* state
            decision = self.decision_module.decide_retrieval(
                current_query,
                current_generated_answer,
                current_log_probs,
                current_retrieved_docs, # Pass current docs used for generation
                round_num
            )

            print(f"[Decision] Should retrieve for next round: {decision['should_retrieve']}. Reason: {decision['reason']}")

            if not decision["should_retrieve"]:
                print("--- Self-RAG 3.0 Pipeline Finished ---")
                return current_generated_answer, current_retrieved_docs

            # If retrieval is needed, perform it
            print(f"[Retrieval] Initiating retrieval from sources: {decision['retrieval_sources']}")
            newly_retrieved_docs = self.decision_module.multi_source_retriever.retrieve(
                current_query, decision['retrieval_sources']
            )
            current_retrieved_docs.extend(newly_retrieved_docs) # Accumulate docs

            # Simulate updating current_query for next retrieval (e.g., query expansion)
            if round_num == 0: # Only expand query if it's the first time retrieving
                current_query = self._expand_query(current_query, current_generated_answer, current_retrieved_docs)
                print(f"[Query Expansion] Next retrieval query: '{current_query}'")

        print("--- Self-RAG 3.0 Pipeline Finished (Max rounds reached) ---")
        return current_generated_answer, current_retrieved_docs

    def _build_prompt_with_docs(self, query: str, docs: List[Dict]) -> str:
        context = "n".join([doc['content'] for doc in docs])
        return f"Given the following context:n{context}nnAnswer the question: {query}"

    def _expand_query(self, original_query: str, current_answer: str, retrieved_docs: List[Dict]) -> str:
        """
        Simulate query expansion based on current answer and retrieved docs.
        In a real system, another LLM would do this.
        """
        if "latest news" in original_query.lower():
            return f"{original_query} - provide more details from recent sources"
        elif "explain" in original_query.lower() and len(current_answer.split()) < 50:
            return f"{original_query} - elaborate with more examples and in-depth analysis"
        return original_query # No expansion

# --- Dummy LLM for simulation ---
class DummyLLM:
    def generate(self, prompt: str, context_docs: List[Dict]) -> str:
        if "capital of France" in prompt:
            if any("Paris" in doc['content'] for doc in context_docs):
                return "The capital of France is indeed Paris."
            return "I believe the capital of France is Paris, but I should double check."
        elif "latest news on AI safety" in prompt:
            if any("breaking news" in doc['content'] for doc in context_docs):
                return f"Breaking news on AI safety: There are new ethical guidelines proposed today, {datetime.now().strftime('%Y-%m-%d')}."
            return "AI safety is an ongoing discussion, with various ethical and technical challenges."
        elif "quantum entanglement" in prompt:
            if any("particles linked" in doc['content'] for doc in context_docs):
                return "Quantum entanglement is a fascinating quantum mechanical phenomenon where two or more particles become inextricably linked, such that they share the same fate regardless of distance. Measuring one instantly influences the other, a concept Einstein famously called 'spooky action at a distance'."
            return "Quantum entanglement refers to a deep connection between particles. I need more specific details to give a comprehensive explanation."
        return f"This is a generated answer for: '{prompt}'. (Context used: {len(context_docs)} docs)."

# Setup and Run Pipeline:
dummy_llm = DummyLLM()
decision_module = DecisionModule(
    confidence_threshold=-0.5, # Make it easier to trigger retrieval
    low_token_confidence_threshold=-2.0,
    factuality_threshold=0.8,
    completeness_threshold=0.7,
    max_retrieval_relevance_threshold=0.7,
    min_relevant_docs=1,
    max_retrieval_rounds=2 # Limit rounds for demo
)
pipeline = SelfRAG3_0_Pipeline(dummy_llm, decision_module)

# Test Cases
print("n--- Test Case 1: Simple factual question with initial low confidence ---")
final_answer1, final_docs1 = pipeline.run("What is the capital of France?")
print(f"nFinal Answer: {final_answer1}")

print("n--- Test Case 2: Complex, real-time query ---")
final_answer2, final_docs2 = pipeline.run("Tell me the latest news on AI safety.")
print(f"nFinal Answer: {final_answer2}")

print("n--- Test Case 3: Explanatory query requiring depth ---")
final_answer3, final_docs3 = pipeline.run("Explain quantum entanglement in detail.")
print(f"nFinal Answer: {final_answer3}")

4.2 学习型策略 (Learned Policy / Critic Model)

更高级的方法是训练一个专门的模型（通常是一个小型LLM或强化学习代理）来学习何时检索以及检索什么。

实现思路：

1. Critic Model (Decision-making version):
   • 输入： 与上述评估器类似，包括当前查询、LLM的初步回答、各项“认知匮乏度”指标的向量表示，以及已检索到的文档摘要。
   • 输出： 一个二元分类结果（Retrieve / Don't Retrieve），以及一个多分类结果（Source1, Source2, Source3等），甚至是一个生成式的检索指令。
   • 训练：
     • 监督学习： 需要大量人工标注的“查询-模型状态-正确决策”数据。例如，给定某种状态，专家判断此时应该检索还是停止，以及检索哪个源。
     • 强化学习： 将检索决策视为一个动作。奖励函数可以基于最终生成答案的质量（由另一个评估器或人工评估）和检索成本。模型通过试错来学习最优的检索策略。

这种方法允许决策模块学习更复杂的模式和上下文依赖关系，从而做出更智能的决策。例如，它可能学习到对于某些类型的医学查询，即使初步置信度不低，也应该优先检索专业数据库以确保准确性。

五、多源检索的挑战与策略

当决定启动“全新的多源检索”时，我们面临的挑战不仅仅是“是否检索”，更是“如何检索”和“从何检索”。

5.1 检索源的选择与优先级

不同的信息源有不同的特点和适用场景：

检索源类型	特点	适用场景
内部知识库	高度可信，领域专一，更新频率可控	公司文档、产品手册、FAQ、特定领域知识
向量数据库	语义检索，高效查找相似文档	大规模非结构化文本、代码库、用户手册
知识图谱	结构化事实，关系丰富，推理性强	实体关系查询、复杂事实组合、推理问答
实时新闻API	最新信息，时效性强	突发事件、市场动态、科技进展
学术数据库	权威性高，深度专业	科学研究、技术原理、综述分析
网页搜索引擎	广度大，覆盖面广，但噪声多	综合信息、长尾问题、最新但非权威信息
结构化数据库	精确查询，数据准确	用户数据、业务报表、产品参数

决策模块需要根据评估出的“认知匮乏度”类型和查询的特征，智能地选择一个或多个最合适的检索源。例如：

• 如果“事实性评分”低，可能需要优先从知识图谱或结构化数据库获取精确事实。
• 如果查询涉及“最新进展”，则应优先查询实时新闻API或网页搜索引擎。
• 如果“完整性评分”低且查询复杂，可能需要从学术数据库或多个通用知识库进行深度挖掘。

5.2 检索策略的动态调整

仅仅选择来源还不够，检索的“方式”也需要动态调整：

1. 查询重写/扩展： LLM可以根据当前的上下文和已有的检索结果，将原始查询重写、分解为子问题，或添加更多上下文信息，以生成更精确的检索查询。
2. 关键词与语义混合检索： 结合传统的关键词匹配和现代的向量语义匹配，以兼顾精确性和召回率。
3. 迭代式检索： 每次检索不仅是为了获取最终答案，也可能是为了获取更多上下文，从而改进下一轮的检索查询。
4. 过滤与排名： 对检索到的文档进行更严格的过滤和排名，例如基于文档的权威性、更新时间、与查询的匹配度等。

代码示例 (Query Expansion, 概念性):

# Part of the SelfRAG3_0_Pipeline (already included above in _expand_query but expanded here)
class QueryExpander:
    def __init__(self, llm_for_expansion):
        self.llm_for_expansion = llm_for_expansion

    def expand_query(self, original_query: str, current_answer_snippet: str, retrieved_docs_summary: List[str], deficit_reason: str) -> str:
        """
        根据当前状态和认知匮乏度原因，使用LLM来扩展或重写查询。
        """
        prompt = f"""
        Original Query: {original_query}
        Current Answer Snippet: {current_answer_snippet}
        Summaries of Retrieved Documents: {'; '.join(retrieved_docs_summary[:3])}
        Cognitive Deficit Reason: {deficit_reason}

        Based on the information above, please generate a more precise or expanded query that would help find
        more relevant information to address the cognitive deficit. Focus on specific details or missing aspects.
        New Query:
        """
        # In a real system, self.llm_for_expansion would be a dedicated LLM call
        # For simulation, we'll use some heuristics

        if "Low factuality score" in deficit_reason:
            return f"Verify facts for '{original_query}' focusing on '{current_answer_snippet}'"
        elif "Low completeness score" in deficit_reason:
            if "explain" in original_query.lower():
                 return f"Provide more detailed examples and applications for '{original_query}' beyond '{current_answer_snippet}'"
            return f"Elaborate on all aspects of '{original_query}', especially details missing from '{current_answer_snippet}'"
        elif "REAL_TIME_NEWS" in deficit_reason:
            return f"Latest breaking news and developments regarding '{original_query}'"

        return original_query # Fallback if no specific expansion logic applies

5.3 避免检索循环与成本控制

无限次的检索不仅浪费资源，还会导致用户体验下降。

• 最大检索轮次限制： 设置一个硬性上限，防止模型陷入无限循环（已在代码中体现）。
• 成本模型： 评估每次检索的成本（时间、API调用费用），只有当预期收益大于成本时才进行检索。这需要一个更复杂的决策模型，可能涉及强化学习。
• 信息增益评估： 每次检索后，评估新信息对回答质量的潜在提升。如果信息增益不显著，则停止检索。

六、Self-RAG 3.0 的未来展望

Self-RAG 3.0代表了RAG技术向更高级自主性迈进的重要一步。它将LLM从一个被动的知识消费者转变为一个主动的知识探索者。展望未来，我们可以预见以下几个发展方向：

1. 更精细的认知匮乏度模型： 结合多模态信息（如图像、音频）来评估理解程度，而不仅仅是文本。
2. 动态检索源管理： 模型不仅能选择检索源，还能动态地发现和集成新的、未预设的检索源。
3. 个性化检索策略： 根据用户偏好、历史交互和领域知识，定制化的检索策略。
4. 端到端可学习的决策模块： 利用更先进的强化学习或元学习技术，让决策模块能够从与环境的交互中持续优化其检索策略。
5. 与Agentic LLM的深度融合： 将这种自主检索能力与LLM的规划、工具使用等Agent能力相结合，构建更强大的智能体。

七、结语

Self-RAG 3.0 提出的自主决策机制，通过量化LLM的“认知匮乏度”，实现了按需、智能的多源检索。这不仅提升了RAG系统的效率和效果，更赋予了LLM一种宝贵的“自我认知”能力，使其能够像人类一样，在面对不确定或不足时，主动寻求和整合外部知识。我们期待这项技术在未来能够解锁更多AI应用的潜能，推动通用人工智能的不断发展。

感谢大家！