MilvusClient 学习指南

面向 Python 开发者，通过 MilvusClient 原生 API 快速上手向量数据库。

一、Milvus 是什么

Milvus 是一个开源的向量数据库，专门用于存储和检索高维向量（embedding）。它在 RAG（检索增强生成）、图像搜索、推荐系统等场景中负责"语义相似度检索"。

核心概念

二、MilvusClient vs ORM API

pymilvus 提供两套 API：

MilvusClient 足够覆盖 95% 的日常场景，且不受连接别名兼容问题困扰。

三、快速入门

3.1 安装

pip install pymilvus

3.2 连接 Milvus

from pymilvus import MilvusClient

# 本地部署
client = MilvusClient(uri="http://localhost:19530")

# 指定数据库
client = MilvusClient(uri="http://localhost:19530", db_name="my_db")

# Milvus Cloud（需要 token）
# client = MilvusClient(
#     uri="https://your-cluster.zillizcloud.com:19530",
#     token="your_token"
# )

3.3 创建 Collection

client.create_collection(
    collection_name="my_vectors",
    dimension=768,     # 向量维度（必须）
    auto_id=True,      # 自动生成主键
)

3.4 插入数据

data = [
    {"text": "今天天气很好", "source": "file1.txt", "vector": [0.1, 0.2, ..., 0.768]},
    {"text": "明天会下雨",   "source": "file2.txt", "vector": [0.3, 0.4, ..., 0.768]},
]
result = client.insert(collection_name="my_vectors", data=data)
print(f"插入 {result['insert_count']} 条数据")

3.5 向量检索

results = client.search(
    collection_name="my_vectors",
    data=[query_vector],              # 必须是列表的列表（支持批量）
    limit=3,                          # 返回条数
    output_fields=["text", "source"], # 额外返回的字段
)

for hits in results:
    for hit in hits:
        print(f"距离: {hit['distance']}")
        print(f"文本: {hit['entity']['text']}")
        print(f"来源: {hit['entity']['source']}")

3.6 完整流程

from pymilvus import MilvusClient
from pymilvus.milvus_client.index import IndexParams, IndexParam

client = MilvusClient(uri="http://localhost:19530")

# 1. 创建 Collection
client.create_collection("demo", dimension=3, auto_id=True)

# 2. 建索引
index_params = IndexParams()
index_params.add_index(
    field_name="vector",
    index_type="IVF_FLAT",
    metric_type="L2",
    params={"nlist": 128},
)
client.create_index("demo", index_params=index_params)

# 3. 插入数据
client.insert("demo", data=[
    {"vector": [0.1, 0.2, 0.3]},
    {"vector": [0.4, 0.5, 0.6]},
    {"vector": [0.7, 0.8, 0.9]},
])

# 4. 强制落盘 + 加载
client.flush("demo")
client.load_collection("demo")

# 5. 搜索
results = client.search("demo", data=[[0.2, 0.3, 0.4]], limit=2)
for hit in results[0]:
    print(f"ID: {hit['id']}, 距离: {hit['distance']}")

# 6. 清理
client.drop_collection("demo")
client.close()

四、核心功能详解

4.1 数据库管理

client.list_databases()              # 列出所有数据库
client.create_database("my_app")     # 创建
client.drop_database("my_app")       # 删除

4.2 Collection 管理

client.list_collections()                    # 列出所有
client.has_collection("my_vectors")          # 检查是否存在
client.describe_collection("my_vectors")     # 查看结构
client.drop_collection("my_vectors")         # 删除
client.get_collection_stats("my_vectors")    # 行数统计: {'row_count': 1000}

4.3 索引

为什么需要索引？ 没有索引就无法 load_collection，也无法搜索。

常用索引类型

创建索引：

index_params = IndexParams()
index_params.add_index(
    field_name="vector",
    index_type="IVF_FLAT",
    metric_type="L2",
    params={"nlist": 128},
)
client.create_index("my_collection", index_params=index_params)

# 加载到内存（搜索前必须）
client.load_collection("my_collection")

索引参数说明

• ** nlist**（IVF 系列）：桶的数量。经验值： sqrt(n)，n 是向量总数。
• ** nprobe**（搜索时设置）：搜索时探查的桶数。值越大精度越高但越慢，通常 nlist/4。
• ** M**（HNSW）：每个节点的连接数，默认 16。
• ** efConstruction**（HNSW）：建图深度，默认 256。

4.4 检索方式

向量检索

results = client.search(
    collection_name="articles",
    data=[query_vector],
    anns_field="embedding",
    limit=5,
    metric_type="L2",
    params={"nprobe": 16},
    output_fields=["title", "content"],
    filter="category == 'tech'",       # 可选：标量过滤
)

标量过滤

# 搜索时过滤
results = client.search(
    collection_name="articles",
    data=[query_vector],
    limit=5,
    filter="category == 'tech' AND year >= 2024",
    output_fields=["title", "year"],
)

支持的操作符：==, !=, >, <, >=, <=, like, in, and, or

主键查询

client.get(collection_name="articles", ids=[1, 2, 3], output_fields=["title"])

纯查询（不走向量搜索）

results = client.query(
    collection_name="articles",
    filter="year >= 2024",
    output_fields=["title", "category"],
    limit=10,
)

4.5 距离度量方式

4.6 数据操作

# 删除
client.delete(collection_name="articles", filter="id in [1, 2, 3]")

# 更新 = 删除 + 插入（Milvus 没有 UPDATE 操作）
client.delete(collection_name="articles", filter="id == 123")
client.insert("articles", data=[{"id": 123, "title": "新标题", ...}])

4.7 分区

client.create_partition("articles", "tech_2024")

# 只在指定分区搜索
results = client.search(
    "articles",
    data=[query_vector],
    limit=5,
    partition_names=["tech_2024"],
)

client.drop_partition("articles", "tech_2023")

4.8 一致性级别

client.search(..., consistency_level="Strong")      # 强一致
client.search(..., consistency_level="Bounded")     # 有界一致（推荐）
client.search(..., consistency_level="Eventually")  # 最终一致

4.9 资源管理

client.release_collection("articles")   # 从内存卸载
client.get_load_state("articles")       # 检查加载状态
client.close()                          # 关闭客户端

五、常见场景实战

5.1 RAG 知识库

from pymilvus import MilvusClient
from pymilvus.milvus_client.index import IndexParams

client = MilvusClient(uri="http://localhost:19530", db_name="knowledge_base")

# 创建
client.create_collection("docs", dimension=1536, auto_id=True)

# 建索引 + 加载
ip = IndexParams()
ip.add_index(field_name="vector", index_type="IVF_FLAT", metric_type="L2", params={"nlist": 128})
client.create_index("docs", index_params=ip)
client.load_collection("docs")

# 写入
chunks = [
    {"text": "校园开放时间为...", "source": "校规手册", "vector": [...]},
]
client.insert("docs", data=chunks)
client.flush("docs")

# 检索
results = client.search("docs", data=[query_vec], limit=3, output_fields=["text", "source"])

5.2 多租户数据隔离

# 方案一：不同数据库
client.create_database("tenant_a")
client_a = MilvusClient(uri="...", db_name="tenant_a")

# 方案二：不同 Collection
client.create_collection("tenant_a_data", dimension=768)

# 方案三：分区隔离
client.create_partition("shared", "tenant_a")
results = client.search("shared", data=[...], partition_names=["tenant_a"])

六、性能调优

七、排错清单

八、与 langchain_milvus 的关系

langchain_milvus 是 LangChain 对 Milvus 的封装，但它内部同时依赖 MilvusClient 和 ORM Collection，在 pymilvus 2.5+ 中存在连接别名不匹配的问题。

推荐实践：

• 写入 + 检索：用 MilvusClient 原生 API
• LangChain 集成：将检索结果封装为 Tool 给 Agent 使用

本文档适用于 pymilvus 2.5+ 版本