解析 ‘Self-Corrective RAG’：当初始检索结果为空时，如何驱动 Agent 自动调整查询参数并重新入图？

各位同仁，各位对RAG技术充满热情的开发者们：

大家好！今天我们齐聚一堂，共同探讨一个在构建健壮、智能RAG系统时不可避免，且极具挑战性的问题：当RAG的初始检索结果为空时，我们的系统应该如何响应？这不仅仅是一个简单的错误处理，更是RAG技术从“被动响应”迈向“主动适应”的关键一步。我们将深入剖析“自修正RAG”（Self-Corrective RAG，简称SCRAG）这一理念，并着重探讨如何驱动一个智能Agent自动调整查询参数并重新执行检索，最终生成有效响应。

RAG架构的基石与潜在的陷阱

首先，让我们快速回顾一下检索增强生成（Retrieval Augmented Generation, RAG）的核心理念。RAG通过将大型语言模型（LLM）的强大生成能力与外部知识库的精确检索能力相结合，旨在解决LLM知识时效性、事实准确性以及幻觉等问题。一个典型的RAG工作流包括以下几个核心阶段：

1. 用户查询（User Query）: 用户提出问题或需求。
2. 检索（Retrieval）: 系统根据用户查询，从预构建的向量数据库或传统搜索引擎中检索相关文档或文本片段。这通常涉及查询嵌入、向量相似度搜索（如余弦相似度）以及潜在的关键词匹配。
3. 增强（Augmentation）: 将检索到的相关信息作为上下文，与用户原始查询一起输入给LLM。
4. 生成（Generation）: LLM利用提供的上下文和查询，生成最终的响应。

这个流程看似完美，但在实际应用中，尤其是在面对复杂、模糊、或者知识库覆盖不全的查询时，检索阶段往往成为整个链条的薄弱环节。最极端的情况就是：检索结果为空。

当检索模块未能找到任何相关文档时，会发生什么？

• LLM将缺乏必要的上下文，可能被迫“自由发挥”，从而产生不准确、泛泛而谈甚至完全错误的“幻觉”内容。
• 用户体验极差，系统似乎“什么都不知道”，无法提供任何帮助。
• 系统无法有效利用其庞大的知识库，失去了RAG的核心价值。

因此，我们需要一种机制来应对这种“检索失败”的局面，让系统能够自我诊断、自我调整，并尝试再次解决问题。这就是我们今天讨论的重点：Self-Corrective RAG。

Self-Corrective RAG：智能Agent驱动的自适应检索

Self-Corrective RAG的核心思想是赋予系统一种“元认知”能力：当它发现初始检索失败时，能够像人类专家一样，反思失败原因，制定新的策略，调整查询参数，然后再次尝试。这个过程由一个或多个“Agent”来驱动和协调。

一个SCRAG系统通常包含以下关键组件：

1. 初始RAG管道: 执行标准的检索和生成流程。
2. 失败检测模块（Failure Detection Module）: 监控检索结果，识别失败情况（例如：返回文档列表为空、所有文档的相关性分数低于预设阈值）。
3. Agent核心（Agent Core）: 这是SCRAG的大脑，负责分析失败原因，制定修正策略，并决定如何调整查询参数。
4. 查询参数调整模块（Query Parameter Adjustment Module）: 根据Agent的策略，具体修改原始查询或检索器的参数。
5. 循环控制与再执行模块（Loop Control & Re-execution Module）: 管理重试次数，执行修正后的检索，并评估结果。

当初始检索结果为空时，Agent如何自动调整查询参数并重新入图？ 这正是我们接下来要深入探讨的环节。我们假设Agent在检测到空结果后被激活，并拥有与LLM交互的能力，以实现智能决策。

Agent如何分析失败原因？

当检索结果为空时，Agent首先需要尝试理解为什么会发生这种情况。可能的内部原因分析包括：

• 查询过于具体或狭隘: 用户查询包含了太多限制性词语或不常见的专有名词，导致数据库中难以直接匹配。
• 查询过于模糊或歧义: 查询缺乏足够的上下文或关键词，导致检索器无法聚焦。
• 知识库覆盖不足: 知识库中确实没有与查询直接相关的信息。
• 检索策略不当: 当前使用的检索方法（如仅关键词、仅向量相似度）不适合当前查询。
• 过滤器设置过于严格: 查询附带的元数据过滤器（如日期、作者、文档类型）将所有潜在文档排除在外。
• top_k 值设置过低: 即使有相关文档，它们也可能因为排名靠后而被截断。

Agent可以利用LLM的推理能力来辅助进行这种分析。例如，Agent可以将原始查询和“检索结果为空”这一事实作为输入，要求LLM推断可能的失败原因。

import openai # 假设使用OpenAI API
from typing import List, Dict, Any

# 模拟一个检索函数
def retrieve_documents(query: str, filters: Dict = None, top_k: int = 5, search_type: str = "vector_search") -> List[Dict]:
    """
    模拟从知识库中检索文档。
    在真实场景中，这将调用向量数据库或搜索引擎API。
    为了演示，我们模拟在某些情况下返回空结果。
    """
    print(f"尝试检索：查询='{query}', 过滤器={filters}, top_k={top_k}, 搜索类型='{search_type}'")

    # 模拟一些预设的失败条件
    if "Apple Q3 2023 财报" in query and filters.get("source") == "official_reports":
        print("模拟：此特定组合导致空结果。")
        return []
    if "区块链扩容方案" in query and search_type == "keyword_search":
        print("模拟：关键词搜索对该查询可能无效。")
        return []
    if "AI伦理准则" in query and filters.get("date_range", {}).get("end") == "2020-01-01":
        print("模拟：日期过滤器过严。")
        return []
    if "量子计算最新进展" in query and top_k < 3:
        print("模拟：top_k过低导致错过相关文档。")
        return []

    # 模拟在其他情况下可能返回一些结果
    # 这里我们只返回一个占位符，因为我们主要关注空结果的处理
    if "AI" in query or "RAG" in query or "区块链" in query:
        if "Apple Q3 2023 财报" not in query: # 避免与上面的空结果冲突
            print("模拟：找到一些文档。")
            return [{"id": "doc_1", "content": f"关于 {query} 的一些信息。", "metadata": {"source": "general_knowledge"}}]

    print("模拟：未找到任何文档。")
    return []

# 假设这是一个LLM API调用函数
def call_llm(prompt: str, model: str = "gpt-4") -> str:
    """
    模拟调用LLM API，用于推理和生成。
    """
    # 实际应用中会调用 openai.ChatCompletion.create 或类似接口
    # 为了演示，我们根据prompt模拟一些响应
    if "分析为什么检索结果为空" in prompt:
        if "Apple Q3 2023 财报" in prompt and "source='official_reports'" in prompt:
            return "分析结果：查询过于具体，且过滤器'official_reports'可能过于严格。建议：放宽过滤器，或泛化查询。"
        elif "区块链扩容方案" in prompt and "keyword_search" in prompt:
            return "分析结果：关键词搜索可能无法捕获语义相关性。建议：尝试向量相似度搜索。"
        elif "AI伦理准则" in prompt and "date_range" in prompt:
            return "分析结果：日期范围过滤器可能排除了相关文档。建议：放宽日期范围。"
        elif "量子计算最新进展" in prompt and "top_k=2" in prompt:
            return "分析结果：top_k值可能过低，导致相关文档被截断。建议：增加top_k值。"
        else:
            return "分析结果：无法明确判断，可能是查询过于具体，也可能是知识库覆盖不足。建议：首先尝试泛化查询。"
    elif "基于此分析，请提出新的查询策略" in prompt:
        if "放宽过滤器" in prompt:
            return "策略：移除或泛宽过滤器。例如，将'source=official_reports'改为'source IN (official_reports, news_articles, press_releases)'。"
        elif "泛化查询" in prompt:
            return "策略：从原始查询中移除不必要的细节或限定词。例如，将'Apple Q3 2023 财报'改为'Apple 财报'或'Apple 财务业绩'。"
        elif "向量相似度搜索" in prompt:
            return "策略：将检索类型从关键词搜索切换为向量相似度搜索。"
        elif "增加top_k值" in prompt:
            return "策略：将top_k值增加到更高的数值，例如从5增加到10。"
        elif "日期范围" in prompt:
            return "策略：移除或扩展日期范围过滤器。"
        else:
            return "策略：尝试移除查询中的特定日期或版本信息，或尝试使用同义词。"
    elif "将原始查询和策略转化为新的查询参数" in prompt:
        if "Apple Q3 2023 财报" in prompt and "泛化查询" in prompt:
            return "{'query': 'Apple 财务业绩', 'filters': {}}"
        elif "AI伦理准则" in prompt and "放宽日期范围" in prompt:
            return "{'query': 'AI伦理准则', 'filters': {'date_range': {'start': '2018-01-01', 'end': '2024-12-31'}}}"
        elif "区块链扩容方案" in prompt and "向量相似度搜索" in prompt:
            return "{'query': '区块链扩容方案', 'search_type': 'vector_search'}"
        elif "量子计算最新进展" in prompt and "增加top_k值" in prompt:
            return "{'query': '量子计算最新进展', 'top_k': 10}"
        else:
            return "{'query': '泛化后的查询', 'filters': {}, 'top_k': 10}"
    return "LLM 模拟响应。"

Agent可以构建一个Prompt，将当前状态（原始查询、已使用的参数、检索结果为空的事实）提供给LLM，请求其进行分析。

def analyze_failure_with_llm(original_query: str, current_params: Dict) -> str:
    prompt = f"""
    检索系统在处理以下查询时未能找到任何相关文档：
    原始查询: "{original_query}"
    当前检索参数: {current_params}
    检索结果: 空

    请分析为什么检索结果为空，并提出可能的修正方向。
    """
    print("n--- Agent请求LLM分析失败原因 ---")
    analysis = call_llm(prompt)
    print(analysis)
    return analysis

Agent的修正策略与查询参数调整

基于LLM的分析或预设的启发式规则，Agent将制定一个或多个修正策略，并将这些策略转化为具体的查询参数调整。

下表总结了几种常见的失败场景、Agent的修正策略以及对应的参数调整：

失败场景	Agent分析	修正策略	查询参数调整示例
查询过于具体/狭隘	包含过多限定词、日期、版本号、或不常见的专有名词。	泛化查询 (Query Generalization)	1. 移除特定限定词: 将 "Apple Q3 2023 财报" 变为 "Apple 财报" 或 "Apple 财务业绩"。 2. 移除日期/版本: 将 "Python 3.9 新特性" 变为 "Python 新特性"。 3. 使用上位词: 将 "特斯拉 Model Y 续航里程" 变为 "电动汽车 续航里程"。
过滤器过于严格	日期范围、文档类型、作者等元数据过滤器排除了所有文档。	放宽/移除过滤器 (Filter Relaxation/Removal)	1. 扩展日期范围: 将 date_range: {start: 2023-07-01, end: 2023-09-30} 变为 date_range: {start: 2022-01-01, end: 2024-12-31}。 2. 拓宽文档类型: 将 doc_type: 'official_report' 变为 doc_type: ['official_report', 'news_article', 'press_release']，或直接移除 doc_type 过滤器。 3. 移除不必要的过滤器。
关键词匹配不足	原始查询词汇与知识库文档词汇不匹配，但语义可能相关。	关键词扩展/同义词替换 (Keyword Expansion/Synonymy)	1. LLM生成同义词: 将 "区块链扩容方案" 扩展为 "区块链扩容方案 OR 区块链吞吐量 OR Layer 2 解决方案"。 2. 使用预设词典/本体: 将 "碳中和" 扩展为 "碳中和 OR 净零排放 OR 气候中和"。
检索类型不匹配	语义查询效果不佳，或关键词查询对概念性问题无效。	切换检索策略 (Toggle Search Strategy)	1. 从向量搜索切换到关键词搜索: 当查询过于字面化或包含大量专有名词时。 2. 从关键词搜索切换到向量搜索: 当查询更侧重概念性或语义相似性时。
top_k 值设置过低	相关文档可能存在，但排名靠后，未被包含在初始结果中。	增加 top_k 值 (Increase top_k)	1. 逐步增加: 将 top_k 从 5 增加到 10，再到 20。通常采用指数或线性递增策略。
复杂查询难以一次检索	查询包含多个子问题或需要多方面信息的综合性问题。	查询分解 (Query Decomposition)	1. LLM分解: 将 "比较A和B的优缺点及其在C领域的应用" 分解为 "A的优缺点", "B的优缺点", "A在C领域的应用", "B在C领域的应用"。 2. 独立检索并聚合: 对每个子查询进行检索，然后将所有检索结果合并，再提供给LLM进行综合生成。

代码实现：Agent如何将策略转化为参数

Agent的核心在于将分析结果和策略转化为实际的Python字典，以供检索函数使用。这通常可以通过LLM的Function Calling能力或基于模式匹配的启发式规则来实现。

def formulate_strategy_with_llm(analysis_result: str) -> str:
    prompt = f"""
    基于此分析：
    {analysis_result}

    请提出新的查询策略。策略应清晰、具体，以便后续转化为参数调整。
    """
    print("n--- Agent请求LLM制定修正策略 ---")
    strategy = call_llm(prompt)
    print(strategy)
    return strategy

def adjust_parameters_with_llm(original_query: str, current_params: Dict, strategy: str) -> Dict:
    prompt = f"""
    原始查询: "{original_query}"
    当前检索参数: {current_params}
    修正策略: "{strategy}"

    请将原始查询和策略转化为新的查询参数。以JSON格式返回，包含 'query', 'filters', 'top_k', 'search_type' 等字段。
    例如：{{'query': '新的查询', 'filters': {{'source': 'all'}}, 'top_k': 10, 'search_type': 'vector_search'}}
    """
    print("n--- Agent请求LLM将策略转化为参数 ---")

    # LLM通常会返回一个字符串化的JSON
    json_string = call_llm(prompt)

    try:
        new_params = eval(json_string) # 使用eval来解析模拟的LLM返回的字典字符串
        # 实际应用中，这里会使用json.loads()并进行严格的schema验证
        # 确保LLM返回的JSON格式正确，并包含预期的字段
        if not isinstance(new_params, dict):
            raise ValueError("LLM返回的参数不是一个字典。")
        # 填充缺失的参数，保持默认值或从当前参数继承
        new_params.setdefault('query', original_query)
        new_params.setdefault('filters', current_params.get('filters', {}))
        new_params.setdefault('top_k', current_params.get('top_k', 5))
        new_params.setdefault('search_type', current_params.get('search_type', 'vector_search'))

        # 对于filters，需要进行合并而不是直接覆盖，除非策略明确指示覆盖
        if 'filters' in new_params and 'filters' in current_params:
            # 简单合并：新参数中的过滤器会覆盖旧参数中同名的过滤器
            # 更复杂的逻辑可能需要根据策略进行智能合并或移除
            merged_filters = current_params['filters'].copy()
            merged_filters.update(new_params['filters'])
            new_params['filters'] = merged_filters

        # 如果策略是泛化查询，则新的query会替换原始query
        # 如果策略是放宽过滤器，新的filters会更新
        # 如果策略是增加top_k，新的top_k会更新
        # 如果策略是切换搜索类型，新的search_type会更新

        print(f"新参数: {new_params}")
        return new_params
    except (json.JSONDecodeError, ValueError) as e:
        print(f"LLM返回的参数解析失败: {e}. 尝试使用启发式回退。")
        # 回退策略：如果LLM无法正确生成JSON，可以采用预设的启发式规则
        return heuristic_parameter_adjustment(original_query, current_params, strategy)

def heuristic_parameter_adjustment(original_query: str, current_params: Dict, strategy: str) -> Dict:
    """
    当LLM解析失败或作为更简单的回退机制时，采用启发式调整参数。
    """
    new_params = current_params.copy()
    new_query = original_query

    if "泛化查询" in strategy:
        # 简单泛化：移除日期和版本号
        new_query = original_query.replace(" Q3 2023", "").replace(" 2023", "").replace(" 最新", "")
        new_params['query'] = new_query
        print(f"启发式泛化查询: '{new_query}'")
    elif "放宽过滤器" in strategy:
        if 'filters' in new_params:
            # 移除所有过滤器作为最宽松的策略
            new_params['filters'] = {}
            print("启发式移除所有过滤器。")
        # 或者可以针对性地放宽，例如：
        # if 'source' in new_params['filters']:
        #     new_params['filters']['source'] = ['official_reports', 'news_articles', 'blogs']
    elif "增加top_k值" in strategy:
        new_params['top_k'] = new_params.get('top_k', 5) + 5 # 每次增加5
        print(f"启发式增加top_k到: {new_params['top_k']}")
    elif "向量相似度搜索" in strategy:
        new_params['search_type'] = 'vector_search'
        print("启发式切换到向量搜索。")
    elif "关键词扩展" in strategy:
        # 简单的关键词扩展：添加一些预设的通用同义词
        if "区块链扩容方案" in new_query:
            new_params['query'] = f"{new_query} OR 区块链吞吐量 OR Layer 2"
            print(f"启发式关键词扩展: '{new_params['query']}'")

    return new_params

完整的Agent驱动的自修正RAG循环

现在，让我们把这些模块整合起来，构建一个完整的自修正RAG工作流。这个工作流将包含一个重试循环，直到找到结果或达到最大重试次数。

import json
import time

def self_corrective_rag_pipeline(
    initial_query: str, 
    max_retries: int = 3,
    initial_filters: Dict = None,
    initial_top_k: int = 5,
    initial_search_type: str = "vector_search"
) -> List[Dict]:

    current_query = initial_query
    current_params = {
        'query': initial_query,
        'filters': initial_filters if initial_filters is not None else {},
        'top_k': initial_top_k,
        'search_type': initial_search_type
    }

    retrieved_docs = []
    retry_count = 0

    print(f"--- 启动自修正RAG流程，原始查询: '{initial_query}' ---")

    while retry_count < max_retries:
        print(f"n--- 第 {retry_count + 1} 次尝试 ---")

        # 1. 执行检索
        retrieved_docs = retrieve_documents(
            query=current_params['query'],
            filters=current_params['filters'],
            top_k=current_params['top_k'],
            search_type=current_params['search_type']
        )

        # 2. 失败检测
        if retrieved_docs:
            print(f"✅ 检索成功！找到 {len(retrieved_docs)} 份文档。")
            # 在真实RAG中，这里会进入LLM生成阶段
            print(f"生成阶段：LLM会使用这些文档生成答案。例如：'根据检索到的信息，{retrieved_docs[0]['content']}'")
            return retrieved_docs # 成功，退出循环
        else:
            print("❌ 检索结果为空。激活Agent进行自修正...")

            # 3. Agent分析失败原因
            analysis_result = analyze_failure_with_llm(initial_query, current_params)

            # 4. Agent制定修正策略
            strategy = formulate_strategy_with_llm(analysis_result)

            # 5. Agent调整查询参数
            # 在这里，我们可以根据分析和策略，逐步、有逻辑地调整参数
            # 例如，可以维护一个优先级列表：先泛化查询，再放宽过滤器，最后增加top_k等

            # 简单示例：直接让LLM基于策略生成新参数
            # 更健壮的系统会有一个专门的参数调整器，根据策略类型进行硬编码或模板化调整
            new_params = adjust_parameters_with_llm(initial_query, current_params, strategy)

            # 更新当前参数
            current_params.update(new_params) 
            # 确保query在每次迭代时都是最新修改过的
            current_params['query'] = new_params.get('query', current_params['query']) 

            retry_count += 1
            time.sleep(0.5) # 模拟思考时间

    print(f"n--- 达到最大重试次数 ({max_retries})，仍未能找到相关文档。---")
    print("建议：告知用户知识库中可能没有相关信息，或尝试手动干预。")
    return [] # 最终失败，返回空

# --- 演示 SCRAG 的不同场景 ---

print("nn--- 场景1: 查询过于具体且过滤器严格 ---")
# 初始查询：Apple Q3 2023 财报 (太具体) + source='official_reports' (太严格)
self_corrective_rag_pipeline(
    initial_query="Apple Q3 2023 财报",
    initial_filters={"source": "official_reports"},
    max_retries=3
)

print("nn--- 场景2: 关键词搜索对语义问题无效 ---")
# 初始查询：区块链扩容方案 + keyword_search
self_corrective_rag_pipeline(
    initial_query="区块链扩容方案",
    initial_search_type="keyword_search",
    max_retries=3
)

print("nn--- 场景3: 日期过滤器过于严格 ---")
# 初始查询：AI伦理准则 + date_range (截止到2020年，而AI伦理更多是近几年发展)
self_corrective_rag_pipeline(
    initial_query="AI伦理准则",
    initial_filters={"date_range": {"start": "2018-01-01", "end": "2020-01-01"}},
    max_retries=3
)

print("nn--- 场景4: top_k 值过低 ---")
# 初始查询：量子计算最新进展 + top_k=2
self_corrective_rag_pipeline(
    initial_query="量子计算最新进展",
    initial_top_k=2,
    max_retries=3
)

print("nn--- 场景5: 综合复杂查询（模拟无法一次性解决，最终可能成功） ---")
self_corrective_rag_pipeline(
    initial_query="人工智能在医疗领域的最新应用及其伦理挑战", # 这个查询本身可能就复杂，需要多轮修正
    initial_filters={"doc_type": "scientific_paper"},
    initial_top_k=3,
    max_retries=4
)

策略优先级与智能回退

在实际的SCRAG系统中，Agent不会盲目地尝试所有策略。通常会有一个策略优先级列表和智能回退机制：

1. 最轻微的调整: 例如，仅仅增加 top_k。
2. 查询泛化: 移除时间、地点、版本等具体限定词。
3. 过滤器放宽: 逐步移除或扩展元数据过滤器。
4. 关键词扩展: 引入同义词或相关概念。
5. 切换检索类型: 从向量搜索切换到关键词搜索，或反之。
6. 查询分解: 将复杂查询拆解为多个简单查询。
7. 最终回退: 如果所有尝试都失败，则向用户返回“未找到信息”的提示，并可能建议用户修改查询。

每次重试时，Agent可以根据当前的失败原因和已尝试的策略，决定下一个要应用的策略。LLM在这一决策过程中发挥着关键作用，它可以根据历史尝试和结果，动态地调整策略。

架构考量

一个生产级的Self-Corrective RAG系统在架构上需要考虑：

• Agent的实现: 可以是一个基于规则的有限状态机，也可以是一个由LLM驱动的更智能的控制器。LLM驱动的Agent更灵活，但成本和延迟更高。
• 状态管理: Agent需要记录每次尝试的查询、参数、结果以及失败原因，以便在多轮修正中做出知情决策。
• 并发与缓存: 频繁的重试可能导致高延迟和高成本。应考虑对检索结果进行缓存，并优化并行检索。
• 可观测性: 记录Agent的决策路径、每次修正的效果，以便调试和优化。
• 用户反馈: 即使Agent最终找到了结果，也需要评估其质量。未来的SCRAG系统甚至可以融入用户反馈，作为Agent学习和改进的信号。

挑战与展望

尽管Self-Corrective RAG前景广阔，但仍面临一些挑战：

• 计算成本与延迟: 每次重试都会增加API调用次数和处理时间，可能影响用户体验。
• 过度泛化: 如果Agent过于积极地泛化查询或放宽过滤器，可能会检索到大量不相关的文档，导致LLM生成质量下降。
• 策略选择的鲁棒性: 如何确保Agent在各种复杂场景下都能选择最有效的修正策略？这依赖于高质量的失败分析和策略制定。
• LLM的可靠性: 如果Agent依赖LLM进行分析和参数调整，LLM的“幻觉”问题可能会导致错误的修正方向。需要精心设计Prompt和验证机制。
• 知识库的局限性: 无论Agent多么智能，如果知识库中根本不存在相关信息，任何修正都无济于事。

展望未来，Self-Corrective RAG将继续朝着更加智能、自适应的方向发展。我们可以预见以下趋势：

• 更深度的LLM集成: LLM不仅用于分析和策略制定，甚至可以直接生成并执行代码来动态调整检索逻辑。
• 强化学习: Agent可以通过与环境的交互（即检索结果的质量）进行学习，优化其修正策略。
• 多模态检索与修正: 将文本、图像、视频等多种模态的数据纳入检索范围，并进行相应的自修正。
• 个性化修正: 根据用户的历史行为和偏好，调整修正策略。

Self-Corrective RAG是构建真正智能、健壮RAG系统的必经之路。它将RAG从一个简单的“查询-检索-生成”管道，提升为一个能够自我诊断、自我学习和自我适应的动态系统，从而显著提升用户体验和系统的可靠性。

提升RAG系统健壮性的关键一步

Self-Corrective RAG通过智能Agent驱动的迭代修正，有效应对了初始检索结果为空的挑战。它将启发式规则与大型语言模型的推理能力相结合，赋予了RAG系统从失败中学习并调整策略的能力，是构建更可靠、更用户友好AI应用的重要进展。