基于 pymilvus 库，本文档涵盖 Milvus 连接、数据库管理、Collection 与 Schema、CRUD 操作等核心内容。
1. 连接 Milvus
PYTHON
from pymilvus import MilvusClient
# 连接本地 Docker Milvus（默认端口 19530）
client = MilvusClient(uri="http://localhost:19530")
参数 | 说明 |
|---|---|
| Milvus 服务地址。本地 Docker 为 |
| 可选，不填为默认数据库 |
| 可选，鉴权 Token（Zilliz Cloud 或开启了认证的 Milvus） |
测试连接
PYTHON
collections = client.list_collections()
print(collections) # 返回当前所有 collection 名称列表
2. 数据库操作
2.1 查看所有数据库
PYTHON
existing_dbs = client.list_databases()
2.2 检查并创建数据库
PYTHON
DB_NAME = "DemoDB"
if DB_NAME in client.list_databases():
print(f"数据库 '{DB_NAME}' 已存在")
else:
client.create_database(db_name=DB_NAME)
print(f"数据库 '{DB_NAME}' 已创建")
2.3 使用数据库
PYTHON
client.use_database(db_name="DemoDB")
切换后，后续所有操作（create_collection, insert, search 等）都在该数据库下进行。
不调用
use_database则默认使用default数据库，所以创建以后一定要指定去哪个数据库操作。
3. Collection（集合）
Collection 是 Milvus 中最基本的数据组织单位，类比关系型数据库中的表（Table）。所有数据操作（插入、查询、删除等）都围绕 Collection 展开。
PYTHON
from pymilvus import CollectionSchema, FieldSchema, DataType
3.1 什么是 Schema
Schema 定义了 Collection 的数据结构，包含所有字段（Field）及其属性。一个 Schema 必须包含：
主键字段（Primary Key）：唯一标识每条数据，类型为 INT64 或 VARCHAR
向量字段（Vector Field）：存储核心向量数据，必须有
dim（维度）标量字段（Scalar Field）：存储元数据，用于过滤（如 VARCHAR, INT32 等）
3.2 创建 Schema
方式一：MilvusClient.create_schema（推荐，新版 API）
PYTHON
schema = client.create_schema(
auto_id=True, # 自动生成主键（True 时插入不需要传 id）
enable_dynamic_field=True, # 开启动态字段：允许插入 schema 未定义的标量字段
)
schema.add_field(field_name="id", datatype=DataType.INT64, is_primary=True, description="主键ID")
schema.add_field(field_name="doc_title", datatype=DataType.VARCHAR, max_length=255, description="文档标题")
schema.add_field(field_name="page_num", datatype=DataType.INT32, description="页码")
schema.add_field(field_name="text_content", datatype=DataType.VARCHAR, max_length=65535, description="文本内容")
schema.add_field(field_name="vector", datatype=DataType.FLOAT_VECTOR, dim=1024, description="向量")
方式二：CollectionSchema + FieldSchema（传统 API）
PYTHON
fields = [
FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=768),
FieldSchema(name="image_path", dtype=DataType.VARCHAR, max_length=512),
]
schema = CollectionSchema(fields, description="多模态图文检索")
3.3 create_schema() 参数详解
参数 | 类型 | 说明 |
|---|---|---|
| bool | 是否自动生成主键 ID。 |
| bool | 是否开启动态字段。为 |
3.4 add_field() 参数详解
参数 | 说明 |
|---|---|
| 字段名称，字符串 |
| 字段类型，使用 |
| 是否为主键，仅一个字段可为 |
| 主键是否自增（仅主键字段可设） |
| VARCHAR 类型必填，最大字符数 |
| FLOAT_VECTOR 类型必填，向量维度 |
| 字段描述，可选 |
常用 DataType
枚举值 | 说明 |
|---|---|
| 64 位整数 |
| 32 位整数 |
| 可变长度字符串（需指定 |
| 浮点型向量（需指定 |
| 二值向量 |
4. 创建 Collection 与冲突处理
4.1 基本创建
PYTHON
client.create_collection(
collection_name="DemoCollection",
schema=schema,
index_params=index_params # 可选，也可创建后再建索引
)
4.2 冲突处理
创建前检查是否存在，若存在则根据场景决定：
场景一：开发测试 — 直接删除重建
PYTHON
if client.has_collection(collection_name=COLLECTION_NAME):
client.drop_collection(collection_name=COLLECTION_NAME)
client.create_collection(collection_name=COLLECTION_NAME, schema=schema, index_params=index_params)
场景二：生产环境 — 存在则跳过，不存在才创建
PYTHON
if not client.has_collection(collection_name=COLLECTION_NAME):
client.create_collection(collection_name=COLLECTION_NAME, schema=schema)
else:
print(f"Collection '{COLLECTION_NAME}' 已存在，跳过创建")
场景三：带 force_recreate 参数控制
PYTHON
def create_collection(force_recreate=False):
if client.has_collection(collection_name=COLLECTION_NAME):
if force_recreate:
client.drop_collection(collection_name=COLLECTION_NAME)
else:
print("已存在，跳过")
return
client.create_collection(collection_name=COLLECTION_NAME, schema=schema)
4.3 create_collection() 参数详解
参数 | 说明 |
|---|---|
| 集合名称 |
| Schema 对象 |
| 可选，IndexParams 对象，一次性创建索引 |
| 一致性级别： |
5. 索引（Index）
5.1 准备索引参数
PYTHON
index_params = client.prepare_index_params()
index_params.add_index(
field_name="vector",
index_name="vector_idx", # 索引名称，可选，默认为字段名
index_type="IVF_FLAT", # 索引类型
metric_type="COSINE", # 距离度量方式
params={"nlist": 128} # 索引参数
)
5.2 常用索引类型
类型 | 说明 | 适用场景 |
|---|---|---|
| 暴力搜索，100% 召回率 | 小数据集、精度要求极高 |
| 倒排文件，速度和精度的平衡 | 通用场景、大数据集高吞吐 |
| IVF + 量化压缩，省内存 | 内存有限的场景 |
| 基于图的索引，速度极快 | 低延迟实时查询 |
| 基于磁盘的图索引 | 超大数据集无法全载入内存 |
5.3 常用距离度量（metric_type）
度量 | 说明 | 值含义 |
|---|---|---|
| 余弦相似度 | 值越大越相似（范围 [-1, 1]） |
| 欧氏距离 | 值越小越相似 |
| 内积 | 值越大越相似 |
5.4 创建与加载
PYTHON
# 方式一：创建 collection 时同时建索引
client.create_collection(collection_name=COLLECTION_NAME, schema=schema, index_params=index_params)
# 方式二：先创建 collection，再单独建索引
client.create_index(collection_name=COLLECTION_NAME, index_params=index_params)
# 加载到内存（搜索前必须执行）
client.load_collection(collection_name=COLLECTION_NAME)
# 释放内存
client.release_collection(collection_name=COLLECTION_NAME)
6. CRUD 操作
6.1 插入数据（Insert）
PYTHON
data = [
{
"doc_title": "Docker核心指南.pdf",
"page_num": 12,
"text_content": "Docker Desktop 在 Windows 下通过 WSL 2 后端运行...",
"vector": np.random.randn(1024).tolist(),
},
]
result = client.insert(
collection_name=COLLECTION_NAME,
data=data
)
print(result['insert_count']) # 插入的行数
print(result['ids']) # 插入产生的主键 ID 列表
关于动态字段：若 Schema 开启了 enable_dynamic_field=True，可直接在 data 中包含 Schema 未定义的字段：
PYTHON
data = [
{
"doc_title": "Docker核心指南.pdf",
"vector": [...],
"author": "张三", # 动态字段！Schema 未定义但可成功插入
}
]
关于主键：
auto_id=True时，data 中不要传 id 字段，Milvus 自动生成auto_id=False时，data 中必须传 id 字段
6.2 Upsert（插入或更新）
Upsert = Insert + Update。根据主键 ID 判断：
如果 ID 不存在 → 执行插入
如果 ID 已存在 → 执行覆盖更新（整行替换）
6.2.1 自定义 ID（auto_id=False）
Schema 设置 auto_id=False，插入时自行传入 id：
PYTHON
# 先插入两条自定义 ID 的数据
data = [
{"id": 1001, "doc_title": "文档A", "vector": [...], "page_num": 1},
{"id": 1002, "doc_title": "文档B", "vector": [...], "page_num": 2},
]
client.insert(collection_name=COLLECTION_NAME, data=data)
# Upsert：id=1001 已存在 → 更新；id=1003 不存在 → 插入
res = client.upsert(collection_name=COLLECTION_NAME, data=[
{
"id": 1001,
"doc_title": "文档A-更新版",
"vector": [...],
"page_num": 10,
},
{
"id": 1003,
"doc_title": "文档C-新增",
"vector": [...],
"page_num": 3,
}
])
print(res) # {'upsert_count': 2, 'ids': [1001, 1003]}
6.2.2 自增 ID（auto_id=True）
当 auto_id=True 时，无法使用 upsert 基于业务 ID 更新，因为主键由 Milvus 自动生成，外部无法预知。此时只能：
先通过搜索/查询拿到自动生成的 id
再通过 id 执行 upsert
PYTHON
# 先查询到目标数据的 id
query_res = client.query(
collection_name=COLLECTION_NAME,
filter='doc_title == "Docker核心指南.pdf"',
output_fields=["id", "doc_title", "text_content"]
)
target_id = query_res[0]["id"] # 取出自动生成的 id
# 再通过该 id upsert 更新
res = client.upsert(collection_name=COLLECTION_NAME, data=[
{
"id": target_id,
"doc_title": "Docker核心指南.pdf",
"page_num": 12,
"text_content": "更新后的文本内容...",
"vector": new_vector,
}
])
重要：对
auto_id=True的 Collection 执行 upsert 时，若不传id字段，Milvus 会生成新 ID 导致行为退化为插入而非更新。
6.2.3 upsert 返回值
PYTHON
{
"upsert_count": 2, # 实际 upsert 的行数
"ids": [1001, 1003] # 受影响的主键 ID 列表
}
6.2.4 动态字段在 Upsert 中的行为
当 Schema 开启了 enable_dynamic_field=True 时，实体的动态字段中的键值对类似于 {"author": "John", "year": 2020, "tags": ["fiction"]}。Upsert 时对动态字段的行为取决于 partial_update 参数：
覆盖模式（partial_update=False / 默认）
动态字段的值整组替换为请求中传入的所有非 Schema 定义字段及其值。
PYTHON
# 假设目标实体当前动态字段为：
# {"author": "John", "year": 2020, "tags": ["fiction"]}
client.upsert(
collection_name=COLLECTION_NAME,
data=[{
"id": 1,
"title": "某标题",
"vector": [...],
"author": "Jane", # 动态字段
"genre": "fantasy", # 新增动态字段
# ⚠️ year 和 tags 未传入，覆盖模式下它们会被移除
}]
)
# upsert 后动态字段变为：
# {"author": "Jane", "genre": "fantasy"}
# → "year" 和 "tags" 被移除，因为覆盖模式下动态字段完全由请求中的非 Schema 字段决定
合并模式（partial_update=True）
动态字段的值合并到原有动态字段中，仅覆盖或新增请求中出现的键，未出现的键保持不变。
PYTHON
# 假设目标实体当前动态字段为：
# {"author": "John", "year": 2020, "tags": ["fiction"]}
client.upsert(
collection_name=COLLECTION_NAME,
data=[{
"id": 1,
"author": "John", # 动态字段 — 保持不变
"year": 2020, # 动态字段 — 保持不变
"tags": ["fiction"], # 动态字段 — 保持不变
"genre": "fantasy", # 动态字段 — 新增
}],
partial_update=True
)
# upsert 后动态字段变为：
# {"author": "John", "year": 2020, "tags": ["fiction"], "genre": "fantasy"}
# → 合并模式，原有的键保留，新增的键追加
6.2.5 Schema 定义 JSON 字段在 Upsert 中的行为
若 Schema 中定义了一个 JSON 类型的字段（例如 extras），该字段的值是作为一个整体存储的。Upsert 时 JSON 字段不支持合并模式，只能整体覆盖。
PYTHON
from pymilvus import DataType
schema = client.create_schema(auto_id=False, enable_dynamic_field=True)
schema.add_field("id", DataType.INT64, is_primary=True)
schema.add_field("vector", DataType.FLOAT_VECTOR, dim=1024)
schema.add_field("extras", DataType.JSON) # Schema 定义的 JSON 字段
PYTHON
# 假设目标实体当前 extras 字段的值为：
# {"author": "John", "year": 2020, "tags": ["fiction"]}
# ❌ 错误：JSON 字段不支持有选择地更新单个键，即使 partial_update=True 也不行
client.upsert(
collection_name=COLLECTION_NAME,
data=[{
"id": 1,
"extras": {"year": 2021}, # 会整体覆盖，不会只改 year
}],
partial_update=True
)
# upsert 后 extras 变为：
# {"year": 2021}
# → "author" 和 "tags" 丢失了！
# ✅ 正确：传入完整的 JSON 值
client.upsert(
collection_name=COLLECTION_NAME,
data=[{
"id": 1,
"extras": {"author": "John", "year": 2021, "tags": ["fiction"]}, # 完整的 JSON
}],
partial_update=True
)
# upsert 后 extras 保持不变（仅 year 被修改）：
# {"author": "John", "year": 2021, "tags": ["fiction"]}
总结：JSON 字段被视为一个不可分割的整体，修改其中任意键都需要传入完整的 JSON 对象。若只想更新 JSON 中的部分键，需先查询出旧值、修改后再整体传回。
6.2.6 partial_update 参数详解
partial_update 控制 upsert 是全量替换还是部分更新，默认为 False。
情况 | 设置 partial_update | 只传部分字段 | 结果 |
|---|---|---|---|
默认情况 | 不设置（即 | 只传部分字段 | 报错（缺少必须字段） |
部分更新模式 |
| 只传部分字段 | 正常部分更新 |
默认行为（partial_update=False）
Milvus 把 Upsert 当作全量替换（Override）
必须提供集合 Schema 中定义的所有字段（除 autoID 主键外）
只传部分字段会报错：
Insert missed an field XXX to collection without set nullable==true or set default_value
部分更新模式（partial_update=True）
只需传入
id+ 想要修改的字段其他 Schema 定义的字段保留原有值
动态字段的行为见 6.2.4
需要 Milvus v2.5+ / v2.6+
代码示例
PYTHON
from pymilvus import MilvusClient
client = MilvusClient("http://localhost:19530")
# ==================== 只想更新部分字段 ====================
data = {
"id": 1,
"issue": "vol.999", # 只修改这个字段
# "title": "新标题", # 可以不传
# "vector": [...] # 可以不传
}
client.upsert(
collection_name="my_collection",
data=[data],
partial_update=True # ← 关键！必须加上这一行
)
总结建议
只想更新个别字段 → 加上
partial_update=True整条记录全部替换 → 不加
partial_update，但必须传所有字段生产环境强烈建议始终显式写上
partial_update参数
6.3 删除数据（Delete）
PYTHON
# 按主键 ID 删除
client.delete(
collection_name=COLLECTION_NAME,
ids=[1001, 1002]
)
# 按过滤条件删除
client.delete(
collection_name=COLLECTION_NAME,
filter='page_num > 10'
)
# 注意：ids 和 filter 只能二选一，不能同时传
6.4 查询数据（Query / Search）
标量查询（Query）
PYTHON
res = client.query(
collection_name=COLLECTION_NAME,
filter='doc_title == "Docker核心指南.pdf"',
output_fields=["id", "doc_title", "page_num", "text_content"],
limit=10,
offset=0
)
向量相似性搜索（Search）
PYTHON
search_params = {
"metric_type": "COSINE",
"params": {"ef": 64} # HNSW 的搜索范围参数
}
results = client.search(
collection_name=COLLECTION_NAME,
data=[query_vector], # 查询向量列表，可传多个
anns_field="vector", # 向量字段名
search_params=search_params,
limit=5, # 返回 top-K 结果数
output_fields=["doc_title", "text_content"], # 返回的标量字段
filter='page_num > 5' # 可选，过滤条件
)
# 解析结果
for hits in results: # results 对应每个查询向量的结果
for hit in hits: # 每个查询向量的 top-K
print(hit["id"]) # 主键
print(hit["distance"]) # 距离分数
print(hit["entity"]) # 标量字段 dict
7. 其他重要函数
7.1 检查与描述
PYTHON
# 检查 collection 是否存在
client.has_collection(collection_name="DemoCollection")
# 查看 collection 详情
client.describe_collection(collection_name="DemoCollection")
# 查看索引详情
client.describe_index(collection_name="DemoCollection", index_name="vector")
# 获取集合统计信息
client.get_collection_stats(collection_name="DemoCollection")
7.2 删除 Collection
PYTHON
client.drop_collection(collection_name="DemoCollection")
7.3 删除数据库
PYTHON
client.drop_database(db_name="DemoDB")
删除数据库前需确保其中没有任何 collection，否则会失败。
7.4 查看所有 Collections
PYTHON
collections = client.list_collections()
8. 完整示例流程
PYTHON
import numpy as np
from pymilvus import MilvusClient, DataType
# 1. 连接
client = MilvusClient(uri="http://localhost:19530")
# 2. 创建/切换数据库
if "DemoDB" not in client.list_databases():
client.create_database("DemoDB")
client.use_database("DemoDB")
# 3. 创建 Schema
schema = client.create_schema(auto_id=False, enable_dynamic_field=True)
schema.add_field("id", DataType.INT64, is_primary=True)
schema.add_field("vector", DataType.FLOAT_VECTOR, dim=1024)
schema.add_field("title", DataType.VARCHAR, max_length=255)
schema.add_field("content", DataType.VARCHAR, max_length=65535)
# 4. 创建索引
index_params = client.prepare_index_params()
index_params.add_index("vector", index_type="IVF_FLAT", metric_type="COSINE", params={"nlist": 128})
# 5. 创建 Collection（含冲突处理）
COLLECTION = "DemoCollection"
if client.has_collection(COLLECTION):
client.drop_collection(COLLECTION)
client.create_collection(COLLECTION, schema=schema, index_params=index_params)
client.load_collection(COLLECTION)
# 6. 插入数据
data = [
{"id": 1, "title": "文档1", "content": "内容1", "vector": np.random.randn(1024).tolist()},
{"id": 2, "title": "文档2", "content": "内容2", "vector": np.random.randn(1024).tolist()},
]
client.insert(COLLECTION, data)
# 7. Upsert 更新
client.upsert(COLLECTION, data=[{"id": 1, "title": "文档1-更新", "content": "新内容", "vector": np.random.randn(1024).tolist()}])
# 8. 搜索
results = client.search(COLLECTION, [np.random.randn(1024).tolist()], "vector", search_params={"metric_type": "COSINE"}, limit=3, output_fields=["title"])
for hit in results[0]:
print(f"ID={hit['id']}, Score={hit['distance']:.4f}, Title={hit['entity']['title']}")
# 9. 删除
client.delete(COLLECTION, ids=[2])
# 10. 清理
client.release_collection(COLLECTION)
client.drop_collection(COLLECTION)
9. 常见问题
Q: auto_id=True 时如何更新指定数据？
先通过 query() 查出目标数据的 id，再用该 id 执行 upsert()。
Q: 如何判断 collection 是否存在？
PYTHON
client.has_collection(collection_name="xxx") # 返回 True / False
Q: 插入时不传某些字段会怎样？
对于 Schema 定义的字段：如果该字段有默认值或可为 null，则自动填充；否则报错
对于动态字段：仅在
enable_dynamic_field=True时才可省略；否则会报错
Q: search 返回的 distance 含义？
取决于创建索引时指定的 metric_type：COSINE 越大越相似；L2 越小越相似；IP 越大越相似。
