PyMilvus 是 Milvus 向量数据库的 Python SDK，它允许开发者通过 Python 接口与 Milvus 进行交互，实现高效的向量相似度搜索和管理。

1. 基本概念 #

Milvus 是一个开源的向量数据库，专为 AI 应用设计，能够处理海量向量数据的存储和检索。PyMilvus 提供了以下核心功能：

• 集合（Collection）管理
• 向量插入和删除
• 向量相似度搜索
• 分区（Partition）管理
• 索引（Index）管理

2. 安装 PyMilvus #

建议使用 Python 3.7 及以上版本。

pip install pymilvus

3. 基本使用 #

3.1 连接 Milvus #

连接前请确保 Milvus 服务端已启动，默认端口为 19530。

# 导入 pymilvus 连接模块
from pymilvus import connections

# 连接到 Milvus 服务器（本地或远程）
# 使用默认连接别名，连接到本地主机的19530端口，数据库名为rensheng
connections.connect("default", host="localhost", port="19530", db_name="rensheng")

# 检查连接状态
# 验证名为"default"的连接是否存在，返回True表示连接成功
print(connections.has_connection("default"))  # True 表示连接成功

常见问题：

• 端口不通或服务未启动会连接失败。
• host/port 填写需与 Milvus 服务端一致。

3.2 创建集合（Collection） #

集合类似于数据库中的"表"，需先定义字段（Field）和集合模式（Schema）。

# 导入 pymilvus 相关模块：集合模式、字段模式、数据类型、集合、连接
from pymilvus import CollectionSchema, FieldSchema, DataType, Collection, connections

# 连接到 Milvus 服务器，使用默认连接别名，连接到本地主机的19530端口，数据库名为rensheng
connections.connect("default", host="localhost", port="19530", db_name="rensheng")

# 定义集合的字段结构，包含主键ID字段和向量字段
fields = [
    # 定义主键字段：64位整数类型，自动生成ID
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
    # 定义向量字段：128维浮点向量
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=128),
]

# 创建集合模式对象，包含字段定义和描述信息
schema = CollectionSchema(fields, description="example")

# 检查集合是否存在，不存在才创建
collection_name = "example"
try:
    # 尝试获取已存在的集合
    collection = Collection(collection_name)
    print(f"集合 {collection_name} 已存在")
except Exception:
    # 集合不存在，创建新集合
    collection = Collection(collection_name, schema)
    print(f"集合 {collection_name} 创建成功")

注意事项：

• 向量字段需指定 dim 维度。
• 主键字段建议用 INT64。
• 集合名需唯一。

3.3 创建索引 #

向量字段需创建索引以加速检索。常用类型有 IVF_FLAT、HNSW、IVF_PQ 等。

# 导入 pymilvus 集合和连接模块
from pymilvus import Collection, connections

# 连接到 Milvus 数据库，指定主机、端口和数据库名称
connections.connect("default", host="localhost", port="19530", db_name="rensheng")
# 获取名为 "example" 的集合对象
collection = Collection("example")

# 定义索引参数配置
index_params = {
    "metric_type": "L2",  # 距离度量方式：L2、IP、COSINE
    "index_type": "IVF_FLAT",  # 索引类型 Inverted File with Flat（倒排文件+暴力遍历）
    "params": {"nlist": 128},  # nlist 越大召回率越高，速度越慢
}

# 为向量字段创建索引
collection.create_index("embedding", index_params)
# 打印索引创建成功信息
print("索引创建成功")

常见问题：

• 未建索引时检索效率低。
• nlist 参数需根据数据量调整。

3.4 加载集合到内存 #

检索前必须先将集合 load 到内存。

# 导入 pymilvus 集合和连接模块
from pymilvus import Collection, connections

# 连接到 Milvus 数据库，指定主机、端口和数据库名称
connections.connect("default", host="localhost", port="19530", db_name="rensheng")
# 获取名为 "example" 的集合对象
collection = Collection("example")

# 将集合加载到内存中以支持搜索操作
collection.load()
# 打印集合加载成功信息
print("集合已加载到内存")

注意事项：

• 未 load 时无法检索。
• 数据量大时 load 可能较慢。

3.5 插入数据 #

插入前需准备好与 schema 对应的数据。

# 导入 Milvus 数据库连接和集合操作所需的模块
from pymilvus import Collection, connections

# 导入随机数生成模块
import random

# 连接到 Milvus 数据库，指定主机、端口和数据库名称
connections.connect("default", host="localhost", port="19530", db_name="rensheng")
# 获取名为 "example" 的集合对象
collection = Collection("example")

# 生成随机向量数据
# 创建 1000 个向量，每个向量包含 128 个随机浮点数
vectors = [[random.random() for _ in range(128)] for _ in range(1000)]

# 只插入向量字段（主键 auto_id=True 可省略）
# 将向量数据组织成插入格式
data = [vectors]

# 插入数据
# 将数据插入到集合中，并获取插入结果
mr = collection.insert(data)
# 打印实际插入的向量数量
print("插入向量数量：", mr.insert_count)

注意事项：

• 数据格式为 List[List[float]]，每个子列表为一个向量。
• 插入后建议 collection.flush() 保证数据落盘。

3.6 向量搜索 #

支持多种距离度量（L2、IP、COSINE），可指定 topK、nprobe 等参数。

# 导入 pymilvus 集合和连接模块
from pymilvus import Collection, connections

# 导入随机数生成模块
import random

# 连接到 Milvus 数据库，指定主机、端口和数据库名称
connections.connect("default", host="localhost", port="19530", db_name="rensheng")
# 获取名为 "example" 的集合对象
collection = Collection("example")

# 定义搜索参数配置
search_params = {"metric_type": "L2", "params": {"nprobe": 10}}

# 生成一个查询向量
query_vector = [random.random() for _ in range(128)]

# 执行向量搜索操作
results = collection.search(
    data=[query_vector], anns_field="embedding", param=search_params, limit=5
)

# 遍历并打印搜索结果
for hits in results:
    for hit in hits:
        print(f"id: {hit.id}, distance: {hit.distance}")

参数说明：

• anns_field：向量字段名
• param：索引参数（如 nprobe）
• limit：返回 topK 个结果

常见问题：

• 未 load 集合会报错。
• 查询向量维度需与 schema 一致。

3.7 删除集合 #

删除集合会清空所有数据，谨慎操作。

# 导入 pymilvus 集合和连接模块
from pymilvus import Collection, connections

# 连接到 Milvus 数据库，指定主机、端口和数据库名称
connections.connect("default", host="localhost", port="19530", db_name="rensheng")
# 获取名为 "example" 的集合对象
collection = Collection("example")

# 删除集合
collection.drop()
# 打印集合删除成功信息
print("集合已删除")

4. 高级特性 #

4.1 分区管理 #

分区可用于数据分组、分层管理。

# 导入 pymilvus 集合和连接模块
from pymilvus import Collection, connections

# 导入随机数生成模块
import random

# 连接到 Milvus 数据库，指定主机、端口和数据库名称
connections.connect("default", host="localhost", port="19530", db_name="rensheng")
# 获取名为 "example" 的集合对象
collection = Collection("example")

# 定义搜索参数配置
search_params = {"metric_type": "L2", "params": {"nprobe": 10}}

# 生成一个查询向量
query_vector = [random.random() for _ in range(128)]

# 执行向量搜索操作
results = collection.search(
    # data: 查询向量数据，格式为List[List[float]]，每个子列表为一个查询向量
    # 这里传入一个128维的随机查询向量
    data=[query_vector],
    # anns_field: 指定要搜索的向量字段名称
    # 必须与集合schema中定义的向量字段名一致
    anns_field="embedding",
    # param: 搜索参数配置，包含距离度量方式和索引相关参数
    # metric_type: 距离度量方式（L2欧氏距离、IP内积、COSINE余弦相似度）
    # nprobe: 搜索时检查的聚类数量，值越大召回率越高但速度越慢
    param=search_params,
    # limit: 返回最相似结果的数量限制
    # 这里返回距离最小的前5个向量
    limit=5,
)
print(results)
# 遍历并打印搜索结果
for hits in results:
    for hit in hits:
        print(f"id: {hit.id}, distance: {hit.distance}")

注意事项：

• 分区名需唯一。
• 插入/检索时指定 partition_name/partition_names。

4.2 混合搜索（向量+标量过滤） #

支持带标量字段的混合检索，可用 expr 过滤。

# 导入 pymilvus 集合模式、字段模式、数据类型、集合和连接模块
from pymilvus import CollectionSchema, FieldSchema, DataType, Collection, connections

# 导入随机数生成模块
import random

# 连接到 Milvus 数据库，指定主机、端口和数据库名称
connections.connect("default", host="localhost", port="19530", db_name="rensheng")

# 定义包含标量字段的 schema
# 创建字段列表，包含主键ID、向量字段和分类字段
fields = [
    # 定义主键字段，使用64位整数类型，自动生成ID
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
    # 定义向量字段，使用128维浮点向量类型
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=128),
    # 定义分类字段，使用64位整数类型
    FieldSchema(name="category", dtype=DataType.INT64),
]
# 创建集合模式对象
schema = CollectionSchema(fields)
# 使用模式创建名为 "mixed_collection" 的集合
collection = Collection("mixed_collection", schema)

# 插入带标量字段的数据
# 生成100个128维随机向量
vectors = [[random.random() for _ in range(128)] for _ in range(100)]
# 生成100个1到5之间的随机分类标签
categories = [random.randint(1, 5) for _ in range(100)]
# 将向量和分类数据组织成插入格式
data = [vectors, categories]
# 将数据插入到集合中
collection.insert(data)

# 创建索引和加载
# 定义索引参数：使用L2距离度量，IVF_FLAT索引类型，聚类数量为64
index_params = {"metric_type": "L2", "index_type": "IVF_FLAT", "params": {"nlist": 64}}
# 为向量字段创建索引
collection.create_index("embedding", index_params)
# 将集合加载到内存中以支持搜索操作
collection.load()

# 带过滤条件的搜索
# 生成一个128维随机查询向量
query_vector = [random.random() for _ in range(128)]
# 定义搜索参数：使用L2距离度量，nprobe参数为10
search_params = {"metric_type": "L2", "params": {"nprobe": 10}}
# 执行带过滤条件的向量搜索，只搜索分类为2、3、4的数据
results = collection.search(
    # 查询向量数据
    data=[query_vector],
    # 指定搜索的向量字段名
    anns_field="embedding",
    # 搜索参数配置
    param=search_params,
    # 返回前5个最相似的结果
    limit=5,
    # 过滤条件：只搜索分类在[2,3,4]范围内的数据
    expr="category in [2, 3, 4]",
)
# 遍历搜索结果
for hits in results:
    # 遍历每个命中结果
    for hit in hits:
        # 打印结果ID和距离值
        print(f"id: {hit.id}, distance: {hit.distance}")

在 Milvus 的 pymilvus 客户端中，collection.insert(data) 的数据格式有两种常见写法：

列表格式（List of Columns）

data = [vectors, categories]
collection.insert(data)

• data 是一个列表，列表中的每一项对应一个字段（列），顺序要和 schema 中字段定义的顺序一致（不包括 auto_id 的主键字段）。
• 每个字段的数据都是一个长度相同的列表（即“列式”存储）。

在这个的 schema 里，字段顺序是：

1. id（auto_id=True，插入时不用提供）
2. embedding
3. category

所以只需要提供 embedding 和 category 两列的数据

字典格式（List of Dicts）

data = [
    {"embedding": [0.1, 0.2, ...], "category": 2},
    {"embedding": [0.3, 0.4, ...], "category": 3},
    # ...
]
collection.insert(data)

这种写法每一行为一个字典，字段名和 schema 对应，适合“行式”插入。

注意事项：

• 列表格式不需要指定字段名，但顺序必须和 schema 一致（auto_id 字段不用管）。
• 字典格式需要指定字段名，顺序无关。

5. 最佳实践 #

1. 批量插入：尽量批量插入数据而不是单条插入，以提高性能。
2. 合理设置索引参数：根据数据量和查询需求调整索引参数（如 nlist、nprobe）。
3. 预加载集合：在查询前加载集合到内存。
4. 使用连接池：在高并发场景下使用连接池管理连接。
5. 监控性能：定期监控查询性能和资源使用情况。
6. 异常处理：所有操作建议加 try/except 捕获异常，便于排查问题。