Milvus的使用

先抛出一段代码，用的是python sdk,然后我们再逐个分析

from pymilvus import (
    connections,
    utility,
    FieldSchema,
    CollectionSchema,
    DataType,
    Collection,
)
import random
​
# 1. 连接服务
# 连接到本地运行的 Milvus 服务，默认端口为 19530
connections.connect("default", host="localhost", port="19530")
​
# 2. 创建一个集合 (Collection)
# 定义字段 schema
fields = [
    # 主键字段：INT64 类型，非自动生成 (auto_id=False)，需要手动指定 ID
    FieldSchema(name="pk", dtype=DataType.INT64, is_primary=True, auto_id=False),
    # 普通标量字段：DOUBLE 类型
    FieldSchema(name="random", dtype=DataType.DOUBLE),
    # 向量字段：FLOAT_VECTOR 类型，维度为 8
    FieldSchema(name="embeddings", dtype=DataType.FLOAT_VECTOR, dim=8)
]
​
# 创建 Schema 对象，并添加描述
schema = CollectionSchema(fields, "hello_milvus is the simplest demo to introduce the APIs")
​
# 基于 Schema 创建集合，如果集合已存在会报错（实际生产中通常先检查是否存在）
hello_milvus = Collection("hello_milvus", schema)
​
# 3. 在集合中插入向量数据
# 构造 3000 条实体数据
entities = [
    [i for i in range(3000)],  # pk 列：0 到 2999
    [float(random.randrange(-20, -10)) for _ in range(3000)],  # random 列：-20 到 -10 之间的随机浮点数
    [[random.random() for _ in range(8)] for _ in range(3000)],  # embeddings 列：3000 个 8 维随机向量
]
​
# 执行插入操作
insert_result = hello_milvus.insert(entities)
# 刷新集合，确保数据落盘并可被搜索/查询（重要步骤）
hello_milvus.flush()  
​
print(f"插入了 {insert_result.insert_count} 条数据，ID 范围：{insert_result.primary_keys[0]} 到 {insert_result.primary_keys[-1]}")
​
# 4. 在实体上生成索引 (Index)
# 定义索引参数：IVF_FLAT 算法，L2 欧氏距离，nlist=128 (聚类中心数)
index = {
    "index_type": "IVF_FLAT",
    "metric_type": "L2",
    "params": {"nlist": 128},
}
# 为 embeddings 字段创建索引
hello_milvus.create_index("embeddings", index)
​
# 5. 将集合加载到内存并执行向量相似性搜索
# 加载集合到内存（Milvus 是内存优先的数据库，搜索前必须 load）
hello_milvus.load()
​
# 选取最后两个向量作为查询向量
vectors_to_search = entities[-1][-2:]
​
# 定义搜索参数：nprobe=10 (搜索时探查的聚类中心数量)
search_params = {
    "metric_type": "L2",
    "params": {"nprobe": 10},
}
​
# 执行向量搜索
# limit=3: 返回最相似的 3 个结果
# output_fields: 返回结果中除了主键和向量外，额外返回的标量字段
result = hello_milvus.search(
    vectors_to_search, 
    "embeddings", 
    search_params, 
    limit=3, 
    output_fields=["random"]
)
​
print("向量搜索结果:")
for hits in result:
    for hit in hits:
        print(f"ID: {hit.id}, Distance: {hit.distance}, Random Value: {hit.entity.get('random')}")
​
# 6. 执行标量查询 (Query)
# 基于标量字段过滤：查找 random > -14 的实体
result_query = hello_milvus.query(
    expr="random > -14", 
    output_fields=["random", "embeddings"]
)
print(f"\n标量查询结果数量: {len(result_query)}")
​
# 7. 执行混合搜索 (Hybrid Search)
# 结合向量相似度 + 标量过滤：查找 random > -12 的实体中，与查询向量最相似的 3 个
result_hybrid = hello_milvus.search(
    vectors_to_search, 
    "embeddings", 
    search_params, 
    limit=3, 
    expr="random > -12", 
    output_fields=["random"]
)
print("\n混合搜索结果:")
for hits in result_hybrid:
    for hit in hits:
        print(f"ID: {hit.id}, Distance: {hit.distance}, Random Value: {hit.entity.get('random')}")
​
# 8. 按照实体的主键删除实体
# 构造删除表达式：删除 pk 等于前两个插入数据的 ID
expr = f"pk in [{entities[0][0]}, {entities[0][1]}]"
hello_milvus.delete(expr)
print(f"\n已执行删除操作: {expr}")
​
# 9. 删除集合
utility.drop_collection("hello_milvus")
print("集合 'hello_milvus' 已删除。")
​
# 断开连接（可选，程序结束会自动断开）
connections.disconnect("default")

一、连接服务 (Connection)

connections.connect("default", host="localhost", port="19530")

• 作用：建立客户端与 Milvus 服务器（Server）的 gRPC 连接。

• 关键点

  • host="localhost": 默认连接本地运行的 Milvus。如果是远程服务器，这里填 IP。

  • port="19530": Milvus 默认的 gRPC 监听端口（注意不是 HTTP 的 9091 或 80）。

  • 隐含逻辑：如果连接失败，通常意味着 Milvus 服务未启动（Docker 容器没跑）或防火墙阻挡。

二、定义集合 Schema

下面的的代码展示了用FieldSchema创建pk等字段，然后再用 CollectionSchema创建shema,从而用Collection("hello_milvus", schema)创建了集合hello_milvus。

fields = [
    FieldSchema(name="pk", dtype=DataType.INT64, is_primary=True, auto_id=False),
    FieldSchema(name="random", dtype=DataType.DOUBLE),
    FieldSchema(name="embeddings", dtype=DataType.FLOAT_VECTOR, dim=8)
]
schema = CollectionSchema(fields, "hello_milvus is the simplest demo...")
hello_milvus = Collection("hello_milvus", schema)

1、集合介绍

集合是 Milvus 中通过 Schema（模式） 定义的数据结构。

创建一个集合时，必须定义其 Schema，这相当于定义了数据的“骨架”。

shema包含了列数据+元数据；在 Milvus 中，你不能直接用 fields 建表，必须把它封装进 schema 才能使用。

2、field

一个集合主要由以下几种字段（Field）组成：

主键字段：唯一标识每一条数据（实体）。

向量字段：存储高维特征向量，是检索的核心

标量字段：存储元数据，用于过滤或展示。

下面我们详细来介绍它。

1）pk (主键):

is_primary=True: 标记为主键，类似关系型数据库的 ID。

auto_id=False: 手动指定 ID。代码后续插入数据时使用了 0, 1, 2...。如果设为 True，Milvus 会自动生成 ID，插入时就不需要传这个字段了。

2）random (标量字段):

DataType.DOUBLE: 普通的双精度浮点数。用于后续的标量过滤（如 random > -14）。

3）embeddings (向量字段):

DataType.FLOAT_VECTOR: 声明这是一个浮点型向量。

dim=8: 向量维度。这里为了演示设为 8，实际业务中通常是 768 (BERT), 1536 (OpenAI), 或更高。注意：一旦创建，维度不可修改。

三、插入数据

entities = [
    [i for i in range(3000)],  # pk 列
    [float(random.randrange(-20, -10)) for _ in range(3000)], # random 列
    [[random.random() for _ in range(8)] for _ in range(3000)] # embeddings 列 (3000个 8维向量)
]
insert_result = hello_milvus.insert(entities)
hello_milvus.flush()

1、概念：列式存储

Milvus 的 Python SDK 在插入数据时，采用的是列式格式（List of Lists），而不是常见的行式格式（List of Dicts）。

• 外层列表：代表整个批次的数据。

• 内层列表：代表一列数据（对应 Schema 中的一个 Field）。

• 重要规则：所有内层列表的长度必须一致（这里都是 3000），代表插入了 3000 行数据。

2、构造数据

1）主键列 (pk)插入了：[0, 1, 2, ..., 2999]

2）标量列 (random）：3000 个介于 -20 到 -10 之间的随机浮点数。

这是为了演示标量过滤。在后续的搜索中，我们可以用 random > -14 这样的条件来筛选数据。

3）向量列 (embeddings):[[random.random() for _ in range(8)] for _ in range(3000)]

生成：3000 个向量，每个向量包含 8 个 0.0 到 1.0 之间的随机浮点数。

3、执行插入

• 动作：将构造好的 entities 发送给 Milvus 服务端。

• 返回值 (insert_result)

  ：这是一个包含插入结果的对象，通常包含：

  • insert_count：成功插入的数量（应为 3000）。

  • primary_keys：插入数据的主键列表（这里就是 0 到 2999）。

• 注意：此时数据虽然发送到了服务器，但可能还在服务器的内存缓冲区中，尚未完全持久化到磁盘，也未立即对搜索可见。

4、数据落盘

• 作用：这是一个强制同步操作。

• 为什么要 Flush？

  1. 持久化：强制将内存中的增量数据（Insert Log）写入磁盘（Segment 文件），防止服务重启导致数据丢失。

  2. 生成索引：Milvus 的索引构建通常是基于磁盘上的数据段（Segment）进行的。只有 Flush 后生成了 Segment，后续的 create_index 才能索引到这些数据。

  3. 搜索可见性：在某些配置下，未 Flush 的数据可能无法被搜索到。

四、创建索引

下述代码展示了集合hello_milvus的创建过程。

索引定义指定索引类型为：IVF_FLAT倒排文件索引； 衡量两个向量的直线距离使用L2，欧几里得距离。

nlist为聚类中心的数量。

index = {
    "index_type": "IVF_FLAT",
    "metric_type": "L2",
    "params": {"nlist": 128},
}
hello_milvus.create_index("embeddings", index)

1、IVF_FLAT

训练阶段：算法会将所有向量聚类成 nlist 个簇（Cluster），每个簇有一个中心点。

搜索阶段：当你要搜索时，系统先判断你的查询向量离哪几个簇的中心最近，然后只在这些选中的簇里进行暴力搜索，而不是遍历全库。

2、metric_type: L2

含义：欧几里得距离。

作用：定义“相似度”的计算方式。

L2 计算的是空间中两点间的直线距离。距离越小，代表向量越相似。

3、params: {"nlist": 128}

含义：聚类中心的数量。

调优逻辑：

这个参数决定了将数据分成多少份。

经验公式：通常建议 nlist约等于根号n （N 为向量总数）。

影响：nlist 越大，搜索时排查的范围越精细（速度可能变慢，但精度可能提高），训练索引的时间也会变长。

4、执行创建索引

• 参数 1 "embeddings"：指定要为哪个字段建立索引。这里对应 Schema 中定义的向量字段。

• 参数 2 index：传入上面定义的索引蓝图。

• 执行过程

  1. Milvus 会读取集合中已 flush 的数据。

  2. 在后台异步执行索引构建任务（聚类、生成倒排表）。

  3. 构建完成后，索引文件会被持久化存储。

五、加载到内存 (Load)

我们需要将数据从磁盘加载到内存中

hello_milvus.load()

六、搜索与查询

A. 向量相似性搜索 (Vector Search)

下面的向量相似性上搜索目的是：在 embeddings 字段中，寻找与 vectors_to_search 最相似的 limit=3 个向量。

vectors_to_search = entities[-1][-2:] # 取最后2个向量作为查询目标
result = hello_milvus.search(vectors_to_search, "embeddings", search_params, limit=3, ...)

• distance: 0.0: 因为查询向量本身就是数据库里的最后两个向量（ID 2998, 2999），所以距离为 0（完全匹配）。

• latency = 0.2796s: 对于 3000 条数据，这个延迟略高（通常毫秒级），可能是因为第一次加载索引或 Python 启动开销。在百万级数据下，IVF_FLAT 的优势才会体现出来（依然保持毫秒级）。

B. 标量查询 (Scalar Query)

下面的代码展示了标量查询，即筛选 random 字段大于 -14 的记录。

因为不涉及向量计算，速度极快。

result = hello_milvus.query(expr="random > -14", ...)

在返回的数据中，random field 的值全部是 -11.0。

这说明过滤生效了！只有满足标量条件的数据才被返回，即使有其他向量距离更近但不满足 random > -12 的数据，也被排除了。

七、删除与清理 (Delete & Drop)

下面展示了使用delete删除字段（逻辑删除）和drop删除集合。

expr = f"pk in [{entities[0][0]}, {entities[0][1]}]" 
hello_milvus.delete(expr)
utility.drop_collection("hello_milvus")

1、构造表达式

entities[0] 是主键 pk 的列表（即 [0, 1, 2, ..., 2999]）。
entities[0][0] 和 entities[0][1]：分别取出了列表中的前两个 ID，即 0 和 1。
f"pk in [...]"：利用 Python 的 f-string 格式化字符串，最终生成的表达式为 "pk in [0, 1]"。
含义：这是一个布尔表达式，告诉 Milvus“我要操作主键是 0 或者 1 的数据”。

2、执行删除 (hello_milvus.delete(expr))

作用：从集合中移除符合 expr 条件的行。

原理：Milvus 的删除操作通常是逻辑删除。系统会在后台标记这两条数据为“已删除”，它们在后续的搜索和查询中不可见，但物理文件可能不会立即从磁盘消失（需要通过 Compaction 机制进行物理清理）。

限制：Milvus 的删除操作目前仅支持基于主键或标量字段的过滤，不支持基于向量相似度的删除。

3、删除整个集合（utility.drop_collection）

Drop: 彻底删除集合及其所有数据和索引，释放存储空间。

总结

1、本文首先介绍了Milvus 的基础连接。通过 connections.connect 建立客户端与服务器的通信。

2、用代码展示了用FieldSchema创建pk等字段，然后再用 CollectionSchema创建shema,从而用Collection("hello_milvus", schema)创建了集合hello_milvus。

然后介绍了集合（Collection）的定义；再而提出shema包含了列数据+元数据。并介绍了个字段的含义。

3、接着展示使用insert插入数据的代码。介绍了Milvus 的 Python SDK 列式存储的概念；并分析了代码中pk，random，embeddings列插入的数据；以及insert函数的动作含义(将构造好的 entities 发送给 Milvus 服务端)和返回结果insert_result(insert_counts和primary_keys)

还分析了flush的意义（持久化，生成索引，搜索可见性）

4、还用代码展示了集合hello_milvus的创建过程。并介绍了IVF_FLAT索引类型，相似度计算方式metric_type: L2，聚类中心数量params: {"nlist": 128}等概念，以及执行创建索引逻辑。

5、然后介绍将数据加载到内存。

7、接着介绍搜索与查询。包括了向量相似性搜索，以及标量查询。

6、然后展示了使用delete删除字段（逻辑删除）和drop删除集合。剖析:构造表达式，执行删除delete ，删除整个集合drop等原理。

最后，文章详细对比了三种核心数据操作模式：

向量搜索（Search）：基于向量相似度（如欧氏距离）寻找 Top-K 结果；

标量查询（Query）：类似 SQL 的 WHERE 子句，用于基于元数据的精确过滤；

混合搜索（Hybrid Search）：结合了向量相似度与标量过滤（expr），实现更精细的检索；

此外，还简要说明了基于表达式的数据删除（Delete）与集合清理（drop_collection）操作。