Python脚本数据向量数据库Milvus如何用

wen 实用脚本 2026-07-18 3

Python脚本高效对接Milvus向量数据库：从入门到实战

目录导读

Milvus与Python生态概述
环境搭建与依赖安装
连接与集合管理
向量数据插入与索引构建
相似度搜索与结果解析
性能优化与常见坑点
常见问题问答（FAQ）

Milvus与Python生态概述

Milvus是一款开源的向量数据库，专为海量向量数据的存储、索引与检索设计，广泛应用于推荐系统、图像搜索、自然语言处理（NLP）等场景，Python作为数据科学领域的首选语言，通过pymilvus库能与Milvus高效交互。

Python脚本数据向量数据库Milvus如何用

本教程将基于Milvus 2.x版本与Python 3.8+环境，演示如何编写脚本完成从环境配置到线上检索的全流程，我们将以“商品推荐场景”为例：将商品文本特征转换为向量后存入Milvus,并在用户输入查询时快速返回相似商品ID。

关键概念：Collection（集合类似数据库表）、Partition（分区类似表分区）、Index（索引加速搜索）、Entity（实体即数据行）。

环境搭建与依赖安装

1 启动Milvus服务

推荐使用Docker快速部署Milvus standalone（单机版）：

# 拉取镜像并启动
docker run -d --name milvus \
  -p 19530:19530 \
  -p 9091:9091 \
  milvusdb/milvus:2.3.3-latest

验证服务是否启动：访问 http://localhost:9091/api/v1/health 应返回 {"status":"ok"}。

2 Python依赖安装

pip install pymilvus==2.3.3 numpy

注意：若使用GPU加速索引，需额外安装 pymilvus[gpu],但本教程仅适用CPU版本。

连接与集合管理

1 连接Milvus服务

from pymilvus import connections, utility
# 连接到本地Milvus
connections.connect(host="localhost", port="19530")
print(f"服务状态: {utility.get_server_version()}")

2 创建Collection（集合）

定义商品文本向量维度为128（假设使用Sentence-Bert模型生成），并指定主键（ID）与向量字段：

from pymilvus import CollectionSchema, FieldSchema, DataType, Collection
# 定义字段
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=False),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=128),
    FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=512)  # 原文本保留
]
schema = CollectionSchema(fields, description="商品向量库")
collection = Collection(name="product_vectors", schema=schema)
print(f"集合已创建: {collection.name}")

关键点：auto_id=False 表示手动指定主键，便于业务映射；dim 必须与模型输出维度一致。

向量数据插入与索引构建

1 准备模拟数据

import numpy as np
# 模拟10条商品向量（维度128）
np.random.seed(42)
vectors = np.random.rand(10, 128).astype(np.float32)
ids = list(range(1, 11))
texts = [f"商品{id}: 优质产品示例" for id in ids]
entities = [ids, vectors, texts]
collection.insert(entities)
collection.flush()  # 强制落盘
print(f"已插入 {collection.num_entities} 条数据")

2 创建索引（优化搜索速度）

搜索性能取决于索引类型，本场景使用IVF_FLAT（聚类+暴力搜索平衡方案）：

index_params = {
    "metric_type": "L2",  # 欧氏距离，也可用IP（内积）或COSINE
    "index_type": "IVF_FLAT",
    "params": {"nlist": 128}  # 聚类中心数，越大精度越高但速度略降
}
collection.create_index(field_name="embedding", index_params=index_params)
print("索引创建完成")

调优建议：数据量超过100万时，改用IVF_SQ8或HNSW索引可平衡内存与速度；如果追求极致准确率，可使用FLAT（暴力搜索）。

相似度搜索与结果解析

1 加载集合与执行查询

collection.load()  # 将数据加载到内存
# 模拟一个查询向量（假设来自用户输入）
query_vector = np.random.rand(1, 128).astype(np.float32)
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", "text"]  # 返回原始文本
)

2 解析搜索结果

for hits in results:
    for hit in hits:
        print(f"ID: {hit.id}, 距离: {hit.distance:.4f}, 文本: {hit.entity.get('text')}")

输出示例：

ID: 5, 距离: 2.3456, 文本: 商品5: 优质产品示例
ID: 8, 距离: 3.0123, 文本: 商品8: 优质产品示例
ID: 2, 距离: 3.7890, 文本: 商品2: 优质产品示例

注意事项：limit 参数控制返回数量，但若数据量极大且索引参数不合理,可能返回结果不足。

性能优化与常见坑点

1 批量插入替代逐条插入

# 推荐方案：一次插入数千条
batch_size = 1000
for i in range(0, total_data, batch_size):
    batch_ids = ids[i:i+batch_size]
    batch_vecs = vectors[i:i+batch_size]
    collection.insert([batch_ids, batch_vecs])

2 避免重复创建索引

重复索引会导致资源浪费,建议在脚本中先检查：

if not collection.has_index():
    collection.create_index(...)

3 连接池与超时配置

生产环境需设置连接超时,防止长时间阻塞：

connections.connect(
    host="localhost",
    port="19530",
    timeout=30  # 单位秒
)

4 常见坑点

向量维度不匹配：插入向量的维度必须与Collection定义一致，否则报错DataDimensionMismatch。
浮点精度问题：Milvus只接受float32，如使用float64需手动转换。
索引未加载：搜索前必须执行collection.load()，否则提示No available index。

常见问题问答（FAQ）

Q1：Milvus与Faiss有什么区别？

A：Faiss是一个向量搜索库，Milvus是一个完整的数据库系统，Milvus原生支持数据持久化、分布式、多副本、备份恢复等功能，适合企业级应用；Faiss更适合单机、内存场景的纯计算需求，如果你的项目需要快速原型开发且数据量不大，Faiss更轻量；若涉及数据管理、高可用,Milvus更合适。

Q2：为什么我的搜索结果始终为空？

A：可能原因：

集合中数据量少于查询的limit值（若只有10条数据，却要求返回20条）。
索引类型不适合高精确度场景（例如用IVF_FLAT且nprobe过小，导致未找到任何结果），尝试将limit改为1，并增大nprobe。
查询向量与库中向量距离过大，Milvus会返回空结果,您可先打印查询向量与某个库中向量的距离作为参考。

Q3：如何删除指定ID的向量？

A：使用 collection.delete(expr="id in [1,2,3]")，注意：删除操作后需调用 collection.compact() 释放磁盘空间,否则索引不会立即变小。

Q4：支持多字段过滤吗？

A：支持，Milvus 2.x支持标量字段（如整数、字符串）的过滤，可结合向量搜索实现混合查询,例如只查询价格低于100元的商品向量：

collection.search(
    data=query_vector,
    anns_field="embedding",
    param=search_params,
    expr="price < 100",
    limit=10
)

Q5：如何查看当前Milvus占用的内存和磁盘？

A：通过Milvus的监控接口或API：utility.get_collection_info("product_vectors") 返回集合元数据，但内存使用需通过服务端日志或第三方监控工具（如Prometheus+Grafana）,社区版暂时不提供实时内存查看功能。

通过本文的Python脚本实战，您已掌握连接Milvus、创建集合、插入数据、构建索引并执行搜索的核心流程，实际生产环境中，请结合数据规模（百万级建议分片、千万级建议分布式集群）、查询延迟要求（<10ms用HNSW，<100ms用IVF）以及硬件资源（显存大小）调整索引参数，建议使用pymilvus的bulk_insert接口处理亿级数据,可显著提升吞吐量。

（全文完）