如何创建高质量的本地知识库增强大模型私域任务处理能力_大模型 知识库

1 引言

1.1 目的和范围

基于RAG的大模型应用作为大模型基础建设的必要课题之一，本方案为需要使用大模型知识库辅助推理模式的人员提供可参考的标准化流程和实现形式，增加大模型的领域能力，加速大模型相关课题的落地。

1.2 RAG概述

大模型改变了我们与信息的交互方式，但对于我们提出的要求，这个交互过程也存在许多限制：

• 受训练阶段和提问的表达方式等影响，大模型不能准确理解用户意图
• 受训练数据和时效性影响，大模型无法回答领域知识，当我们需要了解除它们训练数据以外的具体知识时，往往会达不到要求

对于第一个限制，开源基础模型的理解能力不断提升，例如已开源的qwen-72B刷榜各评测榜单，对使用者的提示能力要求进一步降低，且已有能力已满足多样性任务需求，适合直接部署使用。对于第二个限制，使用检索增强生成技术（RAG，Retrieval Augmented Generation）是目前一种经济可行的方案。

RAG通过大语言模型（LLM)+知识召回（Knowledge Retrieval）的方式， 解决通用大语言模型在专业领域回答缺乏依据、存在幻觉的问题，是各类基于大模型的知识问答应用的常用技术。RAG的知识召回基于各类信息检索算法，主要有以下几种类型：

• 向量检索：主要基于faiss、milvus、weaviate、chroma等向量数据库
• 关键字检索：基于文本检索引擎ElasticSearch、OpenSearch和BM25算法
• 图数据库检索：neo4j、nebula等图数据库
• 关系数据库检索：PostgreSQL、MySQL、Oracle等关系数据库
• 搜索引擎：bing、google、duckduckgo、metaphor

针对领域RAG的应用以向量检索为主，其它检索方式辅助强化，其基本思路是把私域知识文档进行切片然后向量化，后续通过向量检索进行召回文档片段，再作为上下文注入到大语言模型进行归纳总结。另外，使用搜索引擎检索增强生成的方式已经成为目前AI Agent框架的重要组成部分。

1.3 RAG应用框架和流程

RAG包括五个关键阶段，它们将成为构建的任何大型应用程序的一部分。分别为：

• 加载：指从数据存在的地方获取数据到我们的管道中，可以包括文本文件、pdf、代码、数据库信息和 API。
• 索引：创建一个允许查询数据的数据结构。对于大模型来说，是指创建向量嵌入，包含数据语义的数字表示，以及许多其他元数据策略，以便轻松准确地找到与上下文相关的数据。
• 存储：一旦数据被索引，我们将希望存储这个索引，以及任何其他元数据，以避免需要重新索引它。
• 查询：对于任何给定的索引策略，都有许多方法可以利用 llm 数据结构进行查询，包括子查询、多步骤查询和混合策略。
• 评估：任何管道中的一个关键步骤是检查它相对于其他策略的有效性，或者当我们进行更改时。评估提供了客观的衡量标准，以衡量大模型对查询的响应有多准确、可信和快速。

在完成数据源加载、数据预处理、索引构建并存入向量数据库后，能够启动RAG应用执行流程，一个通用的完整流程如下图：

上图中每一步所进行的操作内容：

1. 将构建好的prompt（或选择prompt中的关键查询内容）传递给Embedding模型，将其编码为具有语义信息的查询向量
2. 将查询向量传递到向量数据库，准备进行相似度匹配
3. 通过计算查询向量和向量数据库中的所有块之间的距离（欧氏距离、余弦相似度、点积相似度），选择前k个最相关的块作为模型回答时的参考内容，即所提取的文档内容
4. 将提取的信息和查询的问题组合，构造出完整的prompt，传递给LLM
5. 模型推理过程，使用提供的内容生成回复

1.4 应用场景和目标

2 向量数据库建设

2.1 概述

2.1.1 向量数据库介绍和选型标准

构建向量数据库是在构建RAG应用程序之前需要完成的工作。向量数据库作为数据中心，为上游的索引构建提供存储支撑，为下游的检索算法提供向量数据来源和平台服务。

向量数据库普遍支持各类搜索算法，提供了在大量数据中的高效检索。目前Langchain已经提供了对超过15种向量数据库的支持，下表是常用向量数据库介绍：

向量数据库的选择中，主流方案有以下几种：

• Faiss：特点：1.同时支持cpu和GPU两种设备;2. 支持C++,python, go等客户端;3. 支持常见的索引方式，如IVF,HNSW,支持PQ量化;4. in-memory运行;5. self-hosted。缺点：不能处理大规模数据
• Weaviate：特点：1. 文档丰富，容易上手;2. 提供混合索引;3. 支持自托管+云原生;4.支持python,js,ts,go,java等客户端;5. 支持HNSW,HNSW-PQ,DisANN等索引
• Milvus：特点：1. 通过代理、负载均衡器、消息代理、Kafka和Kubernetes的组合实现了高度可扩展性，这使得整个系统变得非常复杂和资源密集;2. 截至2023年，它是唯一一个提供可工作的DiskANN实现的主要供应商;3. 支持在向量相似度检索过程中进行标量字段过滤，实现混合查询;4. 采用存储与计算分离的架构设计;5. 提供python,juava,go,node.js等语言SDK,也提供milvus lite等in-momery运行;6. 提供了图形界面客户端。 缺点：更新频繁，数据备份时只能同一集群备份，权限控制较差。
• Pinecone：特点：1. 完全云原生，非常容易上手;2. 自建复合索引;3. 支持向量+关键词混合召回;4. 易于集成，灵活可扩展。缺点：收费，只支持云原生。

2.1.2 构建流程

向量数据库是储存经Embedding模型向量化后的文本/图像/音视频数据索引，从数据源转换到可储存向量的过程包括下图所示阶段：

2.2 数据工程

2.2.1 数据提取

数据提取是RAG数据库建设的第一步，其核心目的是从各种数据源中获取相关信息。数据提取过程需要保证信息的完整性和时效性，以下是各指标的具体描述

完整性：完整性指的是在数据提取过程中确保收集的信息全面，覆盖所有目标任务知识领域及话题。

时效性：时效性是指确保数据的更新，特别是对于快速变化的信息。

在数据提取阶段，主要任务包括：

1. 数据源识别：确定哪些数据源包含目标任务所依赖的信息。这些数据源可能包括iCenter空间、UDM文档、本地文件等。
2. 数据提取策略：设计高效且可靠的方法来提取这些数据。这可能包括使用爬虫技术从网站提取数据，或者使用APIs从数据库中提取数据。
3. 初步格式化：提取的数据需要被转换成一种适合进一步处理的格式（如分块、向量化等），例如markdown或JSON文件。
4. 初步筛选：在提取过程中，进行初步的数据筛选，排除明显无关或质量过低的数据。
5. 定期更新：对于时间敏感、频繁变化的数据，需要进行识别并定期更新。

数据提取工作主要涉及源文档的获取以及对源文档进行解析。源文档通过接口调用、脚本采集或爬虫的方式获取。下面以5G需求体系中功能体系为例获取和加载icenter数据：

class UrlAPI:
    def __init__(self, account="", token=""):
        self.content_url = ""
        self.sub_tree_url = ""
        self.uac_server = ""
        self.history_url = ""
        self.history_detail_url = ""
        self.get_comments_url = ""
        self.add_comments_url = ""
        self.modify_comment_url = ""
        self.delete_comment_url = ""
        self.template_url = ""
        self.all_tree_url = ""
        self.authority = ""
        self.put_authority = ""
        self.X_Emp_No = '' #账号
        self.password = '' # 密码
        self.X_Auth_Value = ''
        self.loginsyscode = 'Portal'
        self.originsyscode = ''
        self.token_status = 0
        self.space_id = 'b3406ddb052f4c5da716ff9485814817'
        if account == "": 
            token_dic = self.get_token()
            self.account = token_dic['other']['account']
            self.token = token_dic['other']['token']
            self.headers = {'Content-type': 'application/json', 'X-Emp-No': self.account, 'X-Auth-Value': self.token, 'X-Lang-Id': 'zh_CN'}
        else:
            self.account = account
            self.token = token
            self.headers = {'Content-type': 'application/json', 'X-Emp-No': self.account, 'X-Auth-Value': self.token, 'X-Lang-Id': 'zh_CN'}
            self.auth = requests.auth.HTTPDigestAuth(self.account, self.token)
        
        self.proxies = {"http": "", "https": ""}
        
    def get_host_ip(self):
        try:
            s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            s.connect(('8.8.8.8', 80))
            ip = s.getsockname()[0]
        finally:
            s.close()
        return ip
        
    def get_token(self):
        url = ""
        client_ip = self.get_host_ip()
        text = \
        {
            "account": self.X_Emp_No,
            "passWord": self.password,
            "loginClientIp": client_ip,
            "loginSystemCode": self.loginsyscode,
            "originSystemCode": self.originsyscode,
            "other": {
            "networkArea": '1',
            "networkAccessType": '1'
            },
            "verifyCode": hashlib.md5(str(self.X_Emp_No + self.password + client_ip + self.loginsyscode + self.originsyscode).encode(encoding='utf-8')).hexdigest()
        }
        headers = {'Content-type': 'application/json'}
        content = requests.post(url, data=json.dumps(text), headers=headers)
        return content.json()
        
    def get_page_info(self, content_id):
        url = self.content_url % (content_id, self.space_id)
        resp = requests.get(url, auth=self.auth, headers=self.headers, proxies=self.proxies)
        reps_json_data = json.loads(resp.content)
        try:
            if resp.status_code != 200 or reps_json_data["code"]["msgId"] != 'RetCode.Success':
                time.sleep(1)
                resp = requests.get(url, auth=self.auth, headers=self.headers, proxies=self.proxies)
                reps_json_data = json.loads(resp.content)
        except Exception as e:
            return None, None
        if "bo" not in reps_json_data.keys():
            return None, None
        page_txt = reps_json_data['bo']['contentBody']
        title = reps_json_data['bo']['title']
        return page_txt, title

对提取的文本数据按照markdown格式分章节处理，得到初步的原始数据。

原始数据包含大量上下文无关的噪音，下一步需要对原始数据进行清洗，提升召回数据的质量。

2.2.2 数据清洗

数据清洗是提高RAG系统性能的关键，数据清洗旨在保证信息的规范性、必要性、准确性以及一致性，以下是各指标的具体描述：

• 规范性：规范性是指数据内容应该标准化，不包含冗余字符、特殊字符、敏感词汇、病句等；重复文档、冗余信息应被剔除。
• 必要性：必要性是指数据中不应包含与目标任务无关的信息，保证信息必要内容的完整性，非必要内容应被缩减或剔除。
• 准确性：准确性是指信息内容的描述应当保证准确、可靠，符合事实或领域专家的认知，有误的信息应被剔除。
• 一致性：一致性是指数据内容不应存在歧义或二义性，例如实体、缩略词及术语应当进行标准化；存在二义性冲突的内容应被消除或统一。

在数据清洗阶段，主要任务包括：

1. 无效内容过滤：收集数据中包含的无效内容类型，包括特殊字符、冗余字符、填写说明等，使用例如正则表达式等方式进行批量提取与剔除。
2. 敏感信息过滤：收集数据中包含的敏感信息类型，例如工号、IP地址、敏感方案等，使用正则表达式或人工进行识别与剔除。
3. 无关内容过滤：收集数据中与目标任务无关的信息，人工进行标记并剔除。
4. 错误内容过滤：对可靠性较低的内容进行人工审核，剔除不准确的内容。
5. 二义性消除：对缩略词、术语进行标准化，必要时进行补充描述；识别多处信息之间存在二义性冲突的情况，保证一致性。

2.2.3 数据增强

结合目标任务应用的情况，使用同义词、释义或其他自然语言描述方式来增强语料库的多样性。

2.2.4 数据质量评估

规范性指标

下表中的指标范围为参考值，评估实施时，指标范围需要结合数据具体情况进行设定。

人工验证

在语料质量检查中，人工验证效率比较低下，只能通过抽样对部分语料进行检查。但是人工检查是不可缺失的一环，因为目前的技术代码不能够非常准确的判断文本的流畅性、上下文的相关性等

参考论文《A Large-scale Chinese Short-Text Conversation Dataset and Chinese pre-training dialog models》和《Evaluation of Text Generation: A Survey》，按固定采样比例（如0.1%）进行采样后，从以下几个方面人工评估。

• 语法性：生成语句的流畅性；
• 相关性：上下文的相关性；
• 信息量：清洗后语句自身含有的信息量。
• 准确性：文档内容是否准确，需要领域专家评审。

具体来说，不符合语法性或相关性的语句，给予0分；语句流畅、符合相关性但信息量不足的语句，给予1分；语句流畅、符合相关性且信息量充足的语句，给予2分。

人工验证流畅性、相关性的同时，也能顺便检查下文本是否存在乱码、是否有大段空白、是否存在隐私信息等

2.3 文档切分

RAG应用中需要对知识库的数据内容进行分块储存，以达到段落召回的目标。

分块（chunking）是将大块文本分解为多个小段落的过程，当我们使用Embedding模型编码内容时，这是一项必要的技术，分块可以帮助我们减少Embedding编码内容的噪音，优化从向量数据库中被召回内容的准确性，提高RAG系统效率。

各种分块策略的主要参数有以下两点：

• chunk_size：块大小。衡量分块的大小，没有固定的数据单位，依分块方式而定
• over_lap：重叠部分大小。文档中不同块之间的重复部分调节参数，为了尽可能的保留完整上下文语义信息

选择合适的分块参数很重要，合理的参数能够决定召回结果与query的相关性，直接影响模型回答效果，同时也能够控制token数量，保证不超过模型本身的限制。

Langchain中提供了多种文本分割器，适用于不同场景，应该充分理解各方法的优缺点，按照具体情况选择合适的方式。

字符文本分割器

字符文本分割是最常用的方式，只需要决定每一个块的字符或token的数量，以及各块之间的重叠度，以确保语义上下文不会在块之间丢失。

在多数情况下，固定大小的分块是最佳方式，在计算上更加经济且易于使用。

NLTK文本分割器

NLTK（Natural Language Toolkit）是一个python自然语言工具包，包含已经训练好的模型，它能够将一个段落或一篇文章分割成独立的句子。

代码分割器

Langchain中提供了代码分割器，PythonCodeTextSplitter可以将文本按Python类和方法定义进行拆分，它是RecursiveCharacterSplitter的一个简单子类，具有Python特定的分隔符。

递归文本分割器

递归文本分割器它由字符列表参数化。它尝试按顺序在它们上进行分割，直到块足够小。默认列表为[" ", "\n", " ", ""]。这样做的效果是尽可能地保持所有段落（然后是句子，然后是单词)在一起，因为它们通常看起来是最强的语义相关的文本片段。

专用格式分割

• MarkdownTextSplitter将文本沿Markdown标题、代码块或水平线分割。它是递归字符分割器的简单子类，具有Markdown特定的分隔符。默认情况下，查看源代码以查看Markdown语法。
• LatexTextSplitter 可以沿着 Latex 的标题、头部、枚举等分割文本。它实现为 RecursiveCharacterSplitter 的一个简单子类，带有 Latex 特定的分隔符。默认情况下，查看源代码以查看预期的 Latex 语法。

我们选择典型的字符文本分割器来进行实验，之后为了评估不同分割器对RAG系统的影响，在后面的章节将会尝试使用更多的文本分割器。

def read_files_in_directory(directory):
    file_contents = []
    file_paths = []
    metadatas = []
    
    # 遍历指定路径下的所有文件和文件夹
    for root, dirs, files in os.walk(directory):
        for file_name in files:
            file_path = os.path.join(root, file_name)  # 获取文件的绝对路径
            file_paths.append(file_path)
            
            with open(file_path, 'r') as file:
                content = file.read()
                file_contents.append(content)
            # print(file_path)
            metadatas.append(file_path.split("/")[-1].split(".md")[0])
    
    return file_contents, file_paths, metadatas
 
    
def split_character(directory_path):
 
    contents, _, _ = read_files_in_directory(directory_path)
    text_splitter = CharacterTextSplitter(
    separator = "  ",
    chunk_size = 650,
    chunk_overlap  = 200,
    length_function = len,
    )
    content_str = ""
    for content in contents:
        content_str += content
    chunks = text_splitter.create_documents([content_str], metadatas=[{"source": ''}])
    # print(len(chunks))
    # print (chunks[:5])
    return [{"text": chunk.page_content, "source": chunk.metadata["source"]} for chunk in chunks]

设定metadata一般为对应片段内容的来源路径source，能够快速定位原文段。设定为该文档的名称或片段概述等关键词能够在索引过程更加高效。

2.4 Embedding和索引

文本嵌入（Text Embedding）是指将自然格式的信息（如单词、字母等）通过编码模型转换为包含语义信息的向量，这些带有语义信息的向量经过更多NLP任务的训练能够适应不同的需求，例如分类任务或生成任务。选择好的Embedding模型直接决定了待处理的文本意义能否被理解，那么显然向量召回具有最主要的两个优点：

• 语义粒度泛化。能够处理词汇不匹配，但词义匹配的场景
• 能够实现端到端，解决表征学习和下游目标无法保持一致性的问题

大模型时代的Text Embedding模型均是Transformer结构的仅编码器模型，也就是类bert结构。

在选用适合于自己RAG系统的Embedding模型时，需要考虑以下方面：

• 系统复杂度的大小。需要预估索引量的规模，综合考虑知识库的内容数量，复杂度和重复度处于何种规模以及可供部署模型的硬件资源，使用越复杂的模型，往往效果越好，但资源开销就越大
• 知识库数据与模型训练数据差异大小。需要考虑RAG系统中的数据是否包含大量非通用语义词汇，是否上下文语境含义差异与Embedding模型训练数据中语义差异过大，从而评估是否需要进行微调，适配领域需求。数据语义不匹配的模型会导致召回率下降，影响系统性能。
• 模型任务与需求匹配度。选择Embedding模型需要考虑其适用的任务，不同的模型可能适合处理如文本分类、相似度匹配、命名体识别等不同任务，根据任务需求，应该选择能够最好的捕捉语义信息的嵌入模型。

确定Embedding模型选用可以参考MTEB-海量文本嵌入基准的官方榜单（https://huggingface.co/spaces/mteb/leaderboard），MTEB包含8个语义向量任务，涵盖58个数据集和112种语言。

与MTEB对应的中文评估指标C-MTEB也可以参考，包含6种任务类型，35个数据集，较中肯的评估各模型在不同任务的效果。

以上两类benchmark具有类似的评估任务，可以综合二者结果进行模型选取。

涉及到的任务包含多个类型：

• STS（Semantic Textual Similarity）：语义文本相似度，数据形式的两个文本片段，标签为文本对是否相关。该类任务的关键指标采用的是Spearman秩相关性系数，文本对的标签有两类，相关是1，不相关是0。通过计算多对文本对预测结果和标签两个向量间的相关性系数进行评估，属于相似性判断指标。
• Classfication：分类的语料比较简单，基本上都是一段文本对应一个标签。将embedding模型视为特征提取工具，然后用这些特征训练一个线性分类器，训练完成后，预测新数据上的标签，根据分类的结果计算 average precision（二分类） 或 accuracy （多分类）。
• Pair-classfication：问题的定义是给定一对文本，判断其是否具有相同含义。衡量指标是AP（average precision），属于判别性指标。
• Clustering：文本聚类任务评估，语料是个多分类的问题。计算聚类指标时，一般使用的是v-measure指标，它能比较好的衡量聚类的一致性和完整性。
• Re-Ranking：任务采用MAP（Mean Average Precision），是计算一组文档重排的指标。其衡量的是，是否能够根据query正确的区分正样本和负样本。
• Retrival：任务评估信息检索系统效果，综合采用各类指标，例如召回率、准确率、MAP、MRR、NDCG等。

示例选用m3e-base模型进行向量化，使用 Langchain 的 Embedding 相关模块（HuggingFaceEmbeddings和OpenAIEmbeddings）加载模型并编码我们的文档块。

class EmbedChunks:
    def __init__(self, model_name):
        if model_name == "text-embedding-ada-002":
            self.embedding_model = OpenAIEmbeddings(
                model=model_name,
                openai_api_base=os.environ["OPENAI_API_BASE"],
                openai_api_key=os.environ["OPENAI_API_KEY"])
        else:
            self.embedding_model = HuggingFaceEmbeddings(
                model_name=model_name,
                model_kwargs={"device": "cuda"},
                encode_kwargs={"device": "cuda", "batch_size": 100})
    
    def __call__(self, batch):
        embeddings = self.embedding_model.embed_documents(batch["text"])
        return {"text": batch["text"], "source": batch["source"], "embeddings": embeddings}
 
        
def embedding(embedding_model_name):
    
    chunks_ds = split_character()
    embedded_chunks = ray.data.from_items(chunks_ds).map_batches(
        EmbedChunks,
        fn_constructor_kwargs={"model_name": embedding_model_name},
        batch_size=100,
        num_gpus=1,
        compute=ActorPoolStrategy(size=2))
    # print(embedded_chunks)
    embedded_chunks.show(1)

调用m3e-base模型，通过embedding函数对每个文段编码，embed_documents即Embedding模型的推理过程，计算过程采用ray数据集的map_batches模块，计算使用两个工作线程，每个工作线程有 1 个 GPU。第一文本块的Embedding结果如下：

{
    'text': '# PTRS管理\n![](https://icenterapi.zte.com.cn/group3/M00/16/43/CjYTEWGd3MWAQPCXAAAJr9xIJws671.png)\n\n\n# PTRS管理的功能设计\n## PTRS管理的功能设计分析\n### PTRS管理的业务能力\n  相位噪声指射频器件在各种噪声的作用下引起的系统输出信号相位的随机变化。相位噪声会恶化接收端的信噪比或误差向量幅度，造成大量的误码，这样就限制了高阶调制的使用，会严重影响系统的容量...', 
    'source': '', 
    'embeddings': [0.6715492010116577, 1.1172834634780884, 0.7916168570518494, -0.11632557958364487, -0.09606651216745377, -0.19460099935531616, 0.574505627155304, -0.36584919691085815, -0.4636607766151428, 0.5073410272598267, 0.495039701461792, 0.3359251022338867, 0.660287618637085, -0.6332092881202698, -0.48762381076812744, -0.3404944837093353, 0.035247091203927994, 0.6208893060684204, ...]
    
}

获得了完整的所有内容的Embedding编码块后，需要将其索引（存储）起来，以便于能够快速检索和推理。有许多可选的向量数据库，如2.1.1节所示。将(text, source, embeddings)三元组保存在可进行向量检索的数据库中，通过相似度算法（余弦相似度等）计算Embedding和query，如果使用Langchain进行索引生成，那么需要调整三元组为Langchain适用形式，这里以Langchain中集成的Faiss为例进行索引生成：

def index(directory_path):
 
    contents, _, _ = read_files_in_directory(directory_path)
    text_splitter = CharacterTextSplitter(
    separator = "  ",
    chunk_size = 650,
    chunk_overlap  = 200,
    length_function = len,
    )
    content_str = ""
    for content in contents:
        content_str += content
    chunks = text_splitter.create_documents([content_str], metadatas=[{"source": ''}])
    
    model_name = "model/m3e-base"
    model_kwargs = {"device": "cuda"}
    encode_kwargs = {"device": "cuda", "batch_size": 100}
    embedding = HuggingFaceBgeEmbeddings(
        model_name=model_name,
        model_kwargs=model_kwargs,
        encode_kwargs=encode_kwargs,
        query_instruction="为文本生成向量表示用于文本检索"
        )
    db = FAISS.from_documents(chunks, embedding)
    db.save_local("faiss_index")

将faiss_index索引文件保存在本地，方便进行关联查询，对于示例知识库，构建索引和保存过程大约需要3min，文件内容如下。对于较大的知识库，部分向量数据库也提供云服务等。

faiss索引保存的内容

3 查询生成

查询生成阶段包括查询检索和响应生成。查询检索时将问题通过与构建索引时相同的Embedding模型进行编码，然后对查询向量和知识库编码向量之间进行相似度计算（欧氏距离、余弦相似度、点积相似度），当检索到最相关的几个片段，便可以使用各种采样方式选择文本，进而将其作为大模型的上下文来生成响应。

Langchain中提供相似度计算的接口，当构件好索引，便可以传入query直接运行相似度匹配算法，获取查询结果。

def similarity_search(query):
    model_name = "model/m3e-base"
    model_kwargs = {"device": "cuda"}
    encode_kwargs = {"device": "cuda", "batch_size": 100}
    embedding = HuggingFaceBgeEmbeddings(
    model_name=model_name,
    model_kwargs=model_kwargs,
    encode_kwargs=encode_kwargs,
    query_instruction="为文本生成向量表示用于文本检索"
    )
    new_db = FAISS.load_local("faiss_index", embedding)
    
    # 第一种搜索方式
    # docs = new_db.similarity_search(query)
    
    # 第二种：构建检索器
    retriever = new_db.as_retriever(search_type="mmr", search_kwargs={"k":8})
    docs = retriever.get_relevant_documents(query)
    return docs

在Langchain中可以构建检索器来指定search_type，search_type主要包括两种：query- Similarity Search 和 Max Marginal Relevance Search (MMR Search)，其中MMR Search旨在提高搜索结果的多样性，常用于推荐系统。如果想检索更准确则推荐选择query- Similarity Search，默认的search_type也是query- Similarity Search。同时，配置search_kwargs的k参数可以选择召回文段数量，提高系统召回率。

将召回结果组织为大模型的上下文，进行推理服务，这样就实现了典型的RAG系统，修改知识库内容使得模型具有领域知识，其中各模块包括模型均是可插拔式设计，系统可塑性强，具有很高的优化价值。

4 RAG框架评估

4.1 RAG评估目标

RAG应用构建过程并不复杂，但由于 RAG 流程较长，包含不同的组件，要使 RAG 系统的性能调整到令人满意的状态是非常困难的。一个RAG应用主要包含以下两个组件：

• 检索器组件： 从外部数据源中检索与任务相关的上下文知识，以便高质量完成知识密集型或实时型任务。
• 生成器组件： 使用检索到的强关联知识来增强提示词（prompt）并生成答案。

评估 RAG 流程时，必须单独评估两个组件，并综合两个组件的评估结果来判断RAG 流程是否在哪些方面仍需要改进。此外，要了解 RAG 应用程序的性能是否有所改善，必须对其进行定量评估。为此，需要两个要素：评估指标和评估数据集。

4.2 RAGA

RAGAs（Retrieval-Augmented Generation Assessment，检索增强生成评估）是一个框架（GitHub，文档），可提供必要的工具，帮助在组件层面评估 RAG 流程。

4.2.1 评估数据

RAGAs 不需要依赖评估数据集中人工标注的标准答案，而是利用底层的大语言模型进行评估。为了对 RAG 流程进行评估，RAGAs 需要以下几种信息：

• question：RAG 流程的输入，即用户的查询问题。
• answer：RAG 流程的输出，由RAG流程生成的答案。
• contexts：为解答 question 而从外部知识源检索到的相关上下文。
• ground_truths：question 的标准答案，这是唯一需要人工标注的信息。这个信息仅在评估 context_recall 这一指标时才必须（详见 评估指标）。

4.2.2 评估指标

RAGA 提供了一些指标来按组件和端到端评估 RAG 系统。 在组件级别，RAGAs 提供了评价检索组件（包括context_relevancy和context_recall）和生成组件（涉及faithfulness和answer_relevancy）的专门指标。

上下文精度 ：测量检索到的上下文与问题的相关程度。评估所有在上下文(contexts)中呈现的与基本事实(ground-truth)相关的条目是否排名较高。理想情况下，所有相关文档块(chunks)必须出现在顶层。值范围在 0 到 1 之间，其中分数越高表示精度越高。 公式解释

上下文召回率： 衡量是否检索到回答问题所需的所有相关信息。该指标是根据 ground_truth（这是框架中唯一依赖人工注释的指标）和 contexts 计算的。为了根据真实答案(ground truth)估算上下文召回率(Context recall)，分析真实答案中的每个句子以确定它是否可以归因于检索到的Context。 在理想情况下，真实答案中的所有句子都应归因于检索到的Context.

忠实性：衡量生成答案的事实准确性。给定上下文中正确陈述的数量除以生成答案中的陈述总数。如果答案(answer)中提出的所有基本事实(claims)都可以从给定的上下文(context)中推断出来，则生成的答案被认为是忠实的。

答案相关性：衡量生成的答案与问题的相关程度。该指标是使用 question 和 answer 计算的。

例如，答案“法国位于西欧”。对于问题“法国在哪里，它的首都是什么？”的答案相关性较低，因为它只回答了问题的一半。当答案直接且适当地解决原始问题时，该答案被视为相关。重要的是，我们对答案相关性的评估不考虑真实情况，而是对答案缺乏完整性或包含冗余细节的情况进行惩罚。为了计算这个分数，LLM会被提示多次为生成的答案生成适当的问题，并测量这些生成的问题与原始问题之间的平均余弦相似度。基本思想是，如果生成的答案准确地解决了最初的问题，LLM应该能够从答案中生成与原始问题相符的问题。

上下文精度和召回率是对检索器组件的评估，忠实性和答案相关性是对生成结果质量的评价，通过对各组件指标进行均值归一化，得到两个评价标准：

• 检索质量得分（retrieval_score）：综合评价检索器的文段召回效果。
• 生成质量得分（generate_score）：综合评价根据上下文的模型生成回复质量。

2.2.3 使用RAGAS评估RAG应用

2.2.3.1 环境配置

使用ragas包评估RAG应用效果，首先需要安装ragas包

pip install ragas

安装完成后，可以使用以下代码测试安装的包是否可用

from ragas import evaluate

此出引入后如没有报错，则表明安装的包可以用，如出现以下情况的报错则需要处理

1. 错误场景1：

cannot import name 'AzureOpenAIEmbeddings' from 'langchain.embeddings'

此类属于ragas包版本和langchain版本不匹配问题，需要写在langchain、openai、ragas包，重新安装ragas

2. 错误场景2：

ValueError: Unknown scheme for proxy URL URL('socks://xxxx.com:80')

此类属于代理问题引起，在终端上执行：unset all_proxy; unset ALL_PROXY

2.2.3.2 构建测试数据

测试数据包含以下信息：

• question：用户的查询问题。
• answer：RAG 流程的输出，由RAG流程生成的答案。
• contexts：为解答 question 而从外部知识源检索到的相关上下文。
• ground_truths：question 的标准答案，这是唯一需要人工标注的信息。

构建数据集过程中， question和ground_truths是需要人工标注的。 context为检索知识库后获取到的上下文内容，answer是下游应用结合context对question的回答，下面给出了一个自动化构建的测试数据的代码：

questions = []
 
ground_truths = []
 
answers = []
 
contexts = []
def load_dataset():
    # 加载数据集（question和ground_truths）
    data = pd.read_csv("rag/test_dataset.csv")
    
    for i, row in data.iterrows():
        print(i, row["question"], row["ground_truths"])
        questions.append(row["question"])
        ground_truths.append([row["ground_truths"]])
        # 检索知识库获取检索到的上下文内容
        context = retrieve_context("vector", "gte-large-zh", "feature_gte_large_zh", row["question"], row["question"], 5, 0.5)
        
        contexts.append(context)
        # 将上下文内容与用户问题注入prompt模板，输入给下游模型，获取模型分析的结果
        answer = retrieve_answer(row["question"], context)
        answers.append(answer)
    evaluate_model()

2.2.3.3 使用ragas包评估

将构建完成的评估数据注入给评估模型内，选择评估指标。此时将调用openai的模型开发对数据集进行评估，评估结果拿到后可保存在csv文件中。

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_recall,
    context_precision,
)
 
def evaluate_model():
    # 构建评估数据
    data = {
        "question": questions,
        "answer": answers,
        "contexts": contexts,
        "ground_truths": ground_truths
    }
    dataset = Dataset.from_dict(data)
    
    #选择评估指标，进行评估
    result = evaluate(
        dataset = dataset,
        metrics=[
            context_precision,
            context_recall,
            faithfulness,
            answer_relevancy,
        ],
    )
    df = result.to_pandas()
    df.to_csv("gte.csv", index=False)
    print(df)

直接使用evaluate方法，会默认使用openai的GPT3.5模型进行评估，评估过程受限于apikey的使用次数限制，因此建议使用自己的大模型进行评估

from ragas.llms import RagasLLM
from langchain.schema import LLMResult
from langchain.schema import Generation
from langchain.callbacks.base import Callbacks
from langchain.schema.embeddings import Embeddings
from FlagEmbedding import FlagModel
from ragas.metrics import AnswerRelevancy
 
class MyLLM(RagasLLM):
 
    def __init__(self, llm):
        self.base_llm = llm
    
    @property
    def llm(self):
        return self.base_llm
    
    def generate(
    self,
    prompts: List[str],
    n: int = 1,
    temperature: float = 0,
    callbacks: t.Optional[Callbacks] = None,
    ) -> LLMResult:
        generations = []
        llm_output = {}
        token_total = 0
        for prompt in prompts:
        content = prompt.messages[0].content
        text = self.retrieve_answer(content)  # 调用模型的接口获取输出
        generations.append([Generation(text=text)])
        token_total += len(text)
        llm_output['token_total'] = token_total
        
        return LLMResult(generations=generations, llm_output=llm_output)
    
    async def agenerate(
    self,
    prompts: List[str],
    n: int = 1,
    temperature: float = 0,
    callbacks: t.Optional[Callbacks] = None,
    ) -> LLMResult:
        generations = []
        llm_output = {}
        token_total = 0
        for prompt in prompts:
        content = prompt.messages[0].content
        text = self.retrieve_answer(content)
        generations.append([Generation(text=text)])
        token_total += len(text)
        llm_output['token_total'] = token_total
        
        return LLMResult(generations=generations, llm_output=llm_output)
        
        
    def retrieve_answer(self, prompt):
        url = ""
        payload = json.dumps({
        "model": "",
        "messages": [
        {
        "role": "user",
        "content": prompt
        }
        ],
        "max_tokens": 4096,
        "temperature": 0.1,
        "stream": False
        }, ensure_ascii=False).encode("utf-8")
        headers = {
        'Authorization': 'TEST-46542881-54d4-4096-b93d-6d5a3db326ac',
        'Content-Type': 'application/json'
        }
        
        response = requests.request("POST", url, headers=headers, data=payload)
        res = json.loads(response.text)
    return res['choices'][0]['message']['content']
 
 
class BaaiEmbedding(Embeddings):
 
    def __init__(self,model_path, max_length=512, batch_size=256):
        self.model = FlagModel(model_path, query_instruction_for_retrieval="为这个句子生成表示以用于检索相关文章：")
        self.max_length = max_length
        self.batch_size = batch_size
    
    def embed_documents(self, texts: List[str]) -> List[List[float]]:
        return self.model.encode_corpus(texts, self.batch_size, self.max_length).tolist()
    
    def embed_query(self, text: str) -> List[float]:
        return self.model.encode_queries(text, self.batch_size, self.max_length).tolist()
    
# 使用阶段
def direct_evaluate():
    data = pd.read_csv("gte_raw.csv")
    for i, row in data.iterrows():
        print(row["contexts"])
        questions.append(row["question"])
        answers.append(row["answer"])
        contexts.append(eval(row["contexts"]))
        ground_truths.append(eval(row["ground_truths"]))
    data = {
        "question": questions,
        "answer": answers,
        "contexts": contexts,
        "ground_truths": ground_truths
    }
    dataset = Dataset.from_dict(data)
    my_llm = MyLLM("")
    ans_relevancy = AnswerRelevancy(embeddings=BaaiEmbedding(""))
    faithfulness.llm = my_llm
    context_recall.llm = my_llm
    context_precision.llm = my_llm
    ans_relevancy.llm = my_llm
    result = evaluate(
        dataset = dataset,
        metrics=[
            context_precision,
            context_recall,
            faithfulness,
            ans_relevancy,
        ],
    )
    df = result.to_pandas()
    df.to_csv("gte.csv", index=False)
    print(df)

5 RAG系统优化

5.1 基础模块优化

5.1.1 上下文

为了验证RAG系统在召回相关文段上下文对模型回复效果确实有提升，我们可以通过设置num_chunks=0（无上下文）将其与num_chunks=5(chunk_size=500,m3e-base)来进行比较。

5.1.2 分块大小

分块的大小影响每个上下文段中的概念完备性，更大的块可能更加能够封装完整的模块化知识，但会受到更多噪声的影响，分块策略上可以考虑：

• 使用更小的chunk_size，但选取其附近的文段也进行召回（可能包含相关信息）
• 为知识库添加多种不同分块大小的索引
• 增加每一块的摘要信息并添加索引

为了验证分块大小对RAG系统的影响程度，采用300-900的分块大小构建索引，并采用retrieval_score和generate_score指标进行评估，评估中使用m3e-base模型。

随着分块大小的增加，检索质量也在提升，并且可能会继续增加，而模型回复质量先增后减，说明块大小一定程度能够带来更多有利的信息，提升回答质量，但相应的会引入更多噪音，一味的增加chunk_size不可行，需要按需选择最佳分块大小。

5.1.3 Embedding模型

模型对召回效果影响明显，为了验证不同Embedding模型在我们RAG系统的表现，将采用以下三种开源模型进行评估：

• acge-large-zh（0.65GB）
• gte-large-zh（0.65GB）
• m3e-base（0.41GB）

三种模型评估得分如下（chunk_size=700）：

整体上来看检索得分与生成得分是成正相关的，说明模型输出较为稳定（temperature=0.1），且Embedding模型的选择对检索质量有明显提升，但检索质量达到一定程度，模型的生成质量则受模型基础能力影响更大。

5.2 Embedding模型微调

仅关注数据层面的优化对相关文档的召回率有促进作用，但直接探索使用实际场景中的数据微调Embedding模型，有助于更好的数据表示，提高RAG系统的检索质量和得分。学习比默认的Embedding更符合上下文的token是一个直观的优化点，因为默认的tokenization可能存在问题：

• 默认的分词过程可能会将私域专有词汇拆分，丧失其原本含义
• 在实际使用场景中词汇具有上下文上不同的含义

Embedding模型已经经过自监督学习任务（word2vec）的训练，微调的任务是使模型确认数据集中的哪些部分能够最匹配输入，使得嵌入模型能够学习到数据集中更准确的向量表示。

微调的关键需要构造合理的领域数据集，可以使用模型selfQA合成数据或人工标注。

{'train_runtime': 274.6573, 'train_samples_per_second': 22.421, 'train_steps_per_second': 1.398, 'train_loss': 0.21397568702620143, 'epoch': 2.0}

5.3 混合检索

当我们描述问题时，有时并不清楚我们的提问是否与提供的文本表达存在歧义，或者当我们只有模糊的记忆，只知道一两个词时，向量库的语义搜索往往无法满足我们的需求。此时可以结合关键词和语义进行同时检索，能够进一步提高我们的RAG的准确度。这种混合检索方法可以在保证搜索效率的同时，提高搜索的准确性和全面性。

传统的关键词搜索一定程度弥补的语义搜索的不足，传统关键词搜索擅长：

• 精确匹配（如产品名称、姓名、产品编号）
• 少量字符的匹配（通过少量字符进行向量检索时效果非常不好，但很多用户恰恰习惯只输入几个关键词）
• 倾向低频词汇的匹配（低频词汇往往承载了语言中的重要意义，比如“你想跟我去喝咖啡吗？”这句话中的分词，“喝”“咖啡”会比“你”“想”“吗”在句子中承载更重要的含义）

“混合检索”没有明确的定义，如果我们使用其他搜索算法的组合，也可以被称为“混合检索”。比如，我们可以将用于检索实体关系的知识图谱技术与向量检索技术结合。不同的检索系统各自擅长寻找文本（段落、语句、词汇）之间不同的细微联系，这包括了精确关系、语义关系、主题关系、结构关系、实体关系、时间关系、事件关系等。没有任何一种检索模式能够适用全部的情景。混合检索通过多个检索系统的组合，实现了多个检索技术之间的互补。

6 总结

本方案主要介绍了RAG系统的概念、组成结构、执行流程，并分别对流程的每个部分进行展开，详述了各阶段的实现流程，使用实际案例进行分析。介绍了RAG系统的两个评估指标和计算方式，并针对基础优化点开展了实验，对复杂优化点提供了方案思路，说明了RAG系统优化的必要性。

以大模型为中心的RAG是高度动态化的系统，同时系统中各组件间分工明确且几乎无相互波及，真正实现“可插拔式”。RAG系统也是基于大模型的Agent形式的重要组成部分，而RAG中的“增强”的过程也可以看做Agent中的超级工具之一，为大模型提供上下文参考信息查询的服务，因此对RAG的深入探索也是Agent应用探索的重要部分。