Docker Compose 运行 Milvus (Mac) 并与 python 连接测试

Milvus 简单来说是一个向量化数据库，有 lite、standalone、distributed 三个版本，其中 lite 轻量化版本直接安装 python SDK 运行即可（pip install pymilvus），本文讲述如何借助 docker，在 mac 环境下配置并建立与 python 的连接。

安装 Docker

在官网下载 dmg 包 Mac | Docker Docs

看清楚自己的 mac 是什么芯片，m系列就和我一样选第一个

下载后跟其他软件一样安装即可，安装完成后打开终端，输入 docker -v 验证是否安装成功

下载 Docker 软件会看到类似的界面（我这里是运行了一些容器，大家刚安装正常是没有东西的）

安装 Milvus

拉取 docker compose 文件

wget https://github.com/milvus-io/milvus/releases/download/v2.5.14/milvus-standalone-docker-compose.yml -O docker-compose.yml

在终端输入上面的指令并回车，会下载一个 ocker-compose.yml

启动 Milvus

在终端输入下面的指令

docker compose up -d

运行后打开刚才下载的 docker 软件，就开始看到名为milvus- standalone、milvus-minio 和milvus-etcd的容器已经启动。

简单介绍一下这三个组件（截取自官网）

• milvus-etcd容器不向主机暴露任何端口，并将其数据映射到当前文件夹中的volumes/etcd。
• milvus-minio容器使用默认身份验证凭据在本地为端口9090和9091提供服务，并将其数据映射到当前文件夹中的volumes/minio。
• Milvus-standalone容器使用默认设置为本地19530端口提供服务，并将其数据映射到当前文件夹中的volumes/milvus。

在终端运行 docker-compose ps 也能看到当前容器状态

Python 连接测试

首先在终端运行下面指令安装 python sdk（sdk 其实就是个工具包）

pip install pymilvus==2.6.0b0

如果 PyMilvus 安装正确，运行以下命令时不会出现异常。

python3 -c "from pymilvus import Collection"

贴出我的测试代码

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Milvus 连接测试代码
演示基本的连接、创建集合、插入数据、搜索等功能
"""

import time
import random
import numpy as np
from pymilvus import (
connections,
utility,
FieldSchema,
CollectionSchema,
DataType,
Collection,
)

def test_milvus_connection():
    """测试Milvus连接"""
    print("正在连接Milvus...")

    try:
        # 连接到Milvus服务器
        connections.connect(
            alias="default",  # 连接别名
            host="localhost",  # 主机地址
            port="19530"       # 端口号
        )
        print("Milvus连接成功！")
        return True
    except Exception as e:
        print(f"Milvus连接失败: {e}")
        return False

def create_collection():
    """创建测试集合"""
    print("\n正在创建测试集合...")

    # 定义字段模式
    fields = [
        FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
        FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=128),  # 128维向量
        FieldSchema(name="label", dtype=DataType.VARCHAR, max_length=100)
    ]

    # 创建集合模式
    schema = CollectionSchema(
        fields=fields,
        description="测试集合 - 用于演示Milvus功能"
    )

    # 创建集合
    collection_name = "test_collection"
    collection = Collection(
        name=collection_name,
        schema=schema,
        using="default"
    )

    # 创建索引
    print("正在创建向量索引...")
    index_params = {
        "metric_type": "L2",           # 距离度量类型：L2欧几里得距离
        "index_type": "IVF_FLAT",      # 索引类型：倒排文件
        "params": {"nlist": 128}       # 聚类中心数量
    }
    collection.create_index(
        field_name="vector",
        index_params=index_params
    )

    print(f"集合 '{collection_name}' 创建成功！")
    return collection

def insert_sample_data(collection):
    """插入示例数据"""
    print("\n正在插入示例数据...")

    # 生成随机向量数据
    num_entities = 1000
    vectors = np.random.rand(num_entities, 128).astype(np.float32)
    labels = [f"样本_{i}" for i in range(num_entities)]

    # 准备插入数据
    entities = [
        vectors,  # 向量数据
        labels    # 标签数据
    ]

    # 插入数据
    collection.insert(entities)
    collection.flush()  # 确保数据持久化

    print(f"成功插入 {num_entities} 条数据！")
    return vectors

def search_similar_vectors(collection, query_vector):
    """搜索相似向量"""
    print("\n正在搜索相似向量...")

    # 加载集合到内存
    collection.load()

    # 搜索参数
    search_params = {
        "metric_type": "L2",
        "params": {"nprobe": 10}
    }

    # 执行搜索
    results = collection.search(
        data=[query_vector],
        anns_field="vector",
        param=search_params,
        limit=5,  # 返回前5个最相似的结果
        output_fields=["label"]
    )

    print("搜索结果:")
    for i, hits in enumerate(results):
        print(f"查询向量 {i + 1}:")
        for j, hit in enumerate(hits):
            print(f"  排名 {j + 1}: ID={hit.id}, 距离={hit.distance:.4f}, 标签={hit.entity.get('label')}")

    return results

def get_collection_stats(collection):
    """获取集合统计信息"""
    print("\n集合统计信息:")

    # 获取实体数量
    num_entities = collection.num_entities
    print(f"  实体数量: {num_entities}")
    
    # 获取索引信息
    index_info = collection.index().params
    print(f"  索引类型: {index_info.get('index_type', 'N/A')}")
    print(f"  距离度量: {index_info.get('metric_type', 'N/A')}")

def cleanup(collection_name):
    """清理测试数据"""
    print(f"\n正在清理测试集合 '{collection_name}'...")
    
    try:
        utility.drop_collection(collection_name)
        print(f"集合 '{collection_name}' 已删除")
    except Exception as e:
        print(f"清理时出现警告: {e}")

def main():
    """主函数 - 执行完整的测试流程"""
    print("开始Milvus功能测试演示")
    print("=" * 50)
    
    # 1. 测试连接
    if not test_milvus_connection():
        return
    
    # 2. 创建集合
    collection = create_collection()
    
    # 3. 插入示例数据
    vectors = insert_sample_data(collection)
    
    # 4. 获取统计信息
    get_collection_stats(collection)
    
    # 5. 搜索相似向量
    # 使用第一个向量作为查询向量
    query_vector = vectors[0].tolist()
    search_similar_vectors(collection, query_vector)
    
    # 6. 清理测试数据（可选）
    # cleanup("test_collection")
    
    print("\nMilvus功能测试完成！")
    print("提示：如需保留测试数据，请注释掉cleanup函数调用")

if __name__ == "__main__":
    main()
    

运行效果：

到此 mivlus 就是拉取并启动完成啦，后面可以参考官方文档去学习怎么样开发

参考资料：

Docker—苹果Mac安装Docker的两种方式-CSDN博客

Build RAG with Milvus | Milvus Documentation