当前位置：首页 > news >正文

Milvus 向量数据库基础操作解析

news 2025/8/10 5:37:33

文章目录

Milvus 向量数据库新手入门：基础操作全解析
- 一、基础概念
- 二、基础操作实战
- - 1. 连接到 Milvus 服务器
  - - （1）`connect()`方法详解：
    - （2）`get_connection_addr()`方法详解：
  - 2. 创建集合（Collection）
  - - （1）FieldSchema 类
    - （2）CollectionSchema 类
    - （3）Collection 类：
    - （4）Collection 属性
  - 3. 插入数据
  - - （1）`insert()`方法详解：
    - （2）`flush()` 方法讲解：
  - 4. 创建索引
  - - （1）`create_index()` 方法
  - 5. 加载集合到内存
  - - （1）`load()` 方法：
  - 6. 向量相似性查询
  - - （1）`search()`方法详解
  - 7. 标量查询
  - 8. 删除数据
  - 9. 释放集合和断开连接
  - 10. 删除集合（谨慎操作）
- 总结

Milvus 向量数据库新手入门：基础操作全解析

前言：本文中使用的 Milvus 版本为 2.6.0

一、基础概念

Milvus 是一个开源的向量数据库，专门用于存储、索引和检索大规模的向量数据。它广泛应用于图像识别、自然语言处理、推荐系统等 AI 领域。与传统关系型数据库不同，Milvus 针对高维向量的相似性搜索进行了优化，能够在毫秒级时间内处理数十亿向量的查询。其核心优势在于高效的向量索引算法和分布式架构，可轻松应对海量数据场景。

在开始使用 Milvus 之前，需要了解几个核心概念：

Collection（集合）：类似于传统数据库中的表，是存储向量和标量字段的逻辑单元。每个集合有独立的 schema（结构定义），包含字段信息和描述。
Partition（分区）：集合的子集，用于将数据按一定规则分组（如时间、类别），提高查询效率。查询时可指定分区，减少扫描范围。
Vector（向量）：由浮点数或二进制数组成的数组，是 Milvus 存储的核心数据类型。向量维度需在创建集合时定义，且所有向量需保持一致。
Embedding（嵌入向量）：将非结构化数据（如文本、图像）转换为向量的过程，转换后的向量称为嵌入向量。例如，通过 BERT 模型将文本转换为 768 维向量。
Index（索引）：用于加速向量相似性搜索的数据结构。Milvus 支持多种索引类型（如 IVF_FLAT、HNSW 等），需根据场景选择。
Metric（度量方式）：用于计算向量之间相似度的方法。常用的有欧氏距离（L2）、内积（IP）、余弦相似度等，需在创建索引和查询时保持一致。

二、基础操作实战

下面我们通过实际代码来学习 Milvus 的基础操作，每个操作都会详细说明涉及的方法、参数和返回值。

1. 连接到 Milvus 服务器

首先，需要与已安装的 Milvus 服务器建立连接。connections 模块提供了连接管理功能，其中最核心的是 connect() 方法。

（1）`connect()`方法详解：

方法原型：connections.connect(alias="default", host="localhost", port="19530", user="", password="", db_name="")
功能：建立与 Milvus 服务器的连接，并通过别名标识该连接。
参数说明：
- alias：连接的别名，用于在后续操作中标识这个连接，默认为"default"。
- host：Milvus 服务所在的 IP 地址，本地部署通常为"localhost"，远程部署需填写服务器 IP。
- port：Milvus 服务的端口号，默认端口为 19530（gRPC 协议），HTTP 协议默认端口为 9091。
- user：用户名，如需身份验证时使用（Milvus 企业版支持）。
- password：密码，如需身份验证时使用（Milvus 企业版支持）。
- db_name：数据库名称，默认为"default"。
返回值：无返回值，连接失败会抛出异常。

（2）`get_connection_addr()`方法详解：

此外，get_connection_addr() 方法可用于验证连接是否成功：

方法原型：connections.get_connection_addr(alias="default")
功能：获取指定别名连接的详细信息。
参数：alias 为连接别名。
返回值：包含连接信息的字典（如 host、port、user 等）。

# 从 pymilvus 库中导入 connections 模块，用于管理与 Milvus 服务器的连接
from pymilvus import connections# 连接到 Milvus 服务
connections.connect(alias="default",  # 连接的别名host="localhost",  # Milvus 服务的 IP 地址port="19530"       # Milvus 服务的端口号
)# 检查连接是否成功
print(connections.get_connection_addr("default"))

预期输出：

{'host': 'localhost', 'port': '19530', 'user': '', 'password': '', 'db_name': ''}

结果分析：输出显示了连接的详细信息，表明我们已成功连接到本地的 Milvus 服务器。如果连接失败，会抛出 MilvusException 异常，需检查服务器是否启动、IP 和端口是否正确。

2. 创建集合（Collection）

集合是存储数据的基本单元，类似于关系数据库中的表。创建集合时需要定义其结构（schema），涉及 FieldSchema、CollectionSchema 和 Collection 三个核心类。

类与方法详解：

（1）FieldSchema 类

用于定义单个字段的属性。

构造函数原型：FieldSchema(name, dtype, description="", is_primary=False, auto_id=False, dim=None)
参数说明：
- name：字段名称（字符串，需唯一）。
- dtype：字段数据类型（DataType 枚举，如 INT64、FLOAT_VECTOR 等）。
- description：字段描述（可选）。
- is_primary：是否为主键字段（布尔值，一个集合只能有一个主键）。
- auto_id：是否自动生成主键值（布尔值，仅主键字段有效）。
- dim：向量维度（仅向量类型字段有效，如 FLOAT_VECTOR）。

（2）CollectionSchema 类

用于定义集合的整体结构。

构造函数原型：CollectionSchema(fields, description="", auto_id=False)
参数说明：
- fields：字段列表（FieldSchema 对象的列表）。
- description：集合描述（可选）。
- auto_id：全局自动生成主键开关（已废弃，建议在字段级别设置 auto_id）。

（3）Collection 类：

用于操作集合，其构造函数用于创建集合。

构造函数原型：Collection(name, schema, using="default", shards_num=1)
参数说明：
- name：集合名称（字符串，需唯一）。
- schema：集合的 schema（CollectionSchema 对象）。
- using：连接别名（指定使用哪个连接，默认为"default"）。
- shards_num：分片数量（分布式部署时有效，默认为 1）。
返回值：Collection 对象，用于后续操作该集合。

（4）Collection 属性

name：获取集合名称。
is_loaded：判断集合是否已加载到内存（布尔值）。

# 从 pymilvus 库中导入相关类，用于定义集合结构和创建集合
from pymilvus import FieldSchema, CollectionSchema, DataType, Collection# 定义集合中的字段
fields = [# 主键字段，用于唯一标识每条记录FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),# 向量字段，用于存储向量数据FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=2)
]# 创建集合的 schema（模式），用于定义集合的整体结构
schema = CollectionSchema(fields, "这是一个用于演示的简单集合")# 创建集合
collection_name = "demo_collection"
collection = Collection(name=collection_name, schema=schema)# 查看集合信息
print(f"集合名称: {collection.name}")
print(f"集合是否加载到内存: {collection.is_loaded}")

预期输出：

集合名称: demo_collection
集合是否加载到内存: False

结果分析：我们创建了一个名为 demo_collection 的集合，包含一个自增的主键字段（id）和一个 2 维向量字段（embedding）。此时集合尚未加载到内存，因此 is_loaded 为 False。创建集合时如果名称已存在，会抛出异常，需先删除现有集合或使用新名称。

3. 插入数据

（1）`insert()`方法详解：

创建集合后，我们可以通过 insert() 方法向其中插入数据，插入的数据需与集合的 schema 匹配。

方法原型：collection.insert(data, partition_name="")
功能：向集合（或指定分区）插入数据。
参数说明：
- data：要插入的数据，格式为列表，每个元素是与字段对应的列表或字典（如 [vectors] 表示插入向量字段，若有多个字段需按顺序传入）。
- partition_name：分区名称（可选，不指定则插入默认分区）。
返回值：InsertResult 对象，包含插入的主键列表和插入数量。

InsertResult 对象属性：

primary_keys：获取插入数据自动生成的主键列表（列表类型）。
insert_count：获取成功插入的记录数（整数）。

（2）`flush()` 方法讲解：

flush() 方法用于将内存中的数据刷写到磁盘：

方法原型：collection.flush(partition_names=None)
参数：partition_names 为分区名称列表（可选，不指定则刷新所有分区）。
返回值：无。

# 导入 random 模块，用于生成随机数
import random# 生成 10 个 2 维随机向量
vectors = [[random.random() for _ in range(2)] for _ in range(10)]# 插入数据到集合中
insert_result = collection.insert([vectors])# 查看插入结果
print(f"插入的主键列表: {insert_result.primary_keys}")
print(f"插入的行数: {insert_result.insert_count}")# 强制刷新，确保数据被写入持久化存储
collection.flush()

预期输出（示例，具体数值会不同）：

插入的主键列表: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
插入的行数: 10

结果分析：我们插入了 10 个随机生成的 2 维向量。由于主键字段 id 设为 auto_id=True，Milvus 自动为每条记录分配了唯一的 ID（1 到 10）。flush() 方法确保数据从内存缓存写入磁盘，避免因服务器故障导致数据丢失。插入失败会抛出异常，需检查数据格式是否与 schema 匹配（如向量维度是否正确）。

4. 创建索引

（1）`create_index()` 方法

为了提高向量查询效率，需要为向量字段创建索引。create_index() 方法用于创建索引。

方法原型：collection.create_index(field_name, index_params, index_name="")
功能：为指定字段创建索引。
参数说明：
- field_name：要创建索引的字段名称（字符串，需为向量字段）。
- index_params：索引参数（字典，包含 index_type、metric_type、params）。
  - index_type：索引类型，如 “IVF_FLAT”（常用的倒排文件索引）。
  - metric_type：距离度量方式，如 “L2”（欧氏距离）、“IP”（内积）。
  - params：索引的具体参数，如 IVF 类索引的 nlist（聚类中心数量）。
- index_name：索引名称（可选，默认为字段名+索引类型）。
返回值：布尔值，True 表示创建成功。

索引信息查看：

collection.indexes 属性：获取集合中所有字段的索引信息（列表 of Index 对象）。
Index 对象属性：
- field_name：索引对应的字段名。
- index_type：索引类型。

# 定义索引参数
index_params = {"index_type": "IVF_FLAT",  # 索引类型，IVF_FLAT 是常用的倒排文件索引，适用于大规模数据"metric_type": "L2",       # 距离度量方式，L2 表示欧氏距离"params": {"nlist": 1024}  # 索引的具体参数，nlist 表示聚类中心数量（IVF 类索引必填）
}# 为向量字段创建索引
collection.create_index(field_name="embedding",  # 要创建索引的字段名index_params=index_params
)# 查看索引信息
indexes = collection.indexes
print("索引信息:")
for index in indexes:print(f"字段名: {index.field_name}, 索引类型: {index.index_type}")

预期输出：

索引信息:
字段名: embedding, 索引类型: IVF_FLAT

结果分析：我们为 embedding 字段创建了 IVF_FLAT 类型的索引，使用欧氏距离（L2）作为度量方式。IVF_FLAT 索引通过将向量聚类到多个中心（nlist=1024），查询时仅搜索部分聚类，从而提高效率。创建索引需要一定时间，取决于数据量和索引类型。一个向量字段只能有一个索引，重复创建会覆盖现有索引。

5. 加载集合到内存

（1）`load()` 方法：

在进行查询前，需要将集合的数据和索引加载到内存中，否则查询会失败或效率极低。load() 方法用于加载集合。

方法原型：collection.load(partition_names=None, replica_number=1)
功能：将集合的数据和索引加载到内存。
参数说明：
- partition_names：分区名称列表（可选，不指定则加载所有分区）。
- replica_number：副本数量（分布式部署时有效，默认为 1）。
返回值：无。

加载状态查看：

collection.is_loaded 属性：判断集合是否已加载到内存（布尔值）。

# 加载集合到内存
collection.load()# 查看集合是否已加载
print(f"集合是否加载到内存: {collection.is_loaded}")

预期输出：

集合是否加载到内存: True

结果分析：集合成功加载到内存后，is_loaded 变为 True，此时可以进行高效查询。加载操作的时间取决于集合大小和服务器性能。对于大规模集合，可通过加载指定分区（partition_names 参数）减少内存占用。

6. 向量相似性查询

向量相似性查询是 Milvus 的核心功能，用于查找与目标向量最相似的向量。这一功能通过 search() 方法实现，该方法能高效地在大规模向量数据中检索出相似度最高的结果。

（1）`search()`方法详解

方法原型：collection.search(data, anns_field, param, limit=10, expr="", output_fields=None, partition_names=None)
功能：在指定的向量字段中，搜索与查询向量相似的记录，并按相似度由高到低返回结果。
参数说明：
- data：查询向量列表，每个元素是一个向量（维度需与集合中向量字段的维度一致），支持批量查询。
- anns_field：用于查询的向量字段名称（字符串），必须是集合中已定义的向量字段。
- param：查询参数（字典类型），需与创建索引时的参数匹配。例如，对于 IVF 类索引，需指定 nprobe（查询时探测的聚类中心数量），该值越大，查询精度越高，但速度越慢，默认值为 10。
- limit：返回结果的最大数量（整数），默认值为 10，根据实际需求调整。
- expr：标量过滤条件（字符串，可选），用于在向量搜索前先筛选出符合标量条件的记录，例如 "subject == 'history'"。
- output_fields：需要返回的字段列表（列表 of 字符串，可选），不指定时仅返回主键（id）和距离（distance）。
- partition_names：分区名称列表（可选），指定仅在某些分区中进行查询，可提高查询效率。
返回值：SearchResult 对象，该对象是一个列表，每个元素对应一个查询向量的检索结果，每个结果包含多个 Hit 对象（按距离升序排列，距离越小相似度越高）。

结果处理相关：

Hit 对象属性：
- id：匹配记录的主键值。
- distance：查询向量与该记录向量的距离值，距离越小表示相似度越高。
- entity：包含返回字段的字典，字段由 output_fields 指定。

# 生成一个查询向量，维度与集合中向量字段的维度一致（2 维）
query_vector = [[random.random() for _ in range(2)]]# 定义查询参数
search_params = {"metric_type": "L2",  # 距离度量方式，需与创建索引时的一致（此处为欧氏距离）"params": {"nprobe": 10}  # 查询时探测的聚类中心数量，影响查询精度和速度
}# 执行向量相似性查询
results = collection.search(data=query_vector,anns_field="embedding",  # 用于查询的向量字段名称param=search_params,limit=3,  # 返回最相似的 3 条结果output_fields=["id"]  # 指定返回 id 字段
)# 处理查询结果
print("查询向量:", query_vector)
print("最相似的 3 个向量:")
for result in results[0]:  # results[0] 对应第一个查询向量的结果print(f"ID: {result.id}, 距离: {result.distance}")

预期输出（示例，具体数值会不同）：

查询向量: [[0.4567, 0.7890]]
最相似的 3 个向量:
ID: 5, 距离: 0.0234
ID: 8, 距离: 0.0567
ID: 3, 距离: 0.0890

结果分析：查询返回了与随机生成的查询向量最相似的 3 条记录，按距离从小到大排序（距离越小，向量相似度越高）。通过调整 nprobe 参数可以平衡查询精度和速度：当数据量较大时，可适当增大 nprobe 以提高精度，或减小 nprobe 以加快查询速度。批量查询时，只需在 data 参数中传入多个向量，结果列表会按顺序对应每个查询向量的检索结果。

7. 标量查询

除了向量相似性查询，Milvus 还支持基于标量字段（如主键、整数、字符串等）的查询，类似于传统数据库的 SQL 查询，通过 query() 方法实现。

方法详解：

方法原型：collection.query(expr, output_fields=None, partition_names=None, limit=10000)
功能：根据标量条件筛选出符合要求的记录。
参数说明：
- expr：查询条件表达式（字符串，必填），支持多种运算符，如 ==、!=、>、<、>=、<=、in、not in 等，例如 "id > 5"、"subject in ['history', 'science']"。
- output_fields：需要返回的字段列表（列表 of 字符串，可选），默认返回所有字段。
- partition_names：分区名称列表（可选），指定仅在某些分区中查询。
- limit：返回结果的最大数量（整数），默认值为 10000，防止返回过多结果占用大量内存。
返回值：查询结果列表，每个元素是一个字典，包含指定字段及其对应的值。

# 定义基于主键的查询条件表达式
expr = "id in [1, 3, 5]"  # 查询 id 为 1、3、5 的记录# 执行标量查询
result = collection.query(expr=expr,output_fields=["id", "embedding"]  # 指定返回 id 和 embedding 字段
)# 打印查询结果
print(f"查询条件: {expr}")
for item in result:print(f"ID: {item['id']}, 向量: {item['embedding']}")

预期输出（示例，具体数值会不同）：

查询条件: id in [1, 3, 5]
ID: 1, 向量: [0.1234, 0.5678]
ID: 3, 向量: [0.9012, 0.3456]
ID: 5, 向量: [0.7890, 0.4567]

结果分析：标量查询成功返回了所有符合条件（id 为 1、3、5）的记录，并包含了指定的 id 和 embedding 字段。该功能适用于需要通过标量字段快速筛选数据的场景，例如查询某个类别下的所有记录。查询表达式支持复杂的逻辑组合，例如 expr="id > 5 and publication_year == 2023"（需集合中存在 publication_year 字段），可满足多样化的查询需求。

8. 删除数据

Milvus 支持根据标量条件删除集合中的数据，通过 delete() 方法实现。删除操作是不可逆的，执行前需仔细确认条件。

方法详解：

方法原型：collection.delete(expr, partition_name="")
功能：删除符合指定标量条件的记录。
参数说明：
- expr：删除条件表达式（字符串，必填），语法与 query() 方法的 expr 参数一致，例如 "id == 1"、"subject == 'history'"。
- partition_name：分区名称（可选），指定仅删除某个分区中的符合条件的记录。
返回值：DeleteResult 对象，包含删除操作的结果信息。

删除结果查看：

DeleteResult.delete_count 属性：获取成功删除的记录数量（整数）。

# 定义要删除数据的主键列表
ids_to_delete = [1]# 执行删除操作
delete_result = collection.delete(expr=f"id in {ids_to_delete}")print(f"删除的行数: {delete_result.delete_count}")# 验证删除结果
result = collection.query(expr="id in [1]", output_fields=["id"])
print(f"删除后 ID 为 1 的数据是否存在: {len(result) > 0}")

预期输出：

删除的行数: 1
删除后 ID 为 1 的数据是否存在: False

结果分析：我们成功删除了 id 为 1 的记录，通过后续查询验证，该记录已不存在。删除操作基于 expr 条件，支持批量删除，例如 expr="id in [1,2,3]" 可一次性删除多条记录。需要注意的是，删除的数据不会立即从磁盘中物理移除，而是被标记为“已删除”，后续会通过 Milvus 的后台合并操作清理，以提高存储和查询效率。

9. 释放集合和断开连接

操作完成后，应释放内存中的集合并断开与 Milvus 服务器的连接，以节省系统资源，这是良好的资源管理习惯。

相关方法详解：

释放集合：
- 方法原型：collection.release()
- 功能：将集合从内存中释放，释放其占用的内存资源。
- 参数：无。
- 返回值：无。
- 释放状态查看：collection.is_loaded 属性（布尔值），释放后变为 False。
断开连接：
- 方法原型：connections.disconnect(alias="default")
- 功能：断开与指定别名的 Milvus 服务器连接。
- 参数：alias 为连接别名，默认为 “default”。
- 返回值：无。
- 连接状态查看：connections.list_connections() 方法，返回当前活跃的连接别名列表。

# 释放内存中的集合
collection.release()# 查看集合是否已释放
print(f"集合是否加载到内存: {collection.is_loaded}")# 断开与 Milvus 的连接
connections.disconnect("default")# 检查连接是否还存在
print("是否还存在连接: ", "default" in connections.list_connections())

预期输出：

集合是否加载到内存: False
是否还存在连接:  False

结果分析：release() 方法成功将集合从内存中释放，is_loaded 属性变为 False，表明集合不再占用内存资源；disconnect("default") 方法断开了与 Milvus 服务器的连接，活跃连接列表中已不存在 “default” 别名，说明连接已成功关闭。在处理多个集合或长时间运行的程序时，及时释放资源和断开连接可避免内存泄漏和不必要的资源占用。

10. 删除集合（谨慎操作）

如果某个集合不再需要，可将其删除，删除后集合中的所有数据、索引和元信息都会被永久移除，操作不可逆。utility 模块提供了集合删除相关的方法。

方法详解：

检查集合是否存在：
- 方法原型：utility.has_collection(collection_name, using="default")
- 功能：判断指定名称的集合是否存在。
- 参数说明：
  - collection_name：集合名称（字符串）。
  - using：连接别名，默认为 “default”。
- 返回值：布尔值，True 表示存在，False 表示不存在。
删除集合：
- 方法原型：utility.drop_collection(collection_name, using="default")
- 功能：删除指定名称的集合。
- 参数说明：
  - collection_name：集合名称（字符串）。
  - using：连接别名，默认为 “default”。
- 返回值：无。

# 从 pymilvus 库中导入 utility 模块，用于执行工具类操作
from pymilvus import utility# 检查集合是否存在
if utility.has_collection(collection_name):# 删除集合utility.drop_collection(collection_name)print(f"集合 {collection_name} 已删除")
else:print(f"集合 {collection_name} 不存在")# 验证删除结果
print(f"集合 {collection_name} 是否还存在: {utility.has_collection(collection_name)}")

预期输出：

集合 demo_collection 已删除
集合 demo_collection 是否还存在: False

结果分析：我们先通过 utility.has_collection() 方法确认集合 demo_collection 存在，然后使用 utility.drop_collection() 方法将其删除，最后验证发现该集合已不存在。删除集合是不可逆的操作，在生产环境中需格外谨慎，建议先备份重要数据或通过权限控制防止误操作。