【向量检索】之向量数据库Milvus,Faiss详解及应用案例_向量搜索

Reference

https://www.modb.pro/db/509268

笔记︱几款多模态向量检索引擎：Faiss 、milvus、Proxima、vearch、Jina等 - 知乎 (zhihu.com)

向量数据库入坑指南：聊聊来自元宇宙大厂 Meta 的相似度检索技术 Faiss - 苏洋的文章 - 知乎

常用的三种索引方式及原理-CSDN

向量搜索应用

向量检索技术，其主要的应用领域如人脸识别、推荐系统、图片搜索、视频指纹、语音处理、自然语言处理、文件搜索

背景

向量检索需要用到向量数据库，而普通的关系型数据库（如 MySQL、PostgreSQL）通常不适合处理向量检索任务。这是因为**【向量检索有其特殊的需求和性能要求】**。

原因如下：

1、向量检索的特性

• 高维度数据：向量通常是高维数据，可能有数百到数千个维度。处理高维数据需要特殊的数据结构和算法，以确保检索的效率。
• 相似性度量：向量检索基于相似性度量（如欧氏距离、余弦相似度等）来查找相似向量。计算高维向量之间的相似性需要高效的算法，而普通数据库没有优化这些操作。
• 近似最近邻搜索（ANN）：在大规模数据集上，精确最近邻搜索的计算开销非常大，因此通常使用近似最近邻搜索算法（如 LSH、HNSW、IVF 等）来提高检索效率。向量数据库通常内置了这些算法。

2、普通数据库的限制

• 索引结构：普通数据库的索引（如 B 树、哈希索引等）设计用于处理标量数据（如整数、字符串等），而不是高维向量。它们无法有效地支持高维相似性检索。
• 查询优化：普通数据库的查询优化器针对 SQL 查询进行了优化，但对高维向量相似性检索的优化有限，可能导致查询性能低下。
• 扩展性：处理大规模向量数据需要分布式存储和计算能力，普通数据库在这方面的支持不如专门设计的向量数据库。

3、向量数据库的优势

向量数据库（如 FAISS、Milvus、Annoy、Weaviate 等）专门为向量检索设计，具备以下优势：

• 高效的索引结构：支持多种高效的索引结构（如 HNSW、IVF、PQ 等），能够快速处理高维向量的相似性检索。
• 优化的相似性计算：针对高维向量的相似性计算进行了优化，能够在大规模数据集上高效执行最近邻搜索。
• 扩展性和分布式支持：许多向量数据库支持分布式存储和计算，能够处理海量向量数据，并在多个节点之间均衡负载。

4、结论

向量数据库通过专门的索引结构、优化的相似性计算以及分布式支持，显著提高了高维向量相似性检索的效率和扩展性。因此，对于需要处理大规模向量数据的应用场景（如推荐系统、图像搜索、自然语言处理等），使用向量数据库是更好的选择。

索引类型

——from ChatGPT

相似度指标

• Euclidean distance (L2): This metric is generally used in the field of computer vision (CV).
• Inner product (IP): This metric is generally used in the field of natural language processing (NLP). The metrics that are widely used for binary embeddings include:
• Hamming: This metric is generally used in the field of natural language processing (NLP).
• Jaccard: This metric is generally used in the field of molecular similarity search.
• Tanimoto: This metric is generally used in the field of molecular similarity search.
• Superstructure: This metric is generally used to search for similar superstructure of a molecule.
• Substructure: This metric is generally used to search for similar substructure of a molecule.

向量数据库

以下是主流向量数据库的对比：

—— ChatGPT

Milvus 是一个功能强大、支持多模态和分布式架构的向量数据库，适用于需要处理大规模、多样化数据的场景。

FAISS 则以其高性能和灵活性著称，特别适合需要高效向量检索和 GPU 加速的场景。选择合适的向量数据库需要根据具体的应用需求、数据规模和性能要求来决定。

Milvus

【向量数据库】

【支持高效的向量检索】Milvus 采用了高效的算法和数据结构，如倒排索引、HNSW、量化等，优化搜索性能和存储效率，适合于机器学习、人工智能和相似性搜索的应用场景。

【高度可扩展】Milvus 支持容器化部署（如 Kubernetes）

【易用性和灵活性】【强大的社区支持】

官方Doc

https://milvus.io/docs/v2.2.x

背景

Milvus 是由 Zilliz 开发并开源的高性能分布式向量数据库，最初是由阿里巴巴孵化，旨在处理大规模、多模态的向量检索任务。Milvus 是 LF AI & Data 基金会的托管项目之一。

功能

• 多模态向量检索：支持文本、图像、视频、音频等多种数据类型的向量检索。
• 多种索引类型：支持 IVF（Inverted File）、HNSW（Hierarchical Navigable Small World）、ANNOY、PQ（Product Quantization）等。
• 分布式架构：支持横向扩展，能够处理海量数据并提供高并发性能。
• 数据持久化：支持数据持久化和动态加载，确保数据的持久性和可恢复性。（Milvus 使用 RocksDB 作为持久化存储）
• 高级查询：支持复杂查询，如布尔查询、批量查询等。
• 图形化界面：提供 Milvus Insight 图形化管理界面，方便用户管理和监控集群。
• API 支持：提供丰富的 API，包括 RESTful API、Python SDK、Java SDK 等。

【多模态向量检索支持】

• 特征提取：结合预训练模型进行特征提取，将多模态数据（文本、图像、视频等）转化为向量表示。
• 统一接口：所有模态的数据在 Milvus 中统一表示为向量，用户可以通过统一接口进行检索操作。

【分布式架构】

• 集群架构：包括 Coordinator、Proxy、Data Node、Query Node、Index Node 等组件，支持横向扩展和高并发。
• 数据分片和分区：支持数据水平切分（Sharding）和逻辑分区（Partitioning），便于管理和检索。
• 索引管理：支持多种索引类型的分布式构建和维护，提高大规模数据的检索性能。
• 负载均衡和故障恢复：使用负载均衡机制和故障恢复机制，保证系统的高可用性。
• 数据持久化：支持数据持久化和恢复，确保数据在节点故障或重启时不会丢失。

架构

Milvus 的架构主要由以下几个组件组成：

• Coordinator：负责集群管理、元数据管理和任务调度。
• Proxy：处理客户端请求，负责数据路由和负载均衡。
• Data Node：负责数据存储和管理，包括数据持久化和索引构建。
• Query Node：处理查询请求，执行向量检索和结果返回。
• Index Node：负责索引构建和维护，提高查询效率。
• Meta Store：存储元数据，如数据表结构、索引信息等。

【节点】

query node 负责查询，data node 负责数据写入和持久化、index node负责索引建立和加速查询

the query node is in charge of data query; the data node is responsible for data insertion and data persistence; and the index node mainly deals with index building and query acceleration.

设计原则

The design principles of Milvus 2.0

数据流程

【logs --> log snapshot --> segment】

Both the table and the log are data, In the case of Milvus, it aggregates logs using a processing window from TimeTick. Based on log sequence, multiple logs are aggregated into one small file called log snapshot. Then these log snapshots are combined to form a segment, which can be used individually for load balance.

第三方依赖：MinIO, etcd, and Pulsar.

Milvus cluster includes eight microservice components and three third-party dependencies: MinIO, etcd, and Pulsar.

Storage层有三个部分组成：

Meta store、Log broker、Object storage

数据处理过程

Reference

**数据写入和数据持久化：**https://milvus.io/blog/deep-dive-4-data-insertion-and-data-persistence.md

加载数据到内存（实时查询）：https://milvus.io/blog/deep-dive-5-real-time-query.md#Load-data-to-query-node

1、数据写入：可以为每个 collection 设置分片数量（每个分片对应一个虚拟通道 vchannel），数据将根据主键的哈希值写入相应的分片。

You can specify a number of shards for each collection in Milvus, each shard corresponding to a virtual channel (vchannel). Any incoming insert/delete request is routed to shards based on the hash value of primary key.

当segment满了，会自动出发data flush (512 MB by default)

• **【segment】**Milvus中用于数据存储的最小单元。
• indexes are built on segments.
• Automatically flush segment data If the segment is full, the data coord automatically triggers data flush.

三种类型的segment

• There are three types of segments with different status in Milvus: growing, sealed, and flushed segment.
• growing：可以插入数据
• sealed：不再插入数据 A sealed segment is a closed segment
• flushed ：已经被写入磁盘 A flushed segment is a segment that has already been written into disk. A segment can only be flushed when the allocated space in a sealed segment expires.

2、索引建立：index node 从segmeng中将数据加载到内存，建立索引后再写回 object storage

The index node loads the log snapshots to index from a segment (which is in object storage) to memory , deserializes the corresponding data and metadata to build index, serializes the index when index building completes, and writes it back to object storage.

向量检索维度太高，传统基于树的索引不再适应，取而代之的有基于聚类和基于图的索引。

Vectors cannot be efficiently indexed with traditional tree-based indexes due to their high-dimensional nature, but can be indexed with techniques that are more mature in this subject, such as cluster- or graph-based indexes.

3、查询：query node在加载到内存的segment中执行查询

A collection in Milvus is split into multiple segments, and the query nodes loads indexes by segment.

每个query 节点只负责两个任务：按照 query coord 的指令加载或释放段；在本地段中进行搜索。

Each node is responsible only for two tasks: Load or release segments following the instructions from query coord; conduct a search within the local segments.

有两种类型的segment： growing segments 和sealed segments

There are two types of segments, growing segments (for incremental data), and sealed segments (for historical data). Query nodes subscribe to vchannel to receive recent updates (incremental data) as growing segments.

两只类型的数据被加载到query node : growing segments 和sealed segments

There are two types of data that are loaded to query node: streaming data from log broker, and historical data from object storage (also called persistent storage below).

query node 1从持久化存储中加载历史数据S1，从订阅日志代理中的channel 1加载G1

In query node 1 in the image, historical data (batch data), are loaded via the allocated S1 and S3 from persistent storage. In the meanwhile, query node 1 loads incremental data (streaming data) by subscribing to channel 1 in log broker.

存储结构（逻辑）

Collection

In Milvus, a collection is equivalent to a table in a relational database management system (RDBMS)

数据存储在集合（Collection）中，每个集合包含多个实体（Entity），每个实体包含多个字段（Field），但最重要的是向量字段（Vector Field）。

【分组存储】: 将不同类型的数据分别存储在不同的 collections 中。例如，一个用于存储图像特征向量的 collection，另一个用于存储文本特征向量的 collection。

index

索引构建由索引节点完成。为了避免数据更新时频繁构建索引，Milvus 中的集合被进一步划分为多个段，每个段都有自己的索引。

• 检索加速: 索引通过优化数据存储和访问路径，提高查询速度，特别是在处理大规模向量数据时。
• 数据结构优化: 不同的索引类型（如 IVF、HNSW 等）使用不同的技术来组织数据，从而提升检索效率和存储空间利用率。
• 主要索引类型:
  • FLAT: 精确搜索，适用于小数据集。
  • IVF_FLAT: 倒排文件 + FLAT，平衡精度和速度。
  • IVF_SQ8: 倒排文件 + 量化技术，提高速度和降低存储空间。
  • IVF_PQ: 倒排文件 + 产品量化，适用于大规模数据集。
  • HNSW: 小世界图结构，实现高效的近似最近邻搜索。

schema

Schema 用于定义集合的属性以及集合中字段。

Milvus 一个集合中仅支持一个主键字段。

Reference: 字段架构属性

Field 数据类型

https://milvus.io/docs/schema.md

组成部分

查询 节点

【Query Node】

Milvus 将一个集合拆分成多个 Segment，查询节点按 Segment 加载索引。当搜索请求到达时，会广播到所有查询节点并发搜索。然后每个节点剪枝本地 Segment，搜索符合条件的向量，并缩减后返回搜索结果。

Knowhere

Knowhere 是 Milvus 的核心向量执行引擎，它整合了多个向量相似性搜索库，包括Faiss、Hnswlib和Annoy。 Knowhere 还旨在支持异构计算。它控制在哪个硬件（CPU 或 GPU）上执行索引构建和搜索请求。

ANNS 向量索引

Milvus 支持的向量索引类型大多采用近似最近邻搜索（ANNS）算法。相较于通常非常耗时的精确检索，ANNS 的核心思想不再局限于返回最精确的结果，而是只搜索目标的邻居。ANNS 通过在可接受的范围内牺牲准确率来提高检索效率。

优缺点

优点：

• 高性能：优化大规模数据和高并发场景。
• 多模态支持：处理多种数据类型的向量检索。
• 易用性和灵活性：通过提供多语言SDK和RESTful API，Milvus 确保了易用性和灵活性，满足开发者和企业在图像检索、推荐系统、自然语言处理等多种场景下的需求。
• 分布式架构：支持横向扩展，适合大规模数据。

缺点：

• 部分高级功能依赖商业支持。
• 学习曲线较陡峭，需要一定的配置和管理经验。

Milvus 应用场景

使用场景：推荐系统、图像搜索、文本检索、视频检索等。

Milvus 应用场景示例：电商平台的多模态商品搜索

场景描述： 某大型电商平台需要提供多模态商品搜索功能，包括基于图像、文本和视频的搜索。用户可以通过上传图片、输入文本描述或提供视频片段来查找相似的商品。平台需要处理海量商品数据，支持高并发的搜索请求，并提供实时更新和持久化存储。

具体应用：

1. 图像搜索：用户上传一张商品图片，系统通过预训练的图像识别模型（如 ResNet）提取图像特征向量。Milvus 存储和管理这些特征向量，通过向量相似性搜索返回最相似的商品。
2. 文本搜索：用户输入商品描述（如“红色连衣裙”），系统通过 NLP 模型（如 BERT）提取文本特征向量。Milvus 存储文本向量，并基于向量相似性搜索返回相关商品。
3. 视频搜索：用户上传一个视频片段，系统通过视频理解模型提取视频帧的特征向量。Milvus 存储和管理视频特征向量，通过向量相似性搜索返回与视频内容相关的商品。

Milvus 的优势：

• 多模态支持：能够同时处理图像、文本和视频的向量检索，提供统一的检索接口。
• 分布式架构：支持横向扩展，适应电商平台海量商品数据和高并发的搜索请求。
• 数据持久化：支持数据持久化存储，确保数据的安全性和可恢复性。

问答

插入操作和查询操作是分离的吗？

插入操作和查询操作由两个相互独立的模块处理。从客户端的角度来看，当插入的数据进入消息队列时，插入操作就完成了。但是，插入的数据在加载到查询节点之前是不可搜索的。如果段大小未达到索引构建阈值（默认为 512 MB），Milvus 将诉诸暴力搜索，查询性能可能会降低。

Milvus 会将重复主键视为更新操作吗？

不会。Milvus 目前不支持更新操作，也不检查实体主键是否重复。您有责任确保实体主键是唯一的，如果不是，Milvus 可能包含多个具有重复主键的实体。

FAISS

Faiss is a library for efficient similarity search and clustering of dense vectors.

C++ 开发的，一些有用的算法要在GPU上实现

背景

FAISS（Facebook AI Similarity Search）是由 Facebook AI Research 开发的高效向量相似性搜索库，旨在处理大规模、高维向量数据的相似性搜索任务。

功能

• 多种索引类型：支持 Flat、IVF（Inverted File）、HNSW（Hierarchical Navigable Small World）、PQ（Product Quantization）等索引。
• 高效计算：优化了内存和计算资源的使用，支持大规模向量数据的高效检索。
• GPU 加速：支持 GPU 加速，显著提升大规模数据的检索性能。
• 聚类和分类：支持 K-means 聚类、层次聚类等算法，以及最近邻分类等任务。
• 多语言支持：提供 C++ 和 Python 接口，方便集成和使用。

架构

FAISS 的架构相对简单，由以下几个核心组件组成：

• Index：存储和管理向量数据，并提供高效的检索功能。根据不同的索引类型，Index 有不同的实现。
• Vector Storage：存储原始向量数据。
• Distance Computation：计算向量之间的相似性，支持多种相似性度量，如 L2 距离、内积等。

优缺点

优点：

• 高性能：特别是在 GPU 环境下，FAISS 能够显著提升检索速度。
• 灵活性：支持多种索引类型和配置，适应不同应用场景。
• 社区支持：拥有大量用户和丰富的文档资源。

缺点：

• 数据持久化和分布式支持较弱：主要适用于内存内检索，不适合超大规模数据的持久存储和分布式处理。
• 学习曲线较陡峭：需要一定的配置和使用经验，特别是在选择和配置索引类型时。

数据持久化和分布式支持较弱的原因

• 设计目标：FAISS 的设计更关注于内存内计算的效率和性能，而不是数据存储和持久化。
• 内存内操作：主要在内存中操作向量数据和索引，以保证高效的相似性搜索。
• 缺乏内置的持久化机制：虽然支持手动保存和加载索引，但缺乏自动持久化和数据恢复机制。
• 单节点设计：初衷是为单节点设计，缺乏内置的集群管理、负载均衡和分布式一致性支持。

核心：它能够实现“高性能向量检索”的原因

数据类型：Faiss只使用32位浮点矩阵（32-bit floating）。

Faiss 也能够指定数据类型，比如 IndexFlatL2、IndexHNSW、IndexIVF等二十来种类型

FAISS 应用场景

使用场景：推荐系统、图像搜索、文本检索、生物信息学等。

FAISS 应用场景示例：研究机构的大规模基因相似性分析

场景描述： 某研究机构需要进行大规模基因数据的相似性分析，比较数百万条基因序列之间的相似性，以发现潜在的基因关联和突变模式。系统需要高效的计算能力，以在合理时间内完成大规模的相似性计算。

具体应用：

1. 数据准备：研究人员将基因序列数据转化为高维特征向量（如通过 k-mer 频率、序列嵌入等方法）。
2. 索引构建：使用 FAISS 构建大规模基因特征向量的索引。研究机构可以利用 FAISS 的 IVF-PQ（Inverted File with Product Quantization）索引，以在保证高精度的同时显著减少内存占用。
3. 相似性搜索：通过 FAISS 的内存内计算和 GPU 加速，快速进行基因向量的相似性搜索，发现与目标基因序列最相似的其他基因。

FAISS 的优势：

• 高性能计算：通过 GPU 加速和高效的索引结构，能够在大规模数据集上实现极高的相似性搜索性能。
• 内存内操作：适用于需要在内存中进行高效计算的场景，特别是在基因相似性分析中，能够在较短时间内处理大量数据。
• 灵活性：支持多种索引类型和相似性度量，适应不同的数据特征和分析需求。

对比

Elasticsearch 是一个开源的分布式搜索和分析引擎，主要用于全文搜索、结构化搜索、日志和数据分析。尽管 Elasticsearch 强大且多功能，但在替代 Milvus 或 FAISS 用于向量检索方面有其局限性。

Elasticsearch 的优点

1. 全文搜索和结构化搜索：Elasticsearch 在处理全文搜索、布尔查询和结构化数据搜索方面非常强大，提供了丰富的查询DSL。
2. 分布式架构：Elasticsearch 的分布式架构和水平扩展能力使其能够处理大规模数据集和高并发查询。
3. 数据持久化：内置数据持久化和索引恢复机制，保证数据的持久性和可恢复性。
4. 实时索引和查询：支持实时的数据索引和查询，非常适合日志分析和时间序列数据。
5. 生态系统和工具：拥有丰富的生态系统和工具，如 Kibana 用于数据可视化，Logstash 用于数据处理和管道。

Elasticsearch 的向量检索功能

Elasticsearch 引入了向量检索功能，通过插件（如 Elasticsearch k-NN 插件）或直接支持向量字段（如 dense_vector 类型）来实现向量相似性搜索。以下是它在向量检索方面的特点：

1. k-NN 插件：Elasticsearch 提供 k-NN 插件，支持基于 L2 距离和余弦相似度的向量检索。
2. dense_vector 类型：允许将向量作为字段存储在索引中，并使用向量相似度进行搜索。
3. 集成性：可以与 Elasticsearch 的其他功能（如全文搜索、聚合分析）无缝集成。

Elasticsearch 与 Milvus 和 FAISS 的对比

1. 性能

• FAISS：专为高效内存内向量检索设计，特别是在 GPU 环境下，能够提供极高的检索速度。
• Milvus：专注于高性能向量检索，支持多种索引类型和分布式架构，适合大规模、高并发的向量检索任务。
• Elasticsearch：在全文搜索和结构化搜索方面表现出色，但在高维向量检索性能上可能不如 FAISS 和 Milvus。尽管 k-NN 插件提供了向量检索功能，但在处理大规模高维向量数据时，性能可能不如专门设计的向量数据库。

2. 数据持久化和分布式支持

• Milvus：内置数据持久化和分布式支持，适合大规模数据管理和高并发访问。
• Elasticsearch：具备强大的数据持久化和分布式架构，能够处理大规模数据，并提供高可用性和数据恢复机制。
• FAISS：主要设计用于内存内检索，缺乏内置的数据持久化和分布式支持，需要额外的解决方案来实现这些功能。

3. 多模态支持

• Milvus：专为多模态向量检索设计，能够处理文本、图像、视频等多种数据类型。
• Elasticsearch：主要用于文本和结构化数据搜索，通过扩展支持向量检索，但在多模态数据处理上不如 Milvus 专业。

4. 易用性和生态系统

• Elasticsearch：拥有丰富的生态系统和工具，提供直观的界面和广泛的社区支持，易于集成和使用。
• Milvus 和 FAISS：需要一定的配置和使用经验，Milvus 提供了图形化管理界面和丰富的 API，而 FAISS 更适合有特定高性能需求的用户。

5.适用场景

• Milvus 特别适合于需要高效向量搜索的应用，如图像和视频检索、推荐系统、自然语言处理和生物信息学等。它在处理这些需要计算内容相似度的复杂查询时表现出色。
• Elasticsearch 广泛用于日志分析、全文搜索、数据可视化（通过Kibana）和近实时的数据分析等场景。它在处理文本数据和提供灵活、复杂的查询能力方面非常强大。
• Milvus适合高维向量检索（如图像、音频、文本嵌入等相似度搜索），因为它能提供高效的向量检索。Milvus在大规模数据集和复杂索引类型下，会**倾向于将大量数据加载到内存**中。这使得Milvus在处理高维向量检索时能提供低延迟和高吞吐量。
• Elasticsearch适合全文搜索或结构化数据检索，因为它**通过倒排索引和缓存机制**能在较低内存占用下提供高效的查询。

数据规模

具体来说，处理大规模高维向量数据时，Elasticsearch 的性能瓶颈会因数据规模和具体的检索需求而有所不同。然而，可以提供一些指导性的数据规模和考虑因素，帮助判断何时应该使用专门设计的向量数据库（如 Milvus 或 FAISS）而不是 Elasticsearch。

判断数据规模的标准

1. 向量维度

高维向量数据（如 128 维、256 维甚至更高）需要更多的计算资源来进行相似性计算。一般来说，当向量维度超过 100 时，就需要考虑优化计算性能的方案。

2. 向量数量

数据集的规模在向量数量上也是一个关键因素。如果数据集中有数百万到数亿的向量，Elasticsearch 在进行高效检索时可能会遇到性能瓶颈。在这种情况下，专门的向量数据库可能更适合。

3. 检索延迟和并发请求量

对于实时性要求高、并发请求量大的应用场景，Elasticsearch 的性能可能无法满足要求。专门的向量数据库（如 Milvus 和 FAISS）在设计上更适合这种高并发、低延迟的需求。

具体规模建议

1. 小规模数据集：
   • 向量数量：少于 100,000 个向量
   • 向量维度：少于 128 维
   • 适用性：Elasticsearch 可以很好地处理，并提供丰富的搜索功能。
2. 中等规模数据集：
   • 向量数量：100,000 到 1,000,000 个向量
   • 向量维度：128 到 256 维
   • 适用性：Elasticsearch 可以处理，但可能需要优化和调优索引配置。在高并发场景下可能出现性能问题。
3. 大规模数据集：
   • 向量数量：超过 1,000,000 个向量
   • 向量维度：超过 256 维
   • 适用性：需要考虑使用专门的向量数据库（如 Milvus 或 FAISS），以确保检索性能和系统可扩展性。

问答

！！！传统数据库和向量数据库的显著区别在哪里？

传统数据库重IO，向量数据库则需要大量计算资源

传统数据库：

• 重IO操作：传统数据库（如MySQL、PostgreSQL等）在进行数据读写时，主要依赖磁盘的输入输出操作（IO）。这些操作包括将数据从磁盘读取到内存或将数据从内存写入磁盘。
• 关注磁盘IO和CPU使用率等指标：在管理和优化传统数据库时，管理员通常会关注磁盘IO（读写速度和延迟）和CPU使用率。这是因为这两个指标直接影响数据库的性能，尤其是在处理大量读写请求时。

向量数据库：

• 设计计算节点和存储时需要考虑CPU和GPU：向量数据库（如Milvus、FAISS等）在存储和检索向量（高维数据，如图像或文本的嵌入表示）时，不仅依赖CPU，还可能需要使用GPU。这是因为向量检索（如计算向量之间的相似度）通常需要大量的计算资源，尤其是在处理高维向量和大规模数据时，GPU能够显著提高计算效率

实践

Faiss

安装：INSTALL-FAISS

使用

# 构建索引
import faiss                   # make faiss available
index = faiss.IndexFlatL2(d)   # build the index
print(index.is_trained)
index.add(xb)                  # add vectors to the index
print(index.ntotal)

# 搜索
k = 4                          # we want to see 4 nearest neighbors
D, I = index.search(xb[:5], k) # sanity check 输入为query embeddig和neighbors数
1
2
3
4
5
6
7
8
9
10

API Reference

https://faiss.ai/

https://ai.meta.com/tools/faiss/

https://github.com/facebookresearch/faiss/wiki

https://github.com/facebookresearch/faiss/wiki/Faiss-indexes

example Reference

How to Use FAISS to Build Your First Similarity Search-medium

Milvus

Reference

milvus.io/docs

Milvus性能优化提速之道：揭秘优化技巧，避开十大误区- 知乎

https://github.com/milvus-io/pymilvus/tree/2.2/examples

在 Milvus 中，collection 和 index 是两个核心概念，它们分别负责数据存储和数据检索的组织和优化。

• Collection 与 Index 的关系:
  • 每个 collection 可以有多个字段，其中一个字段通常是向量字段。为了加速对该向量字段的检索，可以为其创建索引。
  • 索引是基于 collection 中的向量数据构建的，创建索引后，检索操作会显著加速。

示例应用

1. 创建 Collection:

from pymilvus import CollectionSchema, FieldSchema, DataType, Collection

# 定义字段
id_field = FieldSchema(name="id", dtype=DataType.INT64, is_primary=True)
vector_field = FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=128)

# 定义 schema
schema = CollectionSchema(fields=[id_field, vector_field], description="example collection")

# 创建 collection
collection = Collection(name="example_collection", schema=schema)
1
2
3
4
5
6
7
8
9
10
11

2. 创建 Index:

index_params = {
    "index_type": "IVF_FLAT",
    "params": {"nlist": 128},
    "metric_type": "L2"
}
collection.create_index(field_name="vector", index_params=index_params)
1
2
3
4
5
6

通过理解 collection 和 index 的概念及其功能，可以更好地在 Milvus 中组织和优化向量数据的存储和检索。

查看 collection

# 查看index
from pymilvus import Collection
collection = Collection("book")  # Get an existing collection.

# 查看collection
collection.schema                # Return the schema.CollectionSchema of the collection.
collection.description           # Return the description of the collection.
collection.name                  # Return the name of the collection.
collection.is_empty              # Return the boolean value that indicates if the collection is empty.
collection.num_entities          # Return the number of entities in the collection.
collection.primary_field         # Return the schema.FieldSchema of the primary key field.
collection.partitions            # Return the list[Partition] object.
collection.indexes               # Return the list[Index] object.
collection.properties       # Return the expiration time of data in the collection.
1
2
3
4
5
6
7
8
9
10
11
12
13
14

加载到内存 collection

Milvus 中的所有搜索和查询操作都是在内存中执行的。因此在执行搜索或查询之前需要将集合加载到内存。

这个加载过程发生在Milvus服务所在的服务器上，而不是客户端。

# Get an existing collection.
collection = Collection("book")      
collection.load(replica_number=2)

utility.loading_progress("book")
# Output: {'loading_progress': 100%}
1
2
3
4
5
6

在搜索或查询之后从内存中释放集合以减少内存使用量。

from pymilvus import Collection
collection = Collection("book")      # Get an existing collection.
collection.release()	
1
2
3

数据写入和更新 collection

from pymilvus import Collection
collection = Collection("book")      # Get an existing collection.
mr = collection.insert(data)

# 批量插入 API
# 使用 NumPy 数组将数据集的每一列组织到单独的文件中。在这种情况下，使用每列的字段名称来命名 NumPy 文件。
from pymilvus import utility
task_id = utility.do_bulk_insert(
    collection_name="book",
    partition_name="2022",
    files=["book_id.npy", "word_count.npy", "book_intro.npy", "book_props.npy"]
)

def flush_collection(self):
    """ 刷新数据 """
    self.collection.load() # 这个方法用于将 collection 加载到内存中，以便进行查询，如果 collection 已经在内存中，则此操作不会重复加载。
    self.collection.flush() # 这个方法用于将内存中的数据刷新到磁盘上，确保所有在内存中的变更（例如插入或删除操作）都持久化到磁盘。
    logger.info("Success flush collection")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

Milvus 刷新数据到磁盘

当插入的数据加载到消息队列中时，Milvus 返回成功，但数据还未刷入磁盘。此时，Milvus 的数据节点会将消息队列中的数据以增量日志的形式写入持久化存储。如果flush()调用，则强制数据节点立即将消息队列中的所有数据写入持久化存储。如果您**需要在插入后立即搜索数据**，则可以在数据插入后调用该flush()方法。

Milvus 会自动触发该flush()操作。大多数情况下，无需手动调用此操作。

如果需要**全量更新一个 Collection** 的数据，推荐使用新建表 + 导入数据 + Alias 切换的方案

MilvusHelper

import os
import sys
from loguru import logger
from pymilvus import (
    connections,
    FieldSchema, CollectionSchema, DataType,
    Collection, utility,
    BulkInsertState
)

class MilvusHandler:
    def __init__(
            self,
            host=MILVUS_HOST,
            port=MILVUS_PORT
    ):
        self.host = host
        self.port = port
        self.connect()

    def connect(self):
        connections.connect("default", host=self.host, port=self.port)
        logger.info("Successfully connected to Milvus")

    @staticmethod
    def disconnect():
        # 断开连接
        connections.disconnect("default")
        logger.info("Successfully disconnected from Milvus")

    def create_collection(self, collection_name=None, collection_dim=None):
        # 初始化 Collection 对象
        if self.has_collection(collection_name=collection_name):
            collection = Collection(name=collection_name)
            logger.info(f"Connected to existing collection:{collection_name}")
        else:
            fields = [
                FieldSchema(name="id", dtype=DataType.VARCHAR, description="text name", max_length=500, is_primary=True,
                            auto_id=False),
                FieldSchema(name="vec", dtype=DataType.FLOAT_VECTOR, description="text embedding vectors",
                            dim=collection_dim)
            ]
            schema = CollectionSchema(fields, "Collection for text embeddings")
            collection = Collection(name=collection_name, schema=schema, using='default')
            self.create_index(collection=collection, filed_name="vec")
            logger.info(f"Successfully created collection :{collection_name}")
        return collection

    @staticmethod
    def create_alias(collection_name, alias):
        utility.create_alias(collection_name=collection_name, alias=alias)

    @staticmethod
    def drop_alias(alias):
        utility.drop_alias(alias=alias)

    @staticmethod
    def drop_collection(collection_name):
        utility.drop_collection(collection_name)
        logger.info(f"Dropped collection: {collection_name}")

    def has_collection(self, collection_name):
        return utility.has_collection(collection_name)

    @staticmethod
    def list_collections():
        collections = utility.list_collections()
        logger.info(f"List of collections: {collections}")
        return collections

    @staticmethod
    def create_index(collection, filed_name=None, index_param=None):
        if not index_param:
            index_param = {
                "index_type": "FLAT",
                "metric_type": "IP",
                "params": {}
            }
        collection.create_index(filed_name, index_param)
        logger.info(f"Success create index")

    @staticmethod
    def drop_index(collection):
        collection.drop_index()
        logger.info(f"Success drop index")

    @staticmethod
    def load_collection(collection):
        collection.load()

    @staticmethod
    def flush_collection(collection):
        collection.flush()

    @staticmethod
    def load_and_flush_collection(collection):
        """ 刷新数据 """
        # 测试加载速度
        import time
        start_time = time.time()
        collection.load()
        end_time = time.time()
        print(f"Loading time: {end_time - start_time} seconds")
        collection.flush()
        logger.info("Success flush collection")

    @staticmethod
    def release_collection(collection):
        """释放内存中加载的collection"""
        collection.release()
        logger.info("Success release collection")

    @staticmethod
    def compact_collection(collection):
        """合并segement"""
        collection.compact()
        logger.info(f" release collection status：{collection.get_compaction_state()}")

    @staticmethod
    def insert_entity(collection, entities):
        """
        单次写入小于 10MB 以下，建议选择流式写入
        """
        # entities = [
        #     {"name": "id", "values": ids, "type": DataType.VARCHAR},
        #     {"name": "vec", "values": vectors, "type": DataType.FLOAT_VECTOR}
        # ]
        mr = collection.insert(entities)
        # (insert count: 10, delete count: 0, upsert count: 0, timestamp: {self._timestamp}, success count: {self.succ_count}, err count: {self.err_count})
        logger.info(f"Data inserted successfully.{mr}")
        return mr

    @staticmethod
    def bulk_insert(collection_name=None, files: list = ["id.npy", "vec.npy"]):
        """
         BulkInsert 不会对查询性能造成太大的影响
        :param collection_name:
        :param files:
        :return:
        """
        task_id = utility.do_bulk_insert(
            collection_name=collection_name,
            files=files
        )

        task = utility.get_bulk_insert_state(task_id=task_id)
        print("Task state:", task.state_name)
        print("Imported files:", task.files)
        print("Collection name:", task.collection_name)
        print("Partition name:", task.partition_name)
        print("Start time:", task.create_time_str)
        print("Imported row count:", task.row_count)
        print("Entities ID array generated by this task:", task.ids)

        if task.state == BulkInsertState.ImportFailed:
            print("Failed reason:", task.failed_reason)

    @staticmethod
    def delete_entity(collection, ids):
        expr = f'id in {ids}'
        dr = collection.delete(expr)
        logger.info(f"Data deleted successfully.{dr}")
        return dr

    def update_entity(self, ids, collection, vectors):
        self.delete_entity(collection, ids)
        self.insert_entity(collection, entities=[ids, vectors])
        logger.info(f"Data update successfully")

    @staticmethod
    def get_entity_num(collection):
        entity_num = collection.num_entities
        logger.info(f"Number of entities: {entity_num}")
        return entity_num

    @staticmethod
    def search(collection, query_vector, top_k=10):

        # "param": {"metric_type":  "IP", "params": {"nprobe": _NPROBE}},
        search_param = {
            "data": [query_vector],
            "anns_field": "vec",
            "param": {"metric_type": "IP"},
            "limit": top_k,
            "expr": None,
            "output_fields": ["id"]}

        results = collection.search(**search_param)

        # results[0].ids
        #
        # results[0].distances
        #
        # hit = results[0][0]
        # hit.entity.get('title')
        return results

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197

• collection.num_entities 返回的是逻辑上的总实体数，包括了已删除但尚未物理移除的实体。compaction 操作会合并数据段，并在此过程中移除标记为删除的数据。当前存在问题：删除未被计算 https://github.com/milvus-io/milvus/issues/17193

Milvus Lite 和 Milvus

1. 架构：

   • Milvus：采用了分布式架构，适用于需要处理大规模数据和高并发请求的场景。它能够分布在多个节点上，提供高可用性和扩展性。
   • Milvus Lite：是一个轻量级版本，适用于单节点部署，主要面向开发、测试和小规模应用场景。

2. 资源需求：

   • Milvus：由于其分布式架构，通常需要更多的计算和存储资源来运行，需要集群环境和专业的运维支持。
   • Milvus Lite：设计为轻量级，占用资源少，可以在单机上运行，适合资源有限的环境。

3. 性能和扩展性：

   • Milvus：在性能和扩展性方面更强大，能够处理大规模数据和高并发请求，可以根据需要水平扩展集群。
   • Milvus Lite：性能和扩展性有限，主要适用于小规模数据和低并发请求的场景。

4. 运维复杂度：

   • Milvus：需要专业的运维管理，包括节点管理、负载均衡和故障恢复等。
   • Milvus Lite：运维相对简单，适合开发者个人或小团队进行开发和测试。

适用场景

• Milvus：适用于需要处理大量数据、复杂查询以及高并发请求的大型生产环境，如搜索引擎、大数据分析、推荐系统等。
• Milvus Lite：适用于开发、测试、小规模应用场景或资源受限的环境，适合个人开发者、小团队、初创企业进行向量检索相关的开发和测试。