一、先安装 Python 客户端

pip install pymilvus==2.6.0b0

这里最好让 pymilvus 版本和 Milvus 服务版本尽量对应。

二、连接到 Milvus

最基础的连接方式如下：

from pymilvus import connections

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

三、创建集合

示例里使用的是 8 维向量，便于演示。

核心步骤包括：

  1. 定义字段；
  2. 定义集合 Schema；
  3. 创建 Collection。

字段示例：

FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True)
FieldSchema(name="title", dtype=DataType.VARCHAR, max_length=200)
FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=8)

四、准备和插入测试数据

示例里准备了 5 条产品数据，例如：

  • iPhone 15
  • MacBook Pro
  • AirPods Pro
  • iPad Air
  • Apple Watch

插入方式通常是按字段列表组织：

insert_data = [
    titles,
    test_vectors
]
product_collection.insert(insert_data)
product_collection.flush()

flush() 的作用是让数据落盘并可搜索。

五、创建索引

索引是 Milvus 检索性能的关键。

示例里用了：

  • index_type = IVF_FLAT
  • metric_type = L2
index_params = {
    "index_type": "IVF_FLAT",
    "metric_type": "L2",
    "params": {"nlist": 8}
}

创建索引后，还需要把集合加载到内存：

product_collection.load()

六、向量搜索

搜索时你需要提供：

  • 查询向量；
  • 搜索参数；
  • 返回条数；
  • 需要输出的字段。

示例：

search_vector = [[0.13, 0.24, 0.46, 0.68, 0.88, 0.02, 0.35, 0.57]]
search_params = {"metric_type": "L2", "params": {"nprobe": 3}}

这一步的结果会返回最接近的若干条记录。

七、更新和删除数据

Milvus 也支持 upsert 和删除操作。

7.1 更新

product_collection.upsert(
    data=[
        [existing_id],
        ["iPhone 15 Pro Max"],
        [[0.14, 0.25, 0.47, 0.69, 0.87, 0.03, 0.36, 0.58]]
    ]
)

7.2 删除

product_collection.delete(f"id in {delete_ids}")

八、做标量过滤查询

除了向量检索，还可以结合普通字段做过滤：

result = product_collection.query(
    expr="title like 'iPhone%'",
    output_fields=["id", "title", "embedding"]
)

这让 Milvus 不只是“查相似向量”，还可以搭配标量字段一起使用。

九、列出集合和查看数据

你还可以通过 utility.list_collections() 列出所有集合，或者用 collection.query() 查看指定集合里的数据。

这对调试特别有帮助，尤其是在刚开始建表和导入样本的时候。

十、为什么这个实验很有价值

一旦你把这条链路跑通了，RAG 里的很多抽象概念就会具体起来：

  • 向量长什么样；
  • 集合怎么存；
  • 检索怎么做；
  • 更新和过滤怎么配合。

这比只看原理图更能帮助你真正理解向量数据库在 RAG 里的角色。