Milvus 向量数据库

Milvus 是一款开源的向量数据库，专为 AI 应用设计，支持海量向量数据的高性能检索。

为什么选择 Milvus

• 开源（Apache 2.0）
• 支持十亿级向量
• 支持混合检索（向量 + 标量过滤）
• 支持多向量索引类型
• 有云服务（Zilliz Cloud）

安装（Docker）

# 下载 docker-compose 文件
curl -sfL https://raw.githubusercontent.com/milvus-io/milvus/master/scripts/standalone_embed.sh -o standalone_embed.sh
bash standalone_embed.sh start

# 或者使用 Docker Compose
git clone https://github.com/milvus-io/milvus.git
cd milvus/deployments/docker
docker-compose up -d

基本概念

Collection（集合）≈ 关系数据库的 Table
Partition（分区）≈ Table 的分区（加速查询）
Index（索引）≈ 加速检索的数据结构
Entity（实体）≈ 一行数据（包含向量和标量字段）

Python SDK 使用

安装

pip install pymilvus

连接和创建集合

from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType

# 连接
connections.connect("default", host="localhost", port="19530")

# 定义 Schema
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=768),
    FieldSchema(name="title", dtype=DataType.VARCHAR, max_length=200),
    FieldSchema(name="category", dtype=DataType.VARCHAR, max_length=50)
]

schema = CollectionSchema(fields, description="文档向量库")

# 创建集合
collection = Collection(name="docs", schema=schema)

插入数据

import numpy as np

# 生成随机向量（实际应该用 Embedding 模型）
embeddings = np.random.rand(100, 768).tolist()
titles = [f"文档 {i}" for i in range(100)]
categories = ["技术"] * 100
ids = list(range(100))

collection.insert([ids, embeddings, titles, categories])

创建索引

# 为向量字段创建 IVF_FLAT 索引
index_params = {
    "metric_type": "L2",
    "index_type": "IVF_FLAT",
    "params": {"nlist": 128}
}

collection.create_index(field_name="embedding", index_params=index_params)

搜索

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

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

results = collection.search(
    data=[np.random.rand(768).tolist()],  # 查询向量
    anns_field="embedding",
    param=search_params,
    limit=5,
    expr="category == '技术'"  # 标量过滤
)

for hits in results:
    for hit in hits:
        print(f"ID: {hit.id}, 距离: {hit.distance}")

混合检索（标量过滤）

# 只在 category == "技术" 的记录中搜索
results = collection.search(
    data=[query_vector],
    anns_field="embedding",
    param=search_params,
    limit=5,
    expr="category == '技术'"
)

与 LangChain 集成

from langchain.vectorstores import Milvus
from langchain.embeddings import OpenAIEmbeddings

# 创建向量存储
vectorstore = Milvus(
    embedding_function=OpenAIEmbeddings(),
    collection_name="my_docs",
    connection_args={"host": "localhost", "port": "19530"}
)

# 插入文档
vectorstore.add_texts(
    texts=["文档内容1", "文档内容2"],
    metadatas=[{"category": "tech"}, {"category": "news"}]
)

# 搜索
results = vectorstore.similarity_search("查询内容", k=5)

性能优化

1. 索引选择

索引类型	精度	速度	内存
FLAT	100%	慢	低
IVF_FLAT	95-99%	快	低
HNSW	95-99%	很快	高
ANNOY	90-95%	快	中

推荐：追求精度用 IVF_FLAT，追求速度用 HNSW。

2. 分区（Partition）

按时间或类别分区，减少搜索范围：

collection.create_partition(partition_name="2024")
collection.insert(data, partition_name="2024")

3. 标量索引

为常用的过滤字段建索引：

collection.create_index(
    field_name="category",
    index_params={"index_type": "Trie"}
)

相关资源

• Milvus 官方文档
• Zilliz Cloud - Milvus 云服务
• PyMilvus GitHub