3_搭建向量知识库

1. 向量及向量知识库

1.1 词向量与向量

词向量（word embedding）是一种以单词为单位将每个单词转化为实数向量的技术。这些实数向量可以被计算机更好地理解和处理。词向量背后的主要理念是相似或相关的对象在向量空间中的距离应该很近。

词向量实际上是将单词转化为固定的静态的向量，虽然可以在一定程度上捕捉并表达文本中的语义信息，但忽略了单词在不同语境中的意思会受到影响这一现实。因此在RAG应用中使用的向量技术一般为通用文本向量(Universal text embedding)，该技术可以对一定范围内任意长度的文本进行向量化，与词向量不同的是向量化的单位不再是单词而是输入的文本，输出的向量会捕捉更多的语义信息。

在RAG（Retrieval Augmented Generation，检索增强生成）方面向量的优势主要有两点：

• 向量比文字更适合检索。当我们在数据库检索时，如果数据库存储的是文字，主要通过检索关键词（词法搜索）等方法找到相对匹配的数据，匹配的程度取决于数据库中的文档中是否含有查询句中的关键词；而向量中包含了原文本的语义信息，可以通过计算问题与数据库中数据的点积、余弦距离、欧几里得距离等指标，直接获取问题与数据在语义层面上的相似度；
• 向量比其它媒介的综合信息能力更强，当传统数据库存储文字、声音、图像、视频等多种媒介时，很难去将上述多种媒介构建起关联与跨模态的查询方法；但是向量却可以通过多种向量模型将多种数据映射成统一的向量形式。

在搭建 RAG 系统时，我们往往可以通过使用向量模型来构建向量，我们可以选择：

• 使用各个公司的 Embedding API；
• 在本地使用向量模型将数据构建为向量。

1.2 向量数据库

专门用于存储和检索向量数据（embedding）的数据库系统。它与传统的基于关系模型的数据库不同，它主要关注的是向量数据的特性和相似性。

在向量数据库中，数据被表示为向量形式，每个向量代表一个数据项。这些向量可以是数字、文本、图像或其他类型的数据。向量数据库使用高效的索引和查询算法来加速向量数据的存储和检索过程。

向量数据库中的数据以向量作为基本单位，对向量进行存储、处理及检索。向量数据库通过计算与目标向量的余弦距离、点积等获取与目标向量的相似度。当处理大量甚至海量的向量数据时，向量数据库索引和查询算法的效率明显高于传统数据库。

• Chroma：是一个轻量级向量数据库，拥有丰富的功能和简单的 API，具有简单、易用、轻量的优点，但功能相对简单且不支持GPU加速，适合初学者使用。
• Weaviate：是一个开源向量数据库。除了支持相似度搜索和最大边际相关性（MMR，Maximal Marginal Relevance）搜索外还可以支持结合多种搜索算法（基于词法搜索、向量搜索）的混合搜索，从而提高搜索结果的相关性和准确性。
• Qdrant：Qdrant使用 Rust 语言开发，有极高的检索效率和RPS（Requests Per Second），支持本地运行、部署在本地服务器及Qdrant云三种部署模式。且可以通过为页面内容和元数据制定不同的键来复用数据。

2. 使用Embedding API

GPT有封装好的接口，我们简单封装即可。目前GPT embedding mode有三种，性能如下所示：

模型	每美元页数	MTEB得分	MIRACL得分
text-embedding-3-large	9,615	64.6	54.9
text-embedding-3-small	62,500	62.3	44.0
text-embedding-ada-002	12,500	61.0	31.4

• MTEB得分为embedding model分类、聚类、配对等八个任务的平均得分。
• MIRACL得分为embedding model在检索任务上的平均得分。

import os
from openai import OpenAI

def openai_embedding(text:str, model:str=None):
	
	client = OpenAI(
		api_key = os.environ['SILICONFLOW_API_KEY'],
		base_url = 'https://api.siliconflow.cn/v1'
	)
	if model == None:
		model = "BAAI/bge-m3"
	
	response = client.embeddings.create(
		input = text,
		model = model
	)
	return response
response = openai_embedding(text='要生成 embedding 的输入文本，字符串形式。')
print(response.model_dump_json(indent=2))
print(f'返回的embedding类型为：{response.object}')
print(f'embedding长度为：{len(response.data[0].embedding)}') 
print(f'embedding（前10）为：{response.data[0].embedding[:10]}')
print(f'本次embedding model为：{response.model}') 
print(f'本次token使用情况为：{response.usage}')

API返回的数据为json格式，除object向量类型外还有存放数据的data、embedding model 型号model以及本次 token 使用情况usage等数据，

3. 数据处理

为构建我们的本地知识库，我们需要对以多种类型存储的本地文档进行处理，读取本地文档并通过前文描述的 Embedding 方法将本地文档的内容转化为词向量来构建向量数据库。

3.1 源文档选取

• 应该包含pdf, txt 等多种形式的数据

3.2 数据读取

1. 使用LangChain 的 PyMuPDFLoader 来读取知识库的 PDF 文件。PyMuPDFLoader 是 PDF 解析器中速度最快的一种，结果会包含 PDF 及其页面的详细元数据，并且每页返回一个文档。

from langchain_community.document_loaders import PyMuPDFLoader

loader = PyMuPDFLoader(
	"../data_base/knowledge_db/pumkin_book/pumpkin_book.pdf"
)
pdf_pages = loader.load()

print(pdf_pages)
print(f"载入后的变量类型为：{type(pdf_pages)}，",  f"该 PDF 一共包含 {len(pdf_pages)} 页")
pdf_page = pdf_pages[1] #取一页
print(f"每一个元素的类型：{type(pdf_page)}.",
    f"该文档的描述性数据：{pdf_page.metadata}",
    f"查看该文档的内容:\n{pdf_page.page_content}",
    sep="\n------\n")

2. 读取markdown

from langchain_community.document_loaders.markdown import UnstructuredMarkdownLoader
loader = UnstructuredMarkdownLoader("../data_base/knowledge_db/prompt_engineering/1. 简介 Introduction.md")
md_pages = loader.load()
md_page = md_pages[0]
print(md_pages) #读取出来的对象是一样的

3.3 数据清洗

1. 使用正则表达式匹配并删除掉\n。

import re
pattern = re.compile(
	r'[^\u4e00-\u9fff](\n)[^\u4e00-\u9fff]', 
	re.DOTALL
)
pdf_page.page_content = re.sub(
	pattern,
	lambda match: match.group(0).replace('\n',''),
	pdf_page.page_content
)
print(pdf_page.page_content)

2. 发现数据中还有不少的•和空格，我们使用replace方法去掉即可。

pdf_page.page_content = pdf_page.page_content.replace('•','')
pdf_page.page_content = pdf_page.page_content.replace(' ', '')
print(pdf_page.page_content)

3. md文件中每一段中间隔了一个换行符，我们同样可以使用replace方法去除

md_page.page_content = md_page.page_content.replace('\n\n', '\n') 
print(md_page.page_content)

3.4 文档分割

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

在检索时，我们会以 chunk 作为检索的元单位，也就是每一次检索到 k 个 chunk 作为模型可以参考来回答用户问题的知识，这个 k 是我们可以自由设定的。

Langchain 中文本分割器都根据 chunk_size (块大小，每个块包含的字符或 Token 的数量)和 chunk_overlap (块与块之间的重叠大小)进行分割。

• RecursiveCharacterTextSplitter(): 按字符串分割文本，递归地尝试按不同的分隔符进行分割文本。
  • 按照这个优先级["\n\n", "\n", " ", ""])，这样就能尽量把所有和语义相关的内容尽可能长时间地保留在同一位置。
  • 该函数有四个参数：*
    • separators - 分隔符字符串数组
    • chunk_size - 每个文档的字符数量限制
    • chunk_overlap - 两份文档重叠区域的长度
    • length_function - 长度计算函数
• CharacterTextSplitter(): 按字符来分割文本。
• MarkdownHeaderTextSplitter(): 基于指定的标题来分割markdown 文件。
• TokenTextSplitter(): 按token来分割文本。
• SentenceTransformersTokenTextSplitter(): 按token来分割文本
• Language(): 用于 CPP、Python、Ruby、Markdown 等。
• NLTKTextSplitter(): 使用 NLTK（自然语言工具包）按句子分割文本。
• SpacyTextSplitter(): 使用 Spacy按句子的切割文本。

from langchain_text_splitters import RecursiveCharacterTextSplitter
text_splitter = RecursiveCharacterTextSplitter(
	chunk_size = 500,
	chunk_overlap = 50
)
text_splitter.split_text(pdf_page.page_content[0:1000])
print(text_splitter)
split_docs = text_splitter.split_documents(pdf_pages)
print(f"切分后的文件数量：{len(split_docs)}")
print(f"切分后的字符数（可以用来大致评估 token 数）：{sum([len(doc.page_content) for doc in split_docs])}")

4. 搭建并使用向量数据库

4.1 前序配置

1. 收集所有数据

import os

file_paths = []
folder_path = './data_base/knowledge_db'
for root, dirs, files in os.walk(folder_path):
	for file in files:
		file_path = os.path.join(root, file)
		file_paths.append(file_path)
print(file_paths)

2. 加载所有数据

from langchain_community.document_loaders import PyMuPDFLoader
from langchain_community.document_loaders import UnstructuredMarkdownLoader

loaders = []
for file_path in file_paths:
	file_type = file_path.split('.')[-1]
	if file_type == 'pdf':
		loaders.append(PyMuPDFLoader(file_path))
	elif file_type == 'md':
		loaders.append(UnstructuredMarkdownLoader(file_path))
texts = []
for loader in loaders:
	texts.extend(loader.load()) #extend将返回的list展开然后放进texts
# 每个元素变量类型为`langchain_core.documents.base.Document`
text = texts[1]
print(f"每一个元素的类型：{type(text)}.", 
	f"该文档的描述性数据：{text.metadata}", 
	f"查看该文档的内容:\n{text.page_content[0:]}", sep="\n------\n")

3. 切分文档

from langchain_text_splitters import RecursiveCharacterTextSplitter
text_splitter = RecursiveCharacterTextSplitter( 
	chunk_size=500, 
	chunk_overlap=50
) 
split_docs = text_splitter.split_documents(texts)

4.2 构建Chroma向量库

Langchain 集成了超过 30 个不同的向量存储库。我们选择 Chroma 是因为它轻量级且数据存储在内存中，这使得它非常容易启动和开始使用。

#from langchain.embeddings.openai import OpenAIEmbeddings
from langchain_openai import OpenAIEmbeddings
embedding = OpenAIEmbeddings(
    model="BAAI/bge-m3",
    openai_api_key=os.environ["SILICONFLOW_API_KEY"],
    openai_api_base="https://api.siliconflow.cn/v1",
    check_embedding_ctx_length=False
)
persist_directory = './data_base/my_vector_db/chroma'

from langchain_community.vectorstores import Chroma
vectordb = Chroma.from_documents(
	documents = split_docs,
	embedding = embedding,
	persist_directory=persist_directory
)
print(f"向量库中存储的数量：{vectordb._collection.count()}")

• 注意，由于这里使用的是 BAAI/bge-m3，而 OpenAIEmbeddings 默认会尝试按 OpenAI 模型的方式检查上下文长度并使用 tiktoken 分词。由于 tiktoken 不认识 BAAI/bge-m3，运行时会出现类似 Warning: model not found. Using cl100k_base encoding. 的警告，并回退到 OpenAI 的 cl100k_base 编码，这可能导致传给 BGE-M3 的输入处理方式不正确，进而影响向量检索效果。因此这里设置 check_embedding_ctx_length=False，并使用新的langchain_openai ， 让 LangChain 不再用 OpenAI tokenizer 预处理文本，而是直接将原始文本交给 SiliconFlow 的 BGE-M3 进行向量化。

4.3 向量检索

Chroma的相似度搜索使用的是余弦距离。当你需要数据库返回严谨的按余弦相似度排序的结果时可以使用similarity_search函数。

question="什么是大语言模型"
sim_docs = vectordb.similarity_search(question,k=3)
print(f"检索到的内容数：{len(sim_docs)}")
for i, sim_doc in enumerate(sim_docs): 
	print(f"检索到的第{i}个内容: \n{sim_doc.page_content[:200]}", end="\n--------------\n")

如果只考虑检索出内容的相关性会导致内容过于单一，可能丢失重要信息。最大边际相关性 (MMR, Maximum marginal relevance) 可以帮助我们在保持相关性的同时，增加内容的丰富度。核心思想是在已经选择了一个相关性高的文档之后，再选择一个与已选文档相关性较低但是信息丰富的文档。这样可以在保持相关性的同时，增加内容的多样性，避免过于单一的结果。

mmr_docs = vectordb.max_marginal_relevance_search(question,k=3)
for i, sim_doc in enumerate(mmr_docs):
    print(f"MMR 检索到的第{i}个内容: \n{sim_doc.page_content[:200]}", end="\n--------------\n")

赏

谢谢你请我吃糖果