RAG各类方法python源码解读与实践:利用Jupyter对RAG技术综合评测【3万字长文】

检索增强生成（RAG ）是一种结合信息检索与生成模型的混合方法。它通过引入外部知识来提升语言模型的性能，从而提高回答的准确性和事实正确性。为了简单易学，不使用LangChain框架或FAISS向量数据库，而是利用Jupyter Notebook仅使用python基本库编写所有技术代码。从一个简单的RAG方法开始，然后测试更高级的技术，如CRAG、Fusion、HyDE等！本篇是综合篇，后续会将每种详细技术另篇一一介绍，欢迎关注我！

文章目录

评测结果
评测环境
评测准备
- 导入库
开始评测
- 简单RAG
- 语义分块
- 上下文增强检索
- 上下文块标题
- 文档增强
- 查询转换
- 重排序器
- RSE
- 上下文压缩
- 反馈循环
- 自适应RAG
- 自RAG
- 知识图谱
- 分层索引
- HyDE
- 融合
- 多模型
- Crag
结论

评测结果

Adaptive RAG以0.86的最高分，超过分层索引（0.84）、Fusion（0.83）和CRAG（0.824）成为本轮测评冠军：
通过智能分类查询并为每种问题类型选择最合适的检索策略，Adaptive RAG表现出比其他方法更好的性能。能够动态切换事实性、分析性、观点性和上下文策略，使其能够以显著的准确性处理多样化的信息需求。
在这里插入图片描述

评测环境

测试查询及其真实答案：

test query:
How does AI’s reliance on massive data sets act as a double-edged sword?True Answer:
It drives rapid learning and innovation while also 
risking the amplification of inherent biases, 
making it crucial to balance data volume with fairness and quality.

应用RAG的PDF文档：Claude 3.5推理模型生成一篇16页的AI主题文档+《Attention is all you need 》。
嵌入生成模型：TaylorAI/gte-tiny。
响应和验证的LLM：LLaMA-3.2–3B Instruct。

评测准备

导入库

下载数
请提前下载数据
安装所需的依赖项。

# 安装所需的库
pip install -r requirements.txt

开始评测

简单RAG

让我们从最简单的RAG开始。首先，我们将可视化它的工作原理，然后进行测试和评估。
简单RAG工作流程: 在这里插入图片描述
如图所示，简单RAG管道的工作流程如下：

从PDF中提取文本。
将文本分割成较小的块。
将块转换为数值嵌入。
基于查询搜索最相关的块。
使用检索到的块生成响应。
将响应与正确答案进行比较以评估准确性。

首先，让我们加载文档，提取文本并将其分割为可管理的块：

# 定义PDF文件的路径
pdf_path = "data/AI_information.pdf"# 从PDF文件中提取文本，并创建较小的重叠块。
extracted_text = extract_text_from_pdf(pdf_path)
text_chunks = chunk_text(extracted_text, 1000, 200)print("文本块数量：", len(text_chunks))### 输出 ###
文本块数量：42

此代码使用extract_text_from_pdf从我们的PDF文件中提取所有文本。然后，chunk_text将大块文本分割成较小的重叠块，每个块大约1000个字符。

接下来，我们需要将这些文本块转换为数值表示（嵌入）：

# 为文本块创建嵌入
response = create_embeddings(text_chunks)

在这里，create_embeddings获取我们的文本块列表，并使用嵌入模型为每个块生成数值嵌入。这些嵌入捕捉了文本的含义。

现在我们可以执行语义搜索，找到与测试查询最相关的块：

# 我们的测试查询，并执行语义搜索。
query = '''AI对大规模数据集的依赖如何成为一把双刃剑？'''
top_chunks = semantic_search(query, text_chunks, embeddings, k=2)

然后，semantic_search将查询嵌入与块嵌入进行比较，返回最相似的块。

有了相关块，让我们生成一个响应：

# 定义AI助手的系统提示
system_prompt = "你是一个严格基于给定上下文回答问题的AI助手。如果无法从提供的上下文中直接得出答案，请回答：'我没有足够的信息来回答这个问题。'"# 基于顶部块创建用户提示，并生成AI响应。
user_prompt = "\n".join([f"上下文 {i + 1}:\n{chunk}\n========\n" for i, chunk in enumerate(top_chunks)])
user_prompt = f"{user_prompt}\n问题：{query}"
ai_response = generate_response(system_prompt, user_prompt)
print(ai_response.choices[0].message.content)

此代码将检索到的块格式化为大语言模型（LLM）的提示。generate_response函数将此提示发送给LLM，LLM仅基于提供的上下文生成答案。

最后，让我们看看我们的简单RAG表现如何：

# 定义评估系统的系统提示
evaluate_system_prompt = "你是一个智能评估系统，负责评估AI助手的响应。如果AI助手的响应非常接近真实响应，则分配1分。如果响应不正确或与真实响应相比不令人满意，则分配0分。如果响应与真实响应部分一致，则分配0.5分。"# 创建评估提示并生成评估响应
evaluation_prompt = f"用户查询：{query}\nAI响应：\n{ai_response.choices[0].message.content}\n真实响应：{data[0]['ideal_answer']}\n{evaluate_system_prompt}"
evaluation_response = generate_response(evaluate_system_prompt, evaluation_prompt)
print(evaluation_response.choices[0].message.content)### 输出 ###
... 因此，得分为0.3，与真实响应不太接近，也不完全一致。

嗯……简单RAG的响应低于平均水平

让我们继续下一个方法。

语义分块

在我们的简单RAG方法中，我们只是将文本分割成固定大小的块。这相当粗糙！它可能会将句子分割成两半，或者将不相关的句子组合在一起。

语义分块旨在更智能。它不是固定大小，而是尝试基于含义分割文本，将语义相关的句子组合在一起。
语义分块工作流程: 在这里插入图片描述

其思想是，如果句子谈论的是相似的内容，它们应该在同一块中。我们将使用相同的嵌入模型来判断句子的相似性。

# 将文本分割成句子（基本分割）
sentences = extracted_text.split(". ")# 为每个句子生成嵌入
embeddings = [get_embedding(sentence) for sentence in sentences]print(f"生成了{len(embeddings)}个句子嵌入。")### 输出 ###
233

此代码将我们的extracted_text分割成单独的句子。然后为每个句子创建嵌入。

现在，我们将计算连续句子之间的相似性：

# 计算连续句子之间的相似性
similarities = [cosine_similarity(embeddings[i], embeddings[i + 1]) for i in range(len(embeddings) - 1)]

这个cosine_similarity函数（我们之前定义的）告诉我们两个嵌入有多相似。得分为1表示它们非常相似，0表示它们完全不同。我们为每对相邻句子计算这个得分。

语义分块是决定在哪里将文本分割成块。我们将使用“断点”方法。我们在这里使用百分位数方法，寻找相似性的大幅下降：

# 使用百分位数方法计算断点，阈值为90
breakpoints = compute_breakpoints(similarities, method="percentile", threshold=90)

compute_breakpoints函数使用“百分位数”方法识别句子之间相似性显著下降的点。这些是我们的块边界。

现在我们可以创建我们的语义块：

# 使用split_into_chunks函数创建块
text_chunks = split_into_chunks(sentences, breakpoints)
print(f"语义块数量：{len(text_chunks)}")### 输出 ###
语义块数量：145

split_into_chunks获取我们的句子列表和我们找到的断点，并将句子分组为块。

接下来，我们需要为这些块创建嵌入：

# 使用create_embeddings函数创建块嵌入
chunk_embeddings = create_embeddings(text_chunks)

是时候生成响应了：

# 基于顶部块创建用户提示
user_prompt = "\n".join([f"上下文 {i + 1}:\n{chunk}\n=====================================\n" for i, chunk in enumerate(top_chunks)])
user_prompt = f"{user_prompt}\n问题：{query}"# 生成AI响应
ai_response = generate_response(system_prompt, user_prompt)
print(ai_response.choices[0].message.content)

最后，进行评估：

# 通过组合用户查询、AI响应、真实响应和评估系统提示来创建评估提示
evaluation_prompt = f"用户查询：{query}\nAI响应：\n{ai_response.choices[0].message.content}\n真实响应：{data[0]['ideal_answer']}\n{evaluate_system_prompt}"# 使用评估系统提示和评估提示生成评估响应
evaluation_response = generate_response(evaluate_system_prompt, evaluation_prompt)# 打印评估响应
print(evaluation_response.choices[0].message.content)### 输出
根据评估标准，
我会给AI助手的响应分配0.2分。

评估者只给了0.2分

虽然语义分块听起来不错，但它在这里并没有帮助我们。事实上，与简单的固定大小分块相比，我们的得分下降了！

这表明仅仅改变分块策略并不能保证成功。我们需要更复杂的方法。让我们在下一节中尝试其他方法。

上下文增强检索

我们看到语义分块虽然在原则上是个好主意，但实际上并没有改善我们的结果。

一个问题是，即使是语义定义的块也可能过于集中。它们可能缺少周围文本的关键上下文。

上下文增强检索通过不仅抓取最佳匹配块，还抓取其邻居来解决这个问题。

让我们看看如何在代码中实现这一点。我们需要一个新函数context_enriched_search来处理检索：

def context_enriched_search(query, text_chunks, embeddings, k=1, context_size=1):"""检索最相关的块及其相邻块。"""# 将查询转换为嵌入向量query_embedding = create_embeddings(query).data[0].embeddingsimilarity_scores = []    # 计算查询与每个文本块嵌入之间的相似性得分for i, chunk_embedding in enumerate(embeddings):# 计算查询嵌入与当前块嵌入之间的余弦相似性similarity_score = cosine_similarity(np.array(query_embedding), np.array(chunk_embedding.embedding))# 将索引和相似性得分存储为元组similarity_scores.append((i, similarity_score))    # 按相似性得分降序排序（最高相似性优先）similarity_scores.sort(key=lambda x: x[1], reverse=True)    # 获取最相关块的索引top_index = similarity_scores[0][0]    # 定义上下文包含的范围# 确保我们不低于0或超出text_chunks的长度start = max(0, top_index - context_size)end = min(len(text_chunks), top_index + context_size + 1)    # 返回相关块及其相邻上下文块return [text_chunks[i] for i in range(start, end)]

核心逻辑与我们之前的搜索类似，但不是只返回单个最佳块，而是抓取其周围的“窗口”块。context_size控制我们包含的块数量。

让我们在RAG管道中使用它。我们将跳过文本提取和分块步骤，因为这些与简单RAG中的步骤相同。

我们将使用固定大小的块，就像在简单RAG部分中一样，并且我们保持chunk_size = 1000和overlap = 200。

现在生成响应，与之前相同：

# 基于顶部块创建用户提示
user_prompt = "\n".join([f"上下文 {i + 1}:\n{chunk}\n=====================================\n" for i, chunk in enumerate(top_chunks)])
user_prompt = f"{user_prompt}\n问题：{query}"# 生成AI响应
ai_response = generate_response(system_prompt, user_prompt)
print(ai_response.choices[0].message.content)

最后，进行评估：

# 创建评估提示并生成评估响应
evaluation_prompt = f"用户查询：{query}\nAI响应：\n{ai_response.choices[0].message.content}\n真实响应：{data[0]['ideal_answer']}\n{evaluate_system_prompt}"
evaluation_response = generate_response(evaluate_system_prompt, evaluation_prompt)
print(evaluation_response.choices[0].message.content)### 输出 ###
根据评估标准，
我会给AI助手的响应分配0.6分。

这次，我们得到了0.6的评估得分！

这比简单RAG（0.5）和语义分块（0.1）都有显著提升。

通过包含相邻块，我们为LLM提供了更多的上下文，从而生成了更好的答案。

我们还没有达到完美，但我们肯定在朝着正确的方向前进。这表明上下文对于检索的重要性。

上下文块标题

我们已经看到通过包含相邻块来添加上下文是有帮助的。但如果块本身的内容缺少重要信息怎么办？

通常，文档具有清晰的结构——标题、副标题——这些结构提供了关键的上下文。上下文块标题（CCH）利用了这一结构。

其思想很简单：在我们创建嵌入之前，我们预先为每个块添加一个描述性标题。这个标题就像一个迷你摘要，为检索系统（和LLM）提供了更多信息。

generate_chunk_header函数将分析每个文本块并生成一个简洁、有意义的标题，总结其内容。这有助于高效组织和检索相关信息。

# 对提取的文本进行分块，这次生成标题
text_chunks_with_headers = chunk_text_with_headers(extracted_text, 1000, 200)# 打印一个样本以查看其外观
print("带标题的样本块：")
print("标题：", text_chunks_with_headers[0]['header'])
print("内容：", text_chunks_with_headers[0]['text'])### 输出 ###
带标题的样本块：
标题：关于AI影响的描述
内容：自...以来，AI一直是社会的重要组成部分...

看到每个块现在都有一个标题和原始文本了吗？这是我们将使用的增强数据。

现在为嵌入。我们将为标题和文本创建嵌入：

# 为每个块生成嵌入（标题和文本）
embeddings = []
for chunk in tqdm(text_chunks_with_headers, desc="生成嵌入"):text_embedding = create_embeddings(chunk["text"])header_embedding = create_embeddings(chunk["header"])embeddings.append({"header": chunk["header"], "text": chunk["text"], "embedding": text_embedding, "header_embedding": header_embedding})

我们遍历我们的块，获取标题和文本的嵌入，并将所有内容存储在一起。这为检索系统提供了两种匹配块与查询的方式。

由于semantic_search已经与嵌入一起工作，我们只需要确保我们的标题和文本块都正确嵌入。这样，当我们执行搜索时，模型可以考虑高级摘要（标题）和详细内容（块文本）以找到最相关的信息。

现在，让我们修改我们的检索步骤，以返回匹配的块及其标题，以便更好地生成上下文和响应。

# 使用查询和新嵌入执行语义搜索
top_chunks = semantic_search(query, embeddings, k=2)# 基于顶部块创建用户提示。注意：不需要添加标题
# 因为上下文已经使用标题和块创建
user_prompt = "\n".join([f"上下文 {i + 1}:\n{chunk['text']}\n=====================================\n" for i, chunk in enumerate(top_chunks)])
user_prompt = f"{user_prompt}\n问题：{query}"# 生成AI响应
ai_response = generate_response(system_prompt, user_prompt)
print(ai_response.choices[0].message.content)### 输出 ###
评估得分：0.5

这次，我们的评估得分是0.5！

通过添加上下文标题，我们为系统提供了更好的机会来找到正确的信息，并为LLM提供了更好的机会来生成完整且准确的答案。

这表明在我们将数据输入检索系统之前增强数据的强大之处。我们没有改变核心RAG管道，但我们使数据本身更具信息性。

文档增强

我们已经看到如何通过添加上下文围绕我们的块（通过邻居或标题）来帮助检索。现在，让我们尝试另一种增强方式：从我们的文本块生成问题。

其思想是，这些问题可以作为替代“查询”，可能比原始文本块本身更好地匹配用户的意图。
文档增强工作流程: 在这里插入图片描述

我们在分块和嵌入创建之间添加此步骤。我们可以简单地使用generate_questions函数来实现这一点。它接受一个text_chunk并返回可以使用它生成的若干问题。

让我们首先看看如何通过问题生成实现文档增强：

# 处理文档（提取文本、创建块、生成问题、构建向量存储）
text_chunks, vector_store = process_document(pdf_path,chunk_size=1000,chunk_overlap=200,questions_per_chunk=3
)print(f"向量存储包含{len(vector_store.texts)}个项目")### 输出 ###
向量存储包含214个项目

在这里，process_document函数完成了所有工作。它接受pdf_path、chunk_size、overlap和questions_per_chunk并返回一个vector_store。

现在，vector_store不仅包含文档的嵌入，还包含生成问题的嵌入。

现在，我们可以像以前一样使用这个vector_store执行语义搜索。我们在这里使用一个简单的函数来查找相似的向量。

# 执行语义搜索以查找相关内容
search_results = semantic_search(query, vector_store, k=5)print("查询：", query)
print("\n搜索结果：")# 按类型组织结果
chunk_results = []
question_results = []for result in search_results:if result["metadata"]["type"] == "chunk":chunk_results.append(result)else:question_results.append(result

这里的重要变化是我们如何处理搜索结果。我们现在在向量存储中有两种类型的项目：原始文本块和生成的问题。这段代码将它们分开，以便我们可以看到哪种类型的内容与查询匹配得最好。

最后一步，生成上下文然后进行评估：

# 从搜索结果中准备上下文
context = prepare_context(search_results)# 生成响应
response_text = generate_response(query, context)# 从验证数据中获取参考答案
reference_answer = data[0]['ideal_answer']# 评估响应
evaluation = evaluate_response(query, response_text, reference_answer)print("\n评估：")
print(evaluation)### 输出 ###
根据评估标准，我会给
AI助手的响应分配0.8分。

我们的评估显示得分约为0.8！

生成问题并将它们添加到我们的可搜索索引中，使我们的性能再次提升。

似乎有时，问题比原始文本块更能代表信息需求。

查询转换

到目前为止，我们一直专注于改进RAG系统使用的数据。但是查询本身呢？

通常，用户提问的方式并不是搜索我们知识库的最佳方式。查询转换旨在解决这个问题。我们将探索三种不同的方法：

**查询重写：**使查询更具体和详细。
**后退提示：**创建一个更广泛、更通用的查询以检索背景上下文。
**子查询分解：**将复杂查询分解为多个更简单的子查询。

让我们看看这些转换的实际效果。我们将使用我们的标准测试查询：

# 查询重写
rewritten_query = rewrite_query(query)# 后退提示
step_back_query = generate_step_back_query(query)

generate_step_back_query与重写相反：它创建一个更广泛的查询，可能会检索到有用的背景信息。

最后，子查询分解：

# 子查询分解
sub_queries = decompose_query(query, num_subqueries=4)

decompose_query将原始查询分解为几个更小、更集中的问题。其思想是，这些子查询一起可能比任何单个查询更好地覆盖原始查询的意图。

现在，为了查看这些转换如何影响我们的RAG系统，让我们使用一个结合了所有先前方法的函数：

def rag_with_query_transformation(pdf_path, query, transformation_type=None):"""运行完整的RAG管道，可选择查询转换。    Args:pdf_path (str): PDF文档路径query (str): 用户查询transformation_type (str): 转换类型（None、'rewrite'、'step_back'或'decompose'）    Returns:Dict: 结果包括查询、转换后的查询、上下文和响应"""# 处理文档以创建向量存储vector_store = process_document(pdf_path)    # 应用查询转换并搜索if transformation_type:# 使用转换后的查询执行搜索results = transformed_search(query, vector_store, transformation_type)else:# 执行常规搜索，不进行转换query_embedding = create_embeddings(query)results = vector_store.similarity_search(query_embedding, k=3)    # 从搜索结果中组合上下文context = "\n\n".join([f"段落 {i+1}:\n{result['text']}" for i, result in enumerate(results)])    # 基于查询和组合上下文生成响应response = generate_response(query, context)    # 返回结果，包括原始查询、转换类型、上下文和响应return {"original_query": query,"transformation_type": transformation_type,"context": context,"response": response}

evaluate_transformations函数将原始查询通过不同的查询转换技术——重写、后退和分解，然后比较它们的输出。

这有助于我们查看哪种方法检索到最相关的信息以生成更好的响应。

# 运行评估
evaluation_results = evaluate_transformations(pdf_path, query, reference_answer)
print(evaluation_results)### 输出 ###
评估得分：0.5

评估得分为0.5。

这表明我们的查询转换技术并没有始终优于更简单的方法。

虽然查询转换可以很强大，但它们并不是万能的。有时，原始查询已经很好地表达了意图，试图“改进”它实际上可能会使事情变得更糟。

重排序器

我们已经尝试改进数据（通过分块策略）和查询（通过转换）。现在，让我们专注于检索过程本身。简单的相似性搜索通常会返回相关和不相关结果的混合。

重排序是第二次处理，重新排序最初检索到的结果，将最好的结果放在顶部。

rerank_with_llm函数获取初始检索到的块，并使用LLM根据相关性重新排序。这有助于确保最有用的信息首先出现。

重排序后，一个最终函数——我们称之为generate_final_response——获取重新排序的块，将它们格式化为提示，并将它们发送给LLM以生成最终响应。

def rag_with_reranking(query, vector_store, reranking_method="llm", top_n=3, model="meta-llama/Llama-3.2-3B-Instruct"):"""包含重排序的完整RAG管道。"""# 创建查询嵌入query_embedding = create_embeddings(query)        # 初始检索（获取比我们需要的更多的结果以便重排序）initial_results = vector_store.similarity_search(query_embedding, k=10)        # 应用重排序if reranking_method == "llm":reranked_results = rerank_with_llm(query, initial_results, top_n=top_n)elif reranking_method == "keywords":reranked_results = rerank_with_keywords(query, initial_results, top_n=top_n) # 我们不使用它。else:# 不重排序，仅使用初始检索的顶部结果reranked_results = initial_results[:top_n]        # 从重排序结果中组合上下文context = "\n\n===\n\n".join([result["text"] for result in reranked_results])        # 基于上下文生成响应response = generate_response(query, context, model)        return {"query": query,"reranking_method": reranking_method,"initial_results": initial_results[:top_n],"reranked_results": reranked_results,"context": context,"response": response}

它接受一个query、一个vector_store（我们已经创建）和一个reranking_method。我们使用“llm”进行基于LLM的重排序。该函数执行初始检索，调用rerank_with_llm重新排序结果，然后生成响应。

rerank_with_keywords在笔记本中定义，但我不在这里使用它。

让我们运行这个函数，看看它是否改善了我们的结果：

# 运行带有LLM重排序的RAG
llm_reranked_result = rag_with_reranking(query, vector_store, reranking_method="llm")# 评估。
evaluation_prompt = f"用户查询：{query}\nAI响应：\n{llm_reranked_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.7

我们的评估得分现在约为0.7！

重排序显著改善了我们的结果。通过使用LLM直接评分每个检索到的文档的相关性，我们能够优先考虑最佳信息以生成响应。

这是一种强大的技术，可以显著提高RAG系统的质量。

RSE

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

RSE不是仅仅抓取前k个块，而是尝试识别并提取整个相关文本段。

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

# 运行带有RSE的RAG
rse_result = rag_with_rse(pdf_path, query)

这一行代码做了很多工作！它：

处理文档（提取文本、分块、创建嵌入，所有这些都在rag_with_rse内部处理）。
根据查询的相关性和位置计算“块值”。
使用一种巧妙的算法找到最佳的连续段块。
将这些段组合成上下文。
基于该上下文生成响应。

现在，进行评估：

# 评估
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)### 输出 ###
然而，标准检索的响应包括...
我会给AI响应分配0.8分

这次，我们得到了大约0.8的得分！

通过专注于连续段的相关文本，RSE为LLM提供了更连贯和完整的上下文，从而生成了更准确和全面的响应。

这表明如何选择和呈现信息给LLM与选择什么信息同样重要。

上下文压缩

我们一直在添加上下文，相邻块，生成的问题，整个段。但有时，少即是多。

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（检索增强生成）管道。    Args:pdf_path (str): PDF文档路径。query (str): 用于检索的用户查询。k (int): 要检索的顶部相关块的数量。默认为10。compression_type (str): 应用于检索块的压缩类型。默认为“selective”。model (str): 用于响应生成的语言模型。默认为“meta-llama/Llama-3.2-3B-Instruct”。    Returns:dict: 包含查询、原始和压缩块、压缩统计信息和最终响应的字典。"""        print(f"\n=== 带有压缩的RAG ===\n查询：{query} | 压缩：{compression_type}")        # 处理文档以提取、分块和嵌入文本vector_store = process_document(pdf_path)        # 基于查询相似性检索前k个相关块results = vector_store.similarity_search(create_embeddings(query), k=k)retrieved_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 retrieved_chunks])        # 组合压缩块以形成响应生成的上下文context = "\n\n---\n\n".join(compressed_chunks)        # 使用压缩上下文生成响应response = generate_response(query, context, model)    print(f"\n=== 响应 ===\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提供了不同的压缩类型选项：

“selective” 仅保留直接相关的句子。
“summary” 创建一个专注于查询的简短摘要。
“extraction” 仅提取包含答案的句子（非常严格！）。

现在，运行压缩代码：

# 运行带有上下文压缩的RAG（使用'selective'模式）
compression_result = rag_with_compression(pdf_path, query, compression_type="selective")# 评估。
evaluation_prompt = f"用户查询：{query}\nAI响应：\n{compression_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.75

这次我们得到了大约0.75的得分。

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

通过仅向LLM提供最相关的信息，我们通常能得到更简洁和准确的答案。

反馈循环

到目前为止，我们所见的所有技术都是“静态的”，它们不会从错误中学习。反馈循环改变了这一点。

其思想很简单：

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

我们可以使用一个函数调用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"从{feedback_file}加载了{len(feedback_data)}条反馈条目")        # 步骤2：通过提取、分块和嵌入管道处理文档chunks, vector_store = process_document(pdf_path)        # 步骤3：通过合并高质量的历史交互来微调向量索引# 这从成功的Q&A对中创建增强的可检索内容if fine_tune and feedback_data:vector_store = fine_tune_index(vector_store, chunks, feedback_data)        # 步骤4：执行核心RAG，使用反馈感知检索# 注意：这依赖于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函数做了几件事：

**加载现有反馈：**它检查feedback_data.json文件并加载任何先前的反馈。
**运行RAG管道：**这部分与我们之前所做的类似。
**请求反馈：**它提示用户对响应的相关性和质量进行评分。
**存储反馈：**它将反馈保存到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']}\n真实响应：{reference_answer}\n{evaluate_system_prompt}"
evaluation_response = generate_response(evaluate_system_prompt, evaluation_prompt)
print(evaluation_response.choices[0].message.content)### 输出 ###
评估得分为0.7，因为...

我们看到得分约为0.7！

这并不是一个巨大的飞跃，这是预期的。反馈循环会随着时间的推移改进系统，通过重复的交互。本节只是展示了机制。

真正的力量来自于积累反馈并使用它来优化检索过程。这使得RAG系统自适应并个性化于其接收的查询类型。

自适应RAG

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

但如果最佳技术取决于所提问题的类型呢？这就是自适应RAG的思想。

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

**事实性策略：**专注于检索精确的事实和数据。
**分析性策略：**旨在全面覆盖一个主题，探索不同的方面。
**观点性策略：**尝试收集关于主观问题的多样化观点。
**上下文策略：**结合用户特定的上下文以定制检索。

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

def rag_with_adaptive_retrieval(pdf_path, query, k=4, user_context=None):"""带有自适应检索的完整RAG管道。    """print("\n=== 带有自适应检索的RAG ===")print(f"查询：{query}")        # 处理文档以提取文本、分块并创建嵌入chunks, vector_store = process_document(pdf_path)        # 分类查询以确定其类型query_type = classify_query(query)print(f"查询分类为：{query_type}")        # 使用基于查询类型的自适应检索策略检索文档retrieved_docs = adaptive_retrieval(query, vector_store, k, user_context)        # 基于查询、检索到的文档和查询类型生成响应response = generate_response(query, retrieved_docs, query_type)        # 将结果编译为字典result = {"query": query,"query_type": query_type,"retrieved_documents": retrieved_docs,"response": response}        print("\n=== 响应 ===")print(response)        return result

它首先使用一个名为classify_query的函数对查询进行分类，该函数与其他辅助函数一起定义。

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

该函数返回一个包含结果的字典，包括查询、查询类型、检索到的文档和生成的响应。

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

# 运行自适应RAG管道
result = rag_with_adaptive_retrieval(pdf_path, query)# 评估。
evaluation_prompt = f"用户查询：{query}\nAI响应：\n{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.86

这次我们得到了大约0.856的得分。

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

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

自RAG

到目前为止，我们的RAG系统主要是反应式的。它们接受查询，检索信息，并生成响应。自RAG采取了不同的方法：它是主动式和反思式的。

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

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

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

自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（LLM可能已经知道这个）。
对于一个创造性任务，如**“写一首诗…”**，它也可能返回False。
但对于一个更复杂或小众的查询，它将返回True。

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

def evaluate_relevance(query, context):"""（说明性示例 - 非完全功能）评估上下文与查询的相关性。"""system_prompt = """你是一个AI助手。确定文档是否与查询相关。仅回答“相关”或“不相关”。"""    user_prompt = f"""查询：{query}文档内容：{context[:500]}... [截断]    此文档是否与查询相关？仅回答“相关”或“不相关”。"""    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来判断检索到的文档是否与查询相关。

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

最后调用所有这些：

# 我们可以调用`self_rag`函数进行自RAG，它会自动
# 决定何时检索以及何时不检索。
result = self_rag(query, vector_store)print(result["response"])### 输出 ###
AI响应的评估得分为0.65

我们在这里得到了0.6的得分。

这反映了以下事实：

自RAG具有很大的潜力，但完整的实现很复杂。
即使是“是否需要检索？”这一步，我们展示的，有时也可能是错误的。
我们没有展示完整的“反思”过程，因此我们不能声称更高的得分。

关键要点是，自RAG使RAG系统更智能和自适应。这是朝着LLM能够推理其自身知识和检索需求迈出的一步。

知识图谱

到目前为止，我们的RAG系统将文档视为独立块的集合。但如果信息是相互关联的呢？如果理解一个概念需要理解相关概念呢？这就是图RAG的用武之地。

图RAG不是将信息组织为平面列表，而是将其组织为知识图谱。可以将其视为一个网络：

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

知识图谱工作流程

核心思想是，通过遍历这个图，我们不仅可以找到直接相关的信息，还可以找到间接相关的信息，这些信息提供了关键的上下文。

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

def build_knowledge_graph(chunks):"""使用嵌入和概念提取从文本块构建知识图谱。    Args:chunks (list of dict): 包含“text”字段的文本块列表。    Returns:tuple: （以文本块为节点的图，嵌入列表）"""graph, texts = nx.Graph(), [c["text"] for c in chunks]embeddings = create_embeddings(texts)  # 计算嵌入    # 添加带有提取概念和嵌入的节点for i, (chunk, emb) in enumerate(zip(chunks, embeddings)):graph.add_node(i, text=chunk["text"], concepts := extract_concepts(chunk["text"]), embedding=emb)    # 基于共享概念和嵌入相似性创建边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.number_of_nodes()}个节点，{graph.number_of_edges()}条边")return graph, embeddings

它接受一个query、一个graph和embeddings，并返回相关节点列表和遍历路径。

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

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

让我们使用这个来生成响应：

# 执行图RAG管道以处理文档并回答查询
results = graph_rag_pipeline(pdf_path, query)# 评估。
evaluation_prompt = f"用户查询：{query}\nAI响应：\n{results['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.78

我们得到了大约0.78的得分。

图RAG并没有超越更简单的方法，但它可以捕捉信息片段之间的关系，而不仅仅是单个片段本身。
KG: 在这里插入图片描述

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

分层索引

我们已经探索了各种改进RAG的方法：更好的分块、上下文增强、查询转换、重排序，甚至基于图的检索。但有一个基本的权衡：

小块：适合精确匹配，但会丢失上下文。
大块：保留上下文，但可能导致检索结果不太相关。

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

摘要：文档较大部分的简明概述。
详细块：这些部分中的较小块。
首先，搜索摘要：这可以快速缩小文档的相关部分。
然后，仅在这些部分中搜索详细块：这提供了小块的精确性，同时保持较大部分的上下文。

让我们使用一个函数调用hierarchical_rag来看看这个实际效果：

def hierarchical_rag(query, pdf_path, chunk_size=1000, chunk_overlap=200, k_summaries=3, k_chunks=5, regenerate=False):"""完整的分层检索增强生成（RAG）管道。    Args:query (str): 用户查询。pdf_path (str): PDF文档路径。chunk_size (int): 处理文本块的大小。chunk_overlap (int): 连续块之间的重叠。k_summaries (int): 要检索的顶部摘要数量。k_chunks (int): 每个摘要要检索的详细块数量。regenerate (bool): 是否重新处理文档。    Returns:dict: 包含查询、生成的响应、检索到的块、 摘要和详细块的数量。"""# 定义摘要和详细向量存储的缓存文件名summary_store_file = f"{os.path.basename(pdf_path)}_summary_store.pkl"detailed_store_file = f"{os.path.basename(pdf_path)}_detailed_store.pkl"        # 如果需要重新生成或缓存文件缺失，则处理文档if regenerate or not os.path.exists(summary_store_file) or not os.path.exists(detailed_store_file):print("处理文档并创建向量存储...")summary_store, detailed_store = process_document_hierarchically(pdf_path, chunk_size, chunk_overlap)                # 保存处理后的存储以供将来使用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:# 从缓存加载现有向量存储print("加载现有向量存储...")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)        # 使用分层搜索检索相关块retrieved_chunks = retrieve_hierarchically(query, summary_store, detailed_store, k_summaries, k_chunks)        # 基于检索到的块生成响应response = generate_response(query, retrieved_chunks)        # 返回带有元数据的结果return {"query": query,"response": response,"retrieved_chunks": retrieved_chunks,"summary_count": len(summary_store.texts),"detailed_count": len(detailed_store.texts)}

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

首先，它搜索summary_store以找到最相关的摘要。
然后，它搜索detailed_store，但仅限于属于顶部摘要的块。这比搜索所有详细块要高效得多。

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

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

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

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

# 评估。
evaluation_prompt = f"用户查询：{query}\nAI响应：\n{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.84

我们的得分是0.84 😆

分层检索提供了迄今为止的最佳得分。

我们获得了搜索摘要的速度，以及搜索较小块的精确性，加上知道每个块属于哪个部分所带来的额外上下文。这就是为什么它通常是表现最好的RAG策略。

HyDE

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

流程是：

**生成假设文档：**使用LLM创建一个会回答查询的文档，如果它存在的话。
**嵌入假设文档：**创建这个假设文档的嵌入，而不是原始查询。
**检索：**找到与假设文档嵌入相似的文档。
**生成：**使用检索到的文档（而不是假设的文档！）来回答查询。

其思想是，一个完整的文档，即使是一个假设的文档，也比一个简短的查询具有更丰富的语义表示。这可以帮助弥合查询与嵌入空间中文档之间的差距。

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

我们使用generate_hypothetical_document来实现这一点：

def generate_hypothetical_document(query, desired_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}个最相似的块...")retrieved_chunks = vector_store.similarity_search(hypothetical_embedding, k=k)        # 准备结果字典results = {"query": query,"hypothetical_document": hypothetical_doc,"retrieved_chunks": retrieved_chunks}        # 步骤4：如果请求，生成最终响应if should_generate_response:print("生成最终响应...")response = generate_response(query, retrieved_chunks)results["response"] = response        return results

hyde_rag函数现在：

生成假设文档。
创建该文档的嵌入（而不是查询！）。
使用该嵌入进行检索。
生成响应，与之前相同。

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

# 运行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技术。不同的方法适用于不同的查询和不同的数据。

融合

我们已经看到不同的检索方法有不同的优势。向量搜索擅长语义相似性，而关键词搜索擅长找到精确匹配。如果我们可以结合它们呢？这就是融合RAG的思想。

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

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

以下是融合检索的函数：

import numpy as npdef fusion_retrieval(query, chunks, vector_store, bm25_index, k=5, alpha=0.5):"""通过结合基于向量和BM25的搜索结果执行融合检索。"""        # 为查询生成嵌入query_embedding = create_embeddings(query)    # 执行向量搜索并将结果存储在字典中（索引 -> 相似性得分）vector_results = {r["metadata"]["index"]: r["similarity"] for r in vector_store.similarity_search_with_scores(query_embedding, len(chunks))}    # 执行BM25搜索并将结果存储在字典中（索引 -> BM25得分）bm25_results = {r["metadata"]["index"]: r["bm25_score"] for r in bm25_search(bm25_index, chunks, query, len(chunks))}    # 从向量存储中检索所有文档all_docs = vector_store.get_all_documents()    # 使用向量和BM25得分的加权和计算每个文档的组合得分scores = [(i, alpha * vector_results.get(i, 0) + (1 - alpha) * bm25_results.get(i, 0)) for i in range(len(all_docs))]    # 按组合得分降序排序文档并保留前k个结果top_docs = sorted(scores, key=lambda x: x[1], reverse=True)[:k]    # 返回带有文本、元数据和组合得分的前k个文档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"])# 评估。
evaluation_prompt = f"用户查询：{query}\nAI响应：\n{fusion_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)### 输出
AI响应的评估得分为0.83

最终得分为0.83。

融合RAG通常为我们提供了显著的提升，因为它结合了不同检索方法的优势。

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

多模型

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

这里的关键变化是：

**提取文本和图像：**我们从PDF中提取文本和图像。
**生成图像标题：**我们使用LLM（特别是具有视觉能力的模型）为每个图像生成文本描述（标题）。
**创建嵌入（文本和标题）：**我们为文本块和图像标题创建嵌入。
**嵌入模型：**在这个笔记本中，我们使用BAAI/bge-en-icl嵌入模型。
**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        # 提取嵌入内容contents = [item["content"] for item in all_items]        # 为所有内容创建嵌入print("为所有内容创建嵌入...")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"用户查询：{query}\nAI响应：\n{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.79

我们得到了大约0.79的得分。

多模态RAG具有潜力，特别是对于图像包含关键信息的文档。然而，它并没有击败我们迄今为止看到的其他技术。

Crag

到目前为止，我们的RAG系统相对被动。它们检索信息并生成响应。但如果检索到的信息不好怎么办？如果它不相关、不完整，甚至矛盾呢？纠正性RAG（CRAG）直面这个问题。

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

**高相关性：**如果检索到的文档很好，则照常进行。
**低相关性：**如果检索到的文档不好，则回退到网络搜索！
**中等相关性：**如果文档还行，则结合文档和网络的信息。

这种“纠正”机制使CRAG比标准RAG更加稳健。它不仅仅希望最好；它正在积极检查并适应。

让我们看看这在实践中是如何工作的。我们将使用一个名为rag_with_compression的函数来实现这一点。

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

这个单一的函数调用做了很多：

**初始检索：**像往常一样检索文档。
**相关性评估：**为每个文档的相关性评分。
**决策：**决定是使用文档、进行网络搜索，还是结合两者。
**响应生成：**使用选择的知识源生成响应。

并且，一如既往，进行评估：

# 评估。
evaluation_prompt = f"用户查询：{query}\nAI响应：\n{crag_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.824