如何手搓一个RAG

RAG的原理

RAG 是一个完整的系统，其工作流程可以简单地分为数据处理、检索、增强和生成四个阶段：

数据处理阶段
1. 对原始数据进行清洗和处理。
2. 将处理后的数据转化为检索模型可以使用的格式。
3. 将处理后的数据存储在对应的数据库中。
检索阶段
1. 将用户的问题输入到检索系统中，从数据库中检索相关信息。
增强阶段
1. 对检索到的信息进行处理和增强，以便生成模型可以更好地理解和使用。
生成阶段
1. 将增强后的信息输入到生成模型中，生成模型根据这些信息生成答案。

学习RAG的实现流程

所有的代码均以给出，那么我们应该做什么才能够深度的理解RAG的构造？（本次学习的代码位于 https://github.com/datawhalechina/tiny-universe/blob/main/content/TinyRAG/）

1.跑通示例代码。（运行示例代码，在运行的过程中发现实操RAG的困难，深入了解RAG）

2.按照教程构造RAG的提示走一遍流程，同时尝试对代码进行改进

要有一个向量化模块，用来将文档片段向量化。 --> 选取一批文档进行向量化流程参考repo源码
要有一个文档加载和切分的模块，用来加载文档并切分成文档片段。 -->将文档进行切分
要有一个数据库来存放文档片段和对应的向量表示。–> 创造数据库储存向量
要有一个检索模块，用来根据 Query （问题）检索相关的文档片段。–> 对数据库进行相似度算法设计
要有一个大模型模块，用来根据检索出来的文档回答用户的问题。 -->使用interglm8b模型测试

3.对流程中遇到的问题进行总结，同时寻求解决方案。

4.自己提出对RAG的构思。

思考:

1.问题是一个向量 , 文档中的相关答案也是一个向量。取余弦相似度是为了能够找到和问题语义相近的答案。

2.RAG由很多个部分组合而成，每一个部分都很重要，如同LLM为RAG系统的大脑，对搜索道的资料进行整合归纳，给出合理的解释，Prompt设定模型的角色，规定模型提取数据的方式，告诉大模型什么时候该做什么，向量数据库对比问题以及知识库答案的相似程度，直接影响大模型的回答。

产生的疑问:

1)问题和答案的维度并不是完全匹配的，余弦相似度是如何计算这两个相似度的，从几何或者是代数上这有什么意义？

2)有没有更好的相似度算法能够运算问题和答案的相似度?

3)RAG的prompt是如何影响大模型的回答的，在用户体验角度上如何设计出更好的prompt?

4)RAG系统最重要的是什么？

关于RAG的问题回答

1) 问题和答案的维度并不是完全匹配的，余弦相似度是如何计算这两个相似度的，从几何或者是代数上这有什么意义？

余弦相似度通过计算两个向量的内积来衡量它们之间的相似性。具体公式如下：

$\text{余弦相似度} = \frac{\vec{A} \cdot \vec{B}}{||\vec{A}|| \cdot ||\vec{B}||} = \frac{\sum_{i=1}^{n} A_i B_i}{\sqrt{\sum_{i=1}^{n} A_i^2} \cdot \sqrt{\sum_{i=1}^{n} B_i^2}}$

在几何上，这个值表示两个向量在向量空间中的夹角的余弦值。夹角越小，相似度越高；夹角越大，相似度越低。当两个向量完全相同时，余弦相似度为1；当两个向量完全不相关时，余弦相似度为0；当两个向量完全相反时，余弦相似度为-1。

2) 有没有更好的相似度算法能够运算问题和答案的相似度？

除了余弦相似度外，还有其他相似度算法可以用于计算问题和答案的相似度，例如：

欧氏距离：度量两个点之间的直线距离。
曼哈顿距离：度量两个点之间的城市街区距离。
Jaccard相似度：度量两个集合之间的相似性，特别适用于稀疏数据。
BM25：一种用于信息检索的词频-逆文档频率（TF-IDF）变种，适合于文本相似度计算。
深度学习方法：使用预训练的语言模型（如BERT、RoBERTa）计算嵌入向量，并使用这些嵌入向量计算相似度。

每种方法都有其适用的场景和优缺点，具体选择需要根据应用场景进行调整。

3) RAG的prompt是如何影响大模型的回答的，在用户体验角度上如何设计出更好的prompt？

RAG（Retrieval-Augmented Generation）的prompt会直接影响到模型生成的质量和相关性。一个好的prompt设计可以：

明确问题背景和上下文：提供足够的信息，让模型更好地理解问题背景。
清晰具体：避免模糊的语言，直接指向核心问题。
简洁明了：保持简洁，不要冗长，减少噪音。
包含关键词：确保重要的关键词和概念被包含在prompt中，以提高检索和生成的相关性。

在用户体验角度上，设计好的prompt可以提高回答的准确性和用户满意度。可以通过用户反馈和迭代来不断优化prompt设计。

4) RAG系统最重要的是什么？

RAG系统最重要的几个方面包括：

检索能力：有效地从大规模文档库中检索相关信息的能力。
生成能力：生成高质量、相关性强的回答的能力。
集成和协调：检索和生成模块的无缝集成和高效协调，确保信息的准确传递和有效融合。
用户体验：提供直观、易用的界面和高效的反馈机制，使用户能够方便地获取所需信息。

这几个方面共同决定了RAG系统的性能和用户满意度。

跑通示例代码

完整的代码如下:

from RAG.VectorBase import VectorStore
from RAG.utils import ReadFiles
from RAG.LLM import OpenAIChat, InternLMChat
from RAG.Embeddings import JinaEmbedding  # 建立向量数据库
docs = ReadFiles('./data').get_content(max_token_len=600, cover_content=150) # 获得data目录下的所有文件内容并分割
vector = VectorStore(docs)
embedding = JinaEmbedding(path='/root/autodl-tmp/jinaai/jina-embeddings-v2-base-zh') # 创建EmbeddingModel
vector.get_vector(EmbeddingModel=embedding)
vector.persist(path='storage') # 将向量和文档内容保存到storage目录下，下次再用就可以直接加载本地的数据库vector = VectorStore()
vector.load_vector('./storage') # 加载本地的数据库
embedding = JinaEmbedding(path='/root/autodl-tmp/jinaai/jina-embeddings-v2-base-zh')
question = 'chronos是什么？'
content = vector.query(question, EmbeddingModel=embedding, k=1)[0]
print(content)model = InternLMChat(path='/root/autodl-tmp/Shanghai_AI_Laboratory/internlm2-chat-7b')
print(model.chat(question, [], content))

分解构建RAG步骤，实操各个模块

将数据向量化的过程

docs = ReadFiles('./data').get_content(max_token_len=600, cover_content=150) # 获得data目录下的所有文件内容并分割
vector = VectorStore(docs)
embedding = JinaEmbedding(path='/root/autodl-tmp/jinaai/jina-embeddings-v2-base-zh') # 创建EmbeddingModel
vector.get_vector(EmbeddingModel=embedding)
vector.persist(path='storage') # 将向量和文档内容保存到storage目录下，下次再用就可以直接加载本地的数据库

我们可以看到构建向量数据库的过程。首先读入数据，使用ReadFile模块进行数据的分块以及预处理，同时使用get_vector结合embedding模块向量化数据。

`ReadingFile模块` 读入pdf数据,进行分块

由于单个文档的长度往往会超过模型支持的上下文，导致检索得到的知识太长超出模型的处理能力，因此，在构建向量知识库的过程中，我们往往需要对文档进行分割，将单个文档按长度或者按固定的规则分割成若干个 chunk，然后将每个 chunk 转化为词向量，存储到向量数据库中。

在检索时，我们会以 chunk 作为检索的元单位，也就是每一次检索到 k 个 chunk 作为模型可以参考来回答用户问题的知识，这个 k 是我们可以自由设定hain 中文本分割器都根据 chunk_size (块大小)和 chunk_overlap (块与块之间的重叠大小)er.png)

chunk_size 指每个块包含的字符或 Token （如单词、句子等）的数量
chunk_overlap 指两个块之间共享的字符数量，用于保持上下文的连贯性，避免分割丢失上下文信息

from PyPDF2 import PdfReader
import re
from RAG.utils import ReadFiles
path = 'data/v1.pdf'
doc =  ''# 储存分块后的文字
reader =   PdfReader(open(path,'rb'))# 储存读入的pdf
content = ''
for page_num in range(len(reader.pages)):content += reader.pages[page_num].extract_text()
# 进行一定的数据清洗
pattern = re.compile(r'[^\u4e00-\u9fff](\n)[^\u4e00-\u9fff]', re.DOTALL)
content = re.sub(pattern, lambda match: match.group(0).replace('\n', ''),content)content = content.replace('•', '')
content = content.replace(' ', '')
# 进行分块 
doc = ReadFiles.get_chunk(content,400,100)# token最长600 重叠 150  
doc[0]

README.md2024-04-141/5动⼿学⼤模型应⽤开发\n项⽬简介\n本项⽬是⼀个⾯向⼩⽩开发者的⼤模型应⽤开发教程，旨在基于阿⾥云服务器，结合个⼈知识库助⼿项⽬，通\n过⼀个课程完成⼤模型开发的重点入⻔，主要内容包括：1.⼤模型简介，何为⼤模型、⼤模型特点是什么、LangChain是什么，如何开发⼀个LLM应⽤，针对⼩⽩\n开发者的简单介绍；2.如何调⽤⼤模型API，本节介绍了国内外知名⼤模型产品API的多种调⽤⽅式，包括调⽤原⽣API、封装\n'

如果需要更精细的控制，可以使用Langchain框架提供的chunk分割器,Langchain可以提供多种文档分割方式，区别在怎么确定块与块之间的边界、块由哪些字符/token组成、以及如何测量块大小

RecursiveCharacterTextSplitter(): 按字符串分割文本，递归地尝试按不同的分隔符进行分割文本。
CharacterTextSplitter(): 按字符来分割文本。
MarkdownHeaderTextSplitter(): 基于指定的标题来分割markdown 文件。
TokenTextSplitter(): 按token来分割文本。
SentenceTransformersTokenTextSplitter(): 按token来分割文本
Language(): 用于 CPP、Python、Ruby、Markdown 等。
NLTKTextSplitter(): 使用 NLTK（自然语言工具包）按句子分割文本。
SpacyTextSplitter(): 使用 Spacy按句子的切割文本。

创建Embedding类

embedding类最主要的方法是 get_embedding,

对于小白来说，涉及到的知识,主要为加载模型,以及调用模型的decode方法。

加载与训练Embedding模型的代码

import torch
from transformers import AutoModel
model = AutoModel.from_pretrained(self.path, trust_remote_code=True).to(device)

import numpy as np 
from tqdm import tqdm
from typing import Dict, List, Optional, Tuple, Union # 指定类型class BaseEmbeddings:"""Base class for embeddings"""def __init__(self, path: str, is_api: bool) -> None:self.path = pathself.is_api = is_apidef get_embedding(self, text: str, model: str) -> List[float]:raise NotImplementedError@classmethoddef cosine_similarity(cls, vector1: List[float], vector2: List[float]) -> float:"""calculate cosine similarity between two vectors"""dot_product = np.dot(vector1, vector2)magnitude = np.linalg.norm(vector1) * np.linalg.norm(vector2)if not magnitude:return 0return dot_product / magnitudeclass JinaEmbedding(BaseEmbeddings):"""class for Jina embeddings"""def __init__(self, path: str = 'jinaai/jina-embeddings-v2-base-zh', is_api: bool = False) -> None:super().__init__(path, is_api)self._model = self.load_model()def get_embedding(self, text: str) -> List[float]:return self._model.encode([text])[0].tolist()def load_model(self):import torchfrom transformers import AutoModelif torch.cuda.is_available():device = torch.device("cuda")else:device = torch.device("cpu")model = AutoModel.from_pretrained(self.path, trust_remote_code=True).to(device)return model

最核心的部分就是get_embedding，将数据向量化。

数据库类

class VectorStore:def __init__(self, document: List[str] = ['']) -> None:self.document = documentdef get_vector(self, EmbeddingModel: BaseEmbeddings) -> List[List[float]]:self.vectors = []for doc in tqdm(self.document, desc="Calculating embeddings"):self.vectors.append(EmbeddingModel.get_embedding(doc))return self.vectorsdef persist(self, path: str = 'storage'):if not os.path.exists(path):os.makedirs(path)with open(f"{path}/doecment.json", 'w', encoding='utf-8') as f:json.dump(self.document, f, ensure_ascii=False)if self.vectors:with open(f"{path}/vectors.json", 'w', encoding='utf-8') as f:json.dump(self.vectors, f)def load_vector(self, path: str = 'storage'):with open(f"{path}/vectors.json", 'r', encoding='utf-8') as f:self.vectors = json.load(f)with open(f"{path}/doecment.json", 'r', encoding='utf-8') as f:self.document = json.load(f)def get_similarity(self, vector1: List[float], vector2: List[float]) -> float:return BaseEmbeddings.cosine_similarity(vector1, vector2)def query(self, query: str, EmbeddingModel: BaseEmbeddings, k: int = 1) -> List[str]:query_vector = EmbeddingModel.get_embedding(query)result = np.array([self.get_similarity(query_vector, vector)for vector in self.vectors])return np.array(self.document)[result.argsort()[-k:][::-1]].tolist()

使用 __init__传进来的doc文档块(chunk)进行数据库的初始化，使用get_vector计算每个文本块对应的向量值,使用persist来保存向量化的数据。

实操数据库

# 创建embedding 用于把文本映射成为向量
embedding = JinaEmbedding(path='/root/autodl-tmp/jinaai/jina-embeddings-v2-base-zh')
# 初始化数据库对象 传入之前分块好的数据
vectordb = VectorStore(doc)
# 进行embedding操作
vectordb.get_vector(embedding)
# 测试
q = 'prompt是什么?'
print(vectordb.query(q,embedding,1))
vectordb.persist('storage') # 保存数据库到指定文件夹

Calculating embeddings: 100%|██████████| 799/799 [00:06<00:00, 122.27it/s]
['我们测试以下问题：question="使⽤⼤模型时，构造Prompt的原则有哪些"result=qa_chain({"query":question})print(result["result"])\n在使⽤⼤型语⾔模型时，构造Prompt的原则主要包括编写清晰、具体的指令和给予模型充⾜的思考时间。⾸先，Prompt需要清晰明确地表达需求，提供⾜够的上下文信息，以确保语⾔模型准确理解⽤户\n的意图。这就好比向⼀个对⼈类世界⼀⽆所知的外星⼈解释事物⼀样，需要详细⽽清晰的描述。过于简\n']

LLM类

AutoTokenizer,AutoModelForCausalLM类

import os
import torch
from transformers import AutoTokenizer, AutoModelForCausalLM
# torch.cuda.empty_cache()
q = '请你介绍一下你自己'
path='/root/autodl-tmp/Shanghai_AI_Laboratory/internlm2-chat-7b'
tokenizer = AutoTokenizer.from_pretrained(path, trust_remote_code=True)
model = AutoModelForCausalLM.from_pretrained(path, torch_dtype=torch.bfloat16, trust_remote_code=True).cuda()
print(model.chat(tokenizer,q))

(你好，我是一款名为书生·浦语的人工智能助手，由上海人工智能实验室开发。我能够使用自然语言与人类进行交流，并致力于通过执行常见的基于语言的任务和提供建议来帮助人类。
我能够回答问题、提供定义和解释、将文本从一种语言翻译成另一种语言、总结文本、生成文本、编写故事、分析情感、提供推荐、开发算法、编写代码以及其他任何基于语言的任务。
我希望我的存在能够为人类带来便利和帮助。)

我使用llm时，经常会因为显存不足报cuda out off memory这样的错。可以对模型的加载和推理进行一定的优化，例如可以在代码上加入 device_map='auto'来对显存进行合理的分配。具体的我还未深入的了解。