接上文（前序内容可以查看上一篇文章）

Relevant Segment Extraction RSE

我们一直专注于单个块，但有时最好的信息分散在多个连续的块中。相关段提取 (RSE) 解决了这个问题。

RSE不仅仅抓取前k个块，还尝试识别和提取相关文本的整个片段。

让我们看看如何在现有管道中实现这一点，我们使用已定义的函数RSE。我们正在添加一个函数调用rag_with_rse，它接受pdf_path和query并返回响应。
我们结合几个函数调用来执行RSE。

# 使用 RSE 运行 RAG
 rse_result = rag_with_rse(pdf_path, query)

这一行代码做了很多事情！它：

1. 处理文档（提取文本、分块、创建嵌入，全部在rag_with_rse内部处理）。
2. 根据与查询和位置的相关性计算“块值”。
3. 使用巧妙的算法来找到最佳的连续块段。
4. 将这些片段组合成一个上下文。
5. 根据该上下文生成响应。

现在来评价一下：

# 评估
evaluation_prompt = f"用户查询：{query} \nAI 响应:\n {rse_result[ 'response' ]} \n真实响应：{reference_answer} \n {evaluate_system_prompt} "
evaluation_response = generate_response(evaluate_system_prompt, evaluation_prompt) 
print (evaluation_response.choices[ 0 ].message.content) 


### OUTPUT ###但是，标准检索的
响应包括... 0.8是我分配给 AI 响应的分数
 

而且...我们的得分已经达到了 0.8 左右！

通过关注相关文本的连续片段，RSE 为 LLM 提供了更连贯、更完整的背景，从而带来更准确、更全面的回应。

这表明我们如何选择和向 LLM 呈现信息与我们选择什么信息一样重要。

Contextual Compression 上下文压缩

我们一直在添加越来越多的背景信息、相邻的区块、生成的问题和整个片段。但有时，少即是多。

LLM 的上下文窗口有限，如果在其中填充不相关的信息可能会影响性能。

上下文压缩是关于选择性的。我们检索大量上下文，然后对其进行压缩，仅保留与查询直接相关的部分。

这里的关键区别在于生成之前的“上下文压缩”步骤。我们不会改变我们检索的内容，而是在将其传递给 LLM 之前对其进行改进。

我们在这里使用一个函数调用rag_with_compression，它接受query和其他参数并实现上下文压缩。在内部，它使用 LLM 分析检索到的块并仅提取与直接相关的句子或段落query。

让我们看看它的实际效果：

def  rag_with_compression ( pdf_path, query, k= 10 , compression_type= "selective" , model= "meta-llama/Llama-3.2-3B-Instruct" ): 
    """
    带有上下文压缩的 RAG（检索增强生成）管道。

    参数：
        pdf_path（str）：PDF 文档的路径。query 
        （str）：用于检索的用户查询。k 
        （int）：要检索的相关块数。默认值为 10。compression_type 
        （str）：应用于检索到的块的压缩类型。默认值为“selective”。model 
        （str）：用于生成响应的语言模型。默认值为“meta-llama/Llama-3.2-3B-Instruct”。

    返回：
        dict：包含查询、原始和压缩块、压缩统计数据和最终响应的字典。
    """ 
    
    print ( f"\n=== RAG WITH COMPRESSION ===\nQuery: {query} | 压缩：{compression_type} " ) 
    
    # 处理文档以提取、分块和嵌入文本
    vector_store = process_document(pdf_path) 
    
    # 根据查询相似度检索前 k 个相关块
    results = vector_store.similarity_search(create_embeddings(query), k=k) 
    removed_chunks = [r[ "text" ] for r in results] 

    # 对检索到的块应用压缩
    compressed = batch_compress_chunks(retrieved_chunks, query, compression_type, model) 
    
    # 过滤掉空的压缩块；如果所有都是空的，则回退到原始
    compressed_chunks, compression_ratios = zip ([(c, r) for c, r in compressed if c.strip()] or [(chunk, 0.0 ) for chunk in removed_chunks]) 
    
    # 组合压缩块以形成用于生成响应的上下文
    context = "\n\n---\n\n" .join(compressed_chunks) 
    
    # 使用压缩上下文生成响应 response
     = generate_response(query, context, model) 

    print ( f"\n=== RESPONSE ===\n {response} " ) 
    
    # 返回详细结果
    return {
        "query": query,
        "original_chunks": retrieved_chunks,
        "compressed_chunks": compressed_chunks,
        "compression_ratios": compression_ratios,
        "context_length_reduction": f"{sum(compression_ratios)/len(compression_ratios):.2f}%",
        "response": response
    }

rag_with_compression 为不同的压缩类型提供了选项：

• “选择性”只保留直接相关的句子。
• “summary”创建针对查询的简短摘要。
• “提取”仅提取包含答案的句子（非常严格！）。

现在，要运行压缩，我们使用以下代码：

# 使用上下文压缩运行 RAG（使用“selective”模式）
compression_result = rag_with_compression(pdf_path, query, compression_type= "selective" ) 

# 评估。
evaluate_prompt = f"用户查询：{query} \nAI 响应：\n {compression_result[ 'response' ]} \nTrue 响应：{reference_answer} \n {evaluate_system_prompt} "
 evaluation_response = generate_response(evaluate_system_prompt, evaluation_prompt) 
print (evaluation_response.choices[ 0 ].message.content) 


### OUTPUT ###
评估分数0.75

这给了我们一个大约 0.75 的分数。

上下文压缩是一种强大的技术，因为它平衡了广度（初始检索获得广泛的信息）和焦点（压缩消除噪音）。

通过向 LLM 提供最相关的信息，我们通常可以获得更简洁、更准确的答案。

Feedback Loop 反馈回路

到目前为止，我们看到的所有技术都是“静态的”，它们不会从错误中吸取教训。反馈循环改变了这种情况。

这个想法很简单：

1. 用户对 RAG 系统的响应提供反馈（例如，好/坏、相关/不相关）。
2. 系统存储该反馈。
3. 未来的检索将使用此反馈来改进。

我们可以使用函数full_rag_workflow调用来实现反馈循环。这是函数定义。

def  full_rag_workflow ( pdf_path, query, feedback_data= None , feedback_file= "feedback_data.json" , fine_tune= False ): 
    """
    执行带有反馈集成的完整 RAG 工作流程以持续改进。

    """ 
    # 步骤 1：如果未明确提供，则加载历史反馈以进行相关性调整
    if feedback_data is  None : 
        feedback_data = load_feedback_data(feedback_file) 
        print ( f"Loaded { len (feedback_data)} feedback entry from {feedback_file} " ) 
    
    # 步骤 2：通过提取、分块和嵌入管道处理文档
    chunks, vector_store = process_document(pdf_path) 
    
    # 步骤 3：通过合并高质量的过去交互来微调向量索引
    # 这将从成功的问答对中创建增强的可检索内容
    if fine_tune and feedback_data: 
        vector_store = fine_tune_index(vector_store, chunks, feedback_data) 
    
    # 步骤 4：使用反馈感知检索
    # 注意：这取决于 rag_with_feedback_loop 函数，该函数应在其他地方定义
    result = rag_with_feedback_loop(query, vector_store, feedback_data) 
    
    # 步骤 5：收集用户反馈以提高未来的性能
    print ( “\n=== 您想对此回复提供反馈吗？ ===" ) 
    print ( "评价相关性（1-5，5 为最相关）：" ) 
    relevance = input () 
    
    print ( "评价质量（1-5，5 为最优质）：" ) 
    quality = input () 
    
    print ( "有任何评论吗？（可选，按 Enter 跳过）" ) 
    comments = input () 
    
    # 步骤 6：将反馈格式化为结构化数据
    feedback = get_user_feedback( 
        query=query, 
        response=result[ "response" ], 
        relevance= int (relevance), 
        quality= int (quality), 
        comments=comments 
    ) 
    
    # 步骤 7：保留反馈以实现持续的系统学习
    store_feedback(feedback, feedback_file) 
    print ( "反馈已记录。谢谢！" ) 
    
    return result

这个 full_rag_workflow 函数做了几件事：

1. 加载现有反馈：它检查 feedback_data.json 文件并加载任何以前的反馈。
2. 运行 RAG 管道：这部分与我们之前所做的类似。
3. 征求反馈：它提示用户对响应的相关性和质量进行评分。
4. 存储反馈：将反馈保存到feedback_data.json文件。

这种反馈如何真正用于改善检索的神奇之处更为复杂，并且发生在诸如 之类的函数中fine_tune_index（adjust_relevance_scores为简洁起见，此处未显示）。但关键思想是，好的反馈可以提高某些文档的相关性，而坏的反馈则会降低相关性。

让我们运行一个简化版本，假设我们没有任何现有的反馈：

# 我们没有之前的反馈，因此“fine_tune=False”
result = full_rag_workflow(pdf_path=pdf_path, query=query, fine_tune= False ) 

# 评估。
evaluation_prompt = f“用户查询：{query} \nAI 响应：\n {result[ 'response' ]} \nTrue 响应：{reference_answer} \n {evaluate_system_prompt} ”
evaluation_response = generate_response(evaluate_system_prompt, evaluation_prompt) 
print (evaluation_response.choices[ 0 ].message.content) 


### OUTPUT ###
评估分数为 0.7，因为......

我们看到分数在 0.7 左右！

这不是一个巨大的飞跃，这是意料之中的。反馈回路会随着时间的推移不断改进系统，并不断进行交互。本节仅演示该机制。

真正的力量来自于积累反馈并利用它来改进检索过程。这使得 RAG 系统能够根据收到的查询类型进行自适应和个性化。

Adaptive RAG

我们探索了改进 RAG 的各种方法：更好地分块、添加上下文、转换查询、重新排名，甚至结合反馈。

但如果最佳技术取决于所提问题的类型，该怎么办？这就是 Adaptive RAG 背后的想法。

我们在这里使用四种不同的策略：

1. 事实策略：注重检索精确的事实和数据。
2. 分析策略：旨在全面涵盖某个主题，探索不同的方面。
3. 意见策略：试图收集有关主观问题的不同观点。
4. 语境策略：结合用户特定的语境来定制检索。

让我们看看它是如何工作的。我们将使用一个名为
rag_with_adaptive_retrieval的函数来处理整个过程：

def  rag_with_adaptive_retrieval ( pdf_path, query, k= 4 , user_context= None ): 
    """
    使用自适应检索完成 RAG 管道。

    """ 
    print ( "\n=== RAG WITH ADAPTIVE RETRIEVAL ===" ) 
    print ( f"Query: {query} " ) 
    
    # 处理文档以提取文本、对其进行分块并创建嵌入
    chunks, vector_store = process_document(pdf_path) 
    
    # 对查询进行分类以确定其类型
    query_type = classify_query(query) 
    print ( f"Query classified as: {query_type} " ) 
    
    # 根据查询类型使用自适应检索策略检索文档
    retrieved_docs = adaptive_retrieval(query, vector_store, k, user_context) 
    
    # 根据查询、检索到的文档和查询类型生成响应
    response = generate_response(query, retained_docs, query_type) 
    
    # 编译结果放入字典中
    result = { 
        "query" : query, 
        "query_type" : query_type, 
        "retrieved_documents" :retrieved_docs, 
        "response" : response 
    } 
    
    print ( "\n=== RESPONSE ===" ) 
    print (response)
    
    return result

它首先使用由其他辅助函数定义的函数classify_query对查询进行分类。

根据识别的类型，选择并执行适当的专门检索策略（
factual_retrieval_strategy、
analytical_retrieval_strategy、
opinion_retrieval_strategy或
contextual_retrieval_strategy）。
最后，使用generate_response结合检索到的文档生成响应。

该函数返回一个包含结果的字典，包括query、query type、retrieved documents和generated response。

让我们使用这个函数并评估它：

# 运行自适应 RAG 管道
result = rag_with_adaptive_retrieval(pdf_path, query) 

# 评估。
evaluation_prompt = f"用户查询：{query} \nAI 响应：\n {result[ 'response' ]} \nTrue 响应：{reference_answer} \n {evaluate_system_prompt} "
 evaluation_response = generate_response(evaluate_system_prompt, evaluation_prompt) 
print (evaluation_response.choices[ 0 ].message.content) 


### OUTPUT ###
评估分数为 0.86

这次我们取得了0.856左右的分数。

通过调整我们的检索策略以适应特定类型的查询，我们可以获得比一刀切方法更好的结果。这凸显了了解用户意图并相应地定制 RAG 系统的重要性。

自适应 RAG 不是一个固定的程序，而是一个框架，它能够根据查询选择最佳策略。

Self RAG

到目前为止，我们的 RAG 系统基本上是被动的。它们接受查询、检索信息并生成响应。Self-RAG 采用了不同的方法：它是主动的和反思性的。

它不仅仅是检索和生成，它还会思考是否检索、检索什么以及如何使用检索到的信息。

这些“反射”步骤使 Self-RAG 比传统 RAG 更具动态性和适应性。它可以决定：

• 完全跳过检索。
• 使用不同的策略进行多次检索。
• 丢弃不相关的信息。
• 优先考虑有良好支持且有用的信息。

Self-RAG 的核心在于它能够生成“反射标记”。这些是模型用来推理自身过程的特殊标记。例如，它对retrieval_needed、relevance、support_rating和utility_ratings使用不同的标记。

模型使用这些标记的组合来决定何时必须检索、何时不必检索，以及 LLM 应在什么基础上生成最终响应。

首先，决定是否需要检索：

def  determine_if_retrieval_needed ( query ): 
    """ 
    (说明性示例 - 并非完全功能)
    确定给定查询是否需要检索。
    """
     system_prompt = """您是 AI 助手，可确定是否需要检索来回答查询。
    对于事实问题、具体信息请求或有关事件、人物或概念的问题，请回答“是”。
    对于意见、假设情景或具有常识的简单查询，请回答“否”。
    仅回答“是”或“否”。"""

     user_prompt = f"查询：{query} \n\n是否需要检索才能准确回答此查询？"

     response = client.chat.completions.create( 
        model= "meta-llama/Llama-3.2-3B-Instruct" , 
        messages=[ 
            { "role" : "system" , "content" : system_prompt}, 
            { "role" : "user" , "content" : user_prompt} 
        ], 
        temperature= 0
     ) 
    answer = response.choices[ 0 ].message.content.strip().lower()
    return “yes” in answer

此
determine_if_retrieval_needed功能（再次简化）使用 LLM 来判断是否需要外部信息。

• 对于像“法国的首都是哪里？”这样的事实问题，它可能会返回False（LLMs可能已经知道这一点）。
• 对于“写一首诗...”这样的创造性任务，它也可能会返回False。
• 但对于更复杂或更小众的查询，它会返回True。

以下是相关性评估的一个简化示例：

def  assess_relevance ( query, context ): 
    """ 
    (说明性示例 - 并非完全功能)
    评估上下文与查询的相关性。
    """
     system_prompt = """您是 AI 助手。确定文档是否与查询相关。
    仅用“相关”或“不相关”回答。"""

     user_prompt = f"""查询：{query}
    文档内容：
    {context[: 500 ]} ... [truncated]

    此文档是否与查询相关？仅用“相关”或“不相关”回答。
    """

     response = client.chat.completions.create( 
        model= "meta-llama/Llama-3.2-3B-Instruct" , 
        messages=[ 
            { "role" : "system" , "content" : system_prompt}, 
            { "role" : "user" , "content" : user_prompt} 
        ], 
        temperature= 0
     )
    answer = response.choices[0].message.content.strip().lower()
    return answer

该evaluate_relevance函数（再次简化）使用 LLM 来判断检索到的文档是否与query相关。

这使得Self-RAG能够在生成响应之前过滤掉不相关的文档。

最后，我们可以使用以下方法调用：

# 我们可以调用 `self_rag` 函数进行自我整理，它会自动
# 决定何时检索，何时不检索。
result = self_rag(query, vector_store) 

print (result[ "response" ]) 


### 输出 ### AI 响应的
评估分数为0.65 

我们这里得到的分数是 0.6。

这反映了以下事实：

• Self-RAG 具有巨大的潜力，但全面实施起来很复杂。
• 甚至我们展示的“是否需要检索？”步骤有时也可能是错误的。
• 我们还没有展示完整的“反思”过程，所以我们不能要求获得更高的分数。

关键点在于，Self-RAG 旨在让 RAG 系统更加智能、更具适应性。这是向能够推理自身知识和检索需求的 LLM 迈出的一步。

Knowledge Graph 知识图谱

到目前为止，我们的 RAG 系统将文档视为独立块的集合。但如果信息是相互关联的，该怎么办？如果理解一个概念需要理解相关概念，该怎么办？这就是 Graph RAG 的作用所在。

Graph RAG 不是以扁平的块列表形式组织信息，而是以知识图谱的形式组织信息。可以将其想象成一个网络：

1. 节点：表示概念、实体或信息片段（如我们的文本块）。
2. 边：表示这些节点之间的关系。

核心思想是，通过遍历该图，我们不仅可以找到直接相关的信息，还可以找到提供关键背景的间接相关信息。

让我们看一些核心步骤如何工作的简化代码：首先，构建知识图谱：

def build_knowledge_graph(chunks):
    """
    Build a knowledge graph from text chunks using embeddings and concept extraction.

    Args:
        chunks (list of dict): List of text chunks, each containing a "text" field.

    Returns:
        tuple: (Graph with nodes as text chunks, list of embeddings)
    """
    graph, texts = nx.Graph(), [c["text"] for c in chunks]
    embeddings = create_embeddings(texts)  # Compute embeddings

    # Add nodes with extracted concepts and embeddings
    for i, (chunk, emb) in enumerate(zip(chunks, embeddings)):
        graph.add_node(i, text=chunk["text"], concepts := extract_concepts(chunk["text"]), embedding=emb)

    # Create edges based on shared concepts and embedding similarity
    for i, j in ((i, j) for i in range(len(chunks)) for j in range(i + 1, len(chunks))):
        if shared_concepts := set(graph.nodes[i]["concepts"]) & set(graph.nodes[j]["concepts"]):
            sim = np.dot(embeddings[i], embeddings[j]) / (np.linalg.norm(embeddings[i])  np.linalg.norm(embeddings[j]))
            weight = 0.7 * sim + 0.3 * (len(shared_concepts) / min(len(graph.nodes[i]["concepts"]), len(graph.nodes[j]["concepts"])))
            if weight > 0.6:
                graph.add_edge(i, j, weight=weight, similarity=sim, shared_concepts=list(shared_concepts))

    print(f"Graph built: {graph.number_of_nodes()} nodes, {graph.number_of_edges()} edges")
    return graph, embeddings

它接受查询、图和嵌入，并返回相关节点列表和遍历路径。

最后，我们有使用这两个函数的graph_rag_pipeline：

def  graph_rag_pipeline ( pdf_path, query, chunk_size= 1000 , chunk_overlap= 200 , top_k= 3 ): 
    """
    从文档到答案的完整 Graph RAG 管道。
    """ 
    # 从 PDF 文档中提取文本
    text = extract_text_from_pdf(pdf_path) 
    
    # 将提取的文本拆分为重叠块
    chunks = chunk_text(text, chunk_size, chunk_overlap) 
    
    # 从文本块构建知识图谱
    graph, embeddings = build_knowledge_graph(chunks) 
    
    # 遍历知识图谱以查找与查询相关的信息
    related_chunks, traversal_path = traverse_graph(query, graph, embeddings, top_k) 
    
    # 根据查询和相关块生成响应
    response = generate_response(query, related_chunks) 
        
    # 返回查询、响应、相关块、遍历路径和图形
    return {
        "query": query,
        "response": response,
        "relevant_chunks": relevant_chunks,
        "traversal_path": traversal_path,
        "graph": graph
    }

让我们使用它来生成响应：

# Execute the Graph RAG pipeline to process the document and answer the query
results = graph_rag_pipeline(pdf_path, query)

# Evaluate.
evaluation_prompt = f"User Query: {query}\nAI Response:\n{results['response']}\nTrue Response: {reference_answer}\n{evaluate_system_prompt}"
evaluation_response = generate_response(evaluate_system_prompt, evaluation_prompt)
print(evaluation_response.choices[0].message.content)


### OUTPUT
0.78

我们的得分是0.78左右。

Graph RAG 的表现并不优于更简单的方法，但它可以捕捉信息片段之间的关系，而不仅仅是各个片段本身。

这对于需要理解概念之间联系的复杂查询尤其有用。

Hierarchical Indices 层次索引

我们探索了各种改进 RAG 的方法：更好的分块、上下文丰富、查询转换、重新排序，甚至基于图形的检索。但这存在一个根本性的权衡：

• 小块：适合精确匹配，但会丢失上下文。
• 大块：保留上下文，但会导致检索相关性降低。

分层索引提供了一种解决方案：我们创建两个级别的表示：

1. 摘要：对文档主要内容的简要概述。
2. 详细块：这些部分内的较小块。

1. 首先，搜索摘要：这可以快速缩小文档的相关部分。
2. 然后，仅在这些部分内搜索详细块：这提供了小块的精度，同时保持了较大部分的上下文。

让我们通过函数hierarchical_rag调用来看一下这个操作：

def hierarchical_rag(query, pdf_path, chunk_size=1000, chunk_overlap=200, 
                     k_summaries=3, k_chunks=5, regenerate=False):
    """
    Complete hierarchical Retrieval-Augmented Generation (RAG) pipeline.

    Args:
        query (str): The user query.
        pdf_path (str): Path to the PDF document.
        chunk_size (int): Size of text chunks for processing.
        chunk_overlap (int): Overlap between consecutive chunks.
        k_summaries (int): Number of top summaries to retrieve.
        k_chunks (int): Number of detailed chunks to retrieve per summary.
        regenerate (bool): Whether to reprocess the document.

    Returns:
        dict: Contains the query, generated response, retrieved chunks, 
              and counts of summaries and detailed chunks.
    """
    # Define filenames for caching summary and detailed vector stores
    summary_store_file = f"{os.path.basename(pdf_path)}_summary_store.pkl"
    detailed_store_file = f"{os.path.basename(pdf_path)}_detailed_store.pkl"
    
    # Process document if regeneration is required or cache files are missing
    if regenerate or not os.path.exists(summary_store_file) or not os.path.exists(detailed_store_file):
        print("Processing document and creating vector stores...")
        summary_store, detailed_store = process_document_hierarchically(pdf_path, chunk_size, chunk_overlap)
        
        # Save processed stores for future use
        with open(summary_store_file, 'wb') as f:
            pickle.dump(summary_store, f)
        with open(detailed_store_file, 'wb') as f:
            pickle.dump(detailed_store, f)
    else:
        # Load existing vector stores from cache
        print("Loading existing vector stores...")
        with open(summary_store_file, 'rb') as f:
            summary_store = pickle.load(f)
        with open(detailed_store_file, 'rb') as f:
            detailed_store = pickle.load(f)
    
    # Retrieve relevant chunks using hierarchical search
    retrieved_chunks = retrieve_hierarchically(query, summary_store, detailed_store, k_summaries, k_chunks)
    
    # Generate a response based on the retrieved chunks
    response = generate_response(query, retrieved_chunks)
    
    # Return results with metadata
    return {
        "query": query,
        "response": response,
        "retrieved_chunks": retrieved_chunks,
        "summary_count": len(summary_store.texts),
        "detailed_count": len(detailed_store.texts)
    }

该hierarchical_rag函数处理两阶段检索过程：

1. 首先，它搜索summary_store以找到最相关的摘要。
2. 然后，它会搜索detailed_store，但只在属于顶部摘要的块内进行搜索。这比搜索所有详细块要高效得多。

该函数还有一个regenerate参数，用于创建新的向量存储或使用现有的向量存储。

让我们用它来回答我们的查询并进行评估：

# 运行分层 RAG 管道
result = hierarchical_rag(query, pdf_path)

我们检索并生成响应。最后，让我们看看评估分数：

# Evaluate.
evaluation_prompt = f"User Query: {query}\nAI Response:\n{result['response']}\nTrue Response: {reference_answer}\n{evaluate_system_prompt}"
evaluation_response = generate_response(evaluate_system_prompt, evaluation_prompt)
print(evaluation_response.choices[0].message.content)


### OUTPUT
0.84

我们的分数是 0.84

分层检索提供了迄今为止最好的分数。

我们获得了搜索摘要的速度和搜索较小块的精度，以及通过了解每个块属于哪个部分而获得的附加上下文。这就是为什么它通常是表现最佳的 RAG 策略。

HyDE

到目前为止，我们一直直接嵌入用户的查询或其转换版本。HyDE（假设文档嵌入）采用了不同的方法。它不是嵌入查询，而是嵌入回答查询的假设文档。

流程如下：

1. 生成一个假设文档：使用 LLM 创建一个可以回答查询的文档（如果存在）。
2. 嵌入假设文档：创建此假设文档的嵌入，而不是原始查询。
3. 检索：查找与假设文档的嵌入相似的文档。
4. 生成：使用检索到的文档（而不是假设的文档！）来回答查询。

其理念是，完整文档（即使是假设文档）比简短查询具有更丰富的语义表示。这有助于弥合查询和嵌入空间中的文档之间的差距。

让我们看看它是如何工作的。首先，我们需要一个函数来生成那个假设的文档。

我们使用
generate_hypothetical_document来做这些：

def  generate_hypothetical_document ( query, desire_length= 1000 ): 
    """
    生成一个假设文档来回答查询。
    """ 
    # 定义系统提示以指导模型如何生成文档
    system_prompt = f"""您是专家文档创建者。
    给定一个问题，生成一份详细的文档来直接回答这个问题。
    文档长度应约为{desired_length}个字符，并提供对该问题的深入、
    翔实的答案。写作时要像这份文件来自
    该主题的权威来源一样。包括具体的细节、事实和解释。
    不要提到这是一份假设文件 - 只需直接写出内容即可。""" 

    # 使用查询定义用户提示
    user_prompt = f"问题：{query} \n\n生成一份完全回答此问题的文档：" 
    
    # 向 OpenAI API 发出请求以生成假设文档
    response = client.chat.completions.create( 
        model= "meta-llama/Llama-3.2-3B-Instruct" ,   # 指定要使用的模型
        messages=[ 
            { "role" : "system" , "content" : system_prompt},   # 指导助手的系统消息
            { "role" : "user" , "content" : user_prompt}   # 带有查询的用户消息
        ], 
        temperature= 0.1   # 设置响应生成的温度
    ) 
    
    # 返回生成的文档内容
    return response.choices[ 0 ].message.content

该函数接受查询并使用 LLM 来创建回答该查询的文档。

现在，让我们将其全部放在 hyde_rag 函数中：

def  hyde_rag ( query, vector_store, k= 5 , should_generate_response= True ): 
    """
    使用假设文档嵌入执行 RAG。
    
    """ 
    print ( f"\n=== 使用 HyDE 处理查询：{query} ===\n" ) 
    
    # 步骤 1：生成一个回答查询的假设文档
    print ( "正在生成假设文档..." ) 
    hypothetical_doc = generate_hypothetical_document(query) 
    print ( f"生成{ len (hypothetical_doc)}个字符的假设文档" ) 
    
    # 步骤 2：为假设文档创建嵌入
    print ( "为假设文档创建嵌入..." ) 
    hypothetical_embedding = create_embeddings([hypothetical_doc])[ 0 ] 
    
    # 步骤 3：根据假设文档检索相似的块
    print ( f"检索{k}个最相似的chunks..." ) 
    retrieved_chunks = vector_store.similarity_search(hypothetical_embedding, k=k) 
    
    # 准备结果字典
    results = { 
        "query" : query, 
        "hypothetical_document" : hypothetical_doc, 
        "retrieved_chunks" : withdrawed_chunks 
    } 
    
    # 步骤 4：如果要求，生成响应
    if should_generate_response: 
        print ( "正在生成最终响应..." ) 
        response = generate_response(query, withdrawed_chunks) 
        results[ "response" ] = response 
    
    return results

hyde_rag 函数现在：

1. 生成假设文档。
2. 创建该文档的嵌入（不是查询！）。
3. 使用嵌入进行检索。
4. 像以前一样生成响应。

让我们运行它并查看生成的响应：

# 运行 HyDE RAG
hyde_result = hyde_rag(query, vector_store) 

# 评估
evaluation_prompt = f"用户查询：{query} \nAI 响应:\n {hyde_result[ 'response' ]} \n真实响应：{reference_answer} \n {evaluate_system_prompt} "
evaluation_response = generate_response(evaluate_system_prompt, evaluation_prompt) 
print (evaluation_response.choices[ 0 ].message.content) 


### 输出
0.5

我们的评估分数在0.5左右。

HyDE 是一个聪明的主意，但它并不总是能发挥更好的作用。在这种情况下，假设的文档可能与我们实际的文档集合的方向略有不同，导致检索结果相关性较低。

这里的关键教训是，没有单一的“最佳” RAG 技术。不同的方法对不同的查询和不同的数据效果更好。

Fusion

我们已经看到，不同的检索方法各有优势。向量搜索擅长语义相似性，而关键字搜索擅长查找精确匹配。如果我们能将它们结合起来会怎么样？这就是 Fusion RAG 背后的想法。

Fusion RAG不会选择一种检索方法，而是同时执行两种方法，然后合并并重新排列结果。这使我们能够同时捕获语义含义和精确的关键字匹配。

我们实现的核心是 fusion_retrieval 函数。该函数执行基于向量和基于 BM25 的检索，对每个分数进行规范化，使用加权公式将它们组合起来，然后根据组合分数对文档进行排名。

以下是融合检索的函数：

import numpy as np

def fusion_retrieval(query, chunks, vector_store, bm25_index, k=5, alpha=0.5):
    """Perform fusion retrieval by combining vector-based and BM25 search results."""
    
    # Generate embedding for the query
    query_embedding = create_embeddings(query)

    # Perform vector search and store results in a dictionary (index -> similarity score)
    vector_results = {
        r["metadata"]["index"]: r["similarity"] 
        for r in vector_store.similarity_search_with_scores(query_embedding, len(chunks))
    }

    # Perform BM25 search and store results in a dictionary (index -> BM25 score)
    bm25_results = {
        r["metadata"]["index"]: r["bm25_score"] 
        for r in bm25_search(bm25_index, chunks, query, len(chunks))
    }

    # Retrieve all documents from the vector store
    all_docs = vector_store.get_all_documents()

    # Compute combined scores for each document using a weighted sum of vector and BM25 scores
    scores = [
        (i, alpha * vector_results.get(i, 0) + (1 - alpha) * bm25_results.get(i, 0)) 
        for i in range(len(all_docs))
    ]

    # Sort documents by combined score in descending order and keep the top k results
    top_docs = sorted(scores, key=lambda x: x[1], reverse=True)[:k]

    # Return the top k documents with text, metadata, and combined score
    return [
        {"text": all_docs[i]["text"], "metadata": all_docs[i]["metadata"], "score": s} 
        for i, s in top_docs
    ]

它结合了两种方法的优点：

• 向量搜索：使用我们现有的 create_embeddings 和 SimpleVectorStore 实现语义相似性。
• BM25 搜索：使用 BM25 算法（一种标准信息检索技术）实现基于关键字的搜索。
• 分数组合：将两种方法的分数结合起来，给出一个统一的排名。

让我们运行完整的管道并生成响应：

# 首先，处理文档以创建块、向量存储和 BM25 索引
chunks、vector_store、bm25_index = process_document(pdf_path) 

# 使用融合检索运行 RAG
fusion_result = answer_with_fusion_rag(query、chunks、vector_store、bm25_index) 
print (fusion_result[ "response" ]) 

# 评估。
evaluate_prompt = f"用户查询：{query} \nAI 响应：\n {fusion_result[ 'response' ]} \nTrue 响应：{reference_answer} \n {evaluate_system_prompt} "
evaluation_response = generate_response(evaluate_system_prompt, evaluation_prompt) 
print (evaluation_response.choices[ 0 ].message.content) 


### 输出AI 响应的
评估分数为0.83 

最终得分为0.83。

Fusion RAG 通常能给我们带来显著的帮助，因为它结合了不同检索方法的优势。

这就像有两个专家一起工作，一个擅长理解查询的含义，另一个擅长找到完全匹配。

Multi Model

到目前为止，我们只处理文本。但很多信息都被锁定在图像、图表和图解中。多模态 RAG 旨在解锁这些信息并用它来改善我们的反应。

这里的关键变化是：

1. 提取文本和图像：我们从 PDF 中提取文本和图像。
2. 生成图像标题：我们使用 LLM（具体来说，具有视觉功能的模型）为每个图像生成文本描述（标题）。
3. 创建嵌入（文本和标题）：我们为文本块和图像标题创建嵌入。
4. 嵌入模型：在此笔记本中，我们使用 BAAI/bge-en-icl 嵌入模型。
5. LLM 模型：为了生成响应和图像标题，我们将使用 llava-hf/llava-1.5–7b-hf 模型。

这样，我们的向量存储就包含文本和视觉信息，并且我们可以跨两种模式进行搜索。

这里我们定义process_document函数：

def  process_document ( pdf_path, chunk_size= 1000 , chunk_overlap= 200 ): 
    """
    为多模式 RAG 处理文档。

    """ 
    # 为提取的图像创建目录
    image_dir = "extracted_images"
     os.makedirs(image_dir, exist_ok= True ) 
    
    # 从 PDF 中提取文本和图像
    text_data, image_paths = extract_content_from_pdf(pdf_path, image_dir) 
    
    # 将提取的文本分块 
    chunked_text = chunk_text(text_data, chunk_size, chunk_overlap) 
    
    # 处理提取的图像以生成标题
    image_data = process_images(image_paths) 
    
    # 组合所有内容项（文本块和图像标题）
     all_items = chunked_text + image_data 
    
    # 提取用于嵌入的内容
    content = [item[ "content" ] for item in all_items] 
    
    # 为所有内容创建嵌入
    print ( "为所有创建嵌入content..." ) 
    embeddings = create_embeddings(contents) 
    
    # 构建向量存储并添加带有嵌入的项目
    vector_store = MultiModalVectorStore() 
    vector_store.add_items(all_items, embeddings) 
    
    # 准备包含文本块和图像标题计数的文档信息
    doc_info = { 
        "text_count" : len (chunked_text), 
        "image_count" : len (image_data), 
        "total_items" : len (all_items), 
    } 
    
    # 打印添加项目的摘要
    print ( f"向向量存储添加了{ len (all_items)}个项目 ( { len (chunked_text)}个文本块, { len (image_data)}个图像标题)" ) 
    
    # 返回向量存储和文档信息
    return vector_store, doc_info

该函数负责处理图像的提取和字幕以及MultiModalVectorStore 的创建。

我们假设图像字幕效果相当好。（在现实场景中，您需要仔细评估字幕的质量）。

现在，让我们通过查询将所有内容整合在一起：

# 处理文档以创建向量存储。我们为此创建了一个新的 pdf
 pdf_path = "data/attention_is_all_you_need.pdf"
 vector_store, doc_info = process_document(pdf_path) 

# 运行多模式 RAG 管道。这与之前非常相似！
 result = query_multimodal_rag(query, vector_store) 

# 评估。
evaluation_prompt = f"User Query: {query} \nAI Response:\n {result[ 'response' ]} \nTrue Response: {reference_answer} \n {evaluate_system_prompt} "
evaluation_response = generate_response(evaluate_system_prompt, evaluation_prompt) 
print (evaluation_response.choices[ 0 ].message.content) 


### OUTPUT 
0.79

我们的得分是0.79左右。

多模态 RAG 具有非常强大的潜力，尤其是对于图像包含关键信息的文档。然而，它并没有打败我们迄今为止见过的其他技术。

Crag

到目前为止，我们的 RAG 系统相对比较被动。它们检索信息并生成响应。但如果检索到的信息不好怎么办？如果它不相关、不完整甚至自相矛盾怎么办？Corrective RAG (CRAG) 正面解决了这个问题。

CRAG 增加了一个关键步骤：评估。在初始检索之后，它会检查检索到的文档的相关性。而且，至关重要的是，它根据评估有不同的策略：

• 高度相关性：如果检索到的文档良好，则照常进行。
• 相关性低：如果检索到的文档不好，则返回网络搜索！
• 中等相关性：如果文档没有问题，则结合文档和网络的信息。

这种“纠正”机制使 CRAG 比标准 RAG 更加强大。它不只是希望得到最好的结果；它积极地检查和适应。

让我们看看这在实践中是如何运作的。我们将使用一个rag_with_compression为此调用的函数。

# 运行 CRAG
 crag_result = rag_with_compression(pdf_path, query, compression_type= "selective" )

这个简单的函数调用做了很多事情：

1. 初始检索：照常检索文档。
2. 相关性评估：对每个文档与查询的相关性进行评分。
3. 决策：决定是否使用文档、进行网络搜索或两者结合。
4. 响应生成：使用所选的知识源生成响应。

和往常一样，评价如下：

# 评估。
evaluate_prompt = f“用户查询：{query} \nAI 响应：\n {crag_result[ 'response' ]} \n真实响应：{reference_answer} \n {evaluate_system_prompt} ”
evaluate_response = generate_response(evaluate_system_prompt, evaluation_prompt) 
print (evaluation_response.choices[ 0 ].message.content) 


### 输出 ### 
0.824

我们的目标分数是 0.824 左右。

CRAG检测和纠正检索失败的能力使其比标准 RAG 更加可靠。

通过在必要时动态切换到网络搜索，它可以处理更广泛的查询，并避免陷入不相关或不充分的信息。

这种“自我纠正”能力是朝着更加强大和值得信赖的 RAG 系统迈出的重要一步。

结论

经过测试的 18 种 RAG 技术代表了提高检索质量的多种方法，从简单的分块策略到自适应 RAG 等高级方法。

虽然简单 RAG 提供了基线，但更复杂的方法（如分层索引（0.84）、Fusion（0.83）和 CRAG（0.824））通过解决检索挑战的不同方面，表现明显优于它。

Adaptive RAG 通过根据查询类型智能地选择检索策略，成为最佳表现者（0.86），表明情境感知、灵活的系统能够在满足各种信息需求的情况下提供最佳结果。