【Milvus】向量数据库pymilvus利用教程

以下是根据 Milvus 官方文档整理的具体 PyMilvus 利用教程，基于 Milvus 2.5.x 版本：

---

PyMilvus 利用教程

目次

• 安装与情况准备
  
• 毗连 Milvus 服务
  
• 数据模型基础概念
  
• 创建集合（Collection）
  
• 插入数据
  
• 创建索引
  
• 向量搜索
  
• 删除操作
  
• 完备示例
  
• 注意事项
  

---

安装与情况准备

搭建 Milvus 服务 基于Docker

1. # 记得提前安装Docker
2. curl -sfL https://raw.githubusercontent.com/milvus-io/milvus/master/scripts/standalone_embed.sh -o standalone_embed.sh
4. bash standalone_embed.sh start
6. # 数据可视化工具
7. docker run -p 8000:3000 -e MILVUS_URL=你的IP:19530 zilliz/attu:v2.5

复制代码

1. pip install pymilvus

复制代码

要求：

• Python 3.10+
  
• Milvus 2.5.x 服务（单机版或集群）
  

---

毗连 Milvus 服务

1. from pymilvus import connections
3. # 连接单机版
4. connections.connect(
5.     alias="default",
6.     host="localhost",
7.     port="19530"
8. )
10. # 连接集群或云服务（如Zilliz Cloud）
11. # connections.connect(
12. #     alias="cloud",
13. #     uri="https://xxx.api.region.zillizcloud.com",
14. #     token="your_api_key"
15. # )

复制代码

---

数据模型基础概念

• Collection: 类似数据库的表，包罗多个字段
  
• Schema: 界说字段类型和束缚
  
• Partition: 数据分区，用于优化查询性能
  
• Index: 加快向量搜索的索引结构
  

---

创建集合（Collection）

1. from pymilvus import (
2.     FieldSchema, CollectionSchema, DataType,
3.     Collection
4. )
6. # 定义字段
7. fields = [
8.     FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
9.     FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=128),
10.     FieldSchema(name="age", dtype=DataType.INT32)
11. ]
13. # 创建Schema
14. schema = CollectionSchema(fields, description="人脸特征向量库")
16. # 创建Collection
17. collection = Collection(name="face_db", schema=schema)

复制代码

参数说明：

• auto_id: 是否自动生成主键
  
• dim: 向量维度（必须与后续插入数据维度一致）
  

---

插入数据

1. import random
3. # 生成随机数据
4. num_entities = 1000
5. vectors = [[random.random() for _ in range(128)] for _ in range(num_entities)]
6. ages = [random.randint(18, 65) for _ in range(num_entities)]
8. # 构造插入数据
9. data = [
10.     vectors,  # 对应embedding字段
11.     ages       # 对应age字段
12. ]
14. # 插入数据
15. insert_result = collection.insert(data)
17. # 获取自动生成的ID
18. print(insert_result.primary_keys)

复制代码

---

创建索引

1. index_params = {
2.     "index_type": "IVF_FLAT",
3.     "metric_type": "L2",
4.     "params": {"nlist": 128}
5. }
7. collection.create_index(
8.     field_name="embedding",
9.     index_params=index_params
10. )

复制代码

常用索引类型：

• FLAT: 精确搜索
  
• IVF_FLAT: 均衡型
  
• HNSW: 高召回率
  
• DISKANN: 磁盘存储优化
  

---

向量搜索

1. # 加载集合到内存
2. collection.load()
4. # 准备搜索向量
5. search_vector = [random.random() for _ in range(128)]
7. # 构建搜索参数
8. search_params = {
9.     "metric_type": "L2",
10.     "params": {"nprobe": 10}
11. }
13. # 执行搜索
14. results = collection.search(
15.     data=[search_vector],
16.     anns_field="embedding",
17.     param=search_params,
18.     limit=5,
19.     output_fields=["age"]  # 返回的额外字段
20. )
22. # 解析结果
23. for hits in results:
24.     for hit in hits:
25.         print(f"ID: {hit.id}, 距离: {hit.distance}, Age: {hit.entity.get('age')}")

复制代码

---

删除操作

1. # 删除实体
2. expr = "age >= 60"
3. collection.delete(expr)
5. # 删除集合
6. collection.drop()

复制代码

---

完备示例

1. from pymilvus import connections, FieldSchema, CollectionSchema, DataType, Collection
3. # 连接服务
4. connections.connect(host='localhost', port='19530')
6. # 创建集合
7. fields = [
8.     FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
9.     FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=128)
10. ]
11. schema = CollectionSchema(fields)
12. collection = Collection("test_collection", schema)
14. # 插入数据
15. data = [[[random.random() for _ in range(128)] for _ in range(1000)]]
16. collection.insert(data)
18. # 创建索引
19. index_params = {"index_type": "IVF_FLAT", "metric_type": "L2", "params": {"nlist": 100}}
20. collection.create_index("vector", index_params)
21. collection.load()
23. # 搜索
24. search_params = {"metric_type": "L2", "params": {"nprobe": 10}}
25. results = collection.search(
26.     data=[[0.5]*128],
27.     anns_field="vector",
28.     param=search_params,
29.     limit=3
30. )
32. # 输出结果
33. print("搜索结果:")
34. for hits in results:
35.     for hit in hits:
36.         print(f"ID: {hit.id}, 距离: {hit.distance}")
38. # 清理
39. collection.drop()

复制代码

---

注意事项

• 版本兼容性：确保 PyMilvus 版本与 Milvus 服务端版本匹配
  
• 资源管理：
  
  
  
  • 搜索前必须调用 load() 加载集合
    
  • 大数据量时注意内存利用
    
  
  
• 索引选择：根据数据规模和性能需求选择符合索引类型
  
• 数据预处置惩罚：确保向量维度与 schema 界说一致
  
• 分页查询：大数据量查询利用 offset + limit 分页
  

官方文档参考：

• Milvus Documentation
  
• PyMilvus API Reference
  

建议结合具体业务需求调解参数，并针对实际数据量进行性能测试。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。