Milvus 2.4 使用详解：从零构建向量数据库并实现搜索功能（Python 实战）

发布于：2025-05-13 ⋅ 阅读:(13) ⋅ 点赞:(0)

文章目录

🌟 引言

在大规模向量数据检索场景中，Milvus 是一个非常强大的开源向量数据库。它支持高效的相似性搜索、灵活的数据模型设计以及多语言接口。本文将通过 Python 客户端 pymilvus，带你一步步从零开始使用 Milvus 构建一个简单的向量数据库，并实现插入数据与相似性搜索的完整流程。

🧰 环境准备

依赖安装

首先确保你已经安装了 Milvus 及其 Python SDK：

pip install pymilvus

我们使用的是 Milvus 2.x 版本中的轻量级客户端模式（即 MilvusClient），适合本地测试和小型项目部署。

📁 整体代码结构概览

我们将实现以下核心功能模块：

  • 初始化 Milvus 客户端
  • 创建集合（Collection）及其 Schema
  • 构建向量索引
  • 插入数据
  • 执行向量搜索

🛠️ 核心函数详解

1️⃣ 初始化 Milvus 客户端

def initialize_client(db_path):
    """
    初始化 Milvus 客户端。
    :param db_path: 数据库路径
    :return: Milvus 客户端实例
    """
    return MilvusClient(db_path)
  • db_path：指定本地 SQLite 存储路径，用于单机版 Milvus 轻量运行。
  • MilvusClient：适用于嵌入式或轻量级部署的客户端。

2️⃣ 创建集合 Schema

def create_collection_schema():
    """
    创建集合的 schema。
    :return: 集合的 schema
    """
    return CollectionSchema([
        FieldSchema("id", DataType.INT64, is_primary=True),
        FieldSchema("text", DataType.VARCHAR, max_length=20000),
        FieldSchema("embed", DataType.FLOAT_VECTOR, dim=3),
        FieldSchema("source", DataType.VARCHAR, max_length=2000)
    ])
  • CollectionSchema：定义集合的数据结构。
  • FieldSchema：
    • id：主键字段，必须为 INT64 类型。
    • text：文本内容，最大长度 20000。
    • embed：向量字段，维度为 3。
    • source：数据来源信息，如 PDF 文件名。

3️⃣ 准备索引参数

def prepare_index_params(client):
    """
    准备索引参数。
    :param client: Milvus 客户端实例
    :return: 索引参数
    """
    index_params = client.prepare_index_params()
    index_params.add_index(
        field_name="embed",
        metric_type="COSINE",
        index_type="",  # 根据需要设置具体的索引类型
        index_name="vector_index"
    )
    return index_params
  • metric_type：推荐使用 "COSINE" 表示余弦相似度。
  • index_type：可选值如 "IVF_FLAT", "HNSW" 等，不填时由系统自动选择。
  • index_name：给索引起个名字，便于后续管理。

4️⃣ 删除已存在的集合（可选）

def drop_collection_if_exists(client, collection_name):
    """
    如果集合存在，则删除集合。
    :param client: Milvus 客户端实例
    :param collection_name: 集合名称
    """
    if client.has_collection(collection_name):
        client.drop_collection(collection_name)
  • 用于清理旧数据，方便调试。

5️⃣ 创建集合并建立索引

def create_and_index_collection(client, collection_name, schema, index_params):
    """
    创建集合并为其创建索引。
    :param client: Milvus 客户端实例
    :param collection_name: 集合名称
    :param schema: 集合的 schema
    :param index_params: 索引参数
    """
    client.create_collection(collection_name=collection_name, schema=schema)
    client.create_index(collection_name, index_params)
  • create_collection：根据 schema 创建集合。
  • create_index：应用之前配置好的索引参数。

6️⃣ 插入数据

def insert_data_into_collection(client, collection_name, data_list):
    """
    向集合中插入数据。
    :param client: Milvus 客户端实例
    :param collection_name: 集合名称
    :param data_list: 数据列表（字典格式）
    """
    for data in data_list:
        client.insert(collection_name, data)
  • 支持逐条插入，也可以批量插入（传入 list of dict）。

7️⃣ 加载集合并执行搜索

def load_and_search_collection(client, collection_name, query_vector, output_fields, limit):
    """
    加载集合并执行搜索。
    :param client: Milvus 客户端实例
    :param collection_name: 集合名称
    :param query_vector: 查询向量
    :param output_fields: 输出字段
    :param limit: 搜索结果数量限制
    :return: 搜索结果
    """
    client.load_collection(collection_name)
    return client.search(
        collection_name,
        [query_vector],
        output_fields=output_fields,
        limit=limit
    )
  • load_collection：加载集合到内存以便搜索。
  • search：执行向量搜索，返回最相似的结果。

🧪 主程序逻辑

def main():
    # 初始化客户端
    db_path = 'database/milvus_database/milvus.db'
    name = 'demo'

    data_list1 = [
        {"id": 0, "text": 'abc', "embed": [1, 2, 3], 'source': 'pdf'},
        {"id": 1, "text": '123', "embed": [12, 21, 30], 'source': 'pdfd'}
    ]

    client = build_vector_index(db_path, name, data_list1)

    # 加载集合并执行搜索
    query_vector = [12, 3, 4]  # 查询向量
    output_fields = ["text"]   # 指定输出字段
    limit = 1                  # 搜索结果数量限制

    result = search_vector(client, name, query_vector, output_fields, limit)

    print('\n\n\n {}\n\n\n'.format(result))
  • 构建了一个包含两个向量记录的集合。
  • 插入后进行一次基于 [12, 3, 4] 的搜索，返回最相似的一条记录的 text 字段。

🧾 示例输出解析

假设搜索结果如下：

[
    {
        "id": 1,
        "distance": 0.98,
        "entity": {
            "text": "123"
        }
    }
]
  • id: 匹配记录的 ID。
  • distance: 相似度距离（越小越近）。
  • entity.text: 返回的原始文本内容。

📝 总结

通过上述步骤，我们完成了 Milvus 的基本操作流程：

| 步骤 | 功能 |
| ✅ 初始化客户端 | 连接本地 Milvus 数据库 |
| ✅ 创建 Schema | 定义数据结构 |
| ✅ 创建索引 | 提高向量检索效率 |
| ✅ 插入数据 | 添加向量及元数据 |
| ✅ 搜索向量 | 实现高效相似性匹配 |

🔍 延伸学习建议

  1. 多维向量支持：将 dim=3 改为你自己的 Embedding 维度（如 768）。
  2. 索引类型优化：尝试不同索引类型如 IVF_FLAT, HNSW 来提升性能。
  3. 持久化与集群：考虑升级到 Milvus Standalone 或 Cluster 模式。
  4. 结合 LangChain / LlamaIndex：将 Milvus 集成进 RAG 架构中。

Milvus 完整使用示例（Python）

完整代码如下：

from pymilvus import MilvusClient, CollectionSchema, FieldSchema, DataType
def initialize_client(db_path):
    """
    初始化 Milvus 客户端。
    :param db_path: 数据库路径
    :return: Milvus 客户端实例
    """
    return MilvusClient(db_path)
def create_collection_schema():
    """
    创建集合的 schema。
    :return: 集合的 schema
    """
    return CollectionSchema([
        FieldSchema("id", DataType.INT64, is_primary=True),
        FieldSchema("text", DataType.VARCHAR, max_length=20000),
        FieldSchema("embed", DataType.FLOAT_VECTOR, dim=3),
        FieldSchema("source", DataType.VARCHAR, max_length=2000)
    ])
def prepare_index_params(client):
    """
    准备索引参数。
    :param client: Milvus 客户端实例
    :return: 索引参数
    """
    index_params = client.prepare_index_params()
    index_params.add_index(
        field_name="embed",
        metric_type="COSINE",
        index_type="",  # 根据需要设置具体的索引类型
        index_name="vector_index"
    )
    return index_params
def drop_collection_if_exists(client, collection_name):
    """
    如果集合存在，则删除集合。
    :param client: Milvus 客户端实例
    :param collection_name: 集合名称
    """
    if client.has_collection(collection_name):
        client.drop_collection(collection_name)
def create_and_index_collection(client, collection_name, schema, index_params):
    """
    创建集合并为其创建索引。
    :param client: Milvus 客户端实例
    :param collection_name: 集合名称
    :param schema: 集合的 schema
    :param index_params: 索引参数
    """
    client.create_collection(collection_name=collection_name, schema=schema)
    client.create_index(collection_name, index_params)
def insert_data_into_collection(client, collection_name, data_list):
    """
    向集合中插入数据。
    :param client: Milvus 客户端实例
    :param collection_name: 集合名称
    :param data_list: 数据列表（字典格式）
    """
    for data in data_list:
        client.insert(collection_name, data)
def load_and_search_collection(client, collection_name, query_vector, output_fields, limit):
    """
    加载集合并执行搜索。
    :param client: Milvus 客户端实例
    :param collection_name: 集合名称
    :param query_vector: 查询向量
    :param output_fields: 输出字段
    :param limit: 搜索结果数量限制
    :return: 搜索结果
    """
    client.load_collection(collection_name)
    return client.search(
        collection_name,
        [query_vector],
        output_fields=output_fields,
        limit=limit
    )

def build_vector_index(db_path, db_name,data_dict_list):
    """
    为指定字段构建向量索引。
    :param client: Milvus 客户端实例
    :param collection_name: 集合名称
    :param field_name: 字段名称
    :param index_type: 索引类型
    """  
    client = initialize_client(db_path)
    # 创建集合 schema 和索引参数
    schema = create_collection_schema()
    index_params = prepare_index_params(client)

    # 删除已存在的集合（如果存在）
    drop_collection_if_exists(client, db_name)

    # 创建并索引集合
    create_and_index_collection(client, db_name, schema, index_params)

     # 插入数据
    insert_data_into_collection(client, db_name, data_dict_list)
    return client

def search_vector(client, collection_name, query_vector, output_fields, limit):
    """
    加载集合并执行搜索。
    :param client: Milvus 客户端实例
    :param collection_name: 集合名称
    :param query_vector: 查询向量
    :param output_fields: 输出字段

    """
    client.load_collection(collection_name)
    return client.search(
        collection_name,
        [query_vector],
        output_fields=output_fields,
        limit=limit
    )

def main():
    # 初始化客户端
    db_path = 'milvus_database/milvus.db'
    name='demo'
    data_list1 = [
        {"id": 0, "text": 'abc', "embed": [1, 2, 3],'source':'pdf'},
        {"id": 1, "text": '123', "embed": [12, 21, 30],'source':'pdfd'}
    ]
    
    client = build_vector_index(db_path, name, data_list1)

    # 加载集合并执行搜索
    query_vector = [12, 3, 4] # 查询向量
    output_fields = ["text"] # 指定输出字段
    limit = 1 # 搜索结果数量限制

    result = search_vector(client, name, query_vector, output_fields, limit)
    
    print('\n\n\n {}\n\n\n'.format(result))
  
if __name__ == "__main__":
    main()

结果如下：
在这里插入图片描述

网站公告

今日签到

点亮在社区的每一天
去签到

热门文章

最新发布