milvus

Milvus 向量数据库技能

通过 pymilvus SDK 使用 Python 代码直接操作 Milvus 向量数据库。涵盖完整生命周期：连接、模式设计、集合管理、向量 CRUD、搜索、混合搜索、全文搜索、索引、分区、数据库和 RBAC。

适用场景

当用户希望执行以下操作时，请使用此技能：

• 连接 Milvus 实例（本地、独立部署、集群或 Milvus Lite）
• 创建具有自定义模式的集合
• 插入、更新、搜索、查询、获取或删除向量
• 执行带重排序的混合搜索
• 执行全文搜索（BM25）
• 管理索引、分区和数据库
• 设置用户、角色和访问控制（RBAC）
• 使用 Milvus 构建 RAG 工作流、语义搜索或推荐系统
• 使用搜索/查询迭代器遍历大型结果集

要求

• Python 3.8+
• pymilvus（pip install pymilvus）
• 正在运行的 Milvus 实例，或在开发时使用 Milvus Lite（嵌入式、基于文件）

功能概览

领域	可执行的操作
连接	连接到 Milvus Lite、Standalone、Cluster 或 Zilliz Cloud
集合	创建（快速或自定义模式）、列出、描述、删除、重命名、清空、加载、释放
向量	插入、更新、搜索、混合搜索、查询、获取、删除
全文搜索	基于 BM25 的稀疏向量关键词搜索
迭代器	对大型数据集进行分页搜索和查询
索引	创建（AUTOINDEX、HNSW、IVF_FLAT 等）、列出、描述、删除
分区	创建、列出、加载、释放、删除
数据库	创建、列出、切换、删除
RBAC	用户、角色和权限管理

---

连接

**重要：编写任何连接代码之前，必须先询问用户的连接详情。**请询问：

1. 部署类型 — Milvus Lite（本地文件）、Standalone/Cluster（可自托管）还是 Zilliz Cloud（托管）？
2. URI — 对于可自托管部署：主机和端口（例如 http://localhost:19530）。对于 Zilliz Cloud：端点 URL。
3. 身份验证 — 如果需要，提供令牌、API 密钥或用户名/密码。
4. 数据库名称 — 如果不使用默认数据库。

**切勿假设或硬编码连接参数。**只有在用户明确要求开发时使用本地/嵌入式模式时，才使用 Milvus Lite（uri="./milvus.db"）。

from pymilvus import MilvusClient

# Milvus Lite (embedded, file-based — great for dev/test)
client = MilvusClient(uri="./milvus.db")

# Standalone / Cluster Milvus (ask user for actual host:port and credentials)
client = MilvusClient(uri="<USER_URI>", token="<USER_TOKEN>")

# Zilliz Cloud (ask user for endpoint and API key)
client = MilvusClient(uri="<USER_ZILLIZ_ENDPOINT>", token="<USER_API_KEY>")

参数：

参数	类型	描述
uri	str	Milvus Lite 使用 "./file.db"，服务端使用 "http://host:19530"
token	str	API 密钥或 "username:password"
user	str	用户名（替代 token）
password	str	密码（替代 token）
db_name	str	目标数据库（默认值："default"）
timeout	float	操作超时时间，单位为秒

异步客户端

from pymilvus import AsyncMilvusClient

async with AsyncMilvusClient(uri="<USER_URI>") as client:
    results = await client.search(collection_name="my_collection", data=[query_vector], limit=10)

---

集合管理

快速创建（自动模式 + 自动索引 + 自动加载）

client.create_collection(
    collection_name="my_collection",
    dimension=768,
    metric_type="COSINE"  # Optional: "COSINE" (default), "L2", "IP"
)

此操作会自动创建 id 字段（INT64，主键，auto_id）、vector 字段（FLOAT_VECTOR）和 AUTOINDEX，并自动加载集合。

自定义模式创建

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("text", DataType.VARCHAR, max_length=512)
schema.add_field("embedding", DataType.FLOAT_VECTOR, dim=768)

index_params = client.prepare_index_params()
index_params.add_index(field_name="embedding", index_type="AUTOINDEX", metric_type="COSINE")

client.create_collection(collection_name="my_collection", schema=schema, index_params=index_params)

数据类型、add_field 参数和所有集合操作，请参阅 references/collection.md。

其他集合操作

client.list_collections()
client.describe_collection(collection_name="my_collection")
client.has_collection(collection_name="my_collection")
client.rename_collection(old_name="old", new_name="new")
client.drop_collection(collection_name="my_collection")
client.truncate_collection(collection_name="my_collection")
client.load_collection(collection_name="my_collection")
client.release_collection(collection_name="my_collection")
client.get_load_state(collection_name="my_collection")
client.get_collection_stats(collection_name="my_collection")

• 快速创建最适合原型开发；生产环境请使用自定义模式。
• 搜索或查询之前必须先加载集合。
• 使用 enable_dynamic_field=True 可插入模式中未定义的字段。

---

向量操作

混合搜索、全文搜索、迭代器、过滤器语法和详细示例，请参阅 references/vector.md。

插入 / 更新

# Vectors must come from an embedding model — never use fake/placeholder vectors
from pymilvus import model

embedding_fn = model.dense.SentenceTransformerEmbeddingFunction(model_name="all-MiniLM-L6-v2")

docs = ["AI advances in 2024", "ML basics for beginners"]
vectors = embedding_fn.encode_documents(docs)

data = [
    {"id": 1, "text": docs[0], "embedding": vectors[0]},
    {"id": 2, "text": docs[1], "embedding": vectors[1]},
]
client.insert(collection_name="my_collection", data=data)
client.upsert(collection_name="my_collection", data=data)

搜索（向量相似度）

# Use the same embedding model to encode the query
query_vectors = embedding_fn.encode_queries(["What is artificial intelligence?"])

results = client.search(
    collection_name="my_collection",
    data=query_vectors,
    anns_field="embedding",
    limit=10,
    output_fields=["text", "id"],
    filter='age > 20 and status == "active"',
    search_params={"metric_type": "COSINE", "params": {"nprobe": 10}}
)

查询 / 获取 / 删除

# Query by filter
client.query(collection_name="my_collection", filter='id in [1, 2, 3]', output_fields=["text"], limit=100)

# Get by primary key
client.get(collection_name="my_collection", ids=[1, 2, 3], output_fields=["text"])

# Delete
client.delete(collection_name="my_collection", ids=[1, 2, 3])
client.delete(collection_name="my_collection", filter='status == "obsolete"')

• 切勿使用伪造或占位向量（例如，[0.1, 0.2, ...]）。始终使用嵌入模型生成向量。
• 使用 pip install "pymilvus[model]" 安装内置嵌入函数，或使用任意嵌入模型（OpenAI、Cohere 等）。
• 搜索时的向量维度必须与集合架构完全一致。
• 查询嵌入模型必须与生成已存储向量时使用的模型相同。
• 对于大量插入的数据，请将数据分批处理（例如，每批 1000 行）。
• 对于大量结果，请使用迭代器 — 请参阅 references/vector.md。

---

索引管理

有关索引类型、度量类型和参数，请参阅 references/index.md。

index_params = client.prepare_index_params()
index_params.add_index(
    field_name="embedding",
    index_type="HNSW",
    metric_type="COSINE",
    params={"M": 16, "efConstruction": 256}
)
client.create_index(collection_name="my_collection", index_params=index_params)

client.list_indexes(collection_name="my_collection")
client.describe_index(collection_name="my_collection", index_name="my_index")
client.drop_index(collection_name="my_collection", index_name="my_index")

• 大多数用例都推荐使用 AUTOINDEX。
• 加载集合前必须先创建索引。

---

其他功能

功能	参考
分区管理	references/partition.md
数据库管理	references/database.md
用户与角色管理（RBAC）	references/user-role.md
常见模式（RAG、语义搜索）	references/patterns.md

---

通用指南

• 编写连接代码前，始终向用户询问连接详细信息（URI、令牌/凭据）。切勿假设或硬编码连接参数。
• 切勿生成伪造或占位向量。 始终使用嵌入模型生成真实向量。对于内置嵌入函数，建议使用 pip install "pymilvus[model]"。
• 快速原型开发时，可使用 Milvus Lite（uri="./file.db"）— 无需服务端，但仅限用户明确要求本地/嵌入式模式时使用。
• 搜索/查询前，必须先将集合加载到内存中。
• 搜索数据中的向量维度必须与集合架构完全一致。
• 查询嵌入模型必须与生成已存储向量时使用的模型相同。
• 执行任何破坏性操作（删除集合、删除数据库、删除向量）前，始终向用户确认。
• 架构可能发生变化时，使用 enable_dynamic_field=True。
• 除非用户有特定的性能要求，否则优先使用 AUTOINDEX。
• 使用 truncate_collection 清除所有数据，而不删除集合。
• 处理大型数据集时，使用迭代器（search_iterator、query_iterator），不要增加限制数量。