当前位置：首页 > news >正文

Milvus：开源向量数据库的初识

news 2025/7/22 8:40:13

随着人工智能、自然语言处理和推荐系统的发展，海量非结构化数据（如文本、图像、音频等）的相似度搜索变得至关重要。传统的关系型数据库难以高效处理这种高维向量数据。Milvus 作为一款开源向量数据库，专为海量向量数据的高效存储和近似最近邻（ANN）搜索设计，广泛应用于语义搜索、智能问答、图像检索等领域。

一、Milvus 理论基础

1. 向量搜索和近似最近邻（ANN）问题

向量搜索的目标是快速找到与查询向量在高维空间中最相似的向量集合。常用的相似度度量包括：

欧氏距离（L2）
余弦相似度（COSINE）
内积（Inner Product，IP）

精确搜索的计算量随数据量呈线性增长，效率低下。ANN算法通过索引结构在精度和速度间做权衡，实现高效的近似搜索。

2. Milvus 的索引结构

Milvus 支持多种索引类型：

FLAT：暴力搜索，精确但慢，适合小规模数据。
IVF（Inverted File）系列：包括 IVF_FLAT、IVF_PQ 等，利用聚类将数据分桶，先粗筛再精检，实现高效搜索。
HNSW（Hierarchical Navigable Small World Graph）：图结构索引，查询速度快，精度高。
DISKANN：支持超大规模磁盘索引。

3. 粗-细粒度搜索机制

在 IVF 索引中，Milvus 通过设置参数 nlist 将数据划分成多个“簇”（大片），查询时通过 nprobe 选择部分簇（小片）进行精细搜索，这种“先切大片后切小片”的两级过滤显著提升查询效率。

4. 混合搜索

Milvus 支持同时对向量字段和结构化字段进行过滤，实现更精准的多条件检索。比如在向量搜索基础上加条件筛选日期、类别等。

5. 多语言和多平台支持

Milvus 提供 Python、Java、Go SDK 和 RESTful API，方便集成到不同业务系统。同时支持可视化管理工具 Attu，简化运维。

二、Milvus 实践操作指南

下面我们使用 Python SDK pymilvus，以一个简单的文本向量存储和检索示例，演示 Milvus 的核心操作。

1. 安装 Milvus 和 Python SDK

# 安装 pymilvus
pip install pymilvus# 启动 Milvus（以 Docker 方式）
docker run -d --name milvus-standalone -p 19530:19530 -p 19121:19121 milvusdb/milvus:v2.2.9-20230913-d3ab1ca

2. 连接 Milvus

from pymilvus import connectionsconnections.connect("default", host="localhost", port="19530")

3. 定义 Collection（包含 ID 和向量字段）

from pymilvus import FieldSchema, CollectionSchema, DataType, Collectionfields = [FieldSchema(name="id", dtype=DataType.INT64, is_primary=True),FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=1024),FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=768)
]schema = CollectionSchema(fields, description="文本向量检索示例")
collection = Collection("text_collection", schema)

4. 插入数据

import randomembeddings = [[random.random() for _ in range(768)] for _ in range(3)]
ids = [1, 2, 3]
texts = ["登录失败问题", "页面UI异常", "密码重置功能"]collection.insert([ids, texts, embeddings])

5. 创建索引

collection.create_index(field_name="embedding",index_params={"index_type": "IVF_FLAT","metric_type": "COSINE","params": {"nlist": 128}}
)

6. 加载 Collection

collection.load()

7. 向量搜索

query_embedding = [random.random() for _ in range(768)]results = collection.search(data=[query_embedding],anns_field="embedding",param={"nprobe": 10},limit=3,output_fields=["id", "text"]
)for hit in results[0]:print(f"id: {hit.entity.get('id')}, text: {hit.entity.get('text')}, distance: {hit.distance}")