1.迭代式检索增强(Iterative Retrieval) #

传统的RAG流程通常为一次性检索：用户提出问题后，向知识库检索相关文档，再结合文档内容与问题用大语言模型（LLM）进行回答。这种方式遇到难以覆盖或理解的问题时，往往会因为文档缺失、信息覆盖不全或初步检索偏差，导致回答准确率不理想。

迭代式检索增强机制通过多步、递进式检索来优化这一流程，核心思想如下：

1. 循环改进检索：

   • 首轮检索获得初步相关文档，基于返回内容与原始问题生成初步答案。
   • 利用该答案或中间推理过程，引导新的补充检索，寻找遗漏、查证或加深理解的上下文。
   • 结合新获得的信息，不断完善和修正最终答案。

2. 优势：

   • 能发现一次检索难以覆盖的关键信息，提高召回率和答案可信度。
   • 支持复杂、开放性问题的逐步解析，特别适合知识分布广泛、文档互相关联的场景。

3. 常见实现策略：

   • 首先用原始 Query 检索，再用生成内容提取新 Query 或补充说明进行迭代。
   • 每轮聚合所有文档片段，或仅按需追加新片段，防止信息冗余和模型上下文超载。
   • 终止条件可以设定迭代轮数、答案置信度、信息重叠率等。

4. 应用效果：

   • 在QA系统、企业级知识库、法律/医疗文档等复杂检索场景下，显著提升答案的完整性和专业性。

迭代式检索增强即为一种模块化的迭代RAG实现，其将传统RAG流程封装为单步迭代单元，并可循环调用以实现多轮逐步增强。实例化时可以按需指定检索器、LLM及答案模板，实现灵活扩展和定制优化。

这一方法为面向复杂任务的智能问答系统提供了强有力的系统优化手段，是RAG系统发展过程中极具价值的技术进阶。

1.1 IterativeRetrieval.py #

# 导入List、Dict、Any、Optional这些类型，用于类型注解
from typing import List, Dict, Any, Optional

# 导入ConfigDict，Pydantic配置专用
from pydantic import ConfigDict

# 导入基础的检索器基类
from langchain_core.retrievers import BaseRetriever

# 导入Document文档类
from langchain_core.documents import Document

# 导入基础LLM类型
from langchain_core.language_models import BaseLanguageModel

# 导入PromptTemplate提示模版类
from langchain_core.prompts import PromptTemplate

# 导入自定义llm对象
from llm import llm

# 导入自定义的get_vector_store函数
from vector_store import get_vector_store

# 定义单次RAG迭代模块：可重复使用的RAG流程
class SingleRAGIteration:
    """单次RAG迭代模块：完整的检索-增强-生成过程"""

    # 初始化方法，设置检索器、LLM与可选的答案模板
    def __init__(self, retriever: BaseRetriever, llm: BaseLanguageModel, answer_template: Optional[str] = None):
        # 保存检索器对象
        self.retriever = retriever
        # 保存LLM对象
        self.llm = llm
        # 若未指定答案模板，则使用默认模板
        template = answer_template or (
            "已知以下相关信息：\n\n{context}\n\n"
            "用户查询：{query}\n\n"
            "请根据上述信息，准确全面地回答用户问题。\n\n答案："
        )
        # 构建答案生成模板
        self.answer_template = PromptTemplate(
            input_variables=["context", "query"], template=template
        )

    # 执行单次RAG迭代
    def execute(self, query: str, k: int = 4) -> Dict[str, Any]:
        """
        执行单次RAG迭代

        流程：
        1. 在知识库中进行检索，获得上下文信息
        2. 大语言模型基于问题和上下文生成答案
        """
        # 步骤1：检索相关文档
        docs = self.retriever._get_relevant_documents(query, k=k)
        # 步骤2：构建上下文字符串
        context = "\n\n".join([f"文档 {i+1}:\n{doc.page_content}" for i, doc in enumerate(docs)])
        # 步骤3：构造prompt，并通过LLM生成答案
        prompt = self.answer_template.format(context=context, query=query)
        answer = self.llm.invoke(prompt).content.strip()
        # 返回结果字典，包括查询、答案、检索到的文档及数目
        return {
            "query": query,
            "answer": answer,
            "retrieved_documents": docs,
            "num_documents": len(docs)
        }

# 定义迭代式检索增强RAG系统
class IterativeRetrievalRAG:
    """迭代式检索增强RAG系统：通过多次完整的RAG流程迭代来逐步提升答案质量"""

    # 初始化方法，设置检索器、LLM与最大迭代次数
    def __init__(self, retriever: BaseRetriever, llm: BaseLanguageModel, max_iterations: int = 2):
        # 保存检索器对象
        self.retriever = retriever
        # 保存LLM对象
        self.llm = llm
        # 最大迭代次数
        self.max_iterations = max_iterations
        # 创建单次RAG迭代模块实例
        self.single_iteration = SingleRAGIteration(retriever, llm)
        # 构建用于生成新查询的提示模板
        self.iterative_query_template = PromptTemplate(
            input_variables=["original_query", "previous_answer"],
            template=(
                "原始问题：{original_query}\n\n"
                "基于上一次的回答：{previous_answer}\n\n"
                "请基于上述信息，生成一个更精确的查询，用于检索更相关的文档。"
                "新查询应该结合原始问题和上一次答案中的关键信息。\n\n新查询："
            )
        )

    # 构建迭代查询：根据上次答案与原始问题生成新查询
    def _build_iterative_query(self, original_query: str, previous_answer: str) -> str:
        """
        构建迭代查询：将第一次迭代的答案与原始问题合并，形成新的查询

        虽然前一次答案可能不够准确，但它们为后续检索提供了重要的线索和方向
        """
        # 使用模板格式化生成prompt
        prompt = self.iterative_query_template.format(
            original_query=original_query,
            previous_answer=previous_answer
        )
        # 通过LLM生成新查询
        new_query = self.llm.invoke(prompt).content.strip()
        return new_query

    # 生成最终答案（包含多步迭代）
    def generate_answer(self, original_query: str, k: int = 4) -> Dict[str, Any]:
        """
        执行迭代式检索增强生成

        流程：
        第一次迭代：
        1. 接收用户原始问题
        2. 在知识库中进行检索，获得初始上下文信息
        3. 大语言模型基于问题和上下文生成初步答案

        第二次迭代（及后续迭代）：
        1. 将上一次迭代的答案与原始问题合并，形成新的查询
        2. 使用新查询在知识库中重新检索，获得更精确的上下文信息
        3. 大语言模型基于原始问题和新上下文生成最终答案
        """
        # 用于保存每轮迭代的结果
        iteration_results = []
        # 当前查询初始化为原始用户问题
        current_query = original_query

        # 按照设定的迭代次数循环
        for iteration in range(1, self.max_iterations + 1):
            # 打印当前迭代轮次
            print(f"\n【第 {iteration}/{self.max_iterations} 次迭代】")
            print(f"当前查询：{current_query}")
            # 执行一次RAG流程，得到答案和上下文
            result = self.single_iteration.execute(current_query, k=k)
            # 保存单轮结果
            iteration_results.append({
                "iteration": iteration,
                "query": current_query,
                "answer": result["answer"],
                "retrieved_documents": result["retrieved_documents"],
                "num_documents": result["num_documents"]
            })
            # 打印部分答案预览
            print(f"生成答案：{result['answer'][:100]}...")
            # 如果不是最后一次迭代，根据当前回答和原始问题生成下一次查询
            if iteration < self.max_iterations:
                current_query = self._build_iterative_query(original_query, result["answer"])
                print(f"构建下一次迭代查询：{current_query}")

        # 获取最后一轮结果用于最终输出
        final_result = iteration_results[-1]
        # 返回最终答案及所有迭代过程
        return {
            "original_query": original_query,
            "final_answer": final_result["answer"],
            "final_retrieved_documents": final_result["retrieved_documents"],
            "num_documents": final_result["num_documents"],
            "iteration_results": iteration_results,
            "total_iterations": len(iteration_results)
        }

# 定义简单的向量检索器（用于演示）
class SimpleVectorRetriever(BaseRetriever):
    """简单的向量检索器"""

    # 向量库对象（用于检索）
    vector_store: Any
    # 默认返回文档数
    k: int = 4
    # Pydantic配置，允许任意类型
    model_config = ConfigDict(arbitrary_types_allowed=True)

    # 获取相关文档方法
    def _get_relevant_documents(self, query: str, **kwargs) -> List[Document]:
        """执行向量检索"""
        # 可通过k参数控制返回数目，否则使用默认值
        k = kwargs.get("k", self.k)
        # 调用向量库的相似度查询方法
        docs = self.vector_store.similarity_search_with_score(query, k=k)
        # 存储最终文档对象
        result_docs = []
        # 遍历每个检索结果
        for doc, distance in docs:
            # 构造Document对象，附带得分等元数据
            result_docs.append(Document(
                page_content=doc.page_content,
                metadata={
                    **doc.metadata,
                    "score": float(distance),
                    "retrieval_method": "vector"
                }
            ))
        # 返回所有文档对象
        return result_docs

# 初始化组件
# 创建向量库
vector_store = get_vector_store(
    persist_directory="chroma_db",
    collection_name="iterative_retrieval"
)

# 创建简单的向量检索器
simple_retriever = SimpleVectorRetriever(
    vector_store=vector_store,
    k=4
)

# 创建迭代式检索增强RAG系统（默认2次迭代）
rag_system = IterativeRetrievalRAG(
    retriever=simple_retriever,
    llm=llm,
    max_iterations=2
)

# 示例文档集合，每个字符串为一个独立的大段文本
documents = [
    # 文档1：关于人工智能与神经网络
    "人工智能（AI）是计算机科学的一个分支，旨在创建能够执行通常需要人类智能的任务的系统。"
    "机器学习是人工智能的核心技术之一，它使计算机能够从数据中学习，而无需明确编程。"
    "深度学习是机器学习的一个子集，使用人工神经网络来模拟人脑的工作方式。"
    "神经网络由多个层组成，每层包含多个神经元，通过反向传播算法进行训练。",
    # 文档2：关于NLP与Transformer
    "自然语言处理（NLP）是人工智能的一个分支，专注于使计算机能够理解、解释和生成人类语言。"
    "Transformer架构是NLP领域的重要突破，它使用自注意力机制来处理序列数据。"
    "BERT和GPT是Transformer架构的两个重要应用，分别用于理解任务和生成任务。"
    "BERT通过双向编码器理解上下文，GPT通过自回归生成器生成文本。",
    # 文档3：关于计算机视觉与卷积神经网络
    "计算机视觉是人工智能的另一个重要分支，专注于使计算机能够理解和分析视觉信息。"
    "卷积神经网络（CNN）是计算机视觉的核心技术，特别适合处理图像数据。"
    "图像分类、目标检测和图像分割是计算机视觉的三个主要任务。"
    "ResNet、YOLO和U-Net分别是这三个任务的代表性模型。",
]

# 索引文档到向量库
print("正在初始化知识库...")
vector_store.add_texts(
    documents,
    metadatas=[{"topic": "人工智能"}, {"topic": "自然语言处理"}, {"topic": "计算机视觉"}]
)
print("知识库初始化完成！\n")

# 分隔线，输出迭代示例头信息
print("="*60)
print("迭代式检索增强示例")
print("="*60)

# 定义示例查询问题
query = "什么是深度学习？它和神经网络有什么关系？"

# 执行迭代式检索增强生成
result = rag_system.generate_answer(query, k=4)

# 输出分隔线和最终结果
print("\n" + "="*60)
print("最终结果")
print("="*60)
# 输出原始用户问题
print(f"\n原始查询：{result['original_query']}")
# 输出最终答案
print(f"\n最终答案：\n{result['final_answer']}")
# 输出迭代总数
print(f"\n总迭代次数：{result['total_iterations']}")
# 输出最终返回的文档数
print(f"\n最终检索到的文档数量：{result['num_documents']}")

# 输出每次迭代的详细信息
print("\n" + "="*60)
print("迭代过程详情")
print("="*60)
for iter_result in result['iteration_results']:
    # 输出当前迭代轮次
    print(f"\n【第 {iter_result['iteration']} 次迭代】")
    # 输出当轮查询内容
    print(f"查询：{iter_result['query']}")
    # 输出当轮生成的答案（只显示前200字）
    print(f"答案：{iter_result['answer'][:200]}...")
    # 输出当轮检索到文档数
    print(f"检索到的文档数量：{iter_result['num_documents']}")

1.2 执行流程 #

1.2.1 核心思想 #

迭代式检索增强采用“迭代优化”策略：

• 第一次迭代：使用原始查询进行检索，生成初步答案
• 后续迭代：基于上一次答案和原始问题，生成新的查询，重新检索
• 逐步优化：通过多次迭代逐步提升答案质量和检索精度
• 最终答案：使用最后一轮迭代的结果作为最终答案

1.2.2 执行流程 #

阶段一：初始化

# 1. 创建向量存储实例
vector_store = get_vector_store(
    persist_directory="chroma_db",
    collection_name="iterative_retrieval"
)

# 2. 创建简单的向量检索器
simple_retriever = SimpleVectorRetriever(
    vector_store=vector_store,
    k=4
)

# 3. 创建迭代式检索增强RAG系统
rag_system = IterativeRetrievalRAG(
    retriever=simple_retriever,
    llm=llm,
    max_iterations=2  # 默认2次迭代
)

初始化时：

• 创建向量存储实例
• 创建简单的向量检索器
• 创建迭代式RAG系统，配置最大迭代次数
• 系统内部会创建单次RAG迭代模块

阶段二：知识库初始化

# 添加文档到向量库
vector_store.add_texts(
    documents,
    metadatas=[...]
)

索引过程：

• 将示例文档添加到向量库
• 每个文档附带元数据（topic）
• 文档会被向量化并存储

阶段三：迭代式查询处理

query = "什么是深度学习？它和神经网络有什么关系？"
result = rag_system.generate_answer(query, k=4)

完整流程：

1. 用户提交原始查询
2. 第一次迭代：
   • 使用原始查询进行检索
   • 生成初步答案
3. 构建迭代查询（如果不是最后一次迭代）：
   • 调用 _build_iterative_query(original_query, previous_answer)
   • 使用LLM基于原始问题和上一次答案生成新查询
4. 第二次迭代（及后续迭代）：
   • 使用新查询进行检索
   • 生成更精确的答案
5. 返回最终结果：
   • 包含原始查询、最终答案、所有迭代过程等

1.2.3 类图 #

1.2.4 时序图 #

1.2.4.1 完整迭代式RAG流程时序图 #

1.2.4.2 单次RAG迭代详细流程 #

1.2.4.3 迭代查询构建详细流程 #

1.2.5 关键设计要点 #

1. 迭代式检索流程

原始查询
    ↓
第一次迭代:
  检索 → 生成初步答案
    ↓
构建迭代查询 (基于原始查询 + 初步答案)
    ↓
第二次迭代:
  使用新查询检索 → 生成更精确答案
    ↓
最终答案

2. 迭代查询生成示例

原始查询: "什么是深度学习？它和神经网络有什么关系？"

第一次迭代答案: "深度学习是机器学习的一个子集，使用人工神经网络来模拟人脑的工作方式。神经网络由多个层组成，每层包含多个神经元..."

构建的迭代查询: "深度学习和神经网络的具体关系，神经网络的层结构和神经元工作原理，深度学习如何通过神经网络实现"

第二次迭代答案: (基于新查询检索到更相关的文档，生成更精确的答案)

3. 迭代优化原理

• 第一次迭代：
  • 使用原始查询，可能检索到相关但不完全匹配的文档
  • 生成初步答案，包含部分关键信息
• 迭代查询构建：
  • 结合原始问题和初步答案中的关键信息
  • 生成更精确、更具体的查询
• 后续迭代：
  • 使用新查询检索，可能找到更相关的文档
  • 生成更准确、更全面的答案

4. 迭代查询模板

默认迭代查询模板：

原始问题：{original_query}

基于上一次的回答：{previous_answer}

请基于上述信息，生成一个更精确的查询，用于检索更相关的文档。
新查询应该结合原始问题和上一次答案中的关键信息。

新查询：

该模板：

• 包含原始问题
• 包含上一次的答案
• 要求生成更精确的查询
• 要求结合原始问题和答案中的关键信息

5. 单次RAG迭代模块

SingleRAGIteration 封装了完整的RAG流程：

• 检索：使用检索器获取相关文档
• 增强：整合文档内容作为上下文
• 生成：使用LLM基于上下文生成答案

这样可以：

• 代码复用：多次迭代复用相同的RAG流程
• 模块化：清晰的职责分离
• 易于测试：可以单独测试单次迭代

6. 迭代结果结构

返回结果包含：

• original_query：原始用户查询
• final_answer：最终答案（最后一轮迭代的答案）
• final_retrieved_documents：最终检索到的文档
• num_documents：最终检索到的文档数量
• iteration_results：所有迭代过程的详细结果
  • 每轮迭代包含：iteration、query、answer、retrieved_documents、num_documents
• total_iterations：总迭代次数

7. 优势与应用场景

优势：

• 逐步优化：通过多次迭代逐步提升答案质量
• 查询精化：基于初步答案生成更精确的查询
• 信息融合：结合原始问题和初步答案中的关键信息
• 可配置：支持自定义最大迭代次数

适用场景：

• 复杂查询：需要多轮检索才能找到完整答案
• 信息不足：第一次检索可能信息不完整
• 答案优化：需要逐步完善答案质量
• 研究场景：需要深入挖掘相关信息

8. 技术细节

• 迭代次数控制：
  • 默认最大迭代次数为2
  • 可通过 max_iterations 参数配置
  • 最后一次迭代不生成新查询
• 查询构建：
  • 使用LLM基于原始问题和上一次答案生成
  • 要求结合关键信息，生成更精确的查询
• 结果保存：
  • 保存每轮迭代的完整结果
  • 便于分析和调试

9. 迭代过程示例

假设查询："什么是深度学习？它和神经网络有什么关系？"

第一次迭代：

查询: "什么是深度学习？它和神经网络有什么关系？"
检索: 4个相关文档（关于AI、机器学习、深度学习的基础介绍）
答案: "深度学习是机器学习的一个子集，使用人工神经网络..."

构建迭代查询：

基于原始问题和初步答案，生成新查询:
"深度学习和神经网络的具体关系，神经网络的层结构和神经元工作原理，深度学习如何通过神经网络实现"

第二次迭代：

查询: "深度学习和神经网络的具体关系，神经网络的层结构和神经元工作原理..."
检索: 4个相关文档（更聚焦于深度学习和神经网络的详细内容）
答案: "深度学习是机器学习的一个子集，它使用多层神经网络来学习数据的表示..."

1.2.6 与其他方法的对比 #

该设计通过“迭代优化+查询精化”策略，在复杂查询场景下提供更高质量的答案，适用于需要逐步完善答案的RAG应用。

2. 智能自判别检索增强(SelfRAG) #

SelfRAG在传统的检索生成流程基础上，引入了大模型对检索结果与答案支持性的自判别能力。系统不仅要检索相关文档、基于上下文生成答案，还要让大模型对生成的答案进行自我验证，包括：

• 上下文信息是否充分支撑生成的答案？
• 是否还有与问题紧密相关但未覆盖的信息需要补充检索？
• 答案是否全面且没有关键遗漏？

SelfRAG的主要流程

核心环节包括：

1. 智能相关性判别
   对检索到的文档逐条判别其与问题的相关性，仅保留高相关内容，过滤掉无关或噪音信息。

2. 上下文增强与答案生成
   利用过滤后的高相关文档作为上下文，由大语言模型生成第一版答案。

3. 自判别校验
   利用大语言模型再生成一段自检prompt，主动判别上述答案是否被上下文充分支持，有无遗漏，结果分为“有支撑/无支撑”“完整/不完整”。

4. 智能补充检索
   若判别为“无支撑”或“不完整”，系统据此构造补充性检索请求，再次检索&增强上下文，然后再次生成/修正答案，循环上述过程，直至答案完善或达到设定的最大尝试次数。

优势及适用场景

SelfRAG与简单RAG相比，能大幅提升答案正确率、完整性和鲁棒性，尤其适用于：

• 知识分布广且不均匀、文档片段质量参差不齐的场景
• 结果要求高准确性、高覆盖率，如企业知识库、医疗/法律QA等
• 多文档、多轮追问需要智能聚合与查缺补漏的问题

2.1 SelfRAG.py #

# 导入typing模块中的List、Dict、Any、Optional类型，用于类型注解
from typing import List, Dict, Any, Optional

# 导入pydantic中的ConfigDict，用于模型参数配置
from pydantic import ConfigDict

# 导入langchain_core.retrievers中的BaseRetriever，作为检索器基类
from langchain_core.retrievers import BaseRetriever

# 导入langchain_core.documents中的Document类，用于封装文档内容
from langchain_core.documents import Document

# 导入langchain_core.language_models中的BaseLanguageModel，作为LLM基类
from langchain_core.language_models import BaseLanguageModel

# 导入langchain_core.prompts中的PromptTemplate，用于构建提示模版
from langchain_core.prompts import PromptTemplate

# 导入自定义llm对象
from llm import llm

# 导入自定义的get_vector_store函数，用于获取向量库
from vector_store import get_vector_store

# 导入正则表达式模块
import re

# 定义文档相关性判别模块
class DocumentRelevanceJudge:
    """文档相关性判别模块：智能判断返回的文档与用户问题的相关程度"""

    # 构造函数，初始化llm及判别模版
    def __init__(self, llm: BaseLanguageModel):
        self.llm = llm

        # 构建相关性判别的PromptTemplate
        self.judge_template = PromptTemplate(
            input_variables=["query", "documents"],
            template=(
                "请判断以下文档是否与用户问题相关。请只回答\"相关\"或\"不相关\"。\n\n"
                "用户问题：{query}\n\n"
                "文档内容：\n{documents}\n\n"
                "请逐个判断每个文档的相关性，格式如下：\n"
                "文档1: [相关/不相关]\n"
                "文档2: [相关/不相关]\n"
                "...\n\n"
                "请直接给出判断结果："
            )
        )

    # 判别文档相关性
    def judge(self, query: str, documents: List[str]) -> List[str]:
        """
        判断文档相关性
        返回：相关文档列表
        """
        # 格式化所有文档，按“文档<i>: 内容”排列
        docs_text = "\n".join([f"文档{i+1}: {doc}" for i, doc in enumerate(documents)])

        # 用prompt模版格式化出最终判别文本
        prompt = self.judge_template.format(query=query, documents=docs_text)
        # 通过llm模型进行判别，得到结果字符串
        judgment = self.llm.invoke(prompt).content.strip()

        # 解析返回的判别结果
        relevant_docs = []
        lines = judgment.strip().split("\n")

        # 遍历每一行判别结果
        for i, line in enumerate(lines):
            # 如果包含“相关”但不包含“不相关”，认为第i个文档相关
            if "相关" in line and "不相关" not in line and i < len(documents):
                relevant_docs.append(documents[i])

        # 返回所有相关文档
        return relevant_docs

# 定义答案可信度验证模块
class AnswerCredibilityVerifier:
    """答案可信度验证模块：验证答案是否有上下文支持，是否完整解答问题"""

    # 构造函数，初始化llm和各类判别模版
    def __init__(self, llm: BaseLanguageModel):
        self.llm = llm

        # 构建上下文支持度检查的PromptTemplate
        self.support_template = PromptTemplate(
            input_variables=["query", "context", "answer"],
            template=(
                "请判断以下答案是否能在提供的上下文中找到依据。请只回答\"有依据\"或\"无依据\"。\n\n"
                "用户问题：{query}\n\n"
                "上下文信息：\n{context}\n\n"
                "生成的答案：\n{answer}\n\n"
                "判断：这个答案是否能在上下文中找到支持依据？\n\n"
                "请直接回答："
            )
        )

        # 构建问题解答完整性检查的PromptTemplate
        self.completeness_template = PromptTemplate(
            input_variables=["query", "answer"],
            template=(
                "请判断以下答案是否完整回答了用户问题。请只回答\"完整\"或\"不完整\"。\n\n"
                "用户问题：{query}\n\n"
                "生成的答案：\n{answer}\n\n"
                "判断：这个答案是否完整回答了用户问题？\n\n"
                "请直接回答："
            )
        )

    # 验证答案可信度
    def verify(self, query: str, context: str, answer: str) -> Dict[str, bool]:
        """
        验证答案可信度
        返回：包含has_support和is_complete的字典
        """
        # 生成上下文支持度判定的prompt
        support_prompt = self.support_template.format(query=query, context=context, answer=answer)
        # 用llm进行支持度判别
        support_result = self.llm.invoke(support_prompt).content.strip()
        # 只要包含“有依据”且不包含“无依据”，支持即为True
        has_support = "有依据" in support_result and "无依据" not in support_result

        # 生成答案完整性判别的prompt
        completeness_prompt = self.completeness_template.format(query=query, answer=answer)
        # 用llm进行完整性判别
        completeness_result = self.llm.invoke(completeness_prompt).content.strip()
        # 只要包含“完整”且不包含“不完整”，完整性即为True
        is_complete = "完整" in completeness_result and "不完整" not in completeness_result

        # 返回支持与完整性判别的结果
        return {
            "has_support": has_support,
            "is_complete": is_complete
        }

# 定义查询增强模块
class QueryEnhancer:
    """查询增强模块：当检索结果不理想时，触发查询增强策略"""

    # 构造函数，初始化llm和增强模板
    def __init__(self, llm: BaseLanguageModel):
        self.llm = llm

        # 构建查询增强的PromptTemplate
        self.enhancement_template = PromptTemplate(
            input_variables=["query"],
            template=(
                "请对以下问题进行查询重写，生成更精确的检索查询。要求：\n"
                "1. 保持原问题的核心含义\n"
                "2. 添加更多关键词和细节\n"
                "3. 使用更具体的表达方式\n\n"
                "原问题：{query}\n\n"
                "请生成1个增强后的查询（只返回查询内容，不要其他说明）："
            )
        )

    # 查询增强（重写查询）
    def enhance(self, query: str) -> str:
        """增强查询"""
        # 用prompt格式化用户查询
        prompt = self.enhancement_template.format(query=query)
        # 调用llm进行增强，去除首尾空格
        enhanced_query = self.llm.invoke(prompt).content.strip()
        # 返回增强后的查询
        return enhanced_query

# 定义SelfRAG检索器
class SelfRAGRetriever(BaseRetriever):
    """SelfRAG检索器：在检索阶段嵌入智能判别模块"""

    # 成员变量：向量库、相关性判别器、查询增强器、top_k值、最大重试次数、模型配置
    vector_store: Any
    relevance_judge: DocumentRelevanceJudge
    query_enhancer: QueryEnhancer
    k: int = 4
    max_retry: int = 2
    model_config = ConfigDict(arbitrary_types_allowed=True)

    # 内部方法：检索相关文档流程
    def _get_relevant_documents(self, query: str, **kwargs) -> List[Document]:
        """执行检索和相关性判别"""
        # 获取本次检索使用的k和最大重试次数
        k = kwargs.get("k", self.k)
        max_retry = kwargs.get("max_retry", self.max_retry)

        # 当前查询
        current_query = query
        # 当前重试次数
        retry_count = 0

        # 在最大重试次数范围内循环
        while retry_count <= max_retry:
            # 步骤1：执行向量检索，检索数量为2k
            candidate_docs = self.vector_store.similarity_search_with_score(current_query, k=k * 2)
            # 仅提取文档正文
            documents = [doc.page_content for doc, _ in candidate_docs]

            # 如果未检索到文档
            if not documents:
                # 若还可重试，则增强原始query再检索
                if retry_count < max_retry:
                    print(f"[SelfRAG] 检索失败，尝试查询增强（第{retry_count + 1}次）...")
                    current_query = self.query_enhancer.enhance(query)
                    retry_count += 1
                    continue
                else:
                    # 没有文档，返回空结果
                    return []

            # 步骤2：智能判别相关性
            relevant_docs = self.relevance_judge.judge(current_query, documents)

            # 如果有相关文档
            if relevant_docs:
                # 转为Document对象，并补充元数据
                result_docs = []
                for doc_content in relevant_docs:
                    # 找回原始文档以保留其metadata
                    original_doc = None
                    for doc, _ in candidate_docs:
                        if doc.page_content == doc_content:
                            original_doc = doc
                            break
                    # 合成Document对象，增加检索方式和相关性信息
                    result_docs.append(Document(
                        page_content=doc_content,
                        metadata={
                            **(original_doc.metadata if original_doc else {}),
                            "retrieval_method": "self_rag",
                            "judged_relevant": True
                        }
                    ))
                # 截取k条并返回
                return result_docs[:k]
            else:
                # 如没有相关文档且还能重试，进行查询增强
                if retry_count < max_retry:
                    print(f"[SelfRAG] 没有找到相关文档，尝试查询增强（第{retry_count + 1}次）...")
                    current_query = self.query_enhancer.enhance(query)
                    retry_count += 1
                else:
                    # 不能再重试，返回空
                    return []
        # 如果所有重试都失败，返回空
        return []

# 定义SelfRAG系统
class SelfRAG:
    """SelfRAG系统：智能自判别检索增强生成"""

    # 构造函数：初始化retriever与llm，以及最大迭代次数
    def __init__(self, retriever: SelfRAGRetriever, llm: BaseLanguageModel, max_iterations: int = 3):
        self.retriever = retriever
        self.llm = llm
        self.max_iterations = max_iterations

        # 创建答案可信度验证器
        self.verifier = AnswerCredibilityVerifier(llm)

        # 构建答案生成PromptTemplate
        self.answer_template = PromptTemplate(
            input_variables=["context", "query"],
            template=(
                "请根据以下上下文信息回答用户问题。要求：\n"
                "1. 答案必须基于提供的上下文信息\n"
                "2. 如果上下文中没有相关信息，请明确说明\"无法从提供的信息中找到答案\"\n"
                "3. 答案要准确、完整、有条理\n\n"
                "上下文信息：\n{context}\n\n"
                "用户问题：{query}\n\n"
                "答案："
            )
        )

    # 生成答案主流程
    def generate_answer(self, query: str, k: int = 4) -> Dict[str, Any]:
        """
        执行SelfRAG流程
        流程：
        1. 检索和文档相关性判别
        2. 生成答案
        3. 答案可信度验证
        4. 动态策略选择（根据验证结果决定是否重新生成或返回）
        """
        # 当前迭代次数
        iteration = 0

        # 在最大迭代次数内循环
        while iteration < self.max_iterations:
            iteration += 1
            print(f"\n[SelfRAG] ===== 第{iteration}轮迭代 =====")

            # 步骤1：检索相关文档
            docs = self.retriever._get_relevant_documents(query, k=k)

            # 如果未检索到文档
            if not docs:
                if iteration < self.max_iterations:
                    print("[SelfRAG] 未检索到相关文档，继续下一轮迭代...")
                    continue
                else:
                    # 最后一轮也未命中，直接返回失败信息
                    return {
                        "query": query,
                        "answer": "抱歉，无法找到相关信息来回答您的问题。",
                        "retrieved_documents": [],
                        "num_documents": 0,
                        "iterations": iteration,
                        "verification": {"has_support": False, "is_complete": False}
                    }

            # 步骤2：构建上下文字符串(按“文档N:
            #            内容”拼接)
            context = "\n\n".join([f"文档 {i+1}:\n{doc.page_content}" for i, doc in enumerate(docs)])

            # 步骤2-3：最多尝试3次生成答案
            for attempt in range(3):
                print(f"[SelfRAG] 第{attempt + 1}次生成尝试...")

                # 构造prompt进行答案生成
                prompt = self.answer_template.format(context=context, query=query)
                answer = self.llm.invoke(prompt).content.strip()

                # 步骤3：进行答案可信度验证
                verification = self.verifier.verify(query, context, answer)

                print(f"[SelfRAG] 验证结果：上下文支持={verification['has_support']}, 完整性={verification['is_complete']}")

                # 步骤4：动态处理，如果答案合格直接返回
                if verification["has_support"] and verification["is_complete"]:
                    print("[SelfRAG] 答案通过验证，返回最终结果")
                    return {
                        "query": query,
                        "answer": answer,
                        "retrieved_documents": docs,
                        "num_documents": len(docs),
                        "iterations": iteration,
                        "verification": verification
                    }
                # 若仅缺乏上下文支持，做下次生成尝试
                elif not verification["has_support"]:
                    print("[SelfRAG] 答案缺乏上下文支持，继续尝试生成...")
                    if attempt < 2:
                        continue
                # 若不完整，则进入下一轮迭代
                elif not verification["is_complete"]:
                    print("[SelfRAG] 答案不完整，继续下一轮迭代...")
                    break

            # 若三次都没有令人满意答案，则进入下一轮迭代
            if iteration < self.max_iterations:
                print("[SelfRAG] 当前迭代未找到满意答案，继续下一轮迭代...")

        # 所有迭代都失败，返回最后一次生成结果
        return {
            "query": query,
            "answer": answer if 'answer' in locals() else "抱歉，无法生成满意的答案。",
            "retrieved_documents": docs if 'docs' in locals() else [],
            "num_documents": len(docs) if 'docs' in locals() else 0,
            "iterations": iteration,
            "verification": verification if 'verification' in locals() else {"has_support": False, "is_complete": False}
        }

# 初始化组件
# 创建向量库对象，指定持久化路径和集合名
vector_store = get_vector_store(
    persist_directory="chroma_db",
    collection_name="self_rag"
)

# 创建相关性判别器实例
relevance_judge = DocumentRelevanceJudge(llm)
# 创建查询增强器实例
query_enhancer = QueryEnhancer(llm)

# 创建SelfRAG检索器实例，注入向量库和相关模块
self_rag_retriever = SelfRAGRetriever(
    vector_store=vector_store,
    relevance_judge=relevance_judge,
    query_enhancer=query_enhancer,
    k=4,
    max_retry=2
)

# 创建SelfRAG系统实例
self_rag = SelfRAG(
    retriever=self_rag_retriever,
    llm=llm,
    max_iterations=3
)

# 示例文档数据，用于构建向量库
documents = [
    "人工智能（AI）是计算机科学的一个分支，旨在创建能够执行通常需要人类智能的任务的系统。"
    "机器学习是人工智能的核心技术之一，它使计算机能够从数据中学习，而无需明确编程。"
    "深度学习是机器学习的一个子集，使用人工神经网络来模拟人脑的工作方式。"
    "神经网络由多个层组成，每层包含多个神经元，通过反向传播算法进行训练。",
    "自然语言处理（NLP）是人工智能的一个分支，专注于使计算机能够理解、解释和生成人类语言。"
    "Transformer架构是NLP领域的重要突破，它使用自注意力机制来处理序列数据。"
    "BERT和GPT是Transformer架构的两个重要应用，分别用于理解任务和生成任务。"
    "BERT通过双向编码器理解上下文，GPT通过自回归生成器生成文本。",
    "计算机视觉是人工智能的另一个重要分支，专注于使计算机能够理解和分析视觉信息。"
    "卷积神经网络（CNN）是计算机视觉的核心技术，特别适合处理图像数据。"
    "图像分类、目标检测和图像分割是计算机视觉的三个主要任务。"
    "ResNet、YOLO和U-Net分别是这三个任务的代表性模型。",
]

# 初始化知识库，批量把示例文档添加到向量库，顺带设定主题元数据
print("正在初始化知识库...")
vector_store.add_texts(
    documents,
    metadatas=[{"topic": "人工智能"}, {"topic": "自然语言处理"}, {"topic": "计算机视觉"}]
)
print("知识库初始化完成！\n")

# 打印分隔线及模块说明，开始展示示例
print("="*60)
print("SelfRAG智能自判别检索增强示例")
print("="*60)

# 构造示例用户查询
query = "什么是深度学习？它和神经网络有什么关系？"

# 执行SelfRAG主流程，得到问答结果
result = self_rag.generate_answer(query, k=4)

# 打印最终结果及检索与验证过程的详细信息
print("\n" + "="*60)
print("最终结果")
print("="*60)
print(f"\n用户查询：{result['query']}")
print(f"\n生成的答案：\n{result['answer']}")
print(f"\n总迭代次数：{result['iterations']}")
print(f"\n检索到的文档数量：{result['num_documents']}")
print(f"\n验证结果：")
print(f"  上下文支持：{result['verification']['has_support']}")
print(f"  完整性：{result['verification']['is_complete']}")

2.2 执行过程 #

2.2.1 核心思想 #

SelfRAG 采用“自判别+动态迭代”策略：

• 文档相关性判别：智能判断检索到的文档是否与用户问题相关
• 查询增强：当检索结果不理想时，自动触发查询增强策略
• 答案可信度验证：验证答案是否有上下文支持，是否完整解答问题
• 动态迭代：根据验证结果决定是否重新生成或进入下一轮迭代

2.2.2 执行流程 #

阶段一：初始化

# 1. 创建向量存储实例
vector_store = get_vector_store(
    persist_directory="chroma_db",
    collection_name="self_rag"
)

# 2. 创建相关性判别器
relevance_judge = DocumentRelevanceJudge(llm)

# 3. 创建查询增强器
query_enhancer = QueryEnhancer(llm)

# 4. 创建SelfRAG检索器
self_rag_retriever = SelfRAGRetriever(
    vector_store=vector_store,
    relevance_judge=relevance_judge,
    query_enhancer=query_enhancer,
    k=4,
    max_retry=2
)

# 5. 创建SelfRAG系统
self_rag = SelfRAG(
    retriever=self_rag_retriever,
    llm=llm,
    max_iterations=3
)

初始化时：

• 创建向量存储实例
• 创建文档相关性判别器
• 创建查询增强器
• 创建SelfRAG检索器，配置相关组件
• 创建SelfRAG系统，配置最大迭代次数

阶段二：知识库初始化

# 添加文档到向量库
vector_store.add_texts(
    documents,
    metadatas=[...]
)

索引过程：

• 将示例文档添加到向量库
• 每个文档附带元数据（topic）
• 文档会被向量化并存储

阶段三：SelfRAG查询处理

query = "什么是深度学习？它和神经网络有什么关系？"
result = self_rag.generate_answer(query, k=4)

完整流程：

1. 用户提交查询
2. 迭代循环（最多 max_iterations 次）：
   • 检索与相关性判别：
     • 调用 retriever._get_relevant_documents(query, k=k)
     • 检索 2k 个候选文档
     • 使用相关性判别器判断文档相关性
     • 如果无相关文档，触发查询增强并重试
   • 生成答案：
     • 整合相关文档作为上下文
     • 使用LLM生成答案
   • 答案验证：
     • 调用 verifier.verify(query, context, answer)
     • 检查答案是否有上下文支持
     • 检查答案是否完整解答问题
   • 动态决策：
     • 如果答案合格（有支持且完整），返回结果
     • 如果缺乏支持，继续尝试生成
     • 如果不完整，进入下一轮迭代
3. 返回结果：
   • 包含查询、答案、检索文档、迭代次数、验证结果等

2.2.3 类图 #

2.2.4 时序图 #

2.2.4.1 完整SelfRAG流程时序图 #

2.2.4.2 检索与相关性判别详细流程 #

2.2.4.3 答案验证详细流程 #

2.2.5 关键设计要点 #

1. SelfRAG流程

用户查询
    ↓
检索阶段:
  向量检索(2k个候选) → 相关性判别 → 查询增强(如需要)
    ↓
生成阶段:
  生成答案 → 答案验证(支持度+完整性)
    ↓
动态决策:
  合格 → 返回
  不合格 → 重新生成或下一轮迭代

2. 相关性判别示例

用户查询: "什么是深度学习？它和神经网络有什么关系？"

检索到的8个候选文档:
文档1: "人工智能（AI）是计算机科学的一个分支..."
文档2: "深度学习是机器学习的一个子集，使用人工神经网络..."
文档3: "自然语言处理（NLP）是人工智能的一个分支..."
文档4: "神经网络由多个层组成，每层包含多个神经元..."
...

判别结果:
文档1: 不相关
文档2: 相关 ✓
文档3: 不相关
文档4: 相关 ✓
...

返回的相关文档: [文档2, 文档4, ...]

3. 答案验证标准

• 上下文支持度（has_support）：
  • 检查答案是否能在提供的上下文中找到依据
  • 防止生成无依据的答案
• 完整性（is_complete）：
  • 检查答案是否完整回答了用户问题
  • 确保答案覆盖问题的所有方面

验证结果组合：

• has_support=True, is_complete=True：答案合格，直接返回
• has_support=False：缺乏支持，继续尝试生成
• is_complete=False：不完整，进入下一轮迭代

4. 查询增强策略

当检索失败或无相关文档时触发：

• 保持原问题的核心含义
• 添加更多关键词和细节
• 使用更具体的表达方式

示例：

原始查询: "什么是深度学习？"

增强查询: "深度学习定义、深度学习与机器学习的关系、深度学习与神经网络的关系、深度学习的应用"

5. 动态迭代策略

• 检索阶段迭代：
  • 如果检索失败或无相关文档，触发查询增强
  • 最多重试 max_retry 次
• 生成阶段迭代：
  • 如果答案缺乏支持，继续尝试生成（最多3次）
  • 如果答案不完整，进入下一轮迭代
• 系统级迭代：
  • 最多进行 max_iterations 轮迭代
  • 每轮迭代包含完整的检索-生成-验证流程

6. 元数据设计

检索返回的 Document 对象包含：

• score：相似度分数（距离，越小越相似）
• retrieval_method: "self_rag"：检索方法标识
• judged_relevant: True：相关性判别结果
• 继承原始文档的元数据（topic 等）

7. 答案生成模板

默认答案生成模板：

请根据以下上下文信息回答用户问题。要求：
1. 答案必须基于提供的上下文信息
2. 如果上下文中没有相关信息，请明确说明"无法从提供的信息中找到答案"
3. 答案要准确、完整、有条理

上下文信息：
{context}

用户问题：{query}

答案：

该模板：

• 强调答案必须基于上下文
• 要求明确说明无法找到答案的情况
• 要求答案准确、完整、有条理

8. 优势与应用场景

优势：

• 智能判别：自动判断文档相关性和答案质量
• 动态优化：根据验证结果动态调整策略
• 质量保证：通过验证确保答案有依据且完整
• 自适应：自动处理检索失败和答案不合格的情况

适用场景：

• 高质量问答：需要确保答案准确性和完整性
• 复杂查询：需要多轮迭代才能找到满意答案
• 专业领域：需要严格验证答案的可信度
• 生产环境：需要可靠的质量保证机制

9. 技术细节

• 检索策略：
  • 检索 2k 个候选文档，提高召回率
  • 通过相关性判别筛选，提高精确率
• 相关性判别：
  • 使用LLM判断每个文档的相关性
  • 解析返回的"文档N: 相关/不相关"格式
• 答案验证：
  • 分别检查支持度和完整性
  • 使用简单的关键词匹配判断结果
• 迭代控制：
  • 检索阶段最多重试 max_retry 次
  • 生成阶段最多尝试3次
  • 系统级最多迭代 max_iterations 次

2.2.6 与其他方法的对比 #

该设计通过“自判别+动态迭代+质量验证”策略，在高质量问答场景下提供更可靠的答案，适用于需要严格质量保证的RAG应用。