免费做字体的网站,无锡网知名网站,wordpress 图片页面,自己如何做简单网站在上一篇教程中#xff0c;我们了解了 Naive RAG 的基本原理和实现。它就像一个刚刚学会查找资料的新手#xff0c;虽然能找到一些信息#xff0c;但有时候找到的并不够精准#xff0c;甚至会有一些无关的干扰。
今天#xff0c;我们将介绍 Retrieve-and-Rerank RAG…在上一篇教程中我们了解了 Naive RAG 的基本原理和实现。它就像一个刚刚学会查找资料的新手虽然能找到一些信息但有时候找到的并不够精准甚至会有一些无关的干扰。
今天我们将介绍 Retrieve-and-Rerank RAG它就像给那位新手配了一个经验丰富的“审阅员”和一位“分类能手”。它不再是“找到什么就给什么”而是会 先广泛查找快速从海量信息中捞出一大堆“可能相关”的资料召回。 再精挑细选让“审阅员”仔细阅读这些资料并根据与问题的真实相关性进行打分和排序最终选出最精准的几份精排。
这样一来Retrieve-and-Rerank RAG 就能显著提升检索的精准度让大语言模型 (LLM) 拿到更高质量的参考资料从而生成更准确、更可靠的答案。 1. Retrieve-and-Rerank RAG 的工作流程两阶段“精选”战略 与 Naive RAG 的三阶段流水线相比Retrieve-and-Rerank RAG 的核心创新在于其两阶段的检索过程 第一阶段广泛召回Retrieve 这个阶段的目标是“宁可错杀一千不可放过一个”尽可能多地召回与查询可能相关的文档片段。为了实现这一点我们通常会采用混合检索策略 向量检索基于文档和查询的“数字指纹”向量相似度进行匹配。它能捕捉语义上的相似性即使关键词不完全匹配也能找到相关内容。 关键词检索如 BM25基于传统的关键词匹配算法查找包含查询中特定词语的文档。这对于精确匹配和专有名词的查找非常有效。
这两种方法互补向量检索弥补了关键词检索无法理解语义的缺点而关键词检索则弥补了向量检索在某些精确匹配上的不足。通过混合检索我们可以得到一个相对较大的“Top-K 候选”文档集合比如 50-100 个文档片段。 第二阶段精确重排序Rerank 这是 Retrieve-and-Rerank RAG 的精髓所在。上一步召回的文档片段数量多但质量参差不齐。这时我们就需要“审阅员”登场了 交叉编码器 (Cross-Encoder) 重排它是一种特殊的深度学习模型能够同时理解查询和每个文档片段的上下文然后给出一个精确的相关性分数。与向量检索的“独立打分”先给查询打分再给文档打分然后比较不同交叉编码器是“联合打分”它能更好地捕捉查询和文档之间的复杂关系。 根据交叉编码器给出的分数我们对所有候选文档进行重新排序并从中选出最相关的“Top-N 精选”文档通常 N 远小于 K比如 5-10 个。这些精选出的文档质量更高与查询的匹配度也更强。 第三阶段生成答案Generate 最后与 Naive RAG 类似我们将这些经过精选的高质量文档作为上下文输入给大语言模型由它来生成最终的答案。由于 LLM 获得的参考资料质量更高生成的答案也会更加准确和可靠。 2. 动手实践升级你的 RAG 系统 现在我们将深入代码看看如何实现 Retrieve-and-Rerank RAG 的核心组件。 2.1 核心组件概览 我们将构建以下关键组件 HybridRetriever (混合检索器)负责结合向量检索和关键词检索实现第一阶段的广泛召回。 CrossEncoderReranker (交叉编码器重排序器)利用交叉编码器模型对召回的文档进行精确排序实现第二阶段的精排。 RetrieveRerankRAG (主控制器)整合检索器和重排序器调度整个 RAG 流程。 2.2 混合检索器 (HybridRetriever)
from typing import List, Tuple, Dict
import numpy as np
from rank_bm25 import BM25Okapi # 用于关键词检索
from sentence_transformers import SentenceTransformer # 用于向量嵌入
import faiss # 用于向量索引class HybridRetriever:def __init__(self, embedding_model: str all-MiniLM-L6-v2):# 初始化用于向量化的嵌入模型self.embedder SentenceTransformer(embedding_model)self.vector_index None # Faiss 向量索引self.bm25_index None # BM25 关键词索引self.documents [] # 存储原始文档文本self.doc_embeddings None # 存储文档的向量嵌入def build_index(self, documents: List[str]):构建向量索引和BM25索引self.documents documentsprint(开始构建向量索引...)# 1. 构建向量索引将文档转换为向量并添加到 Faissdoc_embeddings self.embedder.encode(documents, convert_to_numpyTrue)self.doc_embeddings doc_embeddingsembedding_dim doc_embeddings.shape[1]self.vector_index faiss.IndexFlatIP(embedding_dim) # 使用内积相似度self.vector_index.add(doc_embeddings.astype(float32))print(向量索引构建完成。)print(开始构建BM25索引...)# 2. 构建BM25索引对文档进行分词并构建 BM25 索引# BM25 通常对小写和分词后的文本效果更好tokenized_docs [doc.lower().split() for doc in documents] self.bm25_index BM25Okapi(tokenized_docs)print(BM25索引构建完成。)def vector_search(self, query: str, k: int 50) - List[Tuple[int, float]]:执行向量检索query_embedding self.embedder.encode([query], convert_to_numpyTrue)scores, indices self.vector_index.search(query_embedding.astype(float32), k)# 返回 (文档在原始列表中的索引, 相似度分数)return [(idx, float(score)) for idx, score in zip(indices[0], scores[0]) if idx ! -1]def bm25_search(self, query: str, k: int 50) - List[Tuple[int, float]]:执行BM25关键词检索query_tokens query.lower().split()scores self.bm25_index.get_scores(query_tokens)# 获取 Top-K 结果并确保分数大于0表示有匹配top_indices np.argsort(scores)[::-1] # 降序排列results []for idx in top_indices:if scores[idx] 0 and len(results) k: # 过滤零分结果并限制数量results.append((idx, float(scores[idx])))return resultsdef hybrid_search(self, query: str, k: int 50, vector_weight: float 0.7, bm25_weight: float 0.3) - List[Tuple[int, float]]:执行混合检索融合向量检索和BM25检索的结果# 1. 获取两种检索结果vector_results self.vector_search(query, k)bm25_results self.bm25_search(query, k)# 2. 归一化分数将不同检索方法的分数统一到 [0, 1] 范围vector_scores_only [score for _, score in vector_results]bm25_scores_only [score for _, score in bm25_results]normalized_vector_scores self._normalize_scores(vector_scores_only)normalized_bm25_scores self._normalize_scores(bm25_scores_only)# 3. 融合分数根据权重合并不同检索方法的分数combined_scores {}# 向量检索结果加入for i, (idx, _) in enumerate(vector_results):combined_scores[idx] vector_weight * normalized_vector_scores[i]# BM25检索结果加入如果文档已存在则累加分数for i, (idx, _) in enumerate(bm25_results):if idx in combined_scores:combined_scores[idx] bm25_weight * normalized_bm25_scores[i]else:combined_scores[idx] bm25_weight * normalized_bm25_scores[i]# 4. 排序并返回 Top-K 综合分数最高的文档索引和分数sorted_results sorted(combined_scores.items(), keylambda x: x[1], reverseTrue)[:k]return [(idx, score) for idx, score in sorted_results]def _normalize_scores(self, scores: List[float]) - List[float]:分数归一化到 [0, 1] 范围if not scores:return []min_score, max_score min(scores), max(scores)if max_score min_score: # 避免除以零return [1.0] * len(scores)return [(score - min_score) / (max_score - min_score) for score in scores]
代码解析 HybridRetriever: 这是一个融合了向量检索基于 SentenceTransformer 和 Faiss和关键词检索基于 BM25Okapi的组件。 build_index: 同时构建两种索引为后续的混合检索做准备。 vector_search, bm25_search: 分别执行各自的检索逻辑。 hybrid_search: 这是核心方法它会 分别执行向量检索和 BM25 检索。 对两种检索得到的分数进行归一化确保它们在同一尺度上。 根据预设的 vector_weight 和 bm25_weight 将分数融合。 最后返回综合分数最高的 k 个文档的索引和融合后的分数。 2.3 交叉编码器重排序器 (CrossEncoderReranker)
from sentence_transformers import CrossEncoder # 用于交叉编码器模型
import torch # PyTorchSentence-Transformers 依赖它class CrossEncoderReranker:def __init__(self, model_name: str cross-encoder/ms-marco-MiniLM-L-6-v2):# 初始化交叉编码器模型。这个模型会接收查询和文档对并输出一个相关性分数。self.model CrossEncoder(model_name)# 自动选择 GPU (cuda) 或 CPU 进行计算self.device cuda if torch.cuda.is_available() else cpuself.model.to(self.device) # 将模型加载到指定设备def rerank(self, query: str, documents: List[str], top_k: int 5) - List[Tuple[int, float]]:对文档进行重排序返回 Top-K 结果if not documents:return []# 1. 构建查询-文档对交叉编码器需要查询和每个文档片段的配对作为输入query_doc_pairs [(query, doc) for doc in documents]# 2. 批量计算相关性分数模型会为每个 (查询, 文档) 对输出一个分数# predict 方法会自动处理批量预测并支持 GPU 加速scores self.model.predict(query_doc_pairs, convert_to_numpyTrue)# 3. 排序并返回 Top-K根据分数降序排序并取前 top_k 个scored_docs [(i, float(score)) for i, score in enumerate(scores)]sorted_docs sorted(scored_docs, keylambda x: x[1], reverseTrue)return sorted_docs[:top_k]def rerank_with_threshold(self, query: str, documents: List[str], top_k: int 5, threshold: float 0.1) - List[Tuple[int, float]]:带阈值的重排序过滤低于指定相关性分数的结果# 先对所有文档进行重排序reranked self.rerank(query, documents, len(documents))# 过滤低于阈值的结果filtered [(idx, score) for idx, score in reranked if score threshold]return filtered[:top_k]
代码解析 CrossEncoderReranker: 核心是使用 CrossEncoder 模型。这个模型与 SentenceTransformer 用于向量嵌入的模型不同它是一个交互式模型能够同时处理查询和文档对生成更精确的相关性分数。 rerank: 接收一个查询和一组文档为每个查询-文档对计算相关性分数然后根据分数从高到低排序返回前 top_k 个文档及其分数。 rerank_with_threshold: 增加了阈值过滤功能只有相关性分数达到一定水平的文档才会被保留进一步提升最终结果的质量。 2.4 主控制器 (RetrieveRerankRAG)
from openai import OpenAI
from typing import List, Dict, Anyclass RetrieveRerankRAG:def __init__(self, retriever: HybridRetriever,reranker: CrossEncoderReranker,llm_model: str gpt-3.5-turbo):self.retriever retriever # 混合检索器实例self.reranker reranker # 交叉编码器重排序器实例self.client OpenAI() # OpenAI API 客户端self.llm_model llm_model # 使用的大语言模型def build_index(self, documents: List[str]):构建检索索引交给混合检索器处理print(开始构建检索索引包含向量和BM25...)self.retriever.build_index(documents)print(检索索引构建完成。)def query(self, question: str, retrieval_k: int 50, # 召回阶段的数量rerank_k: int 5, # 精排阶段的数量rerank_threshold: float 0.1) - Dict[str, Any]:处理用户查询执行两阶段检索和生成print(f\n--- 接收到查询: {question} ---)# 第一阶段混合检索召回 Top-K 候选文档print(f执行混合检索召回 {retrieval_k} 个候选文档...)retrieval_results self.retriever.hybrid_search(question, kretrieval_k)if not retrieval_results:print(混合检索未找到相关信息。)return {answer: 未找到相关信息,confidence: 0.0,retrieval_count: 0,rerank_count: 0,final_docs: [],rerank_scores: []}# 获取候选文档的原始文本candidate_docs [self.retriever.documents[idx] for idx, _ in retrieval_results]print(f混合检索召回 {len(retrieval_results)} 个候选文档。)# 第二阶段交叉编码器重排序 Top-N 精选文档print(f执行交叉编码器重排序从 {len(candidate_docs)} 个文档中精选 {rerank_k} 个阈值 {rerank_threshold}...)rerank_results self.reranker.rerank_with_threshold(question, candidate_docs, top_krerank_k, thresholdrerank_threshold)if not rerank_results:print(重排序后未找到高质量匹配文档。)return {answer: 重排序后未找到高质量匹配,confidence: 0.0,retrieval_count: len(retrieval_results),rerank_count: 0,final_docs: [],rerank_scores: []}# 获取最终用于生成答案的文档文本final_docs_with_scores [(candidate_docs[idx], score) for idx, score in rerank_results]final_docs [doc for doc, _ in final_docs_with_scores]print(f重排序后选出 {len(final_docs)} 个高质量文档。)# 构建上下文并添加文档编号以便 LLM 引用context \n\n.join([f文档{i1}: {doc} for i, doc in enumerate(final_docs)])# 生成答案print(正在调用大语言模型生成答案...)answer self._generate_answer(question, context)print(答案生成完成。)# 计算置信度使用重排序的最高分数confidence max(score for _, score in rerank_results) if rerank_results else 0.0return {answer: answer,confidence: confidence,retrieval_count: len(retrieval_results),rerank_count: len(rerank_results),final_docs: final_docs, # 返回最终使用的文档rerank_scores: [score for _, score in rerank_results] # 返回重排序的分数}def _generate_answer(self, question: str, context: str) - str:调用大语言模型生成答案prompt f基于以下文档内容准确回答问题。请确保答案有充分的依据支撑不要提供文档中没有的信息。{context}问题{question}请根据上述文档提供准确、详细的答案try:response self.client.chat.completions.create(modelself.llm_model,messages[{role: system, content: 你是一个专业的问答助手基于提供的文档内容给出准确、有依据的回答。},{role: user, content: prompt}],temperature0.1, # 较低的温度使模型回答更确定和忠实于原文max_tokens1000)return response.choices[0].message.contentexcept Exception as e:print(f调用大语言模型出错: {e})return 生成答案失败请稍后再试。
代码解析 RetrieveRerankRAG: 这个类将 HybridRetriever 和 CrossEncoderReranker 集成起来。 build_index: 委托给 HybridRetriever 来构建混合索引。 query: 这是整个系统的主入口。它首先调用 retriever 进行第一阶段的广泛召回然后将召回的候选文档交给 reranker 进行第二阶段的精确重排序。最后将重排序后精选出的高质量文档作为上下文传递给 LLM 生成答案。 参数 retrieval_k 和 rerank_k 分别控制召回阶段和重排序阶段返回的文档数量。rerank_threshold 用于过滤低相关性的文档。 3. 配置管理让你的系统灵活多变 为了让系统更灵活我们可以使用配置类来管理各种参数。
from dataclasses import dataclassdataclass
class RetrieveRerankConfig:# 检索配置embedding_model: str all-MiniLM-L6-v2 # 用于向量嵌入的模型retrieval_k: int 50 # 混合检索召回的文档数量vector_weight: float 0.7 # 向量检索在混合检索中的权重bm25_weight: float 0.3 # BM25 检索在混合检索中的权重# 重排序配置reranker_model: str cross-encoder/ms-marco-MiniLM-L-6-v2 # 交叉编码器模型rerank_k: int 5 # 重排序后选择的最终文档数量rerank_threshold: float 0.1 # 重排序分数阈值低于此分数的文档将被过滤# 生成配置llm_model: str gpt-3.5-turbo # 使用的大语言模型temperature: float 0.1 # LLM 生成的随机性max_tokens: int 1000 # LLM 生成的最大 token 数classmethoddef high_precision(cls):预设高精度配置召回和精排数量更多阈值更高可能使用更大的重排序模型return cls(retrieval_k100,rerank_k10,rerank_threshold0.2,reranker_modelcross-encoder/ms-marco-MiniLM-L-12-v2 # 示例可能需要下载更大的模型)classmethoddef fast_mode(cls):预设快速模式配置召回和精排数量更少阈值更低追求速度return cls(retrieval_k20,rerank_k3,rerank_threshold0.05)
代码解析 dataclass: Python 的数据类让我们可以简洁地定义只包含数据参数的类。 embedding_model, reranker_model, llm_model: 这些参数让你能够轻松切换不同的模型以适应不同的需求和性能要求。 retrieval_k, rerank_k, rerank_threshold: 这些是控制检索和重排序精度的关键参数。 high_precision 和 fast_mode 类方法提供了两种预设的配置方便用户根据场景快速切换。 4. 使用示例运行你的第一个 Retrieve-and-Rerank RAG import os
# 导入我们刚刚编写的类
from retrieve_rerank_rag_tutorial import RetrieveRerankConfig, HybridRetriever, CrossEncoderReranker, RetrieveRerankRAG# 设置 OpenAI API Key (请替换为你的真实 Key)
# os.environ[OPENAI_API_KEY] YOUR_OPENAI_API_KEY# 创建示例文档以替代真实文件加载
documents [Python是一种高级编程语言广泛用于数据科学、机器学习和Web开发。它的语法简洁易于学习。,机器学习是人工智能的一个重要分支它研究如何让计算机从数据中学习而无需明确编程。,深度学习是机器学习的一个子领域它使用多层神经网络来解决复杂问题如图像识别和自然语言处理。,BM25是一种基于词频的文本检索算法常用于信息检索领域它的全称是Okapi BM25。,Faiss是Facebook AI Research开发的一个用于高效相似性搜索和聚类的库特别适用于处理大规模的向量数据。,Sentence-Transformers是一个Python库可以方便地计算句子、段落和图像嵌入常用于语义相似性搜索和聚类。,交叉编码器模型在问答系统和信息检索中用于重排序搜索结果它能更精确地评估查询和文档之间的相关性。
]# 初始化组件使用默认配置
config RetrieveRerankConfig()# 根据配置初始化检索器和重排序器
retriever HybridRetriever(config.embedding_model)
reranker CrossEncoderReranker(config.reranker_model)# 创建 Retrieve-and-Rerank RAG 系统
rag RetrieveRerankRAG(retriever, reranker, config.llm_model)# 构建索引
rag.build_index(documents)# 查询示例 1
print(\n 查询 1 )
result1 rag.query(什么是机器学习,retrieval_kconfig.retrieval_k,rerank_kconfig.rerank_k,rerank_thresholdconfig.rerank_threshold
)
print(f答案: {result1[answer]})
print(f置信度: {result1[confidence]:.3f})
print(f召回文档数: {result1[retrieval_count]})
print(f重排序后文档数: {result1[rerank_count]})
print(最终使用的文档)
for i, doc in enumerate(result1[final_docs]):print(f 文档 {i1}: {doc})
print(f重排序分数: {[f{score:.3f} for score in result1[rerank_scores]]})# 查询示例 2
print(\n 查询 2 )
result2 rag.query(BM25和Faiss分别是什么它们有什么用,retrieval_kconfig.retrieval_k,rerank_kconfig.rerank_k,rerank_thresholdconfig.rerank_threshold
)
print(f答案: {result2[answer]})
print(f置信度: {result2[confidence]:.3f})
print(f召回文档数: {result2[retrieval_count]})
print(f重排序后文档数: {result2[rerank_count]})
print(最终使用的文档)
for i, doc in enumerate(result2[final_docs]):print(f 文档 {i1}: {doc})
print(f重排序分数: {[f{score:.3f} for score in result2[rerank_scores]]})# 使用高精度配置进行查询
print(\n 使用高精度配置进行查询 )
high_precision_config RetrieveRerankConfig.high_precision()
high_precision_retriever HybridRetriever(high_precision_config.embedding_model)
high_precision_reranker CrossEncoderReranker(high_precision_config.reranker_model)
# 注意这里为了简化示例我们直接重新构建了一个RAG实例。
# 在实际应用中你可能需要考虑如何高效地切换配置或维护多个RAG实例。
high_precision_rag RetrieveRerankRAG(high_precision_retriever, high_precision_reranker, high_precision_config.llm_model)
high_precision_rag.build_index(documents) # 重新构建索引以适应可能变更的嵌入模型result3 high_precision_rag.query(Python 和深度学习有什么关系,retrieval_khigh_precision_config.retrieval_k,rerank_khigh_precision_config.rerank_k,rerank_thresholdhigh_precision_config.rerank_threshold
)
print(f答案: {result3[answer]})
print(f置信度: {result3[confidence]:.3f})
print(f召回文档数: {result3[retrieval_count]})
print(f重排序后文档数: {result3[rerank_count]})
print(最终使用的文档)
for i, doc in enumerate(result3[final_docs]):print(f 文档 {i1}: {doc})
print(f重排序分数: {[f{score:.3f} for score in result3[rerank_scores]]})运行前准备 安装必要的库: pip install faiss-cpu numpy openai sentence-transformers rank_bm25 torchrank_bm25: 用于 BM25 关键词检索。 torch: PyTorch 库CrossEncoder 依赖它。 设置 OpenAI API Key: 确保你的 OPENAI_API_KEY 环境变量已配置。 5. 性能优化让你的 RAG 系统更快更稳 为了应对更高的并发量和响应要求我们可以对重排序阶段进行优化。 5.1 批量重排序 (BatchReranker) 在实际应用中我们可能需要同时处理多个用户的查询或者一个查询可能需要重排序大量文档。批量重排序能显著提高效率。 from typing import List, Tuple
# 继承 CrossEncoderReranker复用其初始化逻辑
class BatchReranker(CrossEncoderReranker): def batch_rerank(self, queries: List[str], documents_list: List[List[str]], top_k: int 5) - List[List[Tuple[int, float]]]:批量重排序多个查询。queries: 多个查询字符串列表。documents_list: 每个查询对应的候选文档列表的列表。results []all_pairs [] # 存储所有 (查询, 文档) 对# pair_indices 记录每个查询对应的 (all_pairs 开始索引, 结束索引)pair_indices [] for q_idx, (query, docs) in enumerate(zip(queries, documents_list)):start_idx len(all_pairs)for doc in docs:all_pairs.append((query, doc))pair_indices.append((start_idx, len(all_pairs)))# 批量计算所有 (查询, 文档) 对的分数这是性能优化的关键print(f批量预测 {len(all_pairs)} 个查询-文档对...)all_scores self.model.predict(all_pairs, convert_to_numpyTrue)print(批量预测完成。)# 根据 pair_indices 分组并排序结果for q_idx, (start, end) in enumerate(pair_indices):query_scores all_scores[start:end]scored_docs [(i, float(score)) for i, score in enumerate(query_scores)]sorted_docs sorted(scored_docs, keylambda x: x[1], reverseTrue)results.append(sorted_docs[:top_k]) # 返回每个查询的 Top-Kreturn results5.2 缓存机制 (CachedReranker) 对于重复出现的查询和文档组合我们可以将重排序的结果缓存起来。下次遇到相同的组合时直接从缓存中读取避免重复计算大大减少延迟。 import hashlib
from typing import Optional# 继承 CrossEncoderReranker复用其核心功能
class CachedReranker(CrossEncoderReranker):def __init__(self, model_name: str, cache_size: int 1000):super().__init__(model_name)self.cache {} # 存储缓存结果self.cache_size cache_size # 缓存最大容量def _get_cache_key(self, query: str, documents: List[str]) - str:生成唯一的缓存键由查询和文档内容哈希得到# 注意文档列表的顺序会影响哈希值确保文档顺序一致content query || ||.join(sorted(documents)) # 排序文档以确保一致性return hashlib.md5(content.encode()).hexdigest()def rerank(self, query: str, documents: List[str], top_k: int 5) - List[Tuple[int, float]]:带缓存的重排序功能cache_key self._get_cache_key(query, documents)if cache_key in self.cache:# print(fCache hit for query: {query})cached_result self.cache[cache_key]return cached_result[:top_k]# 如果缓存中没有则计算重排序结果# 注意这里调用的是父类的 rerank 方法避免死循环result super().rerank(query, documents, len(documents)) # 计算所有结果# 缓存结果并进行简单的 LRU (最近最少使用) 淘汰策略if len(self.cache) self.cache_size:# 删除最旧的缓存项oldest_key next(iter(self.cache)) del self.cache[oldest_key]self.cache[cache_key] resultreturn result[:top_k] 6. 评估指标衡量你的 RAG 系统有多好 光实现还不够我们还需要一套方法来评估 RAG 系统的效果。
# 评估指标辅助函数 (简化版实际应用中会更复杂)
def calculate_precision(retrieved: List[str], relevant: List[str], k: int 5) - float:计算 PrecisionK检索结果中相关文档的比例retrieved_k retrieved[:k]if not retrieved_k:return 0.0num_relevant_in_retrieved sum(1 for doc in retrieved_k if doc in relevant)return num_relevant_in_retrieved / len(retrieved_k)def calculate_recall(retrieved: List[str], relevant: List[str], k: int 5) - float:计算 RecallK所有相关文档中被检索到的比例if not relevant:return 1.0 # 如果没有相关文档召回率视为完美retrieved_k retrieved[:k]num_relevant_in_retrieved sum(1 for doc in retrieved_k if doc in relevant)return num_relevant_in_retrieved / len(relevant)def calculate_ndcg(retrieved: List[str], relevant: List[str], k: int 5) - float:计算 NDCGK (Normalized Discounted Cumulative Gain)考虑排名的相关性dcg 0.0idcg 0.0# 理想情况下的 DCG (所有相关文档排在前面)for i in range(min(k, len(relevant))):idcg 1 / np.log2(i 2) # 相关性分数假设为1# 实际情况下的 DCGfor i, doc in enumerate(retrieved[:k]):if doc in relevant:dcg 1 / np.log2(i 2)return dcg / idcg if idcg 0 else 0.0def calculate_reciprocal_rank(retrieved: List[str], relevant: List[str]) - float:计算 Reciprocal Rank (MRR的一部分)第一个相关文档的排名倒数for i, doc in enumerate(retrieved):if doc in relevant:return 1.0 / (i 1)return 0.0 # 如果没有相关文档被检索到def evaluate_retrieval_quality(rag_system: RetrieveRerankRAG, test_queries: List[str], ground_truth: List[List[str]]) - Dict[str, float]:评估整个 RAG 系统的检索质量metrics {precision_at_5: 0.0,recall_at_5: 0.0,ndcg_at_5: 0.0,mrr: 0.0}total_queries len(test_queries)for i, query in enumerate(test_queries):relevant_docs ground_truth[i]# 调用 RAG 系统进行查询获取最终精排的文档result rag_system.query(query, rerank_k5) # 评估 K5 的情况retrieved_docs result.get(final_docs, [])# 计算各项指标并累加metrics[precision_at_5] calculate_precision(retrieved_docs, relevant_docs, k5)metrics[recall_at_5] calculate_recall(retrieved_docs, relevant_docs, k5)metrics[ndcg_at_5] calculate_ndcg(retrieved_docs, relevant_docs, k5)metrics[mrr] calculate_reciprocal_rank(retrieved_docs, relevant_docs)# 平均化指标for key in metrics:metrics[key] / total_queriesreturn metrics
代码解析 precision_at_K, recall_at_K: 经典的查准率和查全率衡量检索到的文档中有多少是相关的以及所有相关文档中有多少被检索到。 NDCGK: 考虑了文档的相关性分级和排名位置高质量的文档排在前面会获得更高的分数。 MRR (Mean Reciprocal Rank): 衡量第一个相关文档出现的位置排名越靠前越好。
这些指标通常需要人工标注的“测试查询集”和“真实相关文档”来计算。 7. 部署建议将你的 RAG 系统投入生产 7.1 生产环境配置 (production.yaml) 为了在生产环境中稳定运行通常会采用更强大的模型和更优化的参数。
# production.yaml
retrieve_rerank:retrieval:embedding_model: all-MiniLM-L6-v2 # 可以根据需求升级为更大的模型如 BGE-large-en-v1.5retrieval_k: 100 # 召回更多的候选文档vector_weight: 0.6bm25_weight: 0.4reranking:model: cross-encoder/ms-marco-MiniLM-L-12-v2 # 使用更大的交叉编码器模型效果通常更好batch_size: 32 # 批量处理大小优化推理速度top_k: 8 # 最终精选的文档数量threshold: 0.15 # 更严格的重排序阈值cache_size: 5000 # 缓存大小generation:model: gpt-4 # 使用更强大的 LLM如 GPT-4 或 Claude Opustemperature: 0.05 # 更低的温度让答案更严谨、更忠实于文档max_tokens: 1500 # 增加最大生成长度适应更复杂的回答7.2 Docker 部署 使用 Docker 可以将你的应用及其所有依赖项打包成一个独立的、可移植的容器方便部署到任何支持 Docker 的环境中。
# 使用官方 Python 3.9 的轻量级镜像作为基础
FROM python:3.9-slim# 设置工作目录
WORKDIR /app# 复制 requirements.txt 到容器中并安装依赖
# 注意这里需要一个 requirements.txt 文件包含所有依赖例如
# faiss-cpu
# numpy
# openai
# sentence-transformers
# rank_bm25
# torch
# flask
# pyyaml
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt# 预下载模型在构建镜像时下载模型避免运行时下载导致首次启动慢
# 确保这里列出的模型与你配置中使用的模型一致
RUN python -c from sentence_transformers import SentenceTransformer; SentenceTransformer(all-MiniLM-L6-v2)
RUN python -c from sentence_transformers import CrossEncoder; CrossEncoder(cross-encoder/ms-marco-MiniLM-L-6-v2)
# 如果使用了 L-12-v2也需要在这里预下载
RUN python -c from sentence_transformers import CrossEncoder; CrossEncoder(cross-encoder/ms-marco-MiniLM-L-12-v2)# 复制你的应用代码到容器中
COPY . /app# 暴露服务端口
EXPOSE 8000# 定义容器启动时执行的命令
# 假设你的主应用文件是 app.py或者如示例中的 deploy.py
CMD [python, deploy.py] 部署步骤 创建 requirements.txt: 包含所有 Python 依赖库。 创建 Dockerfile: 如上所示。 构建 Docker 镜像: 在项目根目录Dockerfile 所在目录下执行 docker build -t retrieve-rerank-rag . 运行 Docker 容器: docker run -p 8000:8000 retrieve-rerank-rag
这样你的 Retrieve-and-Rerank RAG 服务就会在容器中运行并通过 8000 端口对外提供服务。 8. 性能基准 (Performance Benchmark) 通过引入重排序阶段Retrieve-and-Rerank RAG 在检索质量上相比 Naive RAG 有显著提升但通常也会带来额外的延迟。 指标 Naive RAG Retrieve-Rerank Precision5 0.68 0.82 Recall5 0.71 0.86 NDCG5 0.74 0.89 平均延迟 120ms 280ms
分析 精度提升: Precision, Recall, NDCG 等指标都有 15-20% 的大幅提升这表明重排序阶段有效地筛选出了更相关的文档为 LLM 提供了更优质的上下文。 延迟增加: 重排序模型特别是交叉编码器的计算开销较大导致平均响应延迟有所增加。这是精度提升的代价需要在实际应用中权衡。 9. 总结Retrieve-and-Rerank RAG 的优势与适用场景
Retrieve-and-Rerank RAG 通过引入两阶段检索广泛召回 精确重排序的架构显著解决了 Naive RAG 在检索质量上的局限性。
核心优势 检索精度大幅提升交叉编码器能够更准确地理解查询和文档之间的复杂关系筛选出高质量的上下文。 支持混合检索策略结合向量和关键词检索能够兼顾语义匹配和精确匹配提高召回的全面性。 可解释的重排序过程重排序分数提供了文档相关性的量化依据。 模块化设计便于优化各组件独立便于针对性地进行优化和替换。
适用场景 对准确性要求较高的问答系统例如企业内部知识库、客服机器人需要确保答案的精准度。 法律、医疗等专业领域这些领域对信息准确性和可靠性有极高要求重排序能有效过滤无关信息。 企业知识库检索面对大量异构文档时能更有效地找到所需信息。
注意事项 延迟增加重排序阶段会引入额外的计算开销导致整体响应时间变长需要根据业务需求进行权衡和优化。 需要更多计算资源交叉编码器模型通常比简单的嵌入模型更大需要更多的内存和计算能力尤其是在 GPU 上运行。 重排序模型选择很关键选择一个高质量的、与你的数据领域匹配的交叉编码器模型至关重要。 预告RAG 进阶之旅未完待续...
我们已经从最基础的 Naive RAG 迈向了更强大的 Retrieve-and-Rerank RAG。然而RAG 的潜力远不止于此在接下来的教程中我们将继续探索更高级、更智能的 RAG 架构 Multimodal RAG (多模态 RAG)如何让 RAG 不仅能处理文本还能理解图片、音频等多种信息 Graph RAG (图 RAG)如何利用知识图谱的强大推理能力让 RAG 能够进行复杂的多跳推理回答更深层次的问题 Hybrid RAG (混合 RAG)如何更智能地融合多种检索策略并管理来自不同数据源的信息 Agentic RAG Router (智能体路由)如何引入 LLM 驱动的智能代理让 RAG 系统能根据用户意图动态选择最佳的工具或流程 Agentic RAG Multi-Agent (多智能体)如何让多个专业的智能体协作共同解决极其复杂的问题并通过“辩论”达成共识 持续学习才能玩转AI
这篇RAG入门教程助你启程。想获取 最新AI架构趋势深度解读 RAG、MCP及LLM应用实战教程与代码 精选学习资源与高效工具包 技术答疑与同行交流 欢迎关注 【AI架构笔记】 扫码 / 搜一搜AI架构笔记一起进阶驾驭AI
