任务
使用大模型+RAG技术,缓解大模型幻觉问题。
RAG介绍
在实际业务场景中,通用的基础大模型可能存在无法满足我们需求的情况,主要有以下几方面原因:
-
知识局限性:大模型的知识来源于训练数据,而这些数据主要来自于互联网上已经公开的资源,对于一些实时性的或者非公开的,由于大模型没有获取到相关数据,这部分知识也就无法被掌握。
-
数据安全性:为了使得大模型能够具备相应的知识,就需要将数据纳入到训练集进行训练。然而,对于企业来说,数据的安全性至关重要,任何形式的数据泄露都可能对企业构成致命的威胁。
-
大模型幻觉:由于大模型是基于概率统计进行构建的,其输出本质上是一系列数值运算。因此,有时会出现模型“一本正经地胡说八道”的情况,尤其是在大模型不具备的知识或不擅长的场景中。
针对上述这些问题,研究人员提出了检索增强生成(Retrieval Augmented Generation, RAG)的方法。这种方法通过引入外部知识,使大模型能够生成准确且符合上下文的答案,同时能够减少模型幻觉的出现。
由于RAG简单有效,它已经成为主流的大模型应用方案之一。
如下图所示,RAG通常包括以下三个基本步骤:
-
索引:将文档库分割成较短的 Chunk,即文本块或文档片段,然后构建成向量索引。
-
检索:计算问题和 Chunks 的相似度,检索出若干个相关的 Chunk。
-
生成:将检索到的Chunks作为背景信息,生成问题的回答。
简而言之,就是先把文档分割成一个个的小块,每个小块用模型转为一个向量存储起来作为知识库,当用户提问时,首先在知识库中查找相关的向量,即与问题相关的资料,然后把问题和检索到的资料都输给大模型,让大模型结合自身知识和知识库知识回答问题,这样当大模型不知道问题答案时还能参考知识库知识回答,从而减少大模型幻觉。
实例
按照上面所说,要对已有文本切分构建向量,然后要在回答问题时索引相关向量。所以,核心在于向量库构建和向量索引算法。
1、向量库构建
利用模型将文本转为向量:
# 定义向量模型类
class EmbeddingModel:
"""
class for EmbeddingModel
"""
def __init__(self, path: str) -> None:
# 定义向量模型的tokenizer和model
self.tokenizer = AutoTokenizer.from_pretrained(path)
self.model = AutoModel.from_pretrained(path).cuda()
print(f'Loading EmbeddingModel from {path}.')
def get_embeddings(self, texts: List) -> List[float]:
"""
calculate embedding for text list
"""
encoded_input = self.tokenizer(texts, padding=True, truncation=True, return_tensors='pt')
print(encoded_input)
encoded_input = {k: v.cuda() for k, v in encoded_input.items()}
with torch.no_grad():
model_output = self.model(**encoded_input)
sentence_embeddings = model_output[0][:, 0]
sentence_embeddings = torch.nn.functional.normalize(sentence_embeddings, p=2, dim=1)
return sentence_embeddings.tolist()
(1)加载模型
(2)文本转为向量
get_embeddings
函数的主要作用是为输入的文本列表 (texts
) 计算嵌入向量(embeddings)。它的主要流程如下:
- 分词和编码 (
self.tokenizer
):- 将输入文本列表
texts
使用 tokenizer 进行分词和编码,并将其转换为模型可以处理的格式。使用padding=True
进行填充,确保所有输入句子长度一致,并且truncation=True
用于在句子过长时截断到模型的最大长度。 return_tensors='pt'
表示返回的是 PyTorch 张量(tensor)。
- 将输入文本列表
- 迁移到 GPU (cuda):
- 为了加速处理,所有的输入数据(encoded_input)被迁移到 GPU 上。使用
.cuda()
函数将这些张量转移到 GPU。
- 为了加速处理,所有的输入数据(encoded_input)被迁移到 GPU 上。使用
- 禁用梯度计算:
- 使用
torch.no_grad()
上下文管理器禁用梯度计算,因为这里不需要更新模型的权重,只需要计算句子嵌入。这样可以减少内存消耗并加快计算速度。
- 使用
- 模型前向传递:
- 调用
self.model(**encoded_input)
来执行模型的前向传递,得到模型的输出model_output
。model_output[0]
通常表示每个输入 token 的嵌入,形状为(batch_size, sequence_length, hidden_size)
。
- 调用
- 提取句子级别的嵌入:
model_output[0][:, 0]
取出每个句子对应的[CLS]
token 的嵌入(即0
位置的嵌入)。在 BERT 等模型中,[CLS]
token 的嵌入通常被用作整个句子的语义表示。
- 归一化嵌入:
- 使用
torch.nn.functional.normalize
对句子嵌入进行 L2 归一化,确保每个句子的嵌入向量的模为 1。p=2
指的是 L2 范数,dim=1
是在句子维度上进行归一化。
- 使用
- 转换为 Python 列表:
- 最终将句子嵌入张量转换为 Python 列表,并返回嵌入向量的列表。
2、向量索引算法
其核心在于在一大堆向量中找到与问题有关的向量,这里用余弦相似度判断向量是否与问题有关。
# 定义向量库索引类
class VectorStoreIndex:
"""
class for VectorStoreIndex
"""
def __init__(self, doecment_path: str, embed_model: EmbeddingModel) -> None:
self.documents = []
for line in open(doecment_path, 'r', encoding='utf-8'):
line = line.strip()
self.documents.append(line)
print(self.documents)
self.embed_model = embed_model
self.vectors = self.embed_model.get_embeddings(self.documents)
print(f'Loading {len(self.documents)} documents for {doecment_path}.')
def get_similarity(self, 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 0
return dot_product / magnitude # 点积/模的乘积=余弦相似度
def query(self, question: str, k: int = 1) -> List[str]:
question_vector = self.embed_model.get_embeddings([question])[0]
result = np.array([self.get_similarity(question_vector, vector) for vector in self.vectors])
return np.array(self.documents)[result.argsort()[-k:][::-1]].tolist()
(1)get_similarity
是用于计算两个向量之间的余弦相似度(cosine similarity)
get_similarity(self, vector1: List[float], vector2: List[float]) -> float
- 函数接受两个输入参数:
vector1
和vector2
,它们都是浮点数列表(即向量)。返回值是一个浮点数,即余弦相似度的结果。
dot_product = np.dot(vector1, vector2)
:这一步计算两个向量的点积(即向量之间相应位置元素的乘积和)。点积反映了两个向量在方向上的相似程度。
magnitude = np.linalg.norm(vector1) * np.linalg.norm(vector2)
:这里计算的是两个向量的模(或范数),即向量的长度,并将两个模相乘。模是通过求每个向量的元素平方和再开方得到的。这反映了向量的大小。
if not magnitude: return 0
:如果模为 0(意味着其中一个或两个向量是零向量,即没有方向),则余弦相似度被定义为 0,因为无法比较方向。
return dot_product / magnitude
:最后,函数返回点积除以模的乘积,这就是余弦相似度的定义。这个值位于 -1 到 1 之间,表示两个向量之间的方向相似性,1 表示完全相同的方向,-1 表示完全相反的方向,0 表示它们互相正交。
余弦相似度值越接近 1,表示两个向量的方向越相近。
(2)query
方法,用于从一组文档中查找与问题(question
)最相似的内容。
函数输入与输出:
def query(self, question: str, k: int = 1) -> List[str]
- 参数:
question
:一个字符串,表示要查询的问题或语句。k
:一个整数(默认为1),表示返回的最相似文档的数量。
- 返回值:返回与输入问题最相似的文档列表,长度为
k
。
**question_vector = self.embed_model.get_embeddings([question])[0]**
- 这里通过
self.embed_model.get_embeddings
方法对输入的问题进行嵌入,将文本转换为向量表示。[question]
表示将问题放入列表中,而[0]
则是提取出第一个(也是唯一一个)向量,因为传入的是一个单一问题。 question_vector
是问题的嵌入向量,通常用作进一步计算相似度。
result = np.array([self.get_similarity(question_vector, vector) for vector in self.vectors])
- 这里遍历
self.vectors
中存储的所有文档向量,并逐一与question_vector
进行相似度计算。使用self.get_similarity
方法计算余弦相似度,返回一个相似度分数的列表。 np.array(...)
将所有相似度结果转换为 NumPy 数组,方便后续的排序和索引操作。
return np.array(self.documents)[result.argsort()[-k:][::-1]].tolist()
result.argsort()
:返回result
数组的索引值,按相似度从低到高进行排序。[-k:][::-1]
:选取最后k
个相似度最高的索引,并使用[::-1]
进行逆序排列,以得到从最高相似度到最低相似度的排序。np.array(self.documents)
:将文档转换为 NumPy 数组,使用排好序的索引来提取与问题最相似的文档。.tolist()
:将结果从 NumPy 数组转换为 Python 列表,并将最相似的k
个文档返回。
query函数的作用是计算输入问题与所有文档的相似度,并返回相似度最高的 k
个文档。通过将文本转化为向量并使用余弦相似度进行比较,它能够为文本检索或类似的任务提供有效的结果。
3、大模型输出
最后将问题和检索的内容输入大模型,得到答案。
# 定义大语言模型类
class LLM:
"""
class for Yuan2.0 LLM
"""
def __init__(self, model_path: str) -> None:
print("Creat tokenizer...")
self.tokenizer = AutoTokenizer.from_pretrained(model_path, add_eos_token=False, add_bos_token=False, eos_token='<eod>')
self.tokenizer.add_tokens(['<sep>', '<pad>', '<mask>', '<predict>', '<FIM_SUFFIX>', '<FIM_PREFIX>', '<FIM_MIDDLE>','<commit_before>','<commit_msg>','<commit_after>','<jupyter_start>','<jupyter_text>','<jupyter_code>','<jupyter_output>','<empty_output>'], special_tokens=True)
print("Creat model...")
self.model = AutoModelForCausalLM.from_pretrained(model_path, torch_dtype=torch.bfloat16, trust_remote_code=True).cuda()
print(f'Loading Yuan2.0 model from {model_path}.')
def generate(self, question: str, context: List):
if context:
prompt = f'背景:{context}\n问题:{question}\n请基于背景,回答问题。'
else:
prompt = question
prompt += "<sep>"
inputs = self.tokenizer(prompt, return_tensors="pt")["input_ids"].cuda()
outputs = self.model.generate(inputs, do_sample=False, max_length=1024)
output = self.tokenizer.decode(outputs[0])
print(output.split("<sep>")[-1])
(1)加载大模型
(2)构建prompt
(3)得到结果
将不加RAG和加RAG的输出结果对比:
可以看出,不加RAG的输出虽然很多,但是不准确,比如广州大学的办学时间,简直胡说,加了RAG后有参考资料,输出内容会更准确,从而减少了大模型幻觉。
使用Yuan2-2B完整代码
# 大模型+RAG
# 导入所需的库
from typing import List
import numpy as np
import torch
from transformers import AutoModel, AutoTokenizer, AutoModelForCausalLM
# 向量模型下载
from modelscope import snapshot_download
model_dir = snapshot_download("AI-ModelScope/bge-small-zh-v1.5", cache_dir='.')
# 源大模型下载
from modelscope import snapshot_download
model_dir = snapshot_download('IEITYuan/Yuan2-2B-Mars-hf', cache_dir='.')
# model_dir = snapshot_download('IEITYuan/Yuan2-2B-July-hf', cache_dir='.')
# 定义向量模型类
class EmbeddingModel:
"""
class for EmbeddingModel
"""
def __init__(self, path: str) -> None:
# 定义向量模型的tokenizer和model
self.tokenizer = AutoTokenizer.from_pretrained(path)
self.model = AutoModel.from_pretrained(path).cuda()
print(f'Loading EmbeddingModel from {path}.')
def get_embeddings(self, texts: List) -> List[float]:
"""
calculate embedding for text list
"""
encoded_input = self.tokenizer(texts, padding=True, truncation=True, return_tensors='pt')
encoded_input = {k: v.cuda() for k, v in encoded_input.items()}
print(encoded_input)
with torch.no_grad():
model_output = self.model(**encoded_input)
sentence_embeddings = model_output[0][:, 0]
sentence_embeddings = torch.nn.functional.normalize(sentence_embeddings, p=2, dim=1)
return sentence_embeddings.tolist()
print("> Create embedding model...")
embed_model_path = './AI-ModelScope/bge-small-zh-v1___5'
embed_model = EmbeddingModel(embed_model_path)
# 定义向量库索引类
class VectorStoreIndex:
"""
class for VectorStoreIndex
"""
def __init__(self, doecment_path: str, embed_model: EmbeddingModel) -> None:
self.documents = []
for line in open(doecment_path, 'r', encoding='utf-8'):
line = line.strip()
self.documents.append(line)
# print(self.documents)
self.embed_model = embed_model
self.vectors = self.embed_model.get_embeddings(self.documents)
print(f'Loading {len(self.documents)} documents for {doecment_path}.')
def get_similarity(self, 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 0
return dot_product / magnitude # 点积/模的乘积=余弦相似度
def query(self, question: str, k: int = 1) -> List[str]:
question_vector = self.embed_model.get_embeddings([question])[0]
result = np.array([self.get_similarity(question_vector, vector) for vector in self.vectors])
return np.array(self.documents)[result.argsort()[-k:][::-1]].tolist()
print("> Create index...")
doecment_path = './knowledge.txt'
index = VectorStoreIndex(doecment_path, embed_model)
question = '介绍一下广州大学'
print('> Question:', question)
context = index.query(question)
print('> Context:', context)
# 定义大语言模型类
class LLM:
"""
class for Yuan2.0 LLM
"""
def __init__(self, model_path: str) -> None:
print("Creat tokenizer...")
self.tokenizer = AutoTokenizer.from_pretrained(model_path, add_eos_token=False, add_bos_token=False, eos_token='<eod>')
self.tokenizer.add_tokens(['<sep>', '<pad>', '<mask>', '<predict>', '<FIM_SUFFIX>', '<FIM_PREFIX>', '<FIM_MIDDLE>','<commit_before>','<commit_msg>','<commit_after>','<jupyter_start>','<jupyter_text>','<jupyter_code>','<jupyter_output>','<empty_output>'], special_tokens=True)
print("Creat model...")
self.model = AutoModelForCausalLM.from_pretrained(model_path, torch_dtype=torch.bfloat16, trust_remote_code=True).cuda()
print(f'Loading Yuan2.0 model from {model_path}.')
def generate(self, question: str, context: List):
if context:
prompt = f'背景:{context}\n问题:{question}\n请基于背景,回答问题。'
else:
prompt = question
prompt += "<sep>"
inputs = self.tokenizer(prompt, return_tensors="pt")["input_ids"].cuda()
outputs = self.model.generate(inputs, do_sample=False, max_length=1024)
output = self.tokenizer.decode(outputs[0])
print(output.split("<sep>")[-1])
print("> Create Yuan2.0 LLM...")
model_path = './IEITYuan/Yuan2-2B-Mars-hf'
# model_path = './IEITYuan/Yuan2-2B-July-hf'
llm = LLM(model_path)
print('> Without RAG:')
llm.generate(question, [])
print('> With RAG:')
llm.generate(question, context)
使用Qwen2-1.5B完整代码
# 大模型+RAG
# 导入所需的库
from typing import List
import numpy as np
import torch
from transformers import AutoModel, AutoTokenizer, AutoModelForCausalLM
# 向量模型下载
from modelscope import snapshot_download
model_dir = snapshot_download("AI-ModelScope/bge-small-zh-v1.5", cache_dir='.')
# 源大模型下载
from modelscope import snapshot_download
# model_dir = snapshot_download('IEITYuan/Yuan2-2B-Mars-hf', cache_dir='.')
model_dir = snapshot_download('qwen/Qwen2-1.5B-Instruct', cache_dir='.')
# 定义向量模型类
class EmbeddingModel:
"""
class for EmbeddingModel
"""
def __init__(self, path: str) -> None:
# 定义向量模型的tokenizer和model
self.tokenizer = AutoTokenizer.from_pretrained(path)
self.model = AutoModel.from_pretrained(path).cuda()
print(f'Loading EmbeddingModel from {path}.')
def get_embeddings(self, texts: List) -> List[float]:
"""
calculate embedding for text list
"""
encoded_input = self.tokenizer(texts, padding=True, truncation=True, return_tensors='pt')
encoded_input = {k: v.cuda() for k, v in encoded_input.items()}
print(encoded_input)
with torch.no_grad():
model_output = self.model(**encoded_input)
sentence_embeddings = model_output[0][:, 0]
sentence_embeddings = torch.nn.functional.normalize(sentence_embeddings, p=2, dim=1)
return sentence_embeddings.tolist()
print("> Create embedding model...")
embed_model_path = './AI-ModelScope/bge-small-zh-v1___5'
embed_model = EmbeddingModel(embed_model_path)
# 定义向量库索引类
class VectorStoreIndex:
"""
class for VectorStoreIndex
"""
def __init__(self, doecment_path: str, embed_model: EmbeddingModel) -> None:
self.documents = []
for line in open(doecment_path, 'r', encoding='utf-8'):
line = line.strip()
self.documents.append(line)
# print(self.documents)
self.embed_model = embed_model
self.vectors = self.embed_model.get_embeddings(self.documents)
print(f'Loading {len(self.documents)} documents for {doecment_path}.')
def get_similarity(self, 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 0
return dot_product / magnitude # 点积/模的乘积=余弦相似度
def query(self, question: str, k: int = 1) -> List[str]:
question_vector = self.embed_model.get_embeddings([question])[0]
result = np.array([self.get_similarity(question_vector, vector) for vector in self.vectors])
return np.array(self.documents)[result.argsort()[-k:][::-1]].tolist()
print("> Create index...")
doecment_path = './knowledge.txt'
index = VectorStoreIndex(doecment_path, embed_model)
question = '介绍一下广州大学'
print('> Question:', question)
context = index.query(question)
print('> Context:', context)
# 定义大语言模型类
class LLM:
"""
class for Yuan2.0 LLM
"""
def __init__(self, model_path: str) -> None:
print("Creat tokenizer...")
self.tokenizer = AutoTokenizer.from_pretrained(model_path, add_eos_token=False, add_bos_token=False, eos_token='<eod>')
self.tokenizer.add_tokens(['<sep>', '<pad>', '<mask>', '<predict>', '<FIM_SUFFIX>', '<FIM_PREFIX>', '<FIM_MIDDLE>','<commit_before>','<commit_msg>','<commit_after>','<jupyter_start>','<jupyter_text>','<jupyter_code>','<jupyter_output>','<empty_output>'], special_tokens=True)
print("Creat model...")
self.model = AutoModelForCausalLM.from_pretrained(model_path, torch_dtype=torch.bfloat16, trust_remote_code=True).cuda()
print(f'Loading Yuan2.0 model from {model_path}.')
def generate(self, question: str, context: List):
if context:
prompt = f'背景:{context}\n问题:{question}\n请基于背景,回答问题。'
else:
prompt = question
messages = [
{"role": "system", "content": "You are Qwen, created by Alibaba Cloud. You are a helpful assistant."},
]
messages.append({"role": "user", "content": prompt},)
text = self.tokenizer.apply_chat_template(
messages,
tokenize=False,
add_generation_prompt=True,
)
inputs = self.tokenizer([text], return_tensors="pt").to(self.model.device)
outputs = self.model.generate(**inputs, max_length=1024) # 设置解码方式和最大生成长度
generated_ids = [
output_ids[len(input_ids):] for input_ids, output_ids in zip(inputs.input_ids, outputs)
]
response = self.tokenizer.batch_decode(generated_ids, skip_special_tokens=True)[0]
print(response)
print("> Create Yuan2.0 LLM...")
# model_path = './IEITYuan/Yuan2-2B-Mars-hf'
model_path = './qwen/Qwen2-1___5B-Instruct'
llm = LLM(model_path)
print('> Without RAG:')
llm.generate(question, [])
print('> With RAG:')
llm.generate(question, context)
参考文献:
1、Task3:源大模型RAG实战 - 飞书云文档 (feishu.cn)
2、Qwen官方文档
标签:初试,RAG,模型,question,print,path,model,self,向量 From: https://blog.csdn.net/qq_42755230/article/details/142762364