1 向量数据库

1.1 向量数据库简介

向量数据库就是一种专门用于存储和处理向量数据的数据库系统，传统的关系型数据库通常不擅长处理向量数据，因为它们需要将数据映射为结构化的表格形式，而向量数据的维度较高、结构复杂，导致传统数据库存储和查询效率低下，所以向量数据库应运而生。

1.2 传统数据库与向量数据库的差异

传统数据库采用基于行的存储方式，传统数据库将数据存储为行记录，每一行包含多个字段，并且每个字段都有固定的列。传统数据库通常使用索引来提高查询性能，例如下方就是一个典型的传统数据库表格

这种方式在处理结构化数据时非常高效，但在处理非结构化或半结构化数据时效率低下。

向量数据库将数据以列形式存储，即每个列都有一个独立的存储空间，这使得向量数据库可以更加灵活地处理复杂的数据结构。向量数据库还可以进行列压缩（稀疏矩阵），以减少存储空间和提高数据的访问速度。
并且在向量数据库中，将数据表示为高维向量，其中每个向量对应于数据点。这些向量之间的距离表示它们之间的相似性。这种方式使得非结构化或半结构化数据的存储和检索变得更加高效。

以电影数据库为例，我们可以将每部电影表示为一个特征向量。假设我们使用四个特征来描述每部电影：动作、冒险、爱情、科幻。每个特征都可以在0到1的范围内进行标准化，表示该电影在该特征上的强度。

例如，电影”阿凡达”的向量表示可以是 [0.9, 0.8, 0.2, 0.9]，其中数字分别表示动作、冒险、爱情、科幻的特征强度。其他电影也可以用类似的方式表示。这些向量可以存储在向量数据库中，如下所示：

现在，如果我们想要查找与电影”阿凡达”相似的电影，我们可以计算向量之间的距离，找到最接近的向量，从而实现相似性匹配，而无需复杂的SQL查询。这就像使用地图找到两个地点之间的最短路径一样简单。

1.3 传统数据库与向量数据库优缺点

1.4 相似度搜索算法

1.4.1 余弦相似度与欧氏距离

在向量数据库中，支持通过多种方式来计算两个向量的相似度，例如：余弦相似度、欧式距离、曼哈顿距离、闵可夫斯基距离、汉明距离、Jaccard相似度等多种。其中最常见的就是余弦相似度和欧式距离。
例如下图，左侧就是欧式距离，右侧就是余弦相似度：

余弦相似度主要用于衡量向量在方向上的相似性，特别适用于文本、图像和高维空间中的向量。它不受向量长度的影响，只考虑方向的相似程度，余弦相似度的计算公式如下（计算两个向量夹角的余弦值，取值范围为[-1, 1]）：
欧式距离衡量向量之间的直线距离，得到的值可能很大，最小为 0，通常用于低维空间或需要考虑向量各个维度之间差异的情况。欧氏距离较小的向量被认为更相似，欧式距离的计算公式如下：

1.4.2 相似性搜索加速算法

在向量数据库中，数据按列进行存储，通常会将多个向量组织成一个 M×N 的矩阵，其中 M 是向量的维度（特征数），N 是向量的数量（数据库中的条目数），这个矩阵可以是稠密或者稀疏的，取决于向量的稀疏性和具体的存储优化策略。

这样计算相似性搜索时，本质上就变成了向量与 M×N 矩阵的每一行进行相似度计算，这里可以用到大量成熟的加速算法：

1. 矩阵分解方法：

· SVD（奇异值分解）：可以通过奇异值分解将原始矩阵转换为更低秩的矩阵表示，从而减少计算量。
· PCA（主成分分析）：类似地，可以通过主成分分析将高维矩阵映射到低维空间，减少计算复杂度。

索引结构和近似算法：

· LSH（局部敏感哈希）：LSH 可以在近似相似度匹配中加速计算，特别适用于高维稀疏向量的情况。
· ANN（近似最近邻）算法：ANN 算法如KD-Tree、Ball-Tree等可以用来加速对最近邻搜索的计算，虽然主要用于向量空间，但也可以部分应用于相似度计算中。

GPU 加速：使用图形处理单元（GPU）进行并行计算可以显著提高相似度计算的速度，尤其是对于大规模数据和高维度向量。
分布式计算：由于行与行之间独立，所以可以很便捷地支持分布式计算每行与向量的相似度，从而加速整体计算过程。

向量数据库底层除了在算法层面上针对相似性搜索做了大量优化，在存储结构、索引机制等方面均做了大量的优化，这才使得向量数据库在处理高维数据和实现快速相似性搜索上展示出巨大的优势

1.5 向量数据库的配置和使用

按照部署方式和提供的服务类型进行划分，向量数据库可以划分成几种：
1. 本地文件向量数据库：用户将向量数据存储到本地文件系统中，通过数据库查询的接口来检索向量数据，例如：Faiss。
2. 本地部署 API 向量数据库：这类数据库不仅允许本地部署，而且提供了方便的 API 接口，使用户可以通过网络请求来访问和查询向量数据，这类数据库通常提供了更复杂的功能和管理选项，例如：Milvus、Annoy、Weaviate 等。
3. 云端 API 向量数据库：将向量数据存储在云端，通过 API 提供向量数据的访问和管理功能，例如：TCVectorDB、Pinecone 等。

1.5.1 Faiss 向量数据库

1.5.1.1 Faiss 基本使用

Faiss 是 Facebook 团队开源的向量检索工具，针对高维空间的海量数据，提供高效可靠的相似性检索方式，被广泛用于推荐系统、图片和视频搜索等业务。Faiss 支持 Linux、macOS 和 Windows 操作系统，在百万级向量的相似性检索表现中，Faiss 能实现 < 10ms 的响应（需牺牲搜索准确度）。
CPU环境下使用

1	pip install faiss-cpu

GPU环境下使用并且已经安装了CUDA，则可以使用GPU版本

1	pip install faiss-gpu

代码示例：

import dotenv  
from langchain_community.vectorstores import FAISS  
# from langchain_openai import OpenAIEmbeddings  
from langchain_huggingface import HuggingFaceEmbeddings  
  
dotenv.load_dotenv()  
  
# embedding = OpenAIEmbeddings(model="text-embedding-3-small")  
  
embeddings = HuggingFaceEmbeddings(  
    model_name="sentence-transformers/all-MiniLM-L12-v2",  
    cache_folder="../22-其他Embedding嵌入模型的配置与使用/embeddings/"  
)  
  
# db = FAISS.from_texts([  
#     "笨笨是一只很喜欢睡觉的猫咪",  
#     "我喜欢在夜晚听音乐，这让我感到放松。",  
#     "猫咪在窗台上打盹，看起来非常可爱。",  
#     "学习新技能是每个人都应该追求的目标。",  
#     "我最喜欢的食物是意大利面，尤其是番茄酱的那种。",  
#     "昨晚我做了一个奇怪的梦，梦见自己在太空飞行。",  
#     "我的手机突然关机了，让我有些焦虑。",  
#     "阅读是我每天都会做的事情，我觉得很充实。",  
#     "他们一起计划了一次周末的野餐，希望天气能好。",  
#     "我的狗喜欢追逐球，看起来非常开心。",  
# ], embeddings)  
#  
# print(db.index.ntotal)  
# print(db.save_local("vector-store/"))  
  
db = FAISS.load_local("vector-store/", embeddings, allow_dangerous_deserialization=True)  
  
print(db.similarity_search_with_score("我养了一只猫，叫笨笨"))

1.5.1.2 删除指定数据

在 Faiss 中，支持删除向量数据库中特定的数据，目前仅支持传入数据条目 id 进行删除，并不支持条件筛选（但是可以通过条件筛选找到符合的数据，然后提取 id 列表，然后批量删除）。

代码示例：

print("删除前数量:", db.index.ntotal)
# 获取向量数据库的索引id列表信息
db.delete([db.index_to_docstore_id[0]])
print("删除后数量:", db.index.ntotal)

输出结果：

1 2	删除前数量: 10 删除后数量: 9

1.5.1.3 带过滤的相似性搜索

在绝大部分向量数据库中，除了存储向量数据，还支持存储对应的元数据，这里的元数据可以是文本原文、扩展信息、页码、归属文档id、作者、创建时间等等任何自定义信息，一般在向量数据库中，会通过元数据来实现对数据的检索。

1	向量数据库记录 = 向量(vector)+元数据(metadata)+id

Faiss 原生并不支持过滤，所以在 LangChain 封装的 FAISS 中对过滤功能进行了相应的处理。首先获取比 k 更多的结果 fetch_k（默认为 20 条），然后先进行搜索，接下来再搜索得到的 fetch_k 条结果上进行过滤，得到 k 条结果，从而实现带过滤的相似性搜索。
而且 Faiss 的搜索都是针对元数据的，在 Faiss 中执行带过滤的相似性搜索非常简单，只需要在搜索时传递 filter 参数即可，filter 可以传递一个元数据字典，也可以接收一个函数（函数的参数为元数据字典，返回值为布尔值）。

import dotenv  
from langchain_community.vectorstores import FAISS  
# from langchain_openai import OpenAIEmbeddings  
from langchain_huggingface import HuggingFaceEmbeddings  
  
dotenv.load_dotenv()  
  
# embedding = OpenAIEmbeddings(model="text-embedding-3-small")  
  
embeddings = HuggingFaceEmbeddings(  
    model_name="sentence-transformers/all-MiniLM-L12-v2",  
    cache_folder="../22-其他Embedding嵌入模型的配置与使用/embeddings/"  
)  
  
metadatas: list = [  
    {"page": 1},  
    {"page": 2},  
    {"page": 3},  
    {"page": 4},  
    {"page": 5},  
    {"page": 6},  
    {"page": 7},  
    {"page": 8},  
    {"page": 9},  
    {"page": 10},  
]  
  
db = FAISS.from_texts([  
    "笨笨是一只很喜欢睡觉的猫咪",  
    "我喜欢在夜晚听音乐，这让我感到放松。",  
    "猫咪在窗台上打盹，看起来非常可爱。",  
    "学习新技能是每个人都应该追求的目标。",  
    "我最喜欢的食物是意大利面，尤其是番茄酱的那种。",  
    "昨晚我做了一个奇怪的梦，梦见自己在太空飞行。",  
    "我的手机突然关机了，让我有些焦虑。",  
    "阅读是我每天都会做的事情，我觉得很充实。",  
    "他们一起计划了一次周末的野餐，希望天气能好。",  
    "我的狗喜欢追逐球，看起来非常开心。",  
], embeddings, metadatas)  
  
print(db.index_to_docstore_id)  
print(db.similarity_search_with_score("我养了一只猫，叫笨笨", filter=lambda x: x["page"] > 5))

1.5.2 Pinecone 向量数据库

1.5.2.1 Pinecone 配置

Pinecone 是一个托管的、云原生的向量数据库，具有极简的 API，并且无需在本地部署即可快速使用，Pinecone 服务提供商还为每个账户设置了足够的免费空间，在开发阶段，可以快速基于 Pinecone 快速开发 AI 应用。
相关资料：
1. Pinecone 官网：https://www.pinecone.io/
2. Pinecone 翻译文档：https://www.pinecone-io.com/
3. langchain-pinecone 翻译文档：http://imooc-langchain.shortvar.com/docs/integrations/vectorstores/pinecone/
Pinecone 向量数据库的设计架构与 Faiss 差异较大，Pinecone 由于是一个面向商业端的向量数据库，在功能和概念上会更加丰富，有几个核心概念+架构图如下：

概念的解释如下：
1. 组织：组织是使用相同结算方式的一个或者多个项目的集合，例如个人账号、公司账号等都算是一个组织。
2. 项目：项目是用来管理向量数据库、索引、硬件资源等内容的整合，可以将不同的项目数据进行区分。
3. 索引：索引是 Pinecone 中数据的最高组织单位，在索引中需要定义向量的存储维度、查询时使用的相似性指标，并且在 Pinecone 中支持两种类型的索引：无服务器索引（根据数据大小自动扩容）和 Pod 索引（预设空间/硬件）。
4. 命名空间：命名空间是索引内的分区，用于将索引中的数据区分成不同的组，以便于在不同的组内存储不同的数据，例如知识库、记忆的数据可以存储到不同的组中，类似 Excel 中的 Sheet表。
5. 记录：记录是数据的基本单位，一条记录涵盖了 ID、向量(values)、元数据(metadata) 等。

所以在 Pinecone 中使用向量数据库，要确保组织、项目、索引、命名空间、记录等内容均配置好才可以使用，并且由于 Pinecone 是云端向量数据库，使用时还需配置对应的 API 秘钥（可在注册好 Pinecone 后管理页面的 API Key 中设置）。
对于 Pinecone，LangChain 团队也封装了响应的包，安装命令：

1	pip install -U langchain-pinecone

然后在 .env 文件中配置对应的 API 秘钥，如下

1	PINECONE_API_KEY=xxx

1.5.2.2 Pinecone 使用

import dotenv  
# from langchain_openai import OpenAIEmbeddings  
from langchain_pinecone import PineconeVectorStore  
from langchain_huggingface import HuggingFaceEmbeddings  
  
dotenv.load_dotenv()  
  
# embedding = OpenAIEmbeddings(model="text-embedding-3-small")  
  
embeddings = HuggingFaceEmbeddings(  
    model_name="sentence-transformers/all-MiniLM-L12-v2",  
    cache_folder="../22-其他Embedding嵌入模型的配置与使用/embeddings/"  
)  
  
texts: list = [  
    "笨笨是一只很喜欢睡觉的猫咪",  
    "我喜欢在夜晚听音乐，这让我感到放松。",  
    "猫咪在窗台上打盹，看起来非常可爱。",  
    "学习新技能是每个人都应该追求的目标。",  
    "我最喜欢的食物是意大利面，尤其是番茄酱的那种。",  
    "昨晚我做了一个奇怪的梦，梦见自己在太空飞行。",  
    "我的手机突然关机了，让我有些焦虑。",  
    "阅读是我每天都会做的事情，我觉得很充实。",  
    "他们一起计划了一次周末的野餐，希望天气能好。",  
    "我的狗喜欢追逐球，看起来非常开心。",  
]  
metadatas: list = [  
    {"page": 1},  
    {"page": 2},  
    {"page": 3},  
    {"page": 4},  
    {"page": 5},  
    {"page": 6, "account_id": 1},  
    {"page": 7},  
    {"page": 8},  
    {"page": 9},  
    {"page": 10},  
]  
db = PineconeVectorStore(index_name="llmops", embedding=embeddings, namespace="dataset",  
                         pinecone_api_key="pcsk_Qz5bt_JMBCg1A6oJPbnceUnhwYf6CA1M57kBTxgVTDda96FkwCECAAhwPYrUvyytinYE2")  
db.add_texts(texts, metadatas, namespace="dataset")  
  
query = "我养了一只猫，叫笨笨"  
print(db.similarity_search_with_relevance_scores(query))

1.5.3 Weaviate 向量数据库

1.5.3.1 Weaviate 介绍

Weaviate 是完全使用 Go 语言构建的开源向量数据库，提供了强大的数据存储和检索功能。并且 Weaviate 提供了多种部署方式，以满足不同用户和用例的需求，部署方式如下：
1. Weaviate 云：使用 Weaviate 官方提供的云服务，支持数据复制、零停机更新、无缝扩容等功能，适用于评估、开发和生产场景。
2. Docker 部署：使用 Docker 容器部署 Weaviate 向量数据库，适用于评估和开发等场景。
3. K8s 部署：在 K8s 上部署 Weaviate 向量数据库，适用于开发和生产场景。
4. 嵌入式 Weaviate：基于本地文件的方式构建 Weaviate 向量数据库，适用于评估场景，不过嵌入式 Weaviate 只适用于 Linux、macOS 系统，在 Windows 下不支持。

Weaviate 和 Pinecone/TCVectorDB 一样，也存在着集合的概念，在 Weaviate 中集合类似传统关系型数据库中的表，负责管理一类数据/数据对象，要使用 Weaviate 的流程其实也非常简单：
1. 创建部署 Weaviate 数据库（使用 Weaviate 云、Docker 部署）。
2. 安装 Python 客户端/LangChain 集成包。
3. 连接 Weaviate（本地连接、云端连接）。
4. 创建数据集/集合（代码创建、可视化管理界面创建），在 Weaviate 中，集合的名字必须以大写字母开头，并且只能包含字母、数字和下划线，否则创建的时候会出错，和 Python 的类名规范几乎一致。
5. 添加数据/向量。
6. 相似性搜索/带过滤器的相似性搜索。

参考资料：
1. Weaviate 官网：https://weaviate.io/
2. Weaviate 快速上手指南：https://weaviate.io/developers/weaviate/quickstart

LangChain Weaviate 集成包翻译文档：https://imooc-langchain.shortvar.com/docs/integrations/vectorstores/weaviate

1.5.3.2 Weaviate 向量数据库的使用

Docker 部署 Weaviate 向量数据库：

1	docker run -d --name weaviate-dev -p 8080:8080 -p 50051:50051 cr.weaviate.io/semitechnologies/weaviate:1.24.20

创建好 Weaviate 数据库服务后，接下来就可以安装 Python 客户端/LangChain 集成包，命令如下：

1	pip install -Uqq langchain-weaviate

如果使用的是 Weaviate 云服务，可以直接从可视化界面创建 Collection，亦或者在使用时 LangChain 自动检测对应的数据集是否存在，如果不存在则直接创建。
然后就可以考虑连接 Weaviate 服务了，Weaviate 框架针对不同的部署方式提供的不同的连接方法：
1. weaviate.connect_to_local()：连接到本地的部署服务，需配置连接 URL、端口号。
2. weaviate.connect_to_wcs()：连接到远程的 Weaviate 服务，需配置连接 URL、连接秘钥。

代码示例：

import weaviate

# 连接192.168.2.120:8080并创建weaviate客户端
client = weaviate.connect_to_local("192.168.2.120", "8080")

连接到远程的 Weaviate 服务代码如下

import weaviate
from weaviate.auth import AuthApiKey
client = weaviate.connect_to_wcs(
    cluster_url="https://2j9jgyhprd2yej3c3rwog.c0.us-west3.gcp.weaviate.cloud",
    auth_credentials=AuthApiKey("BAn9bGZdZbdGCmUyfdegQoKFctyMmxaQdDFb")
)

创建好客户端后，接下来可以基于客户端创建 LangChain 向量数据库实例，在实例化 LangChain VectorDB 时，需要传递 client（客户端）、 index_name（集合名字）、text（原始文本的存储键）、embedding（文本嵌入模型），如下

import dotenv
import weaviate
from langchain_openai import OpenAIEmbeddings
from langchain_weaviate import WeaviateVectorStore

dotenv.load_dotenv()

# 1.连接weaviate向量数据库
client = weaviate.connect_to_local("192.168.2.120", "8080")
# 2.实例化WeaviateVectorStore
embedding = OpenAIEmbeddings(model="text-embedding-3-small")
db = WeaviateVectorStore(client=client, index_name="DatasetTest", text_key="text", embedding=embedding)

实例化 LangChain VectorDB 后，就可以像 Faiss、Pinecone、TCVectorDB 一样去使用了，例如执行新增数据后完成检索示例如下

import dotenv
import weaviate
from langchain_openai import OpenAIEmbeddings
from langchain_weaviate import WeaviateVectorStore
dotenv.load_dotenv()

# 1.连接weaviate向量数据库
client = weaviate.connect_to_local("192.168.2.120", "8080")

# 2.实例化WeaviateVectorStore
embedding = OpenAIEmbeddings(model="text-embedding-3-small")
db = WeaviateVectorStore(client=client, index_name="dataset-test", text_key="text", embedding=embedding)

# 3.新增数据
ids = db.add_texts([
    "笨笨是一只很喜欢睡觉的猫咪",
    "我喜欢在夜晚听音乐，这让我感到放松。",
    "猫咪在窗台上打盹，看起来非常可爱。",
    "学习新技能是每个人都应该追求的目标。",
    "我最喜欢的食物是意大利面，尤其是番茄酱的那种。",
    "昨晚我做了一个奇怪的梦，梦见自己在太空飞行。",
    "我的手机突然关机了，让我有些焦虑。",
    "阅读是我每天都会做的事情，我觉得很充实。",
    "他们一起计划了一次周末的野餐，希望天气能好。",
    "我的狗喜欢追逐球，看起来非常开心。",
])

# 4.检索数据
print(db.similarity_search_with_score("笨笨"))

输出内容：
[(Document(page_content=’笨笨是一只很喜欢睡觉的猫咪’), 0.699999988079071), (Document(page_content=’猫咪在窗台上打盹，看起来非常可爱。’), 0.2090398222208023), (Document(page_content=’我的狗喜欢追逐球，看起来非常开心。’), 0.19787956774234772), (Document(page_content=’我的手机突然关机了，让我有些焦虑。’), 0.11435992270708084)]

在 Weaviate 中，也支持带过滤器的相似性筛选，并且 LangChain Weaviate 社区包并没有对筛选过滤器进行二次封装，所以直接传递原生的 weaviate 过滤器即可，参考文档：https://weaviate.io/developers/weaviate/search/filters

例如需要检索 page 属性大于等于 5 的所有数据，可以构建一个 filters 后传递给检索方法，如下：

1
2
3

from weaviate.classes.query import Filter
filters = Filter.by_property("page").greater_or_equal(5)
print(db.similarity_search_with_score("笨笨", filters=filters))

输出结果如下：

[(Document(page_content='我的狗喜欢追逐球，看起来非常开心。', metadata={'page': 10.0, 'account_id': None}), 0.699999988079071), (Document(page_content='我的手机突然关机了，让我有些焦虑。', metadata={'page': 7.0, 'account_id': None}), 0.4045487940311432), (Document(page_content='昨晚我做了一个奇怪的梦，梦见自己在太空飞行。', metadata={'page': 6.0, 'account_id': 1.0}), 0.318904846906662), (Document(page_content='我最喜欢的食物是意大利面，尤其是番茄酱的那种。', metadata={'page': 5.0, 'account_id': None}), 0.2671944797039032)]

如果想获取 Weaviate 原始集合的实例，可以通过 db._collection 快速获得，从而去执行一些原始操作，例如

from weaviate.classes.query import MetadataQuery

collection = db._collection

response = collection.query.near_text(
    query="a sweet German white wine",
    limit=2,
    target_vector="title_country",  # Specify the target vector for named vector collections
    return_metadata=MetadataQuery(distance=True)
)

for o in response.objects:
    print(o.properties)
    print(o.metadata.distance)

1.5.4 自定义向量数据库

向量数据库的发展非常迅猛，几乎间隔几天就有新的向量数据库发布，LangChain 不可能将所有向量数据库都进行集成，亦或者封装的包存在这一些 bug 或错误，这个时候就需要考虑创建自定义向量数据库，去实现特定的方法。

在 LangChain 实现自定义向量数据库的类有两种模式，一种是继承封装好的数据库类，一种是继承基类 VectorStore。前一种一般继承后重写部分方法进行扩展或者修复 bug，后面一种是对接新的向量数据库。

在 LangChain 中，继承 VectorStore 只需实现最基础的 3 个方法即可正常使用：
1. add_texts：将对应的数据添加到向量数据库中。
2. similarity_search：最基础的相似性搜索。
3. from_texts：从特定的文本列表、元数据列表中构建向量数据库。

其他方法因为使用频率并不高，VectorStore 并没有设置成虚拟方法，但是再没有实现的情况下，直接调用会报错，涵盖：
1. delete()：删除向量数据库中的数据。
2. _select_relevance_score_fn()：根据距离计算相似性得分函数。
3. similarity_search_with_score()：携带得分的相似性搜索函数。
4. similarity_search_by_vector()：传递向量进行相似性搜索。
5. max_marginal_relevance_search()：最大边界相似性搜索。
6. max_marginal_relevance_search_by_vector()：传递向量进行最大边界相关性搜索。

代码示例：

import uuid  
from typing import List, Optional, Any, Iterable, Type  
  
import dotenv  
import numpy as np  
from langchain_core.documents import Document  
from langchain_core.embeddings import Embeddings  
from langchain_core.vectorstores import VectorStore  
from langchain_openai import OpenAIEmbeddings  
  
  
class MemoryVectorStore(VectorStore):  
    """基于内存+欧几里得距离的向量数据库"""  
    store: dict = {}  # 存储向量的临时变量  
  
    def __init__(self, embedding: Embeddings):  
        self._embedding = embedding  
  
    def add_texts(self, texts: Iterable[str], metadatas: Optional[List[dict]] = None, **kwargs: Any) -> List[str]:  
        """将数据添加到向量数据库中"""  
        # 1.检测metadata的数据格式  
        if metadatas is not None and len(metadatas) != len(texts):  
            raise ValueError("metadatas格式错误")  
  
        # 2.将数据转换成文本嵌入/向量和ids  
        embeddings = self._embedding.embed_documents(texts)  
        ids = [str(uuid.uuid4()) for _ in texts]  
  
        # 3.通过for循环组装数据记录  
        for idx, text in enumerate(texts):  
            self.store[ids[idx]] = {  
                "id": ids[idx],  
                "text": text,  
                "vector": embeddings[idx],  
                "metadata": metadatas[idx] if metadatas is not None else {},  
            }  
  
        return ids  
  
    def similarity_search(self, query: str, k: int = 4, **kwargs: Any) -> List[Document]:  
        """传入对应的query执行相似性搜索"""  
        # 1.将query转换成向量  
        embedding = self._embedding.embed_query(query)  
  
        # 2.循环和store中的每一个向量进行比较，计算欧几里得距离  
        result = []  
        for key, record in self.store.items():  
            distance = self._euclidean_distance(embedding, record["vector"])  
            result.append({"distance": distance, **record})  
  
        # 3.排序，欧几里得距离越小越靠前  
        sorted_result = sorted(result, key=lambda x: x["distance"])  
  
        # 4.取数据，取k条数据  
        result_k = sorted_result[:k]  
  
        return [  
            Document(page_content=item["text"], metadata={**item["metadata"], "score": item["distance"]})  
            for item in result_k  
        ]  
  
    @classmethod  
    def from_texts(cls: Type["MemoryVectorStore"], texts: List[str], embedding: Embeddings,  
                   metadatas: Optional[List[dict]] = None,  
                   **kwargs: Any) -> "MemoryVectorStore":  
        """从文本和元数据中去构建向量数据库"""  
        memory_vector_store = cls(embedding=embedding)  
        memory_vector_store.add_texts(texts, metadatas, **kwargs)  
        return memory_vector_store  
  
    @classmethod  
    def _euclidean_distance(cls, vec1: list, vec2: list) -> float:  
        """计算两个向量的欧几里得距离"""  
        return np.linalg.norm(np.array(vec1) - np.array(vec2))  
  
  
dotenv.load_dotenv()  
  
# 1.创建初始数据与嵌入模型  
texts = [  
    "笨笨是一只很喜欢睡觉的猫咪",  
    "我喜欢在夜晚听音乐，这让我感到放松。",  
    "猫咪在窗台上打盹，看起来非常可爱。",  
    "学习新技能是每个人都应该追求的目标。",  
    "我最喜欢的食物是意大利面，尤其是番茄酱的那种。",  
    "昨晚我做了一个奇怪的梦，梦见自己在太空飞行。",  
    "我的手机突然关机了，让我有些焦虑。",  
    "阅读是我每天都会做的事情，我觉得很充实。",  
    "他们一起计划了一次周末的野餐，希望天气能好。",  
    "我的狗喜欢追逐球，看起来非常开心。",  
]  
metadatas = [  
    {"page": 1},  
    {"page": 2},  
    {"page": 3},  
    {"page": 4},  
    {"page": 5},  
    {"page": 6, "account_id": 1},  
    {"page": 7},  
    {"page": 8},  
    {"page": 9},  
    {"page": 10},  
]  
embedding = OpenAIEmbeddings(model="text-embedding-3-small")  
  
# 2.构建自定义向量数据库  
db = MemoryVectorStore(embedding=embedding)  
  
ids = db.add_texts(texts, metadatas)  
print(ids)  
  
# 3.执行检索  
print(db.similarity_search("笨笨是谁？"))

2 嵌入模型介绍和使用

2.1 嵌入模型介绍

要想使用向量数据库的相似性搜索，存储的数据必须是向量，那么如何将高维度的文字、图片、视频等非结构化数据转换成向量呢？这个时候就需要使用到 Embedding 嵌入模型了，例如下方就是 Embedding 嵌入模型的运行流程：

Embedding 模型是一种在机器学习和自然语言处理中广泛应用的技术，它旨在将高纬度的数据（如文字、图片、视频）映射到低纬度的空间。Embedding 向量是一个 N 维的实值向量，它将输入的数据表示成一个连续的数值空间中的点。这种嵌入可以是一个词、一个类别特征（如商品、电影、物品等）或时间序列特征等。
而且通过学习，Embedding 向量可以更准确地表示对应特征的内在含义，使几何距离相近的向量对应的物体有相近的含义，甚至对向量进行加减乘除算法都有意义！
一句话理解 Embedding：一种模型生成方法，可以将非结构化的数据，例如文本/图片/视频等数据映射成有意义的向量数据。

目前生成 embedding 方法的模型有以下 4 类：
1. Word2Vec（词嵌入模型）：这个模型通过学习将单词转化为连续的向量表示，以便计算机更好地理解和处理文本。Word2Vec 模型基于两种主要算法 CBOW 和 Skip-gram。
2. Glove：一种用于自然语言处理的词嵌入模型，它与其他常见的词嵌入模型（如 Word2Vec 和 FastText）类似，可以将单词转化为连续的向量表示。GloVe 模型的原理是通过观察单词在语料库中的共现关系，学习得到单词之间的语义关系。具体来说，GloVe 模型将共现概率矩阵表示为两个词向量之间的点积和偏差的关系，然后通过迭代优化来训练得到最佳的词向量表示。
GloVe 模型的优点是它能够在大规模语料库上进行有损压缩，得到较小维度的词向量，同时保持了单词之间的语义关系。这些词向量可以被用于多种自然语言处理任务，如词义相似度计算、情感分析、文本分类等。
3. FastText：一种基于词袋模型的词嵌入技术，与其他常见的词嵌入模型（如 Word2Vec 和 GloVe）不同之处在于，FastText考虑了单词的子词信息。其核心思想是将单词视为字符的 n-grams 的集合，在训练过程中，模型会同时学习单词级别和n-gram级别的表示。这样可以捕捉到单词内部的细粒度信息，从而更好地处理各种形态和变体的单词。
4. 大模型 Embeddings（重点）：和大模型相关的嵌入模型，如 OpenAI 官方发布的第二代模型：text-embedding-ada-002。它最长的输入是 8191 个tokens，输出的维度是 1536。

2.3 Embedding 的价值

1. 降维：在许多实际问题中，原始数据的维度往往非常高。例如，在自然语言处理中，如果使用 Token 词表编码来表示词汇，其维度等于词汇表的大小，可能达到数十万甚至更高。通过 Embedding，我们可以将这些高维数据映射到一个低维空间，大大减少了模型的复杂度。
2. 捕捉语义信息：Embedding 不仅仅是降维，更重要的是，它能够捕捉到数据的语义信息。例如，在词嵌入中，语义上相近的词在向量空间中也会相近。这意味着Embedding可以保留并利用原始数据的一些重要信息。
3. 适应性：与一些传统的特征提取方法相比，Embedding 是通过数据驱动的方式学习的。这意味着它能够自动适应数据的特性，而无需人工设计特征。
4. 泛化能力：在实际问题中，我们经常需要处理一些在训练数据中没有出现过的数据。由于Embedding能够捕捉到数据的一些内在规律，因此对于这些未见过的数据，Embedding仍然能够给出合理的表示。
5. 可解释性：尽管 Embedding 是高维的，但我们可以通过一些可视化工具（如t-SNE）来观察和理解 Embedding 的结构。这对于理解模型的行为，以及发现数据的一些潜在规律是非常有用的。

2.4 CacheBackEmbedding 组件

通过嵌入模型计算传递数据的向量需要昂贵的算力，对于重复的内容，Embeddings 计算的结果肯定是一致的，如果数据重复仍然二次计算，会导致效率非常低，而且增加无用功。

所以在 LangChain 中提供了一个叫 CacheBackEmbedding 的包装类，一般通过类方法 from_bytes_store 进行实例化，它接受以下参数：
1. underlying_embedder：用于嵌入的嵌入模型。
2. document_embedding_cache：用于缓存文档嵌入的任何存储库（ByteStore）。
3. batch_size：可选参数，默认为 None，在存储更新之间要嵌入的文档数量。
4. namespace：可选参数，默认为“”，用于文档缓存的命名空间。此命名空间用于避免与其他缓存发生冲突。例如，将其设置为所使用的嵌入模型的名称。
5. query_embedding_cache：可选默认为 None 或者不缓存，用于缓存查询/文本嵌入的 ByteStore，或这是为 True 以使用与 document_embedding_cache 相同的存储。

CacheBackEmbedding 默认不缓存 embed_query 生成的向量，如果要缓存，需要设置 query_embedding_cache 的值，另外请尽可能设置 namespace，以避免使用不同嵌入模型嵌入的相同文本发生冲突。

代码示例：

import dotenv  
import numpy as np  
from langchain.embeddings import CacheBackedEmbeddings  
from langchain.storage import LocalFileStore  
from langchain_openai import OpenAIEmbeddings  
from numpy.linalg import norm  
  
dotenv.load_dotenv()  
  
def cosine_similarity(vector1: list, vector2: list) -> float:  
    """计算传入两个向量的余弦相似度"""  
    # 1.计算内积/点积  
    dot_product = np.dot(vector1, vector2)  
  
    # 2.计算向量的范数/长度  
    norm_vec1 = norm(vector1)  
    norm_vec2 = norm(vector2)  
  
    # 3.计算余弦相似度  
    return dot_product / (norm_vec1 * norm_vec2)  
  
  
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")  
embeddings_with_cache = CacheBackedEmbeddings.from_bytes_store(  
    embeddings,  
    LocalFileStore("./cache/"),  
    namespace=embeddings.model,  
    query_embedding_cache=True,  
)  
  
query_vector = embeddings_with_cache.embed_query("你好，我是慕小课，我喜欢打篮球")  
documents_vector = embeddings_with_cache.embed_documents([  
    "你好，我是慕小课，我喜欢打篮球",  
    "这个喜欢打篮球的人叫慕小课",  
    "求知若渴，虚心若愚"  
])  
  
print(query_vector)  
print(len(query_vector))  
  
print("============")  
  
print(len(documents_vector))  
print("vector1与vector2的余弦相似度:", cosine_similarity(documents_vector[0], documents_vector[1]))  
print("vector2与vector3的余弦相似度:", cosine_similarity(documents_vector[0], documents_vector[2]))

2.5 HuggingFace Embedding 模型的配置和使用

2.5.1 HuggingFace 本地模型

在某些对数据保密要求极高的场合下，数据不允许传递到外网，这个时候就可以考虑使用本地的文本嵌入模型——Hugging Face 本地嵌入模型，安装 langchain-huggingface 与 sentence-transformers 包，命令如下：

1	pip install -U langchain-huggingface sentence-transformers

其中 langchain-huggingface 是 langchain 团队基于 huggingface 封装的第三方社区包，sentence-transformers 是一个用于生成和使用预训练的文本嵌入，基于 transformer 架构，也是目前使用量最大的本地文本嵌入模型。
配置好后，就可以像正常的文本嵌入模型一样使用了，示例

from langchain_huggingface import HuggingFaceEmbeddings  
  
embeddings = HuggingFaceEmbeddings(  
    model_name="sentence-transformers/all-MiniLM-L12-v2",  
    cache_folder="./embeddings/"  
)  

query_vector = embeddings.embed_query("你好，我是慕小课，我喜欢打篮球游泳")  
  
print(query_vector)  
print(len(query_vector))

2.5.2 HuggingFace远程嵌入模型

部分模型的文件比较大，如果只是短期内调试，可以考虑使用 HuggingFace 提供的远程嵌入模型，首先安装对应的依赖

1	pip install huggingface_hub

然后在 Hugging Face 官网（https://huggingface.co/) 的 setting 中添加对应的访问秘钥，并配置到 .env 文件中

1	HUGGINGFACEHUB_API_TOKEN=xxx

接下来就可以使用 HuggingFace 提供的推理服务，这样在本地服务器上就无需配置对应的文本嵌入模型了。

import dotenv  
from langchain_huggingface import HuggingFaceEndpointEmbeddings  
  
dotenv.load_dotenv()  
  
embeddings = HuggingFaceEndpointEmbeddings(model="sentence-transformers/all-MiniLM-L12-v2")  
  
query_vector = embeddings.embed_query("你好，我是慕小课，我喜欢打篮球游泳")  
  
print(query_vector)  
print(len(query_vector))

相关资料信息：
1. Hugging Face 官网：https://huggingface.co/
2. HuggingFace 嵌入文档：https://python.langchain.com/v0.2/docs/integrations/text_embedding/sentence_transformers/
3. HuggingFace 嵌入翻译文档：http://imooc-langchain.shortvar.com/docs/integrations/text_embedding/sentence_transformers/

3 文档加载器

3.1 Document 与文档加载器

Document 类是 LangChain 中的核心组件，这个类定义了一个文档对象的结构，涵盖了文本内容和相关的元数据，Document 也是文档加载器、文档分割器、向量数据库、检索器这几个组件之间交互传递的状态数据。
在 LangChain 中所有文档加载器的基类为 BaseLoader，封装了统一的 5 个方法：
1. load()/aload()：加载和异步加载文档，返回的数据为文档列表。
2. load_and_split()：传递分割器，加载并将大文档按照传入的分割器进行切割，返回的数据为分割后的文档列表。
3. lazy_load()/alazy_load()：懒加载和异步懒加载文档，返回的是一个迭代器，适用于传递的数据源有多份文档的情况，例如文件夹加载器，可以每次获得最新的加载文档，不需要等到所有文档都加载完毕。

在 LangChain 中封装了上百种文档加载器，几乎所有的文件都可以使用这些加载器完成数据的读取，而不需要手动去封装

代码示例：

from langchain_community.document_loaders import TextLoader  
  
# 1.构建加载器  
loader = TextLoader("./电商产品数据.txt", encoding="utf-8")  
  
# 2.加载数据  
documents = loader.load()  
  
print(documents)  
print(len(documents))  
print(documents[0].metadata)

3.2 内置文档加载器的使用技巧

LangChain 内置文档加载器文档：https://imooc-langchain.shortvar.com/docs/integrations/document_loaders/

3.2.1 Markdown 文档加载器

LangChain 中封装了一个 UnstructuredMarkdownLoader 对象，要使用这个加载器，必须安装 unstructured 包，安装命令

1	pip install unstructured

代码示例：

from langchain_community.document_loaders import UnstructuredMarkdownLoader  
  
loader = UnstructuredMarkdownLoader("./项目API资料.md")  
documents = loader.load()  
  
print(documents)  
print(len(documents))  
print(documents[0].metadata)

3.2.2 Office 文档加载器

from langchain_community.document_loaders import (  
    UnstructuredPowerPointLoader,  
)  
  
# excel_loader = UnstructuredExcelLoader("./员工考勤表.xlsx", mode="elements")  
# excel_documents = excel_loader.load()  
  
# word_loader = UnstructuredWordDocumentLoader("./喵喵.docx")  
# documents = word_loader.load()  
  
ppt_loader = UnstructuredPowerPointLoader("./章节介绍.pptx")  
documents = ppt_loader.load()  
  
print(documents)  
print(len(documents))  
print(documents[0].metadata)

3.2.3 通用文档加载器

在实际的 LLM 应用开发中，由于数据的种类是无穷的，没办法单独为每一种数据配置一个加载器（也不现实），所以对于一些无法判断的数据类型或者想进行通用性文件加载，可以统一使用非结构化文件加载器 UnstructuredFileLoader 来实现对文件的加载。

UnstructuredFileLoader 是所有 UnstructuredXxxLoader 文档类的基类，其核心是将文档划分为元素，当传递一个文件时，库将读取文档，将其分割为多个部分，对这些部分进行分类，然后提取每个部分的文本，然后根据模式决定是否合并（single、paged、elements）。
一个 UnstructuredFileLoader 可以加载多种类型的文件，涵盖了：文本文件、PowerPoint 文件、HTML、PDF、图像、Markdown、Excel、Word 等

例如通过检测文件的扩展名来加载不同的文件加载器，对于没校验到的文件类型，才考虑使用 UnstructuredFileLoader，如下

if file_extension in [".xlsx", ".xls"]:
    loader = UnstructuredExcelLoader(file_path)
elif file_extension == ".pdf":
    loader = UnstructuredPDFLoader(file_path)
elif file_extension in [".md", ".markdown"]:
    loader = UnstructuredMarkdownLoader(file_path)
elif file_extension in [".htm", "html"]:
    loader = UnstructuredHTMLLoader(file_path)
elif file_extension in [".docx", ".doc"]:
    loader = UnstructuredWordDocumentLoader(file_path)
elif file_extension == ".csv":
    loader = UnstructuredCSVLoader(file_path)
elif file_extension in [".ppt", ".pptx"]:
    loader = UnstructuredPowerPointLoader(file_path)
elif file_extension == ".xml":
    loader = UnstructuredXMLLoader(file_path)
else:
    loader = UnstructuredFileLoader(file_path) if is_unstructured else TextLoader(file_path)

3.2.4 自定义文档加载器

代码示例：

from typing import Iterator, AsyncIterator  
from langchain_core.document_loaders import BaseLoader  
from langchain_core.documents import Document  
    
class CustomDocumentLoader(BaseLoader):  
    """自定义文档加载器，将文本文件的每一行都解析成Document"""  
  
    def __init__(self, file_path: str) -> None:  
        self.file_path = file_path  
  
    def lazy_load(self) -> Iterator[Document]:  
        # 1.读取对应的文件  
        with open(self.file_path, encoding="utf-8") as f:  
            line_number = 0  
            # 2.提取文件的每一行  
            for line in f:  
                # 3.将每一行生成一个Document实例并通过yield返回  
                yield Document(  
                    page_content=line,  
                    metadata={"score": self.file_path, "line_number": line_number}  
                )  
                line_number += 1  
  
    async def alazy_load(self) -> AsyncIterator[Document]:  
        import aiofiles  
        async with aiofiles.open(self.file_path, encoding="utf-8") as f:  
            line_number = 0  
            async for line in f:  
                yield Document(  
                    page_content=line,  
                    metadata={"score": self.file_path, "line_number": line_number}  
                )  
                line_number += 1  
    
loader = CustomDocumentLoader("./喵喵.txt")  
documents = loader.load()  
  
print(documents)  
print(len(documents))  
print(documents[0].metadata)

lazy_load() 方法的两个核心步骤就是：读取文件数据、将文件数据解析成Document，并且绝大部分文档加载器都有着两个核心步骤，而且读取文件数据这个步骤大家都大差不差。

就像 *.md、*.txt、*.py 这类文本文件，甚至是 *.pdf、*.doc 等这类非文本文件，都可以使用同一个读取文件数据步骤将文件读取为二进制内容，然后在使用不同的解析逻辑来解析对应的二进制内容，所以很容易可以得出：

1	文档加载器 = 二进制数据读取 + 解析逻辑

因此，在项目开发中，如果大量配置自定义文档解析器的话，将解析逻辑与加载逻辑分离，维护起来会更容易，而且也更容易复用相应的逻辑（具体使用哪种方式取决于开发）。

这样原先的 DocumentLoader 运行流程就变成了如下：

4 文本分割器

4.1 DocumentTransformer 组件

在 LangChain 中针对文档的转换也统一封装了一个基类 BaseDocumentTransformer，所有涉及到文档的转换的类均是该类的子类，将大块文档切割成 chunk 分块的文档分割器也是 BaseDocumentTransformer 的子类实现。

BaseDocumentTransformer 基类封装了两个方法：
1. transform_documents()：抽象方法，传递文档列表，返回转换后的文档列表。
2. atransform_documents()：转换文档列表函数的异步实现，如果没有实现，则会委托 transform_documents() 函数实现。

在 LangChain 中，文档转换组件分成了两类：文档分割器(使用频率高)、文档处理转换器(使用频率低，老版本写法)。

1	pip install -qU langchain-text-splitters

4.2 字符分割器

在文档分割器中，最简单的分割器就是——字符串分割器，这个组件会基于给定的字符串进行分割，默认为 \n\n，并且在分割时会尽可能保证数据的连续性。分割出来每一块的长度是通过字符数来衡量的，使用起来也非常简单，实例化 CharacterTextSplitter 需传递多个参数，信息如下：

1. separator：分隔符，默认为 \n\n。
2. is_separator_regex：是否正则表达式，默认为 False。
3. chunk_size：每块文档的内容大小，默认为 4000。
4. chunk_overlap：块与块之间重叠的内容大小，默认为 200。
5. length_function：计算文本长度的函数，默认为 len。
6. keep_separator：是否将分隔符保留到分割的块中，默认为 False。
7. add_start_index：是否添加开始索引，默认为 False，如果是的话会在元数据中添加该切块的起点。
8. strip_whitespace：是否删除文档头尾的空白，默认为 True。

如果想将文档切割为不超过 500 字符，并且每块之间文本重叠 50 个字符，可以使用 CharacterTextSplitter 来实现，代码如下：

from langchain_community.document_loaders import UnstructuredMarkdownLoader  
from langchain_text_splitters import CharacterTextSplitter  
  
# 1.加载对应的文档  
loader = UnstructuredMarkdownLoader("./项目API文档.md")  
documents = loader.load()  
  
# 2.创建文本分割器  
text_splitter = CharacterTextSplitter(  
    separator="\n\n",  
    chunk_size=500,  
    chunk_overlap=50,  
    add_start_index=True,  
)  
  
# 3.分割文本  
chunks = text_splitter.split_documents(documents)  
  
for chunk in chunks:  
    print(f"块大小:{len(chunk.page_content)}, 元数据:{chunk.metadata}")  
  
print(len(chunks))

4.3 递归字符文本分割器

普通的字符文本分割器只能使用单个分隔符对文本内容进行划分，在划分的过程中，可能会出现文档块过小或者过大的情况，这会让 RAG 变得不可控，例如：
1. 文档块可能会变得非常大，极端的情况下某个块的内容长度可能就超过了 LLM 的上下文长度限制，这样这个文本块永远不会被引用到，相当于存储了数据，但是数据又丢失了。
2. 文档块可能会远远小于窗口大小，导致文档块的信息密度太低，块内容即使填充到 Prompt 中，LLM 也无法提取出有用的信息。

RecursiveCharacterTextSplitter，即递归字符串分割，这个分割器可以传递一组分隔符和设定块内容大小，根据分隔符的优先顺序对文本进行预分割，然后将小块进行合并，将大块进行递归分割，直到获得所需块的大小，最终这些文档块的大小并不能完全相同，但是仍然会逼近指定长度。
RecursiveCharacterTextSplitter 的分隔符参数默认为 ["\n\n", "\n", " ", ""]，即优先使用换两行的数据进行分割，然后在使用单个换行符，如果块内容还是太大，则使用空格，最后再拆分成单个字符。
所以如果使用默认参数，这个字符文本分割器最后得到的文档块长度一定不会超过预设的大小，但是仍然会有小概率出现远小于的情况（目前也没有很好的解决方案）。

代码示例：

from langchain_community.document_loaders import UnstructuredMarkdownLoader  
from langchain_text_splitters import RecursiveCharacterTextSplitter  
  
loader = UnstructuredMarkdownLoader("./项目API文档.md")  
documents = loader.load()  
  
text_splitter = RecursiveCharacterTextSplitter(  
    chunk_size=500,  
    chunk_overlap=50,  
    add_start_index=True,  
)  
  
chunks = text_splitter.split_documents(documents)  
  
for chunk in chunks:  
    print(f"块大小: {len(chunk.page_content)}, 元数据: {chunk.metadata}")

4.4 语义文档分割器

语义相似性分割器，SemanticChunker 在使用上和其他的文档分割器存在一些差异，并且该类并没有继承 TextSplitter，实例化参数含义如下：

1. embeddings：文本嵌入模型，在该分类器底层使用向量的余弦相似度来识别语句之间的相似性。
2. buffer_size：文本缓冲区大小，默认为 1，即在计算相似性时，该文本会叠加前后各 1 条文本，如果不够则不叠加（例如第 1 条和最后 1 条）。
3. add_start_index：是否添加起点索引，默认为 False。
4. breakpoint_threshold_type：断点阈值类型，默认为 percentile 即百分位
5. breakpoint_threshold_amount：断点阈值金额/得分。
6. number_of_chunks：分割后的文档块个数，默认为 None。
7. sentence_split_regex：句子切割正则，默认为 (?<=[.?!])\s+，即以英文的点、问号、感叹号切割语句，不同的文档需要传递不同的切割正则表达式。

代码示例：

import dotenv  
from langchain_community.document_loaders import UnstructuredFileLoader  
from langchain_experimental.text_splitter import SemanticChunker  
from langchain_openai import OpenAIEmbeddings  
  
dotenv.load_dotenv()  
  
# 1.构建加载器和文本分割器  
loader = UnstructuredFileLoader("./科幻短篇.txt")  
text_splitter = SemanticChunker(  
    embeddings=OpenAIEmbeddings(model="text-embedding-3-small"),  
    number_of_chunks=10,  
    add_start_index=True,  
    sentence_split_regex=r"(?<=[。？！.?!])"  
)  
  
# 2.加载文本与分割  
documents = loader.load()  
chunks = text_splitter.split_documents(documents)  
  
# 3.循环打印  
for chunk in chunks:  
    print(f"块大小: {len(chunk.page_content)}, 元数据: {chunk.metadata}")

4.5 自定义文档分割器

代码示例：

from typing import List  
  
import jieba.analyse  
from langchain_community.document_loaders import UnstructuredFileLoader  
from langchain_text_splitters import TextSplitter  
  
  
class CustomTextSplitter(TextSplitter):  
    """自定义文本分割器"""  
  
    def __init__(self, seperator: str, top_k: int = 10, **kwargs):  
        """构造函数，传递分割器还有需要提取的关键词数，默认为10"""  
        super().__init__(**kwargs)  
        self._seperator = seperator  
        self._top_k = top_k  
  
    def split_text(self, text: str) -> List[str]:  
        """传递对应的文本执行分割并提取分割数据的关键词，组成文档列表返回"""  
        # 1.根据传递的分隔符分割传入的文本  
        split_texts = text.split(self._seperator)  
  
        # 2.提取分割出来的每一段文本的关键词，数量为self._top_k个  
        text_keywords = []  
        for split_text in split_texts:  
            text_keywords.append(jieba.analyse.extract_tags(split_text, self._top_k))  
  
        # 3.将关键词使用逗号进行拼接组成字符串列表并返回  
        return [",".join(keywords) for keywords in text_keywords]  
  
  
# 1.创建加载器与分割器  
loader = UnstructuredFileLoader("./科幻短篇.txt")  
text_splitter = CustomTextSplitter("\n\n", 10)  
  
# 2.加载文档并分割  
documents = loader.load()  
chunks = text_splitter.split_documents(documents)  
  
# 3.循环遍历文档信息  
for chunk in chunks:  
    print(chunk.page_content)

4.6 非分割类型的文档分割器

在 LangChain 中，还存在另一种非分割类型的文档转换器，这类转换器也是传递文档列表并返回文档列表，一般是将某种文档按照需求转换成另外一种格式（例如：翻译文档、文档重排、HTML 转文本、文档元数据提取、文档转问答等）

4.6.1 问答转换器

在 RAG 的外挂知识库中，向量存储知识库中使用的文档通常以叙述或对话格式存储。但是，绝大部分用户的查询都是问题格式，所以如果我们在对文档进行向量化之前先将其转换为问答格式，可以在一定程度上增加检索相关文档的可能性，降低检索不相关文档的可能性。

这个技巧也是 RAG 应用开发中常见的一种优化策略，即将原始数据转换成 QA 数据后进行存储，除此之外，对于绝大部分 LLM 的微调，使用的也是 QA问答数据也可以考虑使用该问答转换器进行转换。

在 LangChain 中封装了 Doctran 库并实现了 DoctranQATransformer 类可以快捷实现该功能，这个库底层使用 OpenAI 的函数回调来实现对问答数据的提取，首先安装该库

1	pip install -U doctran

代码示例：

import dotenv  
from doctran import Doctran  
from langchain_community.document_transformers import DoctranQATransformer  
from langchain_core.documents import Document  
  
_ = Doctran  
  
dotenv.load_dotenv()  
  
# 1.构建文档列表  
page_content = """机密文件 - 仅供内部使用  
日期：2023年7月1日  
主题：各种话题的更新和讨论  
亲爱的团队，  
希望这封邮件能找到你们一切安好。在这份文件中，我想向你们提供一些重要的更新，并讨论需要我们关注的各种话题。请将此处包含的信息视为高度机密。  
安全和隐私措施  
作为我们不断致力于确保客户数据安全和隐私的一部分，我们已在所有系统中实施了强有力的措施。我们要赞扬IT部门的John Doe（电子邮件：john.doe@example.com）在增强我们网络安全方面的勤奋工作。未来，我们提醒每个人严格遵守我们的数据保护政策和准则。此外，如果您发现任何潜在的安全风险或事件，请立即向我们专门的团队报告，联系邮箱为security@example.com。  
人力资源更新和员工福利  
最近，我们迎来了几位为各自部门做出重大贡献的新团队成员。我要表扬Jane Smith（社保号：049-45-5928）在客户服务方面的出色表现。Jane一直受到客户的积极反馈。此外，请记住我们的员工福利计划的开放报名期即将到来。如果您有任何问题或需要帮助，请联系我们的人力资源代表Michael Johnson（电话：418-492-3850，电子邮件：michael.johnson@example.com）。  
营销倡议和活动  
我们的营销团队一直在积极制定新策略，以提高品牌知名度并推动客户参与。我们要感谢Sarah Thompson（电话：415-555-1234）在管理我们的社交媒体平台方面的杰出努力。Sarah在过去一个月内成功将我们的关注者基数增加了20%。此外，请记住7月15日即将举行的产品发布活动。我们鼓励所有团队成员参加并支持我们公司的这一重要里程碑。  
研发项目  
在追求创新的过程中，我们的研发部门一直在为各种项目不懈努力。我要赞扬David Rodriguez（电子邮件：david.rodriguez@example.com）在项目负责人角色中的杰出工作。David对我们尖端技术的发展做出了重要贡献。此外，我们希望每个人在7月10日定期举行的研发头脑风暴会议上分享他们的想法和建议，以开展潜在的新项目。  
请将此文档中的信息视为最机密，并确保不与未经授权的人员分享。如果您对讨论的话题有任何疑问或顾虑，请随时直接联系我。  
感谢您的关注，让我们继续共同努力实现我们的目标。  
此致，  
Jason Fan  
联合创始人兼首席执行官  
Psychic  
jason@psychic.dev"""  
documents = [Document(page_content=page_content)]  
  
# 2.构建问答转换器并转换  
qa_transformer = DoctranQATransformer(openai_api_model="gpt-3.5-turbo-16k")  
transformer_documents = qa_transformer.transform_documents(documents)  
  
# 3.输出内容  
for qa in transformer_documents[0].metadata.get("questions_and_answers"):  
    print("问答数据:", qa)

输出内容：

{'question': '文件日期是什么？', 'answer': '2023年7月1日'}
{'question': '文件主题是什么？', 'answer': '各种话题的更新和讨论'}
{'question': '谁是IT部门的网络安全负责人？', 'answer': 'John Doe（电子邮件：john.doe@example.com）'}
{'question': '如果发现安全风险或事件，应该向谁报告？', 'answer': '专门的团队，联系邮箱为security@example.com'}
{'question': '谁在客户服务方面表现出色？', 'answer': 'Jane Smith（社保号：049-45-5928）'}
{'question': '员工福利计划的开放报名期是什么时候？', 'answer': '即将到来'}
{'question': '人力资源代表的联系信息是什么？', 'answer': 'Michael Johnson（电话：418-492-3850，电子邮件：michael.johnson@example.com）'}
{'question': '谁在管理社交媒体平台方面做出了杰出努力？', 'answer': 'Sarah Thompson（电话：415-555-1234）'}
{'question': '产品发布活动的日期是什么时候？', 'answer': '7月15日'}
{'question': '谁在研发部门担任项目负责人角色？', 'answer': 'David Rodriguez（电子邮件：david.rodriguez@example.com）'}
{'question': '研发头脑风暴会议的日期是什么时候？', 'answer': '7月10日'}

4.6.2 翻译转换器

在 RAG 应用开发中，将文档通过嵌入/向量的方式进行比较的好处在于能跨语言工作，例如：你好，世界！、Hello, World! 和こんにちは、世界！分别是中英日三国的语言，但是因为语义相近，所以在向量空间中的位置也是非常接近的。

当一个 RAG 应用需要跨语言工作时，一般有两种策略：
1. 在将文档切块并嵌入存储到向量数据库时，同时将文档翻译成多国语言并进行相同的操作。
2. 在进行检索操作时，将检索出来的文档执行翻译功能，然后使用翻译后的文档。
这两种策略都涉及到一个功能，就是文档的翻译，或者是说将文档转换成另外一种形式的文档，这类操作其实和文档转换器的作用一模一样，所以可以考虑使用该组件来实现这个功能，LangChain 中针对翻译的转换器就提供了不少，例如 Doctran。

5 文档检索器

5.1 带得分阈值的相似性搜索

在 LangChain 的相似性搜索中，无论结果多不匹配，只要向量数据库中存在数据，一定会查找出相应的结果，在 RAG 应用开发中，一般是将高相似文档插入到 Prompt 中，所以可以考虑添加一个相似性得分阈值，超过该数值的部分才等同于有相似性。
在 similarity_search_with_relevance_scores() 函数中，可以传递 score_threshold 阈值参数，过滤低于该得分的文档。
例如没有添加阈值检索我养了一只猫，叫笨笨，示例与输出如下：

import dotenv  
from langchain_community.vectorstores import FAISS  
from langchain_core.documents import Document  
from langchain_openai import OpenAIEmbeddings  
  
dotenv.load_dotenv()  
  
embedding = OpenAIEmbeddings(model="text-embedding-3-small")  
  
documents = [  
    Document(page_content="笨笨是一只很喜欢睡觉的猫咪", metadata={"page": 1}),  
    Document(page_content="我喜欢在夜晚听音乐，这让我感到放松。", metadata={"page": 2}),  
    Document(page_content="猫咪在窗台上打盹，看起来非常可爱。", metadata={"page": 3}),  
    Document(page_content="学习新技能是每个人都应该追求的目标。", metadata={"page": 4}),  
    Document(page_content="我最喜欢的食物是意大利面，尤其是番茄酱的那种。", metadata={"page": 5}),  
    Document(page_content="昨晚我做了一个奇怪的梦，梦见自己在太空飞行。", metadata={"page": 6}),  
    Document(page_content="我的手机突然关机了，让我有些焦虑。", metadata={"page": 7}),  
    Document(page_content="阅读是我每天都会做的事情，我觉得很充实。", metadata={"page": 8}),  
    Document(page_content="他们一起计划了一次周末的野餐，希望天气能好。", metadata={"page": 9}),  
    Document(page_content="我的狗喜欢追逐球，看起来非常开心。", metadata={"page": 10}),  
]  
db = FAISS.from_documents(documents, embedding)  
  
print(db.similarity_search_with_relevance_scores("我养了一只猫，叫笨笨", score_threshold=0.4))

5.2 as_retriever() 检索器

在 LangChain 中，VectorStore 可以通过 as_retriever() 方法转换成检索器，在 as_retriever() 中可以传递一下参数：

1. search_type：搜索类型，支持 similarity(基础相似性搜索)、similarity_score_threshold(携带相似性得分+阈值判断的相似性搜索)、mmr(最大边际相关性搜索)。
2. search_kwargs：其他键值对搜索参数，类型为字典，例如：k、filter、score_threshold、fetch_k、lambda_mult 等，当搜索类型配置为 similarity_score_threshold 后，必须添加 score_threshold 配置选项，否则会报错，参数的具体信息要看 search_type 类型对应的函数配合使用。

并且由于检索器是 Runnable 可运行组件，所以可以使用 Runnable 组件的所有功能（组件替换、参数配置、重试、回退、并行等）。

例如将向量数据库转换成携带得分+阈值判断的相似性搜索，并设置得分阈值为0.5，数据条数为10条，代码示例如下：

import dotenv  
import weaviate  
from langchain_community.document_loaders import UnstructuredMarkdownLoader  
from langchain_openai import OpenAIEmbeddings  
from langchain_text_splitters import RecursiveCharacterTextSplitter  
from langchain_weaviate import WeaviateVectorStore  
from weaviate.auth import AuthApiKey  
  
dotenv.load_dotenv()  
  
# 1.构建加载器与分割器  
loader = UnstructuredMarkdownLoader("./项目API文档.md")  
text_splitter = RecursiveCharacterTextSplitter(  
    separators=["\n\n", "\n", "。|！|？", "\.\s|\!\s|\?\s", "；|;\s", "，|,\s", " ", "", ],  
    is_separator_regex=True,  
    chunk_size=500,  
    chunk_overlap=50,  
    add_start_index=True,  
)  
  
# 2.加载文档并分割  
documents = loader.load()  
chunks = text_splitter.split_documents(documents)  
  
# 3.将数据存储到向量数据库  
db = WeaviateVectorStore(  
    client=weaviate.connect_to_wcs(  
        cluster_url="https://eftofnujtxqcsa0sn272jw.c0.us-west3.gcp.weaviate.cloud",  
        auth_credentials=AuthApiKey("21pzYy0orl2dxH9xCoZG1O2b0euDeKJNEbB0"),  
    ),  
    index_name="DatasetDemo",  
    text_key="text",  
    embedding=OpenAIEmbeddings(model="text-embedding-3-small"),  
)  
db.add_documents(chunks)  
  
# 4.转换检索器（带阈值的相似性搜索，数据为10条，得分阈值为0.5）  
retriever = db.as_retriever(  
    search_type="similarity_score_threshold",  
    search_kwargs={"k": 10, "score_threshold": 0.5},  
)  
  
# 5.检索结果  
documents = retriever.invoke("关于配置接口的信息有哪些")  
  
print(list(document.page_content[:50] for document in documents))  
print(len(documents))

5.3 MMR 最大边际相关性

最大边际相关性（MMR，max_marginal_relevance_search）的基本思想是同时考量查询与文档的相关度，以及文档之间的相似度。相关度确保返回结果对查询高度相关，相似度则鼓励不同语义的文档被包含进结果集。具体来说，它计算每个候选文档与查询的相关度，并减去与已经入选结果集的文档的最大相似度，这样更不相似的文档会有更高分。

而在 LangChain 中MMR 的实现过程和 FAISS 的带过滤器的相似性搜索非常接近，同样也是先执行相似性搜索，并得到一个远大于 k 的结果列表，例如 fetch_k 条数据，然后对搜索得到的 fetch_k 条数据计算文档之间的相似度，通过加权得分找到最终的 k 条数据。

简单来说，MMR 就是在一大堆最相似的文档中查找最不相似的，从而保证结果多样化。

执行一个 MMR 最大边际相似性搜索需要的参数为：搜索语句、k条搜索结果数据、fetch_k条中间数据、多样性系数(0代表最大多样性，1代表最小多样性)，在 LangChain 中也是基于这个思想进行封装，max_marginal_relevance_search() 函数的参数如下：

1. query：搜索语句，类型为字符串，必填参数。
2. k：搜索的结果条数，类型为整型，默认为 4。
3. fetch_k：要传递给 MMR 算法的的文档数，默认为 20。
4. lambda_mult：函数系数，数值范围从0-1，底层计算得分 = lambda_mult *相关性 - (1 - lambda_mult)*相似性，所以 0 代表最大多样性、1 代表最小多样性。
5. kwargs：其他传递给搜索方法的参数，例如 filter 等，这个参数使用和相似性搜索类似，具体取决于使用的向量数据库。

import dotenv  
import weaviate  
from langchain_community.document_loaders import UnstructuredMarkdownLoader  
from langchain_openai import OpenAIEmbeddings  
from langchain_text_splitters import RecursiveCharacterTextSplitter  
from langchain_weaviate import WeaviateVectorStore  
from weaviate.auth import AuthApiKey  
  
dotenv.load_dotenv()  
  
# 1.构建加载器与分割器  
loader = UnstructuredMarkdownLoader("./项目API文档.md")  
text_splitter = RecursiveCharacterTextSplitter(  
    separators=["\n\n", "\n", "。|！|？", "\.\s|\!\s|\?\s", "；|;\s", "，|,\s", " ", "", ],  
    is_separator_regex=True,  
    chunk_size=500,  
    chunk_overlap=50,  
    add_start_index=True,  
)  
  
# 2.加载文档并分割  
documents = loader.load()  
chunks = text_splitter.split_documents(documents)  
  
# 3.将数据存储到向量数据库  
db = WeaviateVectorStore(  
    client=weaviate.connect_to_wcs(  
        cluster_url="https://eftofnujtxqcsa0sn272jw.c0.us-west3.gcp.weaviate.cloud",  
        auth_credentials=AuthApiKey("21pzYy0orl2dxH9xCoZG1O2b0euDeKJNEbB0"),  
    ),  
    index_name="DatasetDemo",  
    text_key="text",  
    embedding=OpenAIEmbeddings(model="text-embedding-3-small"),  
)  
  
# 4.执行最大边际相关性搜索（可以去除重复数据）  
# search_documents = db.similarity_search("关于应用配置的接口有哪些？")  # 会查到重复数据  
search_documents = db.max_marginal_relevance_search("关于应用配置的接口有哪些？")  
  
# 5.打印搜索的结果  
# print(list(document.page_content[:100] for document in search_documents))  
for document in search_documents:  
    print(document.page_content[:100])  
    print("===========")

5.4 检索器组件

在 LangChain 中，传递一段 query 并返回与这段文本相关联文档的组件被称为检索器，并且 LangChain 为所有检索器设计了一个基类——BaseRetriever，该类继承了 RunnableSerializable，所以该类是一个 Runnable 可运行组件，支持使用 Runnable 组件的所有配置，在 BaseRetriever 下封装了一些通用的方法，类图如下

其中 get_relevance_documents() 方法将在 0.3.0 版本开始被遗弃（老版本非 Runnable 写法），使用检索器的技巧也非常简单，按照特定的规则创建好检索器后（通过 as_retriever() 或者构造函数），调用 invoke() 方法即可。

并且针对所有向量数据库，LangChain 都配置了 as_retriever() 方法，便于快捷将向量数据库转换成检索器，不同的检索器传递的参数会有所差异，需要查看源码或者查看文档搭配使用，例如下方是一个向量数据库检索器的使用示例

import dotenv  
import weaviate  
from langchain_core.runnables import ConfigurableField  
from langchain_openai import OpenAIEmbeddings  
from langchain_weaviate import WeaviateVectorStore  
from weaviate.auth import AuthApiKey  
  
dotenv.load_dotenv()  
  
# 1.构建向量数据库  
db = WeaviateVectorStore(  
    client=weaviate.connect_to_wcs(  
        cluster_url="https://eftofnujtxqcsa0sn272jw.c0.us-west3.gcp.weaviate.cloud",  
        auth_credentials=AuthApiKey("21pzYy0orl2dxH9xCoZG1O2b0euDeKJNEbB0"),  
    ),  
    index_name="DatasetDemo",  
    text_key="text",  
    embedding=OpenAIEmbeddings(model="text-embedding-3-small"),  
)  
  
# 2.转换检索器  
retriever = db.as_retriever(  
    search_type="similarity_score_threshold",  
    search_kwargs={"k": 10, "score_threshold": 0.5},  
).configurable_fields(  
    search_type=ConfigurableField(id="db_search_type"),  
    search_kwargs=ConfigurableField(id="db_search_kwargs"),  
)  
  
# 3.修改运行时配置执行MMR搜索，并返回4条数据  
mmr_documents = retriever.with_config(  
    configurable={  
        "db_search_type": "mmr",  
        "db_search_kwargs": {  
            "k": 4,  
        }  
    }  
).invoke("关于应用配置的接口有哪些？")  
print("相似性搜索: ", mmr_documents)  
print("内容长度:", len(mmr_documents))  
  
print(mmr_documents[0].page_content[:20])  
print(mmr_documents[1].page_content[:20])

5.5 自定义检索器

在 LangChain 中实现自定义检索器的技巧其实非常简单，只需要继承 BaseRetriever 类，然后实现 _get_relevant_documents() 方法即可，从 query 到 list[document] 的逻辑全部都在这个函数内部实现，异步的方法也可以不需要实现，底层会委托同步方法来执行。

from typing import List  
  
from langchain_core.callbacks import CallbackManagerForRetrieverRun  
from langchain_core.documents import Document  
from langchain_core.retrievers import BaseRetriever  
    
class CustomRetriever(BaseRetriever):  
    """自定义检索器"""  
    documents: list[Document]  
    k: int  
  
    def _get_relevant_documents(self, query: str, *, run_manager: CallbackManagerForRetrieverRun) -> List[Document]:  
        """根据传入的query，获取相关联的文档列表"""  
        matching_documents = []  
        for document in self.documents:  
            if len(matching_documents) > self.k:  
                return matching_documents  
            if query.lower() in document.page_content.lower():  
                matching_documents.append(document)  
        return matching_documents  
  
  
# 1.定义预设文档  
documents = [  
    Document(page_content="笨笨是一只很喜欢睡觉的猫咪", metadata={"page": 1}),  
    Document(page_content="我喜欢在夜晚听音乐，这让我感到放松。", metadata={"page": 2}),  
    Document(page_content="猫咪在窗台上打盹，看起来非常可爱。", metadata={"page": 3}),  
    Document(page_content="学习新技能是每个人都应该追求的目标。", metadata={"page": 4}),  
    Document(page_content="我最喜欢的食物是意大利面，尤其是番茄酱的那种。", metadata={"page": 5}),  
    Document(page_content="昨晚我做了一个奇怪的梦，梦见自己在太空飞行。", metadata={"page": 6}),  
    Document(page_content="我的手机突然关机了，让我有些焦虑。", metadata={"page": 7}),  
    Document(page_content="阅读是我每天都会做的事情，我觉得很充实。", metadata={"page": 8}),  
    Document(page_content="他们一起计划了一次周末的野餐，希望天气能好。", metadata={"page": 9}),  
    Document(page_content="我的狗喜欢追逐球，看起来非常开心。", metadata={"page": 10}),  
]  
  
# 2.创建检索器  
retriever = CustomRetriever(documents=documents, k=3)  
  
# 3.调用检索器获取搜索结果并打印  
retriever_documents = retriever.invoke("猫")  
print(retriever_documents)  
print(len(retriever_documents))

6 RAG 优化策略

6.1 RAG 开发6个阶段优化策略

在 RAG 应用开发中，无论架构多复杂，接入了多少组件，使用了多少优化策略与特性，所有优化的最终目标都是提升LLM生成内容的准确性，而对于 Transformer架构类型的大模型来说，要实现这个目标，一般只需要 3 个步骤：

1. 传递更准确的内容：传递和提问准确性更高的内容，会让 LLM 能识别到关联的内容，生成的内容准确性更高。
2. 让重要的内容更靠前：GPT 模型的注意力机制会让传递 Prompt 中更靠前的内容权重更高，越靠后权重越低。
3. 尽可能不传递不相关内容：缩短每个块的大小，尽可能让每个块只包含关联的内容，缩小不相关内容的比例。

看起来很简单，但是目前针对这 3 个步骤 N 多研究员提出了不少方案，比较遗憾的是，目前也没有一种统一的方案，不同的场合仍然需要考虑不同的方案结合才能实现相对好一点的效果，并不是所有场合都适合配置很复杂的优化策略。

在 RAG 应用开发中，使用的优化策略越多，单次响应成本越高，性能越差，需要合理使用。映射到 RAG 中，其实就是切割合适的文档块、更准确的搜索语句、正确地排序文档、剔除重复无关的检索内容，所以在 RAG应用开发中，想进行优化，可以针对 query(提问查询)、TextSplitter(文本分割器)、VectorStore(向量数据库)、Retriever(检索器)、Prompt(基础prompt编写) 这几个组件。

在完整的 LLM 应用流程中拆解 RAG 开发阶段并进行优化看起来相对繁琐，可以考虑单独将 RAG 开发阶段的流程拎出来，并针对性对每个阶段进行优化与调整，按照不同的功能模块，共可以划分成 6 个阶段：查询转换、路由、查询构建、索引、检索和生成。

在 RAG 开发的 6 个阶段中，不同的阶段拥有不同的优化策略，需要针对不同的应用进行特定性的优化，目前市面上常见的优化方案有：问题转换、多路召回、混合检索、搜索重排、动态路由、图查询、问题重建、自检索等数十种优化策略，每种策略所在的阶段并不一致，效果也有差异，并且相互影响。
并且 RAG 优化和 LangChain 并没有关系，无论使用任何框架、任何编程语言，进行 RAG 开发时，掌握优化的思路才是最重要的！
将对应的优化策略整理到 RAG 运行流程中，优化策略与开发阶段对应图如下：