大模型核心技术RAG讲解

一.RAG技术原理

1.什么是RAG？

Retrieval-Augmented Generation (RAG-检索增强生成) 是帮助大模型更好地回答问题的一种方式,在大模型的应用场景中，一个非常常见的需求是：手里有一堆文档，当遇到一个问题时，希望通过其中的一段文本来找到答案,这时，RAG（检索增强生成）技术就能派上用场,它的作用就是从这些文档中找到最相关的信息，然后利用大模型生成精准的回答,简单来说，RAG就是帮助大模型‘找到’并‘利用’最相关的信息来回答你的问题，就像一个聪明的助手快速翻阅资料找到最合适的答案,如下图所示：

能这样实现的原因在于已经有非常多的实验论文能够证明:当为大模型提供一定的上下文信息后，其输出会变得更稳定。那么，将知识库中的信息或掌握的信息先输送给大模型，再由大模型服务用户，就是大家普遍达成共识的一个结论和方法。传统的对话系统、搜索引擎等核心依赖于检索技术，如果将这一检索过程融入大模型应用的构建中，既可以充分利用大模型在内容生成上的能力，也能通过引入的上下文信息显著约束大模型的输出范围和结果，同时还实现了将私有数据融入大模型中的目的，达到了双赢的效果。

从技术实现细节上看,RAG的实现是包括两个阶段：检索阶段和生成阶段,在检索阶段，从知识库中找出与问题最相关的知识，为后续的答案生成提供素材,在生成阶段，RAG会将检索到的知识内容作为输入，与问题一起输入到语言模型中进行生成,这样，生成的答案不仅考虑了问题的语义信息，还考虑了相关私有数据的内容,如下图所示：

为什么要搞得这个复杂呢？还是在于大模型本身的以下两个特性：

• 1.如果用户提出的问题，其对应的答案出现在一篇文章中，去知识库中找到一篇与用户输入相关的文章是很容易的，但是将检索到的这整篇文章直接放入Prompt中并不是最优的选择，因为其中一定会包含非常多无关的信息，而无效信息越多，对大模型后续的推理影响越大
• 2.任何一个大模型都存在最大输入的Token限制，如果将一整个文本去全部传入大模型，无法容纳如此多的信息

而RAG构建的关键，则是由检索组件（通常由Embedding模型和向量数据库组成）和生成组件（大模型）组成, 在推理时，用户查询用于对索引文档运行相似性搜索，以检索与查询最相似的文档，并为大模型提供额外的上下文,因此，在构建RAG的过程中，也往往是遵循着这种流程去做定制化的开发,那么每个环节中应该掌握哪些知识呢？来看下面的这个思维导图：

[Embedding 模型选择]在Hugging Face上有一个榜单(MTEB-Massive Text Embedding Benchmark),是目前最权威的嵌入模型评估基准之一，目前该榜单主要分为新版和旧版，可以根据需要访问以下地址：

• 新版 MTEB 排行榜（支持更多自定义筛选）：
  https://huggingface.co/spaces/mteb/leaderboard
• 旧版 MTEB 排行榜（包含中文等语言筛选）：
  https://huggingface.co/spaces/mteb/leaderboard_legacy

查看榜单的实用建议：

• 1.按需筛选：MTEB 总分是所有子任务的均分，但如果是做 RAG 应用，建议重点查看Retrieval（检索）子项的得分，而不是盲目追求总分第一
• 2.多维度过滤：在新版榜单中，可以结合语言(如中文)、任务类型（如 BEIR 检索）、模型可用性（开源或闭源）以及是否经过指令微调（Instruction-tuned）来进行精准筛选
• 3.结合实际：榜单分数存在微小差距或过拟合可能，建议结合模型的参数量大小、上下文长度限制以及实际的业务场景(如是否需要处理长文本、特定专业领域等)来综合选择

在实现RAG系统时，主要有三种方式：手动实现、框架实现和手动+框架结合的方式,通常，企业中不会选择完全手动实现，因为这样不仅工作量大，而且维护复杂,而手动+框架结合的方式，因其灵活性和高效性，会是90%以上开发者的首选

2.RAG技术必要性介绍

大型语言模型（LLM）在生成语言时，可能会产生幻觉或不准确的结果,这些问题包括：

• 信息误导：模型生成的内容可能基于不准确或过时的信息
• 知识更新滞后：模型无法访问最新的信息，尤其是在专业领域
• 推理能力不足：LLM 在面对复杂推理任务时，难以提供高效且准确的结果

为了解决这些问题，检索增强生成（Retrieval-Augmented Generation, RAG）应运而生,RAG 通过先从数据库中检索与问题相关的信息，再基于检索到的内容进行回答生成，极大地提升了模型输出的准确性和相关性,RAG 不仅提高了知识更新的效率，还显著增强了生成内容的可追溯性，使其在实际应用中更具实用性和可信度

3.一个RAG系统的核心组件说明

为了实现 RAG 模型，需要以下几个核心模块：

• (1).向量化模块：用于将文档片段转换为向量表示，以便后续检索
• (2).文档加载与切分模块：负责加载文档并将其切分为若干易于处理的文档片段
• (3).数据库模块：用于存储文档片段及其对应的向量表示
• (4).检索模块：根据用户输入的查询，检索与其相关的文档片段
• (5).生成模块：调用语言模型生成基于检索信息的回答

基本构成如下图所示：

而这些组件，也是接下来手动构建一个RAG系统的基础

据此也可以看出一个RAG系统的基本流程：

• 索引：将文档库分割成较短的 Chunk，并通过编码器构建向量索引
• 检索：根据问题和 chunks 的相似度检索相关文档片段
• 生成：以检索到的上下文为条件，生成问题的回答

这些模块共同构成了一个简易 RAG 的核心架构,每个模块在模型的不同阶段发挥作用，确保查询的处理从检索到生成都有条不紊地进行

二.文本向量化模块创建

正如此前所说，在自然语言处理和机器学习领域中，Embedding 是将文本转化为数值向量的常用方法,通过这种方式，模型可以衡量不同文本之间的相似性,进而应用于如搜索、分类、推荐等多个领域,OpenAI 发布的其第三代 Embedding 模型，这些模型具有更高的性能、更低的成本,且在多语言处理上表现出色,以下是关于OpenAI Embedding模型的详细介绍和调用方法

1.OpenAI Embedding模型简介

Embedding是将文本字符串表示为向量（浮点数列表），通过计算向量之间的距离来衡量文本之间的相关性.向量距离越小，表示文本之间的相关性越高；距离越大，相关性越低,常见的 Embedding 应用包括：

• 搜索：根据文本查询的相关性对结果进行排序
• 聚类：根据文本相似性将其分组
• 推荐：根据相关文本字符串推荐项目
• 异常检测：识别与其他内容相关性较低的异常点
• 多样性测量：分析相似性分布
• 分类：将文本字符串根据其最相似的标签进行分类

OpenAI 最新的 Embedding 模型包括text-embedding-3-small和text-embedding-3-large，它们比以往的模型具有更高的性能，且支持更多语言,这两个模型分别生成长度为1536 和 3072的向量,此外，用户可以通过设置维度参数来减少向量的维度，而不损失其表示概念的能力

2.OpenAI Embedding模型获取

要获取文本的 Embedding 向量，可以将文本字符串发送到 OpenAI 的 Embedding API 端点，并指定所使用的模型（例如text-embedding-3-small）.响应结果将包含 Embedding 向量，以及一些额外的元数据信息

截止目前，OpenAI 提供了两个第三代 Embedding 模型，定价基于输入的 token 数量。以下是关于模型性能与定价的概览：

Embedding 向量可以直接用于多种应用场景，例如存储在向量数据库中，进行文本相似度搜索等,默认情况下，text-embedding-3-small生成的向量长度为 1536，text-embedding-3-large的向量长度为 3072

• 余弦相似度介绍与效果简介
• 欧氏距离与余弦相似度计算公式

假设现有a、b两个向量：

例如，余弦相似度可以通过如下方式进行计算和呈现：

import matplotlib.pyplot as plt import numpy as np # 创建两个向量 a = np.array([0, 1]) b = np.array([1, 1]) # 计算两个向量的余弦相似度 cosine_similarity = np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)) # 绘制向量 plt.quiver(0, 0, a[0], a[1], angles='xy', scale_units='xy', scale=1, color='r') plt.quiver(0, 0, b[0], b[1], angles='xy', scale_units='xy', scale=1, color='b') # 设置图表属性 plt.xlim(-0.5, 1.5) plt.ylim(-0.5, 1.5) plt.grid() plt.title(f'Cosine Similarity: {cosine_similarity:.2f}') plt.xlabel('X axis') plt.ylabel('Y axis') # 添加图例 plt.legend(['Vector a', 'Vector b']) # 显示图表 plt.show()

这幅图展示了二维空间中两个向量的方向:红色向量代表\(a→\) ，蓝色向量代表 \(b→\),它们之间的夹角表示了两个向量的余弦相似度,余弦相似度是通过计算两个向量的点积并除以它们各自的范数（即长度）来得到的,在这个示例中，这两个向量的余弦相似度大约为 0.71，意味着它们在方向上有一定程度的相似性。这个值越接近 1，表示两个向量的方向越相似

为了实现 RAG 模型的功能，首先需要一个向量化（Embedding）模块,向量化是 RAG 的基础，它的作用是将文档片段转化为向量表示，便于后续的检索操作,在这个过程中，将实现一个向量化类，用来将文本片段映射成向量

为了便于扩展和未来可能使用不同的模型，首先编写一个Embedding 基类,该基类定义了获取文本向量表示的方法，同时包含一个计算两个向量之间余弦相似度的功能,这样，如果未来使用不同的向量化模型，只需继承该基类并重写向量获取的逻辑，而不需要重复编写相似度计算部分

class BaseEmbeddings: """ 向量化的基类，用于将文本转换为向量表示。不同的子类可以实现不同的向量获取方法。 """ def __init__(self, path: str, is_api: bool) -> None: """ 初始化基类。 参数： path (str) - 如果是本地模型，path 表示模型路径；如果是API模式，path可以为空 is_api (bool) - 表示是否使用API调用，如果为True表示通过API获取Embedding """ self.path = path self.is_api = is_api def get_embedding(self, text: str, model: str) -> List[float]: """ 抽象方法，用于获取文本的向量表示，具体实现需要在子类中定义。 参数： text (str) - 需要转换为向量的文本 model (str) - 所使用的模型名称 返回： list[float] - 文本的向量表示 """ raise NotImplementedError @classmethod def cosine_similarity(cls, vector1: List[float], vector2: List[float]) -> float: """ 计算两个向量之间的余弦相似度，用于衡量它们的相似程度。 参数： vector1 (list[float]) - 第一个向量 vector2 (list[float]) - 第二个向量 返回： float - 余弦相似度值，范围从 -1 到 1，越接近 1 表示向量越相似 """ dot_product = np.dot(vector1, vector2) # 向量点积 magnitude = np.linalg.norm(vector1) * np.linalg.norm(vector2) # 向量的模 if not magnitude: return 0 return dot_product / magnitude # 计算余弦相似度

在这个基类基础上，可以通过继承它来实现具体的模型,例如，可以使用 OpenAI 的 API 来生成文本的向量表示，只需重写get_embedding方法即可

class OpenAIEmbedding(BaseEmbeddings): """ 使用 OpenAI 的 Embedding API 来获取文本向量的类，继承自 BaseEmbeddings。 """ def __init__(self, path: str = '', is_api: bool = True) -> None: """ 初始化类，设置 OpenAI API 客户端，如果使用的是 API 调用。 参数： path (str) - 本地模型的路径，使用API时可以为空 is_api (bool) - 是否通过 API 获取 Embedding，默认为 True """ super().__init__(path, is_api) if self.is_api: # 初始化 OpenAI API 客户端 from openai import OpenAI self.client = OpenAI() self.client.api_key = os.getenv("OPENAI_API_KEY") # 从环境变量中获取 API 密钥 self.client.base_url = os.getenv("OPENAI_BASE_URL") # 从环境变量中获取 API 基础URL def get_embedding(self, text: str, model: str = "text-embedding-3-large") -> List[float]: """ 使用 OpenAI 的 Embedding API 获取文本的向量表示。 参数： text (str) - 需要转化为向量的文本 model (str) - 使用的 Embedding 模型名称，默认为 'text-embedding-3-large' 返回： list[float] - 文本的向量表示 """ if self.is_api: # 去掉文本中的换行符，保证输入格式规范 text = text.replace("\n", " ") # 调用 OpenAI API 获取文本的向量表示 return self.client.embeddings.create(input=[text], model=model).data[0].embedding else: raise NotImplementedError # 如果不是 API 模式，这里未实现本地模型的处理

这样设计的结构让我们可以轻松替换或扩展向量化模型，而不需要改变整体框架

三.文档加载与切分模块创建

在实现了向量化之后，接下来需要编写一个文档加载与切分模块，用于处理不同格式的文档并将其切分为小片段,为什么要进行切分呢？这是为了确保每个文档片段都尽量保持简短且信息集中，以便于后续的向量化和检索

1.文档格式处理函数

目标是支持多种格式的文档，例如 PDF、Markdown、TXT 等,每种文件格式都有不同的读取方式，下面展示一个支持多种格式的简单实现:

def read_file_content(cls, file_path: str): # 根据文件扩展名选择读取方法 if file_path.endswith('.pdf'): return cls.read_pdf(file_path) elif file_path.endswith('.md'): return cls.read_markdown(file_path) elif file_path.endswith('.txt'): return cls.read_text(file_path) else: raise ValueError("Unsupported file type")

2.文档切分函数

这里考虑将文档按Token长度进行切分，设置一个最大的 Token 长度，然后按这个长度进行切分,在这个过程中，也会确保每个片段之间有一定的重叠，避免重要信息被切掉

def get_chunk(cls, text: str, max_token_len: int = 600, cover_content: int = 150): chunk_text = [] curr_len = 0 curr_chunk = '' lines = text.split('\n') # 以换行符为单位切分文本 for line in lines: line = line.replace(' ', '') line_len = len(enc.encode(line)) # 计算当前行的 Token 长度 if line_len > max_token_len: print('warning line_len = ', line_len) if curr_len + line_len <= max_token_len: curr_chunk += line + '\n' curr_len += line_len + 1 else: chunk_text.append(curr_chunk) curr_chunk = curr_chunk[-cover_content:] + line curr_len = line_len + cover_content if curr_chunk: chunk_text.append(curr_chunk) return chunk_text

四.词向量数据库与向量检索模块

接下来将继续构建向量数据库以及检索模块，这是RAG 模型中的核心功能之一,向量数据库用于存储文档片段及其对应的向量表示，而检索模块则根据用户提出的问题（Query）在数据库中检索相关的文档片段,通过这些功能，创建的简易 RAG 能够根据输入的查询快速找到最相关的文档片段

为了构建这个向量数据库，需要以下几个关键功能：

• 持久化存储（persist）：将数据库存储到本地，便于下次加载使用
• 加载数据库（load_vector）：从本地文件加载已经存储的向量和文档
• 获取向量表示（get_vector）：将文档转化为向量表示并存储
• 检索（query）：根据用户的 Query，检索数据库中的相关文档片段

五.大模型问答模块编写

现在已经构建了向量数据库和检索模块，接下来将实现大模型模块，用于根据检索到的相关文档片段生成对用户问题的回答,为了简化和便于扩展，会先实现一个基类BaseModel，然后再以GPT4o模型为例，展示如何使用大语言模型来完成这个任务

1.大模型模块的基类

这里先编写一个基类BaseModel，它包含两个主要方法：

• chat：负责处理用户的输入并生成回答
• load_model：如果是使用本地模型，这个方法负责加载模型,如果使用 API 模型（如 OpenAI），可以不用实现这个方法

class BaseModel: """ 基础模型类，作为所有模型的基类。 包含一些通用的接口，如加载模型、生成回答等。 """ def __init__(self, path: str = '') -> None: self.path = path # 用于存储模型文件的路径，默认为空。 def chat(self, prompt: str, history: List[dict], content: str) -> str: """ 使用模型生成回答的抽象方法。 :param prompt: 用户的提问内容 :param history: 之前的对话历史（字典列表） :param content: 提供的上下文内容 :return: 模型生成的答案 """ pass # 具体的实现由子类提供 def load_model(self): """ 加载模型的方法，通常用于本地模型。 """ pass # 如果是 API 模型，可能不需要实现

2.借助GPT4o模型进行对话

class GPT4oChat(BaseModel): """ 基于 GPT-4o 模型的对话类，继承自 BaseModel。 主要用于通过 OpenAI API 来生成对话回答。 """ def __init__(self, api_key: str, base_url: str = "https://ai.devtool.tech/proxy/v1") -> None: """ 初始化 GPT-4o 模型。 :param api_key: OpenAI API 的密钥 :param base_url: 用于访问 OpenAI API 的基础 URL，默认为代理 URL """ super().__init__() self.client = OpenAI(api_key=api_key, base_url=base_url) # 初始化 OpenAI 客户端 def chat(self, prompt: str, history: List = [], content: str = '') -> str: """ 使用 GPT-4o 生成回答。 :param prompt: 用户的提问 :param history: 之前的对话历史（可选） :param content: 可参考的上下文信息（可选） :return: 生成的回答 """ # 构建包含问题和上下文的完整提示 full_prompt = PROMPT_TEMPLATE['GPT4o_PROMPT_TEMPLATE'].format(question=prompt, context=content) # 调用 GPT-4o 模型进行推理 response = self.client.chat.completions.create( model="gpt-4o-mini", # 使用 GPT-4o 小型模型 messages=[ {"role": "user", "content": full_prompt} ] ) # 返回模型生成的第一个回答 return response.choices[0].message.content

3.提示模板

为了方便维护和复用提示语，可以使用一个字典来保存不同模型的提示模板,在这里为GPT4o定义了一个模板：

PROMPT_TEMPLATE = dict( GPT4o_PROMPT_TEMPLATE=""" 下面有一个或许与这个问题相关的参考段落，若你觉得参考段落能和问题相关，则先总结参考段落的内容。 若你觉得参考段落和问题无关，则使用你自己的原始知识来回答用户的问题，并且总是使用中文来进行回答。 问题: {question} 可参考的上下文： ··· {context} ··· 有用的回答:""" )

六.RAG Demo完整流程演示

接下来，展示一个完整的 RAG Demo（物流助手），结合前面实现的向量检索和大模型模块，展示如何在实际应用中使用 RAG 模型来回答问题

# coding:utf-8 # 导入必备的工具包 import time from langchain_core.prompts import PromptTemplate from langchain_ollama import OllamaLLM from langchain_ollama import OllamaEmbeddings from langchain_community.vectorstores import FAISS # 向量数据库 # 加载FAISS向量库 embeddings = OllamaEmbeddings(model="mxbai-embed-large", temperature=0) db = FAISS.load_local("faiss/wuliu", embeddings, allow_dangerous_deserialization=True) start_time = time.time() def get_related_content(related_docs): related_content = [] for doc in related_docs: related_content.append(doc.page_content.replace("\n\n", "\n")) return "\n".join(related_content) def define_prompt(): question = '我的快递出发地是哪？预计几天的时间到达？' docs = db.similarity_search(question, k=2) related_content = get_related_content(docs) PROMPT_TEMPLATE = """ 基于以下已知信息，简洁和专业的来回答用户的问题。不允许在答案中添加编造成分。 已知内容: {context} 问题: {question}""" prompt = PromptTemplate( input_variables=["context", "question"], template=PROMPT_TEMPLATE, ) my_pmt = prompt.format(context=related_content, question=question) return my_pmt def main(): model = OllamaLLM(model="qwen2.5:7b") my_pmt = define_prompt() print(f'>>>提示词模板：{my_pmt}') result = model.invoke(my_pmt) return result if __name__ == '__main__': result = main() print('*' * 80) print(result) print('*' * 80) end_time = time.time() print(f'推理耗时：{end_time - start_time:.4f}s')

RAG相关项目:

• 【AI大模型应用开发】【项目实战】13.RAG智慧问答项目-(一)项目介绍&项目架构&项目环境配置

• 【AI大模型应用开发】【项目实战】8.物流行业信息咨询RAG系统