14.10-Milvus实践

要点

Milvus 面向超大规模向量检索设计——百亿级向量、分布式架构
架构分层：数据节点、查询节点、协调服务、存储引擎，各层独立扩展
依赖 etcd（元数据）、MinIO/S3（对象存储）、Pulsar/Kafka（消息队列）——部署复杂度高
Collection schema 显式定义字段类型，和关系数据库思维不同
索引类型丰富：IVF_FLAT、IVF_SQ8、HNSW、GPU 索引——选型需要理解底层原理
支持标量过滤 + 向量检索混合查询
适合数据量大、性能要求高、团队有能力运维分布式系统的场景

1. 架构概览

Milvus 的架构和前面讲的 pgvector、Qdrant 差异明显。它把存储、计算、协调拆开，各层可以独立扩展。

┌─────────────────────────────────────────────────────────┐
│                      Access Layer                        │
│                  （SDK / REST / gRPC）                    │
├─────────────────────────────────────────────────────────┤
│                Coordinator Service                       │
│    ┌──────────┬──────────┬──────────────┐               │
│    │ Root     │ Query    │ Data         │               │
│    │ Coord    │ Coord    │ Coord        │               │
│    └──────────┴──────────┴──────────────┘               │
├─────────────────────────────────────────────────────────┤
│                 Worker Nodes                             │
│    ┌──────────┬──────────┬──────────────┐               │
│    │ Query    │ Data     │ Index        │               │
│    │ Nodes    │ Nodes    │ Nodes        │               │
│    └──────────┴──────────┴──────────────┘               │
├─────────────────────────────────────────────────────────┤
│                Storage Layer                             │
│    ┌──────────┬──────────┬──────────────┐               │
│    │ etcd     │ MinIO/S3 │ Pulsar/      │               │
│    │ (meta)   │ (object) │ Kafka (log)  │               │
│    └──────────┴──────────┴──────────────┘               │
└─────────────────────────────────────────────────────────┘

Access Layer：SDK 入口，支持 gRPC 和 REST
Coordinator Service：集群管理，包括 Root Coord（元数据）、Query Coord（查询调度）、Data Coord（数据管理）
Worker Nodes：执行实际的查询、数据写入、索引构建
Storage Layer：etcd 存元数据，MinIO 存向量数据，Pulsar/Kafka 存 WAL（Write-Ahead Log）

这种架构的好处是查询节点和数据节点可以独立扩缩容。坏处是组件多——运维成本高。

2. 部署

Docker Compose（开发环境）

Milvus 依赖 etcd、MinIO、Pulsar，用 Docker Compose 部署最方便。

# docker-compose.yml
services:
  etcd:
    image: quay.io/coreos/etcd:v3.5.18
    environment:
      ETCD_AUTO_COMPACTION_MODE: revision
      ETCD_AUTO_COMPACTION_RETENTION: "1000"
      ETCD_QUOTA_BACKEND_BYTES: "4294967296"
    command: etcd -advertise-client-urls=http://127.0.0.1:2379 -listen-client-urls http://0.0.0.0:2379 --data-dir /etcd
 
  minio:
    image: minio/minio:RELEASE.2024-09-22T00-33-43Z
    environment:
      MINIO_ACCESS_KEY: minioadmin
      MINIO_SECRET_KEY: minioadmin
    command: minio server /minio_data --console-address ":9001"
 
  pulsar:
    image: apachepulsar/pulsar:2.11.0
    command: bin/pulsar standalone --no-functions-worker --no-stream-storage
 
  milvus:
    image: milvusdb/milvus:v2.5.4
    depends_on:
      - etcd
      - minio
      - pulsar
    environment:
      ETCD_ENDPOINTS: etcd:2379
      MINIO_ADDRESS: minio:9000
      PULSAR_ADDRESS: pulsar:6650
    ports:
      - "19530:19530"   # gRPC
      - "9091:9091"     # metrics

docker compose up -d

Milvus Standalone（轻量开发）

如果不想跑完整集群，Milvus 提供 Standalone 模式——所有组件在一个进程里。

docker run -d --name milvus-standalone \
  -p 19530:19530 \
  -p 9091:9091 \
  milvusdb/milvus:v2.5.4 standalone

Standalone 模式功能完整，但性能和扩展性不如分布式部署。适合本地开发和小规模数据。

生产部署

生产环境建议用 Kubernetes。Milvus Operator 可以自动化管理集群。

# 安装 Milvus Operator
kubectl apply -f https://raw.githubusercontent.com/milvus-io/milvus-operator/main/manifests/namespace.yaml
kubectl apply -f https://raw.githubusercontent.com/milvus-io/milvus-operator/main/manifests/deployment.yaml
 
# 创建集群
kubectl apply -f milvus-cluster.yaml

也可以用 Zilliz Cloud（托管服务）——省去运维成本。

3. 创建 Collection

Milvus 的 schema 定义比 pgvector 和 Qdrant 更严格——所有字段类型必须显式声明。

import { MilvusClient, DataType } from '@zilliz/milvus2-sdk-node'
 
const client = new MilvusClient({ address: 'localhost:19530' })
 
// 创建 collection
await client.createCollection({
  collection_name: 'documents',
  description: '文档向量集合',
  fields: [
    {
      name: 'id',
      data_type: DataType.VarChar,
      is_primary_key: true,
      type_params: { max_length: 64 },
    },
    {
      name: 'document_id',
      data_type: DataType.VarChar,
      type_params: { max_length: 64 },
    },
    {
      name: 'document_title',
      data_type: DataType.VarChar,
      type_params: { max_length: 256 },
    },
    {
      name: 'chunk_index',
      data_type: DataType.Int64,
    },
    {
      name: 'content',
      data_type: DataType.VarChar,
      type_params: { max_length: 65535 },
    },
    {
      name: 'embedding',
      data_type: DataType.FloatVector,
      type_params: { dim: 768 },
    },
    {
      name: 'user_id',
      data_type: DataType.VarChar,
      type_params: { max_length: 64 },
    },
    {
      name: 'tenant_id',
      data_type: DataType.VarChar,
      type_params: { max_length: 64 },
    },
    {
      name: 'created_at',
      data_type: DataType.Int64,  // Milvus 没有原生时间类型，用时间戳
    },
  ],
  // 开启动态字段——可以存储未预定义的字段
  enable_dynamic_field: true,
})

Milvus 的 schema 思维和关系数据库不同——没有 JSON 类型，字段类型固定。如果 metadata 结构不固定，可以开启 enable_dynamic_field。

4. 创建索引

索引类型决定了检索性能和资源消耗。

IVF_FLAT（倒排文件 + 精确距离）

await client.createIndex({
  collection_name: 'documents',
  field_name: 'embedding',
  index_type: 'IVF_FLAT',
  metric_type: 'COSINE',
  params: { nlist: 1024 },  // 聚类数
})

nlist：聚类中心数量，越大检索越精确但越慢
查询时需要设置 nprobe（探测的聚类数）
适合中等规模数据（百万级）

IVF_SQ8（倒排文件 + 标量量化）

await client.createIndex({
  collection_name: 'documents',
  field_name: 'embedding',
  index_type: 'IVF_SQ8',
  metric_type: 'COSINE',
  params: { nlist: 1024 },
})

每个维度从 32 bit float 压缩到 8 bit
内存占用降到 1/4，但精度略有损失
适合内存有限的大规模场景

HNSW（分层导航小世界）

await client.createIndex({
  collection_name: 'documents',
  field_name: 'embedding',
  index_type: 'HNSW',
  metric_type: 'COSINE',
  params: { M: 16, efConstruction: 256 },
})

M：每个节点的最大连接数
efConstruction：构建索引时的搜索范围
查询时需要设置 ef（搜索时的候选队列大小）
查询速度快，但内存占用高

标量字段索引

标量字段的过滤查询也需要索引：

await client.createIndex({
  collection_name: 'documents',
  field_name: 'tenant_id',
  index_type: 'Trie',
})
 
await client.createIndex({
  collection_name: 'documents',
  field_name: 'created_at',
  index_type: 'STL_SORT',
})

索引类型选择

索引类型	内存占用	查询速度	精度	适用场景
IVF_FLAT	中	中	精确	百万级，内存适中
IVF_SQ8	低	中	近似	大规模，内存有限
IVF_PQ	极低	中	近似	超大规模，精度要求不高
HNSW	高	快	近似	对延迟要求高
GPU_IVF_FLAT	中	快	精确	有 GPU 资源
GPU_CAGRA	中	极快	近似	有 GPU，需要极低延迟

不确定时，先用 HNSW——查询快，不需要调太多参数。如果内存不够，换 IVF_SQ8。

5. 写入数据

单条写入

await client.insert({
  collection_name: 'documents',
  data: [
    {
      id: 'doc-1-chunk-0',
      document_id: 'doc-1',
      document_title: '退款政策',
      chunk_index: 0,
      content: 'chunk 内容...',
      embedding: embeddingArray,
      user_id: 'user-123',
      tenant_id: 'tenant-456',
      created_at: Date.now(),
    },
  ],
})

批量写入

export async function bulkInsert(
  chunks: Array<{
    id: string
    document_id: string
    document_title: string
    chunk_index: number
    content: string
    embedding: number[]
    user_id?: string
    tenant_id?: string
  }>
): Promise<void> {
  // Milvus 单次写入限制：默认 100MB，可以调整
  const BATCH_SIZE = 500
 
  for (let i = 0; i < chunks.length; i += BATCH_SIZE) {
    const batch = chunks.slice(i, i + BATCH_SIZE)
 
    await client.insert({
      collection_name: 'documents',
      data: batch.map((chunk) => ({
        id: chunk.id,
        document_id: chunk.document_id,
        document_title: chunk.document_title,
        chunk_index: chunk.chunk_index,
        content: chunk.content,
        embedding: chunk.embedding,
        user_id: chunk.user_id ?? '',
        tenant_id: chunk.tenant_id ?? '',
        created_at: Date.now(),
      })),
    })
  }
}

Upsert（插入或更新）

Milvus 2.3+ 支持 upsert：

await client.upsert({
  collection_name: 'documents',
  data: [
    {
      id: 'doc-1-chunk-0',
      document_id: 'doc-1',
      document_title: '退款政策（已更新）',
      chunk_index: 0,
      content: '更新后的内容...',
      embedding: newEmbedding,
    },
  ],
})

写入后加载

Milvus 写入数据后需要加载到内存才能查询：

await client.loadCollection({
  collection_name: 'documents',
})

如果数据量大，可以分批加载 partition。

6. 相似度查询

基础查询

const results = await client.search({
  collection_name: 'documents',
  vectors: [queryVector],
  limit: 5,
  // 搜索参数，和索引类型相关
  params: { ef: 128 },  // HNSW 用 ef，IVF 用 nprobe
})
 
// 结果结构
// {
//   results: [
//     { id: '...', score: 0.92, content: '...', document_id: '...' },
//     ...
//   ]
// }

带过滤的查询

Milvus 支持标量过滤 + 向量检索混合查询。

const results = await client.search({
  collection_name: 'documents',
  vectors: [queryVector],
  limit: 5,
  filter: 'tenant_id == "tenant-456" && created_at >= 1704067200000',
  params: { ef: 128 },
})

过滤语法类似 SQL：

// 等于
'tenant_id == "tenant-456"'
 
// 不等于
'status != "archived"'
 
// 范围
'created_at >= 1704067200000 && created_at < 1735689600000'
 
// IN
'document_id in ["doc-1", "doc-2", "doc-3"]'
 
// LIKE（模糊匹配）
'document_title LIKE "%退款%"'
 
// 组合
'tenant_id == "tenant-456" && (status == "active" || status == "draft")'

输出指定字段

const results = await client.search({
  collection_name: 'documents',
  vectors: [queryVector],
  limit: 5,
  output_fields: ['document_id', 'document_title', 'content', 'chunk_index'],
  params: { ef: 128 },
})

7. 高级检索

多向量查询

可以一次传多个查询向量，Milvus 会分别返回每个向量的结果：

const results = await client.search({
  collection_name: 'documents',
  vectors: [queryVector1, queryVector2],
  limit: 5,
})
 
// results 会按查询向量分组

范围搜索

不返回 topK，而是返回所有距离在阈值内的结果：

const results = await client.search({
  collection_name: 'documents',
  vectors: [queryVector],
  params: {
    radius: 0.3,      // 距离阈值
    range_filter: 0.8, // 返回 score 在 [0.8, 0.3) 之间（Cosine 距离，越大越近）
  },
})

带 offset 的查询

用于分页：

const results = await client.search({
  collection_name: 'documents',
  vectors: [queryVector],
  limit: 5,
  offset: 10,  // 跳过前 10 条
})

8. 分区和分区键

Partition

Milvus 支持按 partition 组织数据——类似数据库的分区表。

// 创建 partition
await client.createPartition({
  collection_name: 'documents',
  partition_name: 'tenant_456',
})
 
// 写入指定 partition
await client.insert({
  collection_name: 'documents',
  partition_name: 'tenant_456',
  data: chunks,
})
 
// 查询指定 partition
const results = await client.search({
  collection_name: 'documents',
  partition_names: ['tenant_456'],
  vectors: [queryVector],
  limit: 5,
})

Partition Key

更好的方式是用 partition key——Milvus 自动按字段值分配 partition。

await client.createCollection({
  collection_name: 'documents',
  fields: [
    // ... 其他字段
    {
      name: 'tenant_id',
      data_type: DataType.VarChar,
      type_params: { max_length: 64 },
      is_partition_key: true,  // 设为 partition key
    },
  ],
})

设置 is_partition_key 后，查询时 Milvus 自动按 tenant_id 路由到对应 partition——不需要手动管理 partition。

9. 性能调优

查询参数

不同索引类型的调优参数不同：

// HNSW
{ ef: 128 }  // 默认 64，越大越精确但越慢
 
// IVF_FLAT / IVF_SQ8
{ nprobe: 16 }  // 默认 1，越大越精确但越慢

动态调整：

await client.search({
  collection_name: 'documents',
  vectors: [queryVector],
  limit: 5,
  params: { ef: parseInt(process.env.HNSW_EF ?? '128') },
})

数据量分段

当数据量很大时，按时间或 tenant 分段加载：

// 只加载最近的 partition
await client.loadCollection({
  collection_name: 'documents_recent',
})
 
// 旧数据放在另一个 collection，按需加载
await client.loadCollection({
  collection_name: 'documents_archive',
})

监控

Milvus 暴露 Prometheus metrics：

http://localhost:9091/metrics

关键指标：

milvus_query_latency：查询延迟
milvus_insert_rate：写入速率
milvus_collection_count：collection 数量
milvus_index_size：索引大小

健康检查

app.get('/health/milvus', async (c) => {
  try {
    const res = await client.checkHealth()
    return c.json({
      status: res.isHealthy ? 'ok' : 'degraded',
      reasons: res.reasons,
    })
  } catch (err) {
    return c.json({ status: 'error', error: String(err) }, 503)
  }
})

10. 一个完整的 Milvus Store 实现

// src/services/rag/milvus-store.ts
import { MilvusClient, DataType } from '@zilliz/milvus2-sdk-node'
 
export class MilvusVectorStore {
  private client: MilvusClient
 
  constructor() {
    this.client = new MilvusClient({
      address: process.env.MILVUS_ADDRESS ?? 'localhost:19530',
    })
  }
 
  async initCollection(name: string, dimensions: number): Promise<void> {
    const exists = await this.client.hasCollection({ collection_name: name })
 
    if (!exists.value) {
      await this.client.createCollection({
        collection_name: name,
        fields: [
          {
            name: 'id',
            data_type: DataType.VarChar,
            is_primary_key: true,
            type_params: { max_length: 64 },
          },
          {
            name: 'document_id',
            data_type: DataType.VarChar,
            type_params: { max_length: 64 },
          },
          {
            name: 'content',
            data_type: DataType.VarChar,
            type_params: { max_length: 65535 },
          },
          {
            name: 'embedding',
            data_type: DataType.FloatVector,
            type_params: { dim: dimensions },
          },
          {
            name: 'tenant_id',
            data_type: DataType.VarChar,
            type_params: { max_length: 64 },
            is_partition_key: true,
          },
        ],
        enable_dynamic_field: true,
      })
 
      await this.client.createIndex({
        collection_name: name,
        field_name: 'embedding',
        index_type: 'HNSW',
        metric_type: 'COSINE',
        params: { M: 16, efConstruction: 256 },
      })
 
      await this.client.loadCollection({ collection_name: name })
    }
  }
 
  async upsertChunks(collectionName: string, chunks: ChunkWithEmbedding[]): Promise<void> {
    const BATCH_SIZE = 500
    for (let i = 0; i < chunks.length; i += BATCH_SIZE) {
      const batch = chunks.slice(i, i + BATCH_SIZE)
      await this.client.insert({
        collection_name: collectionName,
        data: batch.map((chunk) => ({
          id: chunk.id,
          document_id: chunk.metadata.documentId,
          content: chunk.content,
          embedding: chunk.embedding,
          tenant_id: chunk.metadata.tenantId ?? '',
        })),
      })
    }
  }
 
  async search(
    collectionName: string,
    queryVector: number[],
    options: { topK?: number; tenantId?: string } = {}
  ): Promise<SearchResult[]> {
    const { topK = 5, tenantId } = options
 
    const filter = tenantId ? `tenant_id == "${tenantId}"` : undefined
 
    const results = await this.client.search({
      collection_name: collectionName,
      vectors: [queryVector],
      limit: topK,
      filter,
      output_fields: ['document_id', 'content'],
      params: { ef: 128 },
    })
 
    return results.results.map((r) => ({
      id: String(r.id),
      score: r.score,
      content: r.content,
      metadata: { documentId: r.document_id },
    }))
  }
 
  async deleteDocument(collectionName: string, documentId: string): Promise<void> {
    await this.client.delete({
      collection_name: collectionName,
      filter: `document_id == "${documentId}"`,
    })
  }
}

总结

回顾这一节的要点：

Milvus 面向超大规模向量检索——百亿级向量，分布式架构
架构分层：Access Layer → Coordinator → Worker Nodes → Storage Layer
部署依赖 etcd + MinIO + Pulsar——Docker Compose 开发，K8s 生产
schema 显式定义所有字段类型，和关系数据库思维不同
索引类型：IVF_FLAT（精确）、IVF_SQ8（量化压缩）、HNSW（快但费内存）
支持标量过滤 + 向量检索混合查询
Partition key 自动路由，适合多租户场景
性能调优：HNSW 调 ef，IVF 调 nprobe
适合数据量大、性能要求高、有能力运维分布式系统的团队

下一篇讲相似度检索——向量距离的计算方法和索引原理。