Milvus学习手册

发表于 更新于 分类于 阅读次数： 本文字数： 4k 阅读时长 ≈ 15 分钟

Milvus 学习手册

文档核心说明

本文档基于 Milvus 2.4 + 稳定版本，面向企业开发、运维、SaaS 平台研发人员，整理了从基础入门到企业级落地的全流程知识，覆盖核心概念、操作实战、场景设计、运维审计等全链路内容，所有代码与方案可直接复制到生产环境。

一、基础入门

1.1 Milvus 核心简介

Milvus 是 LF AI & Data 基金会毕业的开源分布式向量数据库，专为 AI 大模型、海量高维向量数据设计，是 RAG 检索增强生成、语义检索、多模态检索、推荐系统、人脸识别等场景的核心基础设施。

  • 核心优势：支持万亿级向量毫秒级相似性检索、全索引覆盖、多语言 SDK、混合检索、云原生弹性扩缩容。

1.2 环境准备与服务连接

1.2.1 客户端依赖安装

1
pip install pymilvus==2.4.10  # 建议与服务端版本保持一致

1.2.2 服务连接

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from pymilvus import MilvusClient, utility, DataType
import numpy as np

# 1. 单机本地连接（默认端口 19530）
client = MilvusClient(uri="http://localhost:19530")

# 2. 远程服务连接（带认证示例）
# client = MilvusClient(
#     uri="http://your-milvus-host:19530",
#     user="root",
#     password="your-password"
# )

# 验证连接
print("现有集合：", client.list_collections())

二、核心概念与 Schema 设计

2.1 核心概念

概念 极简释义
向量（Vector） AI 模型输出的高维浮点数组，Milvus 的核心处理对象
字段（Field） 数据的最小单元，分为标量字段（用于条件过滤）和向量字段（用于检索）
集合（Collection） 关系型数据库的「表」，数据管理的核心单元
分区（Partition） 集合的子集，用于数据隔离，按高频过滤字段分区提升检索效率
索引（Index） 向量的加速检索结构，牺牲少量精度换取检索速度

2.2 Schema 字段类型全解析

2.2.1 标量字段类型

数据类型（DataType） 说明 适用场景 核心注意事项
BOOL 布尔值，True/False 二分类标签、状态标记 无
INT32 32 位有符号整数 数值 ID、计数、年龄 常用整数类型
INT64 64 位有符号整数 主键 ID、时间戳 主键字段必须用 INT64 或 VARCHAR
FLOAT 32 位浮点数 小范围数值、评分 精度要求不高时使用
DOUBLE 64 位浮点数 高精度数值、金额 精度要求高时使用
VARCHAR 可变长字符串 文本内容、分类标签 必须指定 max_length，最大支持 65535
JSON JSON 格式数据 灵活的结构化数据 Milvus 2.4 + 支持，可对 JSON 内键值创建索引
Array 数组类型 多标签、多值属性 Milvus 2.3 + 支持，元素类型需一致

2.2.2 向量字段类型

数据类型（DataType） 说明 适用场景 核心注意事项
FLOAT_VECTOR 32 位浮点型向量 绝大多数语义检索、图像检索场景 必须指定 dim（维度），需与 Embedding 模型输出维度完全一致
BINARY_VECTOR 二进制向量（0/1 位存储） 超大规模数据、极低内存场景 维度必须是 8 的倍数，精度略低于浮点向量

三、核心操作：集合、索引、数据的增删改查

3.1 集合（表）的增删改查

3.1.1 创建集合

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
collection_name = "crud_demo"

# 1. 定义 Schema
schema = client.create_schema(
    auto_id=False,          # 关闭自增主键，自定义主键；开启后系统自动生成
    enable_dynamic_field=True,  # 开启动态字段，支持插入未定义的字段
    description="CRUD操作示例集合"
)

# 2. 添加字段
schema.add_field(field_name="doc_id", datatype=DataType.INT64, is_primary=True, description="文档唯一主键")
schema.add_field(field_name="category", datatype=DataType.VARCHAR, max_length=64, description="文档分类")
schema.add_field(field_name="publish_time", datatype=DataType.INT64, description="发布时间戳")
schema.add_field(field_name="doc_vector", datatype=DataType.FLOAT_VECTOR, dim=768, description="768维语义向量")

# 3. 创建集合
client.create_collection(collection_name=collection_name, schema=schema)

3.1.2 查看 / 修改 / 删除集合

1
2
3
4
5
6
7
8
9
10
11
12
# 1. 列出所有集合
print("所有集合：", client.list_collections())

# 2. 查看集合详情
collection_info = client.describe_collection(collection_name=collection_name)
print("集合详情：", collection_info)

# 3. 修改集合TTL（数据生命周期，单位：秒）
client.alter_collection(collection_name=collection_name, properties={"collection.ttl.seconds": 30*24*3600})

# 4. 删除集合（谨慎操作，会清除所有数据）
# client.drop_collection(collection_name=collection_name)

3.2 索引的增删改查

3.2.1 创建索引

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# 1. 为向量字段创建HNSW索引（生产首选）
index_params = client.prepare_index_params()
index_params.add_index(
    field_name="doc_vector",
    index_type="HNSW",
    metric_type="IP",
    params={"M": 16, "ef_construction": 200},
    index_name="doc_vector_hnsw"
)

# 2. 为标量字段创建索引（加速过滤）
index_params.add_index(field_name="category", index_type="INVERTED")
index_params.add_index(field_name="publish_time", index_type="STL_SORT")

# 执行创建
client.create_index(collection_name=collection_name, index_params=index_params, async=True)

3.2.2 查看 / 修改 / 删除索引

1
2
3
4
5
6
7
8
9
10
11
12
13
# 1. 列出所有索引
print("所有索引：", client.list_indexes(collection_name=collection_name))

# 2. 查看索引详情
print("索引详情：", client.describe_index(collection_name=collection_name, field_name="doc_vector"))

# 3. 修改索引：直接创建新索引，系统自动替换旧索引
new_index_params = client.prepare_index_params()
new_index_params.add_index(field_name="doc_vector", index_type="IVF_SQ8", metric_type="IP", params={"nlist": 4096})
client.create_index(collection_name=collection_name, index_params=new_index_params, async=True)

# 4. 删除索引
# client.drop_index(collection_name=collection_name, field_name="doc_vector")

3.3 数据的增删改查

3.3.1 插入数据

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# 批量插入数据
data_num = 1000
data = [
    {
        "doc_id": i,
        "category": "技术" if i % 2 == 0 else "产品",
        "publish_time": 1710000000 + i * 86400,
        "doc_vector": np.random.rand(768).tolist()
    } for i in range(data_num)
]

insert_res = client.insert(collection_name=collection_name, data=data)
print(f"插入成功，行数：{insert_res['insert_count']}")
client.flush(collection_name=collection_name)
client.load_collection(collection_name=collection_name)

3.3.2 查询数据

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# 1. 标量查询
query_res = client.query(
    collection_name=collection_name,
    filter="doc_id in [1,2,3,4,5]",
    output_fields=["doc_id", "category"]
)
print("标量查询结果：", query_res)

# 2. 向量检索
query_vector = [np.random.rand(768).tolist()]
search_res = client.search(
    collection_name=collection_name,
    data=query_vector,
    anns_field="doc_vector",
    limit=5,
    search_params={"metric_type": "IP", "params": {"ef": 64}},
    filter="category == '技术'",
    output_fields=["doc_id", "category"]
)

3.3.3 更新 / 删除数据

1
2
3
4
5
6
7
8
9
10
11
12
13
# 1. 更新数据：upsert，主键存在则更新，不存在则插入
upsert_res = client.upsert(
    collection_name=collection_name,
    data=[{"doc_id": 1, "category": "技术（已更新）", "doc_vector": np.random.rand(768).tolist()}]
)
print(f"更新成功，行数：{upsert_res['upsert_count']}")

# 2. 删除数据
delete_res = client.delete(
    collection_name=collection_name,
    filter="doc_id in [1000, 1001]"
)
print(f"删除成功，行数：{delete_res['delete_count']}")

3.4 向量检索常用参数

3.4.1 相似度度量方式

度量方式 规则 适用场景
IP（内积） 值越大，相似度越高 归一化后的语义向量、推荐系统
L2（欧氏距离） 值越小，相似度越高 图像、语音、未归一化的向量
COSINE（余弦相似度） 值越大，相似度越高 NLP 文本语义匹配、多模态检索

3.4.2 主流索引核心参数

索引类型 构建参数 检索参数
HNSW M（邻居数，推荐 16-32）、ef\_construction（构建候选集，推荐 128-512） ef（检索候选集，必须大于 limit，推荐 64-256）
IVF 系列 nlist（分桶数，推荐√N~N/10） nprobe（扫描分桶数，推荐 nlist 的 5%-10%）

3.5 标量过滤条件语法

操作符类型 操作符 说明
比较操作符 ==、\!=、\>、\<、\>=、\<= 基础比较
逻辑操作符 and、or、not 多条件组合
范围操作符 in、not in 列表匹配
字符串操作符 like、array\_contains 模糊匹配、数组包含

四、向量索引深度指南

4.1 向量索引核心基础

  • KNN 精确检索：全量向量暴力比对，100% 召回，对应 FLAT 索引，仅适合百万级以内小数据；

  • ANN 近似检索：通过索引结构仅扫描部分候选向量，用极小精度损失换取速度提升，是生产环境主流选择。

4.2 全量索引类型详解

索引类型 核心原理 召回率 适用场景
FLAT 暴力检索，无索引结构 100% 百万级以内小数据、100% 召回要求
IVF_FLAT 倒排分桶，桶内原始向量 95%+ 百万 - 千万级数据、平衡召回与速度
IVF_SQ8 倒排分桶 + 标量量化，压缩 75% 内存 90%-95% 千万 - 亿级数据、内存有限
IVF_PQ 倒排分桶 + 乘积量化，极致压缩 80%-90% 亿级以上冷归档数据
HNSW 分层导航小世界图，快速跳转检索 98%+ 千万 - 亿级热数据、高并发低延迟（生产首选）
SCANN 图索引 + 非对称量化，比 HNSW 省 30% 内存 95%+ 平衡性能与内存的线上场景
DISKANN 磁盘图索引，支持百亿级数据 95%+ 百亿级超大规模数据、内存有限
AUTOINDEX 智能自动选择最优索引 自适应 快速落地、无专业运维的场景

4.3 多索引特性

Milvus 的多索引特性是广义的多维度索引组合能力，核心能力：

  1. 多字段联合索引：一个集合内，多个向量 / 标量字段分别创建独立索引；

  2. 标量 + 向量混合索引：先通过标量索引过滤，再做向量检索；

  3. 分区级差异化索引：不同分区创建不同索引，实现冷热数据分层；

  4. 多向量复合索引：多个向量字段索引协同，实现加权融合检索。

五、企业级权限体系

5.1 原生 RBAC 权限体系功能边界

5.1.1 原生支持的能力

  • 四级资源粒度：全局级、数据库级、集合级、分区级；

  • 全操作特权覆盖：运维、管理、读写、只读等全生命周期操作；

  • 多角色绑定：一个用户可绑定多个角色，权限取并集；

  • 权限动态生效：修改权限实时生效，无需重启。

5.1.2 原生不支持的硬限制

  • 不支持行级 / 字段级权限控制，最小粒度为分区；

  • 不支持角色继承、条件授权、黑名单授权；

  • 不支持 QPS / 配额限制、外部 SSO/LDAP 原生集成；

  • 原生审计能力不足，需外部日志采集。

5.2 RBAC 全生命周期落地操作

5.2.1 启用 RBAC

修改 Milvus 配置开启授权：

1
2
3
common:
  security:
    authorizationEnabled: true

首次登录修改 root 默认密码：

1
utility.reset_password(user="root", old_password="Milvus", new_password="Root@2026_Prod")

5.2.2 用户 / 角色 / 权限配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
# 1. 创建用户
utility.create_user(user="rag_app_user", password="RagApp@2026")

# 2. 创建角色
utility.create_role(role_name="rag_app_ro")

# 3. 授予权限
privileges = ["Search", "Query", "DescribeCollection"]
for priv in privileges:
    utility.grant_role_privilege(
        role_name="rag_app_ro",
        object_type="Collection",
        object_name="knowledge_base",
        privilege=priv
    )

# 4. 绑定用户与角色
utility.add_user_to_role(user="rag_app_user", role_name="rag_app_ro")

# 5. 权限验证
rag_client = MilvusClient(
    uri="http://localhost:19530",
    user="rag_app_user",
    password="RagApp@2026"
)
# 验证正常权限
query_vector = np.random.rand(768).tolist()
rag_client.search(collection_name="knowledge_base", data=[query_vector], limit=5, search_params={"metric_type": "COSINE", "params": {"ef": 64}})
# 验证越权操作
try:
    rag_client.insert(collection_name="knowledge_base", data=[{"doc_id": 999, "doc_vector": np.random.rand(768).tolist()}])
except Exception as e:
    print("越权拦截成功：", e)

5.3 行级权限管控方案

针对原生不支持的行级权限，采用原生 RBAC + 应用层过滤的组合方案：

  1. Schema 中预埋权限字段（部门、角色、用户、公开标记）；

  2. 所有权限字段创建索引，加速过滤；

  3. 应用层拦截器，所有检索请求必须携带权限过滤条件，禁止无过滤的全量检索。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
def enterprise_kb_search(query_vector, current_user, limit=5):
    # 构建权限过滤条件
    filter_cond = f"""
    is_valid == true
    and (
        is_public == true
        or array_contains(dept_ids, '{current_user['dept_id']}')
        or array_contains_any(role_ids, {current_user['role_ids']})
        or array_contains(authorized_user_ids, '{current_user['user_id']}')
    )
    """

    # 执行检索
    search_res = client.search(
        collection_name="enterprise_kb",
        data=[query_vector],
        anns_field="content_vector",
        limit=limit,
        search_params={"metric_type": "COSINE", "params": {"ef": 64}},
        filter=filter_cond,
        output_fields=["chunk_id", "content"]
    )
    return search_res

六、企业级典型场景落地

6.1 多业务系统集群共享隔离

6.1.1 架构设计

Text
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# 顶层：客户端/业务层
┌─────────────────────────────────────────────────────────────┐
│  业务系统（RAG知识库、推荐、人脸、风控…）                  │
│  ├─────────────┐  ├─────────────┐  ├─────────────┐          │
│  │ RAG 应用    │  │ 推荐系统    │  │ 人脸识别    │          │
│  │ (rag_user)  │  │ (rec_user)  │  │ (face_user) │          │
│  └─────────────┘  └─────────────┘  └─────────────┘          │
└───────────────────────────┬─────────────────────────────────┘
                            │
                            ▼
# 中间层：Milvus Proxy + RBAC鉴权
┌─────────────────────────────────────────────────────────────┐
│  Milvus Proxy 集群（无状态、共享、负载均衡）                │
│  → 身份认证、RBAC权限校验、请求路由                          │
└───────────────────────────┬─────────────────────────────────┘
                            │
                            ▼
# 核心层：Milvus共享集群
┌─────────────────────────────────────────────────────────────┐
│  逻辑数据库（完全隔离）：                                    │
│  ├─────────────┐  ├─────────────┐  ├─────────────┐          │
│  │ rag_db      │  │ rec_db      │  │ face_db     │          │
│  └─────────────┘  └─────────────┘  └─────────────┘          │
└─────────────────────────────────────────────────────────────┘

6.1.2 落地实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
# 1. 为每个业务创建独立数据库
utility.create_database(db_name="rag_knowledge_db")
utility.create_database(db_name="recommend_db")
utility.create_database(db_name="face_recognition_db")

# 2. 为每个业务创建独立用户与角色，仅授予对应数据库权限
utility.create_user(user="rag_biz_user", password="RagBiz@2026")
utility.create_role(role_name="rag_biz_admin")
utility.grant_role_privilege(
    role_name="rag_biz_admin",
    object_type="Database",
    object_name="rag_knowledge_db",
    privilege="All"
)
utility.add_user_to_role(user="rag_biz_user", role_name="rag_biz_admin")

# 3. 业务用户连接验证
rag_client = MilvusClient(
    uri="http://localhost:19530",
    user="rag_biz_user",
    password="RagBiz@2026",
    db_name="rag_knowledge_db"
)
# 验证无法访问其他数据库
try:
    invalid_client = MilvusClient(
        uri="http://localhost:19530",
        user="rag_biz_user",
        password="RagBiz@2026",
        db_name="recommend_db"
    )
    invalid_client.list_collections()
except Exception as e:
    print("越权拦截成功：", e)

6.2 SaaS 平台多租户分区级隔离

采用分区级隔离，适合大量租户的轻量隔离：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
# 1. 平台初始化集合与分区
collection_name = "saas_knowledge_base"
client.create_partition(collection_name=collection_name, partition_name="tenant_001")
client.create_partition(collection_name=collection_name, partition_name="tenant_002")

# 2. 动态新增租户
def create_tenant(tenant_id, tenant_password):
    utility.create_user(user=f"{tenant_id}_user", password=tenant_password)
    utility.create_role(role_name=f"{tenant_id}_role")
    privileges = ["Search", "Query", "DescribeCollection"]
    for priv in privileges:
        utility.grant_role_privilege(
            role_name=f"{tenant_id}_role",
            object_type="Partition",
            object_name=tenant_id,
            privilege=priv,
            collection_name=collection_name
        )
    utility.add_user_to_role(user=f"{tenant_id}_user", role_name=f"{tenant_id}_role")
    return {"tenant_id": tenant_id, "status": "created"}

create_tenant("tenant_001", "Tenant001@2026")

# 3. 租户检索
def tenant_search(tenant_id, tenant_password, query_vector):
    tenant_client = MilvusClient(
        uri="http://localhost:19530",
        user=f"{tenant_id}_user",
        password=tenant_password
    )
    # 双重隔离：分区+tenant_id过滤
    return tenant_client.search(
        collection_name=collection_name,
        data=[query_vector],
        limit=5,
        search_params={"metric_type": "COSINE", "params": {"ef": 64}},
        partition_names=[tenant_id],
        filter=f"tenant_id == '{tenant_id}'"
    )

6.3 开发 / 测试 / 生产环境权限分离

  1. 物理隔离：开发、测试、生产分别部署独立集群；

  2. 权限分层：开发 / 测试环境开发人员拥有读写权限，生产环境仅拥有只读权限；

  3. 最小权限：线上应用仅拥有最小读写 / 只读权限，仅运维拥有生产管理员权限。

七、知识库全生命周期管理

7.1 文档更新方案

  • 全量更新：先插入新版数据，再切换版本，最后删除旧版，实现无停机原子更新；

  • 增量更新：新旧分块对比，仅替换修改的分块，保留未修改分块，提升性能。

7.2 文档失效 / 下架处理

  • 软删除：标记is\_valid=false，检索时过滤，可恢复，适合合规审计场景；

  • 硬删除：物理删除数据，永久下架，适合不再使用的文档。

7.3 生命周期最佳实践

  • 定期清理软删除超过 30 天、超过 3 个历史版本的旧数据，避免存储膨胀；

  • 所有操作记录审计日志，满足合规要求。

八、版本升级与运维管理

8.1 升级前准备

  1. 版本兼容性校验，严禁跨 3 个以上大版本直接升级；

  2. 全量数据备份，使用 Milvus Backup 工具，验证备份有效；

  3. 测试环境预演，验证升级流程、功能、性能；

  4. 规划业务方案与回滚预案。

8.2 升级方案

  • 单机 Docker：低峰期停止服务，替换镜像，启动验证；

  • K8s Helm：滚动升级，业务无中断；

  • 跨大版本：双集群迁移，灰度切换流量。

8.3 升级后处理

  1. 数据完整性校验；

  2. 重建所有向量索引，适配新版本特性；

  3. 全量功能验证，升级客户端 SDK；

  4. 性能优化与配置适配。

九、常见问题与踩坑排查

  1. 索引不生效：检查 metric_type 是否一致、索引是否构建完成、集合是否加载；

  2. 召回率低：检查 HNSW 的 ef 是否大于 limit、IVF 的 nprobe 是否足够、向量是否归一化；

  3. 权限不足：检查用户角色绑定、权限配置、资源名称是否正确；

  4. OOM 内存溢出：检查索引内存占用、是否加载了过多集合、并发是否过高。

核心知识点速览

  1. 最小权限原则是企业权限管控的核心，仅授予业务必需的最小权限。

  2. 检索前必须加载集合到内存，否则无法执行检索。

  3. 索引的metric\_type必须与检索时完全一致，否则会降级为暴力检索。

  4. HNSW 检索的ef必须大于limit，否则召回率会断崖式下跌。

  5. 原生 RBAC 最小粒度为分区，行级权限需结合权限字段 + 强制过滤实现。

  6. 多业务系统隔离推荐数据库级隔离，SaaS 多租户推荐分区级隔离。

  7. 所有权限字段必须创建索引，保障过滤性能。

  8. 升级前必须全量备份数据，无备份不升级。

  9. SDK 与服务端大版本必须完全匹配，否则会出现兼容性问题。

  10. 所有检索请求必须携带权限过滤条件，严禁无过滤的全量检索。