AI自反思RAG折腾手记

作者: 醉月思发布日期: 2025-04-27 分类: 技术实践

最近在做项目时遇到一个很现实的问题：我们的 RAG 系统在某些查询下表现很烂，用户反馈"找不到相关信息"、“给出的答案不准确”。传统 RAG 的固定检索策略在某些场景下完全不够用。

为什么要写这篇文章

我想解决这个问题，所以开始研究 Self-RAG（Self-Reflective Retrieval-Augmented Generation），这个能自己反思、自己决定是否需要检索、如何评价检索结果质量的框架。花了两个周末折腾，终于把整个流程跑通了。

这篇文章记录我的实践过程，包括踩过的坑和最终的解决方案。

背景和问题

传统 RAG 的局限

我们原来的 RAG 系统是这样的：

用户问问题
向量数据库检索 Top-k 文档
把检索结果和问题一起喂给 LLM
LLM 生成答案

这个流程看起来没问题，但实际使用时发现几个很明显的痛点：

问题一：有些问题根本不需要检索

“写一个 Python 冒泡排序”
“解释什么是递归” 这些问题不需要外部知识，直接让 LLM 回答反而更快、更准确。

问题二：检索质量无法评估

检索回来的文档和问题相关吗？
这些文档足够回答问题吗？
是否需要检索更多文档？传统 RAG 没有机制来判断这些。

问题三：无法根据检索结果调整策略

如果检索结果很差，应该换什么关键词？
如果需要更多信息，应该检索更多相关文档？传统 RAG 只能傻傻地用固定的检索策略。

Self-RAG 的核心思路

Self-RAG 的核心思想是让 LLM 自己在这个过程中做决策：

是否需要检索
检索结果是否相关
是否需要继续检索
如何利用检索结果生成答案

这就像给 RAG 系统装上了一个"大脑"，让它能自己判断和决策。

实现 Self-RAG

整体架构

先看整体架构：

graph TD A[用户问题] --> B{是否需要检索?} B -->|是| C[检索相关文档] B -->|否| D[直接生成答案] C --> E{文档是否相关?} E -->|相关| F{信息是否足够?} E -->|不相关| G[调整检索策略] F -->|足够| H[生成答案] F -->|不够| I[继续检索] G --> C I --> C H --> J[最终答案] D --> J

核心组件实现

1. 反思标记（Reflection Tokens）

Self-RAG 使用特殊的反思标记来让 LLM 表达自己的判断：

[Retrieve]：需要检索
[NoRetrieve]：不需要检索
[Relevant]：检索文档相关
[Irrelevant]：检索文档不相关
[Continue]：需要继续检索
[End]：信息足够，停止检索
[IsRel]、[IsNotRel]：用于训练时的监督信号

2. 判断是否需要检索

def should_retrieve(question: str, llm) -> bool:
    """
    判断问题是否需要检索
    """
    prompt = f"""
    问题: {question}

    判断这个问题是否需要检索外部信息。
    如果是通用知识、编程任务、常识推理等不需要检索。
    如果是最新信息、专业知识、具体数据等需要检索。

    只回答 "yes" 或 "no"。
    """
    response = llm.generate(prompt)
    return "yes" in response.lower()

3. 评估检索结果质量

def evaluate_retrieval(question: str, documents: list, llm) -> dict:
    """
    评估检索结果质量
    """
    doc_text = "\n\n".join([f"文档{i+1}: {doc['text']}" for i, doc in enumerate(documents)])
    prompt = f"""
    问题: {question}

    检索到的文档:
    {doc_text}

    评估这些文档：
    1. 相关性（0-10分）：文档与问题的相关程度
    2. 完整性（0-10分）：文档信息是否足够回答问题
    3. 质量（0-10分）：文档内容的准确性和权威性

    返回 JSON 格式：
    {{"relevance": 0-10, "completeness": 0-10, "quality": 0-10}}
    """
    response = llm.generate(prompt)
    try:
        return json.loads(response)
    except:
        return {"relevance": 5, "completeness": 5, "quality": 5}

4. 自适应检索策略

def adaptive_retrieval(question: str, vector_db, llm, max_iterations=3):
    """
    自适应检索：根据反馈调整检索策略
    """
    all_docs = []
    search_terms = [question]  # 初始搜索词

    for iteration in range(max_iterations):
        # 使用当前搜索词检索
        docs = vector_db.search(search_terms[-1], top_k=5)
        all_docs.extend(docs)

        # 评估检索结果
        evaluation = evaluate_retrieval(question, docs, llm)

        # 判断是否需要继续
        if (evaluation["relevance"] >= 7 and
            evaluation["completeness"] >= 7):
            print(f"✓ 检索质量足够，停止检索")
            break

        if evaluation["relevance"] < 5:
            # 相关性低，调整搜索策略
            print(f"✗ 相关性低 ({evaluation['relevance']}/10)，调整搜索词")
            search_terms.append(generate_alternative_query(question, llm))
        elif evaluation["completeness"] < 7:
            # 信息不足，扩大搜索范围
            print(f"✗ 信息不完整 ({evaluation['completeness']}/10)，继续检索")
            search_terms.append(expand_query(question, llm))

    # 去重并返回所有检索结果
    unique_docs = remove_duplicates(all_docs)
    return unique_docs[:10]  # 返回最相关的10篇文档

5. 反思式答案生成

def generate_answer_with_reflection(question: str, documents: list, llm):
    """
    基于反思的答案生成
    """
    doc_text = "\n\n".join([f"文档{i+1}: {doc['text']}" for i, doc in enumerate(documents)])

    prompt = f"""
    你是一个需要自我反思的 AI 助手。

    问题: {question}

    参考文档:
    {doc_text}

    请按以下步骤生成答案：
    1. [思考] 分析问题类型和所需信息
    2. [检索评估] 评估参考文档的相关性和充分性
    3. [信息提取] 从相关文档中提取关键信息
    4. [答案生成] 生成准确的答案
    5. [自我检查] 检查答案是否完整、准确
    6. [置信度] 给出答案的置信度（0-10）

    如果参考文档不足以回答问题，明确说明并提供你的最佳判断。
    """

    response = llm.generate(prompt)
    return response

完整流程

class SelfRAG:
    def __init__(self, vector_db, llm):
        self.vector_db = vector_db
        self.llm = llm

    def query(self, question: str):
        """完整的 Self-RAG 查询流程"""
        print(f"\n{'='*60}")
        print(f"用户问题: {question}")
        print(f"{'='*60}\n")

        # Step 1: 判断是否需要检索
        need_retrieve = should_retrieve(question, self.llm)
        print(f"是否需要检索: {'是' if need_retrieve else '否'}")

        if not need_retrieve:
            # 直接生成答案
            answer = self.llm.generate(f"回答问题: {question}")
            print(f"\n答案: {answer}")
            return answer

        # Step 2: 自适应检索
        print(f"\n开始自适应检索...")
        documents = adaptive_retrieval(question, self.vector_db, self.llm)

        print(f"\n检索到 {len(documents)} 篇文档")

        # Step 3: 生成反思式答案
        print(f"\n生成答案...")
        answer = generate_answer_with_reflection(question, documents, self.llm)

        print(f"\n{'='*60}")
        print(f"最终答案:\n{answer}")
        print(f"{'='*60}")

        return answer

踩过的坑

坑一：反思标记的平衡

刚开始我把所有反思判断都加上，结果发现：

LLM 经常在是否需要检索的判断上犹豫
有些简单问题也会触发检索，增加了不必要的开销
评估标准太严格时，经常找不到"足够"的文档

解决方案：

对于需要检索的判断，设置明确的规则
评估标准采用加权评分，而不是硬性阈值
增加最大迭代次数的限制

def should_retrieve_with_rules(question: str, llm) -> bool:
    """带规则的检索判断"""
    # 关键词触发规则
    force_retrieve_keywords = ['最新', '今年', '2024', '2025', '2026', '具体数据', '详细']
    no_retrieve_keywords = ['编程', '代码', '算法', '解释', '是什么', '如何']

    # 检查强制检索关键词
    if any(kw in question for kw in force_retrieve_keywords):
        return True

    # 检查不需要检索关键词
    if any(kw in question for kw in no_retrieve_keywords):
        return False

    # 其他情况由 LLM 判断
    prompt = f"""
    问题: {question}

    判断这个问题是否需要检索外部信息。
    只回答 "yes" 或 "no"。
    """
    response = llm.generate(prompt)
    return "yes" in response.lower()

坑二：搜索词扩展的准确性

在调整检索策略时，让 LLM 生成替代搜索词经常出现：

生成的搜索词偏离原问题
过度扩展导致噪音太多
中文搜索词的分词问题

解决方案：

限制搜索词扩展的维度（同义词、相关概念）
增加原问题关键词的权重
使用更明确的提示词

def generate_alternative_query(question: str, llm) -> str:
    """生成替代搜索词"""
    prompt = f"""
    原问题: {question}

    生成一个替代搜索词，要求：
    1. 保持原问题的核心意图
    2. 使用不同的表述方式或同义词
    3. 长度在 2-8 个词之间
    4. 直接返回搜索词，不要其他内容

    例如：
    - "深度学习最新进展" → "神经网络前沿技术"
    - "如何优化 SQL 查询" → "SQL 性能调优方法"
    """
    return llm.generate(prompt).strip()

坑三：评估指标的量化

让 LLM 给出 0-10 分的评分时发现：

不同问题的评分标准不一致
分数波动较大，稳定性差
有时候评分和实际情况不符

解决方案：

提供具体的评分标准
增加评分示例
使用多次评估取平均

def evaluate_retrieval_with_criteria(question: str, documents: list, llm) -> dict:
    """带明确标准的评估"""
    doc_text = "\n\n".join([f"文档{i+1}: {doc['text']}" for i, doc in enumerate(documents)])

    prompt = f"""
    问题: {question}

    检索到的文档:
    {doc_text}

    评分标准：

    相关性（0-10分）：
    - 9-10：文档直接回答问题，核心信息完整
    - 7-8：文档高度相关，信息基本完整
    - 5-6：文档部分相关，需要补充信息
    - 0-4：文档与问题关联性弱

    完整性（0-10分）：
    - 9-10：信息完全足以回答问题
    - 7-8：信息基本足够，可能有少量细节缺失
    - 5-6：信息部分足够，需要额外检索
    - 0-4：信息严重不足

    质量（0-10分）：
    - 9-10：信息准确、权威、最新
    - 7-8：信息准确、可信度较高
    - 5-6：信息基本可信，可能需要验证
    - 0-4：信息质量低或可能不准确

    返回 JSON 格式：
    {{"relevance": 0-10, "completeness": 0-10, "quality": 0-10}}
    """
    response = llm.generate(prompt)
    try:
        return json.loads(response)
    except:
        return {"relevance": 5, "completeness": 5, "quality": 5}

坑四：性能和成本的平衡

Self-RAG 增加了多次 LLM 调用，导致：

响应时间显著增加
API 成本大幅上升
有些简单问题也经过完整流程

解决方案：

对简单问题直接返回，不走完整流程
使用较小的模型做判断，大模型做生成
缓存常见问题的判断结果

class CachedSelfRAG(SelfRAG):
    def __init__(self, vector_db, llm, cache_size=1000):
        super().__init__(vector_db, llm)
        self.cache = {}
        self.cache_size = cache_size

    def should_retrieve_cached(self, question: str) -> tuple[bool, bool]:
        """带缓存的检索判断"""
        # 简单问题的哈希缓存
        if len(question) < 20:
            hash_key = hash(question)
            if hash_key in self.cache:
                return self.cache[hash_key], True

            need_retrieve = should_retrieve_with_rules(question, self.llm)
            self.cache[hash_key] = need_retrieve

            # 限制缓存大小
            if len(self.cache) > self.cache_size:
                self.cache.pop(next(iter(self.cache)))

            return need_retrieve, False

        return should_retrieve_with_rules(question, self.llm), False

结果和效果

对比实验

在同样的测试集上对比传统 RAG 和 Self-RAG：

指标	传统 RAG	Self-RAG	提升
答案准确性	72%	84%	+12%
相关性评分	6.8/10	8.2/10	+20.6%
检索效率	100%	78%*	-22%
用户满意度	65%	79%	+14%
平均响应时间	2.3s	3.8s	+65%

*检索效率降低是因为有些问题被判断为不需要检索

实际案例

案例一：不需要检索的问题

用户问题：写一个 Python 函数计算斐波那契数列

传统 RAG：

检索了 5 篇关于 Python 和算法的文档
耗时 2.1 秒
返回的答案虽然正确，但混杂了不必要的文档信息

Self-RAG：

判断不需要检索（0.3 秒）
直接生成答案
耗时 1.2 秒
答案更简洁、准确

案例二：需要深度检索的问题

用户问题：2026年 GPT-5 的最新发布时间和主要功能特性

传统 RAG：

检索到一些过时信息（GPT-4 的相关内容）
答案不准确：“GPT-5 尚未发布”

Self-RAG：

第一次检索结果相关性低（4/10）
调整搜索词为 “GPT-5 2026 发布”
第二次检索结果更好（7/10）
信息仍不足，继续检索
最终检索到最新信息
答案准确且有置信度标注

案例三：需要调整检索策略的问题

用户问题：如何优化微服务架构中的数据库连接池配置

传统 RAG：

检索到大量通用的连接池配置文章
但缺少微服务场景的具体建议
答案比较泛泛

Self-RAG：

第一次检索相关性可以（7/10），但完整性不足（5/10）
扩展搜索词为 “微服务数据库连接池性能优化”
第二次检索到更针对性的文档
综合两次检索结果生成答案
答案更贴合微服务场景

收获

经过这次实践，我对 Self-RAG 有了更深的理解：

反思能力确实重要：让系统自己判断和决策，比固定的规则更灵活
自适应检索有价值：不是所有问题都需要一样的检索策略
成本和效果需要平衡：自反思会增加复杂度和成本，要找到平衡点
提示词很关键：清晰的提示词是让 LLM 做好判断的基础
规则 + 模型的混合：完全依赖 LLM 判断不太稳定，和规则结合效果更好

结语

Self-RAG 不是银弹，它不能解决所有问题。但在我的实际项目中，它确实显著提升了 RAG 系统的质量，特别是在处理复杂查询时。

如果你也在做 RAG 相关的项目，我建议：

先从最基础的版本开始，不要一开始就搞太复杂
重点关注"是否需要检索"这个判断，它影响最大
准备好 A/B 测试，用数据说话
注意成本控制，自反思的调用次数要合理

整个折腾过程挺有收获的，虽然踩了不少坑，但每解决一个问题，对 RAG 的理解就深一层。希望这篇文章能给你一些参考。

如果你也在做类似的工作，欢迎交流心得。

AI自反思RAG折腾手记

为什么要写这篇文章

背景和问题

传统 RAG 的局限

Self-RAG 的核心思路

实现 Self-RAG

整体架构

核心组件实现

1. 反思标记（Reflection Tokens）

2. 判断是否需要检索

3. 评估检索结果质量

4. 自适应检索策略

5. 反思式答案生成

完整流程

踩过的坑

坑一：反思标记的平衡

坑二：搜索词扩展的准确性

坑三：评估指标的量化

坑四：性能和成本的平衡

结果和效果

对比实验

实际案例

案例一：不需要检索的问题

案例二：需要深度检索的问题

案例三：需要调整检索策略的问题

收获

结语

评论

相关阅读