【大模型_向量数据库_Milvus-Milvus快速入门】

瀧爷1939

253人浏览 · 2026-05-28 17:30:37

瀧爷1939 · 2026-05-28 17:30:37 发布

Milvus 是当前最主流的开源向量数据库，2.x 版本已完全云原生、生产可用，本教程基于 Milvus 2.4 稳定版 + Python SDK 编写，从环境搭建到可运行Demo全流程覆盖，新手10分钟即可跑通。

一、前置准备

方式1：本地Docker部署（推荐新手，零成本）

Milvus 官方提供了单节点Standalone版本的Docker镜像，一键启动即可：

# 1. 拉取镜像并启动容器，挂载本地目录持久化数据（避免容器删除后数据丢失）
docker run -d \
  --name milvus-standalone \
  -p 19530:19530 \ # Milvus默认gRPC端口，客户端连接用
  -p 9091:9091 \   # 监控端口，可选
  -v /tmp/milvus:/var/lib/milvus \ # 数据持久化到本地/tmp/milvus目录
  milvusdb/milvus:v2.4.0 milvus run standalone

# 2. 验证容器是否启动成功
docker ps | grep milvus
# 看到milvus-standalone容器状态为Up即成功

方式2：云托管部署（免运维，适合快速验证）

直接用Zilliz Cloud（Milvus官方云服务）的免费实例：注册后创建免费实例，会得到集群连接地址、用户名、密码，无需本地部署。

二、环境安装

安装Python SDK和依赖库，建议用Python 3.8+版本：

# 安装Milvus Python SDK，版本和Milvus服务端保持一致（2.4.x）
pip install pymilvus==2.4.0 numpy

三、完整Demo：商品语义检索

我们用「商品向量检索」场景演示全流程：把商品信息向量化存入Milvus，后续输入query可以检索到语义最相似的商品，同时支持类似MySQL的价格过滤。

完整可运行代码

from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType, utility
import numpy as np

# ---------------------- 步骤1：连接Milvus服务 ----------------------
# 本地部署用localhost，云服务替换为集群地址+用户名密码
connections.connect(
    alias="default",
    host="localhost",
    port="19530"
    # 云服务需要额外加：
    # user="your_username",
    # password="your_password"
)

# ---------------------- 步骤2：创建集合（相当于MySQL的表） ----------------------
# 2.1 定义集合的字段结构
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),  # 主键，自增
    FieldSchema(name="product_name", dtype=DataType.VARCHAR, max_length=100),       # 商品名称
    FieldSchema(name="category", dtype=DataType.VARCHAR, max_length=50),            # 商品分类
    FieldSchema(name="price", dtype=DataType.FLOAT),                                # 商品价格
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=128)             # 向量字段，维度128（可根据Embedding模型调整）
]

# 2.2 定义集合Schema，添加描述
schema = CollectionSchema(
    fields=fields,
    description="商品向量检索集合"
)

# 2.3 创建集合，如果已存在则先删除（仅Demo用，生产环境不要随便删）
collection_name = "product_demo"
if utility.has_collection(collection_name):
    utility.drop_collection(collection_name)
collection = Collection(name=collection_name, schema=schema)

print(f"集合 {collection_name} 创建成功")

# ---------------------- 步骤3：插入模拟数据 ----------------------
# 模拟6个商品数据，实际使用时需要把文本通过Embedding模型转为向量
products = [
    {"product_name": "iPhone 15", "category": "手机", "price": 5999},
    {"product_name": "小米14", "category": "手机", "price": 3999},
    {"product_name": "索尼WH-1000XM5", "category": "耳机", "price": 2499},
    {"product_name": "AirPods Pro 2", "category": "耳机", "price": 1899},
    {"product_name": "MacBook Pro 14", "category": "电脑", "price": 14999},
    {"product_name": "联想小新Pro14", "category": "电脑", "price": 4999}
]

# 生成模拟向量（Demo用，实际需要替换为真实Embedding）
# 实际使用时：用Embedding模型把商品描述转为128/1536维向量
def get_embedding(text):
    # 这里用随机向量模拟，实际替换为真实Embedding逻辑
    return np.random.rand(128).tolist()

# 组装插入数据
insert_data = [
    [p["product_name"] for p in products],
    [p["category"] for p in products],
    [p["price"] for p in products],
    [get_embedding(f"{p['category']} {p['product_name']}") for p in products]
]

# 执行插入
collection.insert(insert_data)
collection.flush() # 强制刷盘，确保数据可见
print(f"插入 {len(products)} 条商品数据成功")

# ---------------------- 步骤4：创建向量索引（加速检索） ----------------------
# Milvus默认用暴力检索，数据量超过1万后必须建索引才能毫秒级返回
index_params = {
    "index_type": "HNSW",  # 最常用的ANN索引，平衡精度和速度
    "metric_type": "COSINE", # 距离度量：余弦相似度，适合语义检索
    "params": {"M": 16, "efConstruction": 64} # 索引参数，新手用默认即可
}
collection.create_index(field_name="embedding", index_params=index_params)
print("向量索引创建成功")

# ---------------------- 步骤5：加载集合到内存（检索前必须执行） ----------------------
collection.load()
print("集合加载到内存成功")

# ---------------------- 步骤6：向量检索 ----------------------
# 模拟用户query：想要找「便宜的无线耳机」，实际使用时把query转为向量
query_vector = get_embedding("便宜的无线耳机")

# 基础检索：返回Top3最相似的商品
print("\n=== 基础向量检索结果 ===")
results = collection.search(
    data=[query_vector], # 查询向量，支持批量查询
    anns_field="embedding", # 指定检索的向量字段
    param={"metric_type": "COSINE", "params": {"ef": 64}}, # 检索参数，ef越大精度越高速度越慢
    limit=3, # 返回Top3
    output_fields=["product_name", "category", "price"] # 需要返回的标量字段
)

# 打印结果
for hit in results[0]:
    print(f"商品：{hit.entity.product_name}，分类：{hit.entity.category}，价格：{hit.entity.price}元，相似度：{hit.distance:.4f}")

# ---------------------- 步骤7：混合检索（向量+标量过滤，类似MySQL的WHERE） ----------------------
# 场景：找「价格低于5000元」的相似商品，结合向量相似度和结构化过滤
print("\n=== 带价格过滤的混合检索结果 ===")
results = collection.search(
    data=[query_vector],
    anns_field="embedding",
    param={"metric_type": "COSINE", "params": {"ef": 64}},
    limit=3,
    output_fields=["product_name", "category", "price"],
    filter="price < 5000" # 标量过滤条件，语法类似MySQL的WHERE
)

for hit in results[0]:
    print(f"商品：{hit.entity.product_name}，分类：{hit.entity.category}，价格：{hit.entity.price}元，相似度：{hit.distance:.4f}")

# ---------------------- 步骤8：资源清理（Demo必做，避免内存泄漏） ----------------------
collection.release() # 释放内存中的集合
collection.drop() # 删除集合
connections.disconnect("default") # 断开连接
print("\n资源清理完成")

四、Demo运行说明

1. 为什么Demo用随机向量？

上面的Demo为了简化流程用了随机向量模拟Embedding，实际生产使用时需要把文本/图片通过Embedding模型转为真实向量，替换get_embedding函数即可，比如用国内开源的BGE模型：

# 安装Embedding模型依赖
pip install sentence-transformers

from sentence_transformers import SentenceTransformer

# 加载BGE向量模型，会把文本转为768维向量，注意和集合的dim参数一致
model = SentenceTransformer("BAAI/bge-base-zh")

# 替换原来的get_embedding函数
def get_embedding(text):
    return model.encode(text).tolist()

2. 云服务部署的连接修改

如果用Zilliz Cloud免费实例，只需要修改connections.connect部分的参数，替换为实例的地址、用户名、密码即可，其余代码完全一致。

五、核心概念回顾（呼应之前和MySQL的对比）

概念	对应MySQL概念	说明
Collection（集合）	表	存储同类型数据的容器
Field（字段）	列	集合的属性，比如商品名、价格、向量
主键	主键	唯一标识每条数据
索引	索引	加速向量检索的结构，类似MySQL的B+树索引，但针对高维向量优化
标量过滤	WHERE条件	支持对结构化字段做过滤，和MySQL逻辑一致
向量检索	无对应	输入向量返回TopK最相似的结果，是向量数据库的核心能力