【AI 大模型】RAG 检索增强天生 ⑤ ( 向量数据库 | 向量数据库 索引结构和 ...

上一篇博客 【AI 大模型】RAG 检索增强天生 ④ ( 向量相似度计算 | 余弦间隔 | 欧式间隔 | OpenAI 文本向量模型 | 手动实现的 余弦相似度 和 欧氏间隔 函数计算 ) 中 , 讲解了 向量相似度 的计算方式 , 使用 OpenAI 的 text-embedding-ada-002 文本向量模型 天生了 一组文字的 文本向量 , 分别使用 余弦间隔 和 欧式间隔 计算了 文本向量 之间的相似度 ;




一、向量数据库

---

1、向量数据库引入


在上一篇博客 【AI 大模型】RAG 检索增强天生 ④ ( 向量相似度计算 | 余弦间隔 | 欧式间隔 | OpenAI 文本向量模型 | 手动实现的 余弦相似度 和 欧氏间隔 函数计算 ) 中 , 使用 向量模型 将文本转为 向量 , 如 : OpenAI 的 text-embedding-ada-002 文本向量模型 , 可以将文本转为 1536 维的浮点数向量值 ;
OpenAI 的 text-embedding-ada-002 文本向量模型 可以 跨语言进行相似度计算 , 如 : 计算 英文 和 中文 的相似度 ;
然后 , 通过 循环遍历 对比 目标文本向量 和 每个文本向量 的 余弦间隔 和 欧式间隔 , 得到一个间隔最短的文本 , 这样对比计算量非常大 ;
怎样快速检索出某个 文本 寄义相似的 目标文本 , 这里引入一个新的工具 " 向量数据库 " ;

2、向量数据库简介


向量数据库 ( Vector Database ) 是专门用于 存储、检索 和 管理 高维向量数据的 数据库体系 , 其核心能力是 快速 执行 向量相似性搜索 ;
向量数据库 可以 快速搜索到 与 目标文本 相似的 文本内容 ;

向量数据库 存储 :

• 向量存储 : 将 将 文本、图片、音视频 等数据 通过 机器学习模型 转换为 高维向量 , 然后 存储到 " 向量数据库 " 中 ;
  
• 向量压缩 : 向量数据库 中 使用了 向量压缩技能 , 可 节流 向量 存储空间 ;
  

向量相似度计算 : 在 向量数据库 中的 向量 , 通过 计算向量间隔 衡量相似性 , 如 : 欧氏间隔、余弦间隔 ;

向量数据库查询 : 使用 ANN 近似最近邻搜索 算法 在 高维向量空间 中快速查找与给定 向量点 最接近的 向量 , 该算法可以 在 包管一定检索精度的前提下 , 显著 提高了搜索服从 , 特别实用于处置惩罚大规模、高维数据集 ;

3、向量数据库 索引结构和搜索算法


技能范例代表算法/结构特点算法原理时间复杂度优缺点实用场景树结构KD-Tree, Ball-Tree基于空间划分，结构直观，支持准确搜索递归划分空间（KD-Tree按坐标轴划分，Ball-Tree按超球体划分）O(N log N)✅ 低维准确搜索快
❌ 高维性能急剧下降（维度灾难）维度<20的结构化数据检索哈希方法LSH (局部敏感哈希)牺牲精度换速率，哈希碰撞可控设计哈希函数使相似向量映射到相同桶的概率更高O(N)✅ 速率快、内存低
❌ 精度与哈希函数设计强相干快速去重、近似搜索初筛图索引HNSW, NSG小天下网络优化，层级化搜索路径构建多层概率图结构，通过邻居跳转实现高效近邻搜索O(log N)✅ 高召回率、支持动态更新
❌ 内存消耗较大大规模高维数据（图像/文本）量化方法PQ (乘积量化), SQ (标量量化)有损压缩，向量维度解耦计算将高维向量分解为子空间并分别量化，降低存储和计算复杂度O(N)✅ 内存占用降低80%+
❌ 量化误差导致精度损失十亿级向量内存优化场景倒排索引IVF (倒排文件体系)粗粒度筛选+细粒度比较先聚类（如K-means），搜索时仅扫描最近簇的向量O(√N)✅ 搜索速率提升显著
❌ 需预训练聚类中央配合PQ量化加快搜索混合结构DiskANN磁盘-内存分级存储，减少IO瓶颈基于SSD优化存储，联合图索引与量化技能O(log N)✅ 支持TB级数据
❌ 需要SSD硬件配合超大规模磁盘存储场景深度学习驱动Learned Index数据分布自顺应性，端到端优化使用神经网络预测向量分布，优化索引构建训练后O(1)✅ 自顺应数据分布
❌ 需要大量训练数据数据分布规律的专用场景
4、向量数据库 应用场景


向量数据库应用场景 :

• 保举体系 : 根据用户举动向量 匹配相似商品 ;
  
• 图像检索 : 输入一张图片 , 快速找到相似图片 ;
  
• 语义搜索 : 将文本转换为向量 , 实现 语义级搜索 ;
  
• 生物信息学 : 对 基因序列、蛋白质结构 进行 相似性匹配 ;
  

5、传统数据库 与 向量数据库 对比


对比维度传统数据库向量数据库核心数据范例结构化数据（表格、字段）非结构化数据的向量化表示（高维数值）主要查询方式准确匹配（SQL条件查询）相似性搜索（Top-K最近邻，ANN算法）索引结构B树、哈希索引HNSW图、IVF倒排索引、LSH哈希等性能瓶颈复杂JOIN、事件锁竞争高维向量计算服从与内存占用典范应用场景金融交易、用户管理等结构化业务图像/文本检索、保举体系、语义搜索扩展性垂直扩展（硬件升级）为主分布式架构，自然支持程度扩展处置惩罚规模百万~十亿级结构化记载十亿~万亿级高维向量查询耽误毫秒~秒级（依靠索引优化）亚毫秒~百毫秒级（ANN加快）数据同等性强同等性（ACID）终极同等性为主（分布式场景）代表体系MySQL, PostgreSQL, OracleMilvus, Pinecone, Faiss, Qdrant



二、常见 向量数据库 对比

---

名称开源云服务主要特点长处缺点使用场景FAISS是否专注高性能向量检索，需搭配其他存储使用速率快，支持大规模高维数据，得当研究场景无长期化存储，需自行处置惩罚数据管理研究项目、离线大规模相似性搜索（如保举体系原型）Pinecone否是全托管云服务，开箱即用简单易用，自动扩展，低运维本钱代价高，机动性低，仅支持云服务贸易应用快速摆设（如实时保举、语义搜索）Milvus是是分布式架构，支持海量数据，多索引范例扩展性强，功能全面，社区活跃自托管摆设复杂，资源消耗较高企业级生产环境（如十亿级向量搜索、AI平台后端）Weaviate是是内置模型向量化，支持混合搜索（向量+关键词）自带数据向量化，GraphQL接口机动模型依靠性强，自界说向量需额外设置语义增强搜索（如知识图谱、联合文本和向量的多模态检索）Qdrant是是高性能Rust实现，支持过滤查询低耽误，内存服从高，得当实时场景社区较小，文档相对较少高并发低耽误场景（如实时保举、流式数据处置惩罚）PGVector是否PostgreSQL扩展，支持SQL利用向量无缝兼容PostgreSQL，事件支持，混合查询性能低于专用库，大规模数据需优化已有PostgreSQL的项目添加向量搜索（如联合关系数据的保举体系）RediSearch是是基于Redis的向量检索，内存优先超低耽误，支持实时更新内存本钱高，不得当超大数据集实时性要求极高的场景（如实时个性化广告、会话式AI）ElasticSearch是是联合全文检索与向量搜索，成熟生态体系混合搜索能力强，社区资源丰富向量检索性能较弱，高维数据服从低文本+向量混合搜索（如搜索引擎增强、日志分析联合语义）



三、向量数据库 案例

---

1、安装 向量数据库 chromadb


执行 pip install chromadb 命令 , 安装 向量数据库 chromadb ;

使用 Python 语言 开辟 chromadb 向量数据库 流程如下 :

• 首先 , 需要 导入 chromadb 库 ;
  

1. import chromadb  # ChromaDB 向量数据库

复制代码

• 然后 , 创建 长期化的 Chroma 向量数据库 客户端 实例 ;
  

1. # 初始化 ChromaDB 客户端 (持久化到本地目录)
2. chroma_client = chromadb.PersistentClient(path="chroma_db")

复制代码

• 再后 , 获取聚集 , 相称于 向量数据库 表 ;
  

1. # 创建或获取集合 (相当于数据库表)
2. collection = chroma_client.get_or_create_collection(
3.     name="news_articles",  # 集合名称
4.     metadata={"hnsw:space": "cosine"}  # 使用余弦相似度
5. )

复制代码

• 再后 , 将 文本向量 插入数据库 ;
  

1. # 将文档插入数据库
2. collection.add(
3.     ids=document_ids,               # 唯一ID列表
4.     embeddings=document_embeddings, # 文本向量列表
5.     documents=documents             # 原始文本列表
6. )

复制代码

• 最后 , 查询 向量数据库 ;
  

1. # 执行相似性查询
2. results = collection.query(
3.     query_embeddings=[query_embedding],  # 查询向量
4.     n_results=2                          # 返回前2个最相似结果
5. )

复制代码

2、核心要点 解析



① 创建数据库实例


调用 PersistentClient 构造函数 , 创建 长期化的 Chroma 向量数据库 客户端 实例 , 返回 ClientAPI 实例 ;

1. def PersistentClient(
2.     path: str = "./chroma",  # 路径参数，指定Chroma数据保存的目录，默认为"./chroma"
3.     settings: Optional[Settings] = None,  # 设置参数，可选，默认为None
4.     tenant: str = DEFAULT_TENANT,  # 租户参数，指定此客户端要使用的租户，默认为默认租户
5.     database: str = DEFAULT_DATABASE,  # 数据库参数，指定此客户端要使用的数据库，默认为默认数据库
6. ) -> ClientAPI:
7.     """
8.     创建一个持久化的Chroma实例，并将其数据保存到磁盘。这对于测试和开发很有用，
9.     但不建议在生产环境中使用。
11.     参数：
12.         path: 保存Chroma数据的目录。默认值为"./chroma"。
13.         settings: 客户端的设置配置，如果为None，则使用默认设置。
14.         tenant: 此客户端要使用的租户。默认使用默认租户。
15.         database: 此客户端要使用的数据库。默认使用默认数据库。
16.     """
17.     if settings is None:
18.         settings = Settings()  # 如果设置参数为None，则创建一个默认的设置实例
19.     settings.persist_directory = path  # 设置持久化目录
20.     settings.is_persistent = True  # 标记设置为持久化
22.     # 确保参数是正确的类型 -- 用户可以传递任何类型的数据
23.     tenant = str(tenant)  # 将租户参数转换为字符串
24.     database = str(database)  # 将数据库参数转换为字符串
26.     return ClientCreator(tenant=tenant, database=database, settings=settings)  # 返回创建的客户端实例

复制代码

② 创建数据库表


ClientAPI#get_collection 函数 的 作用是 获取具有给定名称的聚集 , 相称于 创建 向量数据库 的 数据库表 , 得到 Collection 实例对象 ;

1. class ClientAPI(BaseAPI, ABC):
2.     tenant: str
3.     database: str
5. @abstractmethod
6.     def get_collection(
7.         self,
8.         name: str,  # 集合的名称
9.         embedding_function: Optional[
10.             EmbeddingFunction[Embeddable]
11.         ] = ef.DefaultEmbeddingFunction(),  # 嵌入函数，可选，用于将文档嵌入向量空间，默认为默认嵌入函数
12.         data_loader: Optional[DataLoader[Loadable]] = None,  # 数据加载器，可选，用于加载记录（文档、图像等）
13.     ) -> Collection:
14.         """获取具有给定名称的集合。
15.         
16.         参数：
17.             name: 要获取的集合的名称
18.             embedding_function: 可选的嵌入函数，用于将文档嵌入向量空间。
19.                                 如果未提供，则使用默认的嵌入函数。
20.             data_loader: 可选的数据加载器函数，用于加载记录（文档、图像等）。
22.         返回：
23.             Collection: 请求的集合
25.         引发：
26.             ValueError: 如果集合不存在
28.         示例：
29.             ```python
30.             client.get_collection("my_collection")
31.             # 返回一个集合，名称为"my_collection"，元数据为空
32.             ```
33.         """
34.         pass

复制代码

③ 向 向量数据库表 中 插入文本向量


将 文本向量 插入到 数据库表中 ;

1. class Collection(CollectionCommon["ServerAPI"]):
2. def add(
3.         self,
4.         ids: OneOrMany[ID],  # 要添加的嵌入的ID
5.         embeddings: Optional[
6.             Union[
7.                 OneOrMany[Embedding],
8.                 OneOrMany[PyEmbedding],
9.             ]
10.         ] = None,  # 要添加的嵌入向量，可选。如果为None，将根据集合中设置的嵌入函数基于文档或图像计算嵌入。
11.         metadatas: Optional[OneOrMany[Metadata]] = None,  # 与嵌入相关联的元数据，可选。在查询时，可以根据此元数据进行过滤。
12.         documents: Optional[OneOrMany[Document]] = None,  # 与嵌入相关联的文档，可选。
13.         images: Optional[OneOrMany[Image]] = None,  # 与嵌入相关联的图像，可选。
14.         uris: Optional[OneOrMany[URI]] = None,  # 与嵌入相关联的图像的URI，可选。
15.     ) -> None:
16.         """将嵌入添加到数据存储中。
17.         
18.         参数：
19.             ids: 您希望添加的嵌入的ID
20.             embeddings: 要添加的嵌入向量。如果为None，则将根据集合中设置的嵌入函数基于文档或图像计算嵌入。可选。
21.             metadatas: 与嵌入相关联的元数据。在查询时，您可以根据此元数据进行过滤。可选。
22.             documents: 与嵌入相关联的文档。可选。
23.             images: 与嵌入相关联的图像。可选。
24.             uris: 与嵌入相关联的图像的URI。可选。
26.         返回：
27.             None
29.         引发：
30.             ValueError: 如果您既没有提供嵌入也没有提供文档
31.             ValueError: 如果ids、embeddings、metadatas或documents的长度不匹配
32.             ValueError: 如果您没有提供嵌入函数且没有提供嵌入
33.             ValueError: 如果您同时提供了嵌入和文档
34.             ValueError: 如果您提供了一个已经存在的ID
35.         """
37.         # 验证并准备添加请求
38.         add_request = self._validate_and_prepare_add_request(
39.             ids=ids,
40.             embeddings=embeddings,
41.             metadatas=metadatas,
42.             documents=documents,
43.             images=images,
44.             uris=uris,
45.         )
47.         # 调用客户端的添加方法
48.         self._client._add(
49.             collection_id=self.id,  # 集合ID
50.             ids=add_request["ids"],  # 添加请求中的ID
51.             embeddings=add_request["embeddings"],  # 添加请求中的嵌入
52.             metadatas=add_request["metadatas"],  # 添加请求中的元数据
53.             documents=add_request["documents"],  # 添加请求中的文档
54.             uris=add_request["uris"],  # 添加请求中的URI
55.             tenant=self.tenant,  # 租户信息
56.             database=self.database,  # 数据库信息
57.         )

复制代码

④ 从 向量数据库表 中查询 相似数据


通过调用 Collection#query 函数 , 可以从 向量数据库表 中查询 相似数据 , 可设置查询指定个数的相似结果 ;

1. class Collection(CollectionCommon["ServerAPI"]):
2. def query(
3.         self,
4.         query_embeddings: Optional[
5.             Union[
6.                 OneOrMany[Embedding],
7.                 OneOrMany[PyEmbedding],
8.             ]
9.         ] = None,  # 要查询的嵌入向量，可选
10.         query_texts: Optional[OneOrMany[Document]] = None,  # 要查询的文档文本，可选
11.         query_images: Optional[OneOrMany[Image]] = None,  # 要查询的图像，可选
12.         query_uris: Optional[OneOrMany[URI]] = None,  # 用于数据加载器的URI，可选
13.         n_results: int = 10,  # 每个查询嵌入或文本要返回的邻居数量，可选，默认为10
14.         where: Optional[Where] = None,  # 用于过滤结果的Where类型字典，例如：`{"$and": [{"color" : "red"}, {"price": {"$gte": 4.20}}]}`，可选
15.         where_document: Optional[WhereDocument] = None,  # 用于通过文档过滤的WhereDocument类型字典，例如：`{$contains: {"text": "hello"}}`，可选
16.         include: Include = [
17.             IncludeEnum.metadatas,
18.             IncludeEnum.documents,
19.             IncludeEnum.distances,
20.         ],  # 要包含在结果中的内容列表，可以包含`"embeddings"`、`"metadatas"`、`"documents"`、`"distances"`。ID始终包括在内。默认为`["metadatas", "documents", "distances"]`，可选
21.     ) -> QueryResult:
22.         """获取提供的query_embeddings或query_texts的n_results个最近邻嵌入。
24.         参数：
25.             query_embeddings: 要获取最近邻的嵌入向量，可选。
26.             query_texts: 要获取最近邻的文档文本，可选。
27.             query_images: 要获取最近邻的图像，可选。
28.             query_uris: 用于数据加载器的URI，可选。
29.             n_results: 每个查询嵌入或文本要返回的邻居数量，可选，默认为10。
30.             where: 用于过滤结果的Where类型字典，例如：`{"$and": [{"color" : "red"}, {"price": {"$gte": 4.20}}]}`，可选。
31.             where_document: 用于通过文档过滤的WhereDocument类型字典，例如：`{$contains: {"text": "hello"}}`，可选。
32.             include: 要包含在结果中的内容列表，可以包含`"embeddings"`、`"metadatas"`、`"documents"`、`"distances"`。ID始终包括在内。默认为`["metadatas", "documents", "distances"]`，可选。
34.         返回：
35.             QueryResult: 包含结果的QueryResult对象。
37.         引发：
38.             ValueError: 如果您既没有提供query_embeddings，也没有提供query_texts，也没有提供query_images
39.             ValueError: 如果您同时提供了query_embeddings和query_texts
40.             ValueError: 如果您同时提供了query_embeddings和query_images
41.             ValueError: 如果您同时提供了query_texts和query_images
42.         """
44.         # 验证并准备查询请求
45.         query_request = self._validate_and_prepare_query_request(
46.             query_embeddings=query_embeddings,
47.             query_texts=query_texts,
48.             query_images=query_images,
49.             query_uris=query_uris,
50.             n_results=n_results,
51.             where=where,
52.             where_document=where_document,
53.             include=include,
54.         )
56.         # 调用客户端的查询方法
57.         query_results = self._client._query(
58.             collection_id=self.id,  # 集合ID
59.             query_embeddings=query_request["embeddings"],  # 查询请求中的嵌入
60.             n_results=query_request["n_results"],  # 查询请求中的邻居数量
61.             where=query_request["where"],  # 查询请求中的过滤条件
62.             where_document=query_request["where_document"],  # 查询请求中的文档过滤条件
63.             include=query_request["include"],  # 查询请求中要包含的内容
64.             tenant=self.tenant,  # 租户信息
65.             database=self.database,  # 数据库信息
66.         )
68.         # 转换查询响应
69.         return self._transform_query_response(
70.             response=query_results, include=query_request["include"]
71.         )

复制代码

3、完备代码示例


下面的代码是 在 博客 【AI 大模型】RAG 检索增强天生 ④ ( 向量相似度计算 | 余弦间隔 | 欧式间隔 | OpenAI 文本向量模型 | 手动实现的 余弦相似度 和 欧氏间隔 函数计算 ) 的 代码基础上 , 将 多少文本 计算出来的 文本向量 存储到 chromadb 向量数据库中 , 然后再从 向量数据库 中查询 相似的文本数据 ;

完备代码示例 :

1. # 导入所需库import chromadb  # ChromaDB 向量数据库
2. from openai import OpenAI  # OpenAI 客户端# 初始化 OpenAI 客户端 (替换成自己的 API 信息)client = OpenAI(    api_key="sk-i3dHqZygi7757aF6",  # 替换为你的 OpenAI API Key , 这里我把自己的 API-KEY 隐藏了    base_url="https://api.xiaoai.plus/v1"  # 替换为你的 API 端点)def get_embeddings(texts, model="text-embedding-ada-002"):    """使用 OpenAI 的嵌入模型将文本转换为向量"""    # 调用 OpenAI API 获取嵌入向量    response = client.embeddings.create(        input=texts,        model=model    )    # 从响应中提取向量数据    return [item.embedding for item in response.data]# 初始化 ChromaDB 客户端 (持久化到本地目录)
3. chroma_client = chromadb.PersistentClient(path="chroma_db")
4. # 创建或获取集合 (相当于数据库表)
5. collection = chroma_client.get_or_create_collection(
6.     name="news_articles",  # 集合名称
7.     metadata={"hnsw:space": "cosine"}  # 使用余弦相似度
8. )
9. # 原始文本数据documents = [    "李彦宏称大模型本钱每年降低90%",    "乌军大批直升机击落多架俄无人机",    "王力宏回应是否想找新朋侪",    "饺子不知道观众怎么想出的藕饼cp",    "加沙停火协议关键时刻生变",]# 天生文档向量document_embeddings = get_embeddings(documents)# 预备文档 ID (需要唯一标识符)document_ids = [str(i) for i in range(len(documents))]  # 天生 ["0", "1", ..., "4"]# 将文档插入数据库
10. collection.add(
11.     ids=document_ids,               # 唯一ID列表
12.     embeddings=document_embeddings, # 文本向量列表
13.     documents=documents             # 原始文本列表
14. )
15. # 查询文本query_text = "国际争端新闻"# 天生查询向量query_embedding = get_embeddings([query_text])[0]# 执行相似性查询
16. results = collection.query(
17.     query_embeddings=[query_embedding],  # 查询向量
18.     n_results=2                          # 返回前2个最相似结果
19. )
20. # 打印查询结果print("查询内容:", query_text)print("最相似结果:")for doc, score in zip(results['documents'][0], results['distances'][0]):    print(f"相似度 {score:.4f}: {doc}")

复制代码

执行结果 :

1. 查询内容: 国际争端新闻
2. 最相似结果:
3. 相似度 0.1806: 加沙停火协议关键时刻生变
4. 相似度 0.1922: 乌军大批直升机击落多架俄无人机

复制代码

下图是在本地天生 的 向量数据库 文件内容 ;

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。