【AI 大模型】RAG 检索增强生成 ⑤ ( 向量数据库 | 向量数据库索引结构和搜索算法 | 常见向量数据库对比

文章目录

一、向量数据库
- 1、向量数据库引入
- 2、向量数据库简介
- 3、向量数据库索引结构和搜索算法
- 4、向量数据库应用场景
- 5、传统数据库与向量数据库对比
二、常见向量数据库对比
三、向量数据库案例
- 1、安装向量数据库 chromadb
- 2、核心要点解析
- - ① 创建数据库实例
  - ② 创建数据库表
  - ③ 向向量数据库表中插入文本向量
  - ④ 从向量数据库表中查询相似数据
- 3、完整代码示例

在这里插入图片描述

上一篇博客【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, Oracle	Milvus, 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 库 ;

import chromadb  # ChromaDB 向量数据库

然后 , 创建持久化的 Chroma 向量数据库客户端实例 ;

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

再后 , 获取集合 , 相当于向量数据库表 ;

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

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

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

最后 , 查询向量数据库 ;

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

2、核心要点解析

① 创建数据库实例

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

def PersistentClient(
    path: str = "./chroma",  # 路径参数，指定Chroma数据保存的目录，默认为"./chroma"
    settings: Optional[Settings] = None,  # 设置参数，可选，默认为None
    tenant: str = DEFAULT_TENANT,  # 租户参数，指定此客户端要使用的租户，默认为默认租户
    database: str = DEFAULT_DATABASE,  # 数据库参数，指定此客户端要使用的数据库，默认为默认数据库
) -> ClientAPI:
    """
    创建一个持久化的Chroma实例，并将其数据保存到磁盘。这对于测试和开发很有用，
    但不建议在生产环境中使用。

    参数：
        path: 保存Chroma数据的目录。默认值为"./chroma"。
        settings: 客户端的设置配置，如果为None，则使用默认设置。
        tenant: 此客户端要使用的租户。默认使用默认租户。
        database: 此客户端要使用的数据库。默认使用默认数据库。
    """
    if settings is None:
        settings = Settings()  # 如果设置参数为None，则创建一个默认的设置实例
    settings.persist_directory = path  # 设置持久化目录
    settings.is_persistent = True  # 标记设置为持久化

    # 确保参数是正确的类型 -- 用户可以传递任何类型的数据
    tenant = str(tenant)  # 将租户参数转换为字符串
    database = str(database)  # 将数据库参数转换为字符串

    return ClientCreator(tenant=tenant, database=database, settings=settings)  # 返回创建的客户端实例

② 创建数据库表

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

class ClientAPI(BaseAPI, ABC):
    tenant: str
    database: str

@abstractmethod
    def get_collection(
        self,
        name: str,  # 集合的名称
        embedding_function: Optional[
            EmbeddingFunction[Embeddable]
        ] = ef.DefaultEmbeddingFunction(),  # 嵌入函数，可选，用于将文档嵌入向量空间，默认为默认嵌入函数
        data_loader: Optional[DataLoader[Loadable]] = None,  # 数据加载器，可选，用于加载记录（文档、图像等）
    ) -> Collection:
        """获取具有给定名称的集合。
        
        参数：
            name: 要获取的集合的名称
            embedding_function: 可选的嵌入函数，用于将文档嵌入向量空间。
                                如果未提供，则使用默认的嵌入函数。
            data_loader: 可选的数据加载器函数，用于加载记录（文档、图像等）。

        返回：
            Collection: 请求的集合

        引发：
            ValueError: 如果集合不存在

        示例：
            ```python
            client.get_collection("my_collection")
            # 返回一个集合，名称为"my_collection"，元数据为空
            ```
        """
        pass

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

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

class Collection(CollectionCommon["ServerAPI"]):
def add(
        self,
        ids: OneOrMany[ID],  # 要添加的嵌入的ID
        embeddings: Optional[
            Union[
                OneOrMany[Embedding],
                OneOrMany[PyEmbedding],
            ]
        ] = None,  # 要添加的嵌入向量，可选。如果为None，将根据集合中设置的嵌入函数基于文档或图像计算嵌入。
        metadatas: Optional[OneOrMany[Metadata]] = None,  # 与嵌入相关联的元数据，可选。在查询时，可以根据此元数据进行过滤。
        documents: Optional[OneOrMany[Document]] = None,  # 与嵌入相关联的文档，可选。
        images: Optional[OneOrMany[Image]] = None,  # 与嵌入相关联的图像，可选。
        uris: Optional[OneOrMany[URI]] = None,  # 与嵌入相关联的图像的URI，可选。
    ) -> None:
        """将嵌入添加到数据存储中。
        
        参数：
            ids: 您希望添加的嵌入的ID
            embeddings: 要添加的嵌入向量。如果为None，则将根据集合中设置的嵌入函数基于文档或图像计算嵌入。可选。
            metadatas: 与嵌入相关联的元数据。在查询时，您可以根据此元数据进行过滤。可选。
            documents: 与嵌入相关联的文档。可选。
            images: 与嵌入相关联的图像。可选。
            uris: 与嵌入相关联的图像的URI。可选。

        返回：
            None

        引发：
            ValueError: 如果您既没有提供嵌入也没有提供文档
            ValueError: 如果ids、embeddings、metadatas或documents的长度不匹配
            ValueError: 如果您没有提供嵌入函数且没有提供嵌入
            ValueError: 如果您同时提供了嵌入和文档
            ValueError: 如果您提供了一个已经存在的ID
        """

        # 验证并准备添加请求
        add_request = self._validate_and_prepare_add_request(
            ids=ids,
            embeddings=embeddings,
            metadatas=metadatas,
            documents=documents,
            images=images,
            uris=uris,
        )

        # 调用客户端的添加方法
        self._client._add(
            collection_id=self.id,  # 集合ID
            ids=add_request["ids"],  # 添加请求中的ID
            embeddings=add_request["embeddings"],  # 添加请求中的嵌入
            metadatas=add_request["metadatas"],  # 添加请求中的元数据
            documents=add_request["documents"],  # 添加请求中的文档
            uris=add_request["uris"],  # 添加请求中的URI
            tenant=self.tenant,  # 租户信息
            database=self.database,  # 数据库信息
        )

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

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

class Collection(CollectionCommon["ServerAPI"]):
def query(
        self,
        query_embeddings: Optional[
            Union[
                OneOrMany[Embedding],
                OneOrMany[PyEmbedding],
            ]
        ] = None,  # 要查询的嵌入向量，可选
        query_texts: Optional[OneOrMany[Document]] = None,  # 要查询的文档文本，可选
        query_images: Optional[OneOrMany[Image]] = None,  # 要查询的图像，可选
        query_uris: Optional[OneOrMany[URI]] = None,  # 用于数据加载器的URI，可选
        n_results: int = 10,  # 每个查询嵌入或文本要返回的邻居数量，可选，默认为10
        where: Optional[Where] = None,  # 用于过滤结果的Where类型字典，例如：`{"$and": [{"color" : "red"}, {"price": {"$gte": 4.20}}]}`，可选
        where_document: Optional[WhereDocument] = None,  # 用于通过文档过滤的WhereDocument类型字典，例如：`{$contains: {"text": "hello"}}`，可选
        include: Include = [
            IncludeEnum.metadatas,
            IncludeEnum.documents,
            IncludeEnum.distances,
        ],  # 要包含在结果中的内容列表，可以包含`"embeddings"`、`"metadatas"`、`"documents"`、`"distances"`。ID始终包括在内。默认为`["metadatas", "documents", "distances"]`，可选
    ) -> QueryResult:
        """获取提供的query_embeddings或query_texts的n_results个最近邻嵌入。

        参数：
            query_embeddings: 要获取最近邻的嵌入向量，可选。
            query_texts: 要获取最近邻的文档文本，可选。
            query_images: 要获取最近邻的图像，可选。
            query_uris: 用于数据加载器的URI，可选。
            n_results: 每个查询嵌入或文本要返回的邻居数量，可选，默认为10。
            where: 用于过滤结果的Where类型字典，例如：`{"$and": [{"color" : "red"}, {"price": {"$gte": 4.20}}]}`，可选。
            where_document: 用于通过文档过滤的WhereDocument类型字典，例如：`{$contains: {"text": "hello"}}`，可选。
            include: 要包含在结果中的内容列表，可以包含`"embeddings"`、`"metadatas"`、`"documents"`、`"distances"`。ID始终包括在内。默认为`["metadatas", "documents", "distances"]`，可选。

        返回：
            QueryResult: 包含结果的QueryResult对象。

        引发：
            ValueError: 如果您既没有提供query_embeddings，也没有提供query_texts，也没有提供query_images
            ValueError: 如果您同时提供了query_embeddings和query_texts
            ValueError: 如果您同时提供了query_embeddings和query_images
            ValueError: 如果您同时提供了query_texts和query_images
        """

        # 验证并准备查询请求
        query_request = self._validate_and_prepare_query_request(
            query_embeddings=query_embeddings,
            query_texts=query_texts,
            query_images=query_images,
            query_uris=query_uris,
            n_results=n_results,
            where=where,
            where_document=where_document,
            include=include,
        )

        # 调用客户端的查询方法
        query_results = self._client._query(
            collection_id=self.id,  # 集合ID
            query_embeddings=query_request["embeddings"],  # 查询请求中的嵌入
            n_results=query_request["n_results"],  # 查询请求中的邻居数量
            where=query_request["where"],  # 查询请求中的过滤条件
            where_document=query_request["where_document"],  # 查询请求中的文档过滤条件
            include=query_request["include"],  # 查询请求中要包含的内容
            tenant=self.tenant,  # 租户信息
            database=self.database,  # 数据库信息
        )

        # 转换查询响应
        return self._transform_query_response(
            response=query_results, include=query_request["include"]
        )

3、完整代码示例

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

完整代码示例 :

# 导入所需库
import chromadb  # ChromaDB 向量数据库
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 客户端 (持久化到本地目录)
chroma_client = chromadb.PersistentClient(path="chroma_db")

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

# 原始文本数据
documents = [
    "李彦宏称大模型成本每年降低90%",
    "乌军大批直升机击落多架俄无人机",
    "王力宏回应是否想找新伴侣",
    "饺子不知道观众怎么想出的藕饼cp",
    "加沙停火协议关键时刻生变",
]

# 生成文档向量
document_embeddings = get_embeddings(documents)

# 准备文档 ID (需要唯一标识符)
document_ids = [str(i) for i in range(len(documents))]  # 生成 ["0", "1", ..., "4"]

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

# 查询文本
query_text = "国际争端新闻"

# 生成查询向量
query_embedding = get_embeddings([query_text])[0]

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

# 打印查询结果
print("查询内容:", query_text)
print("最相似结果:")
for doc, score in zip(results['documents'][0], results['distances'][0]):
    print(f"相似度 {score:.4f}: {doc}")

执行结果 :

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

在这里插入图片描述

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

在这里插入图片描述