Milvus Vector Database Skill#

Operate Milvus vector databases directly through Python code using the pymilvus SDK. This skill covers the full lifecycle — connecting, schema design, collection management, vector CRUD, search, hybrid search, indexing, partitions, databases, and RBAC.

When to Use#

Use this skill when the user wants to:

• Connect to a Milvus instance (local, standalone, cluster, or Milvus Lite)
• Create collections with custom schemas
• Insert, upsert, search, query, get, or delete vectors
• Perform hybrid search with reranking
• Manage indexes, partitions, databases
• Set up users, roles, and access control (RBAC)
• Build RAG pipelines, semantic search, or recommendation systems with Milvus

Requirements#

• Python 3.8+
• pymilvus ( pip install pymilvus)
• A running Milvus instance, or use Milvus Lite (embedded, file-based) for development

Capabilities Overview#

AreaWhat You Can DoConnectionConnect to Milvus Lite, Standalone, Cluster, or Zilliz CloudCollectionsCreate (quick or custom schema), list, describe, drop, rename, load, releaseVectorsInsert, upsert, search, hybrid search, query, get, deleteIndexesCreate (AUTOINDEX, HNSW, IVF_FLAT, etc.), list, describe, dropPartitionsCreate, list, load, release, dropDatabasesCreate, list, switch, dropRBACUsers, roles, privileges management

---

Connection#

python

from pymilvus import MilvusClient

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

# Standalone / Cluster Milvus
client = MilvusClient(uri="http://localhost:19530", token="root:Milvus")

# Zilliz Cloud
client = MilvusClient(
    uri="https://in03-xxxx.api.gcp-us-west1.zillizcloud.com:19530",
    token="your_api_key"
)

Parameters:

ParameterTypeDescriptionuristr"./file.db" for Milvus Lite, "http://host:19530" for servertokenstrAPI key or "username:password"``userstrUsername (alternative to token)passwordstrPassword (alternative to token)db_namestrTarget database (default: "default")timeoutfloatOperation timeout in seconds

Async Client#

python

from pymilvus import AsyncMilvusClient

async with AsyncMilvusClient(uri="http://localhost:19530") as client:
    results = await client.search(...)

---

Collection Management#

Quick Create (auto schema + auto index + auto load)#

python

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

This automatically creates:

• id field (INT64, primary key, auto_id)
• vector field (FLOAT_VECTOR, dim=dimension)
• AUTOINDEX on vector field
• Collection is auto-loaded

Custom Schema Create#

python

from pymilvus import DataType

# Step 1: Define schema
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)

# Step 2: Define index
index_params = client.prepare_index_params()
index_params.add_index(
    field_name="embedding",
    index_type="AUTOINDEX",
    metric_type="COSINE"
)

# Step 3: Create collection
client.create_collection(
    collection_name="my_collection",
    schema=schema,
    index_params=index_params
)

Supported Data Types#

Scalar types:

DataTypeNotesDataType.BOOLBooleanDataType.INT8 / INT16 / INT32 / INT64IntegersDataType.FLOAT / DOUBLEFloating pointDataType.VARCHARString (requires max_length)DataType.JSONJSON objectDataType.ARRAYArray (requires element_type, max_capacity)

Vector types:

DataTypeNotesDataType.FLOAT_VECTORFloat32 vector (requires dim)DataType.FLOAT16_VECTORFloat16 vector (requires dim)DataType.BFLOAT16_VECTORBFloat16 vector (requires dim)DataType.BINARY_VECTORBinary vector (requires dim)DataType.SPARSE_FLOAT_VECTORSparse vector (no dim needed)

add_field Parameters#

python

schema.add_field(
    field_name="my_field",
    datatype=DataType.VARCHAR,
    is_primary=False,
    auto_id=False,
    max_length=256,          # Required for VARCHAR
    dim=768,                 # Required for vector types (except sparse)
    element_type=DataType.INT64,  # Required for ARRAY
    max_capacity=100,        # Required for ARRAY
    nullable=False,
    default_value=None,
    is_partition_key=False,
    description=""
)

Other Collection Operations#

python

# List all collections
collections = client.list_collections()

# Describe a collection
info = client.describe_collection(collection_name="my_collection")

# Check if collection exists
exists = client.has_collection(collection_name="my_collection")

# Rename a collection
client.rename_collection(old_name="old_name", new_name="new_name")

# Drop a collection
client.drop_collection(collection_name="my_collection")

# Load collection into memory (required before search/query)
client.load_collection(collection_name="my_collection")

# Release collection from memory
client.release_collection(collection_name="my_collection")

# Get load state
state = client.get_load_state(collection_name="my_collection")

# Get collection statistics
stats = client.get_collection_stats(collection_name="my_collection")

Collection Guidance#

• Quick create is best for prototyping; use custom schema for production.
• A collection must be loaded before search or query operations.
• Before dropping a collection, confirm with the user — this deletes all data.
• Use enable_dynamic_field=True to allow inserting fields not defined in the schema.

---

Vector Operations#

Target collection must exist and be loaded.

Insert#

python

data = [
    {"id": 1, "text": "AI advances", "embedding": [0.1, 0.2, ...]},
    {"id": 2, "text": "ML basics", "embedding": [0.3, 0.4, ...]},
]
res = client.insert(collection_name="my_collection", data=data)
# Returns: {"insert_count": 2, "ids": [1, 2]}

Upsert (insert or update if PK exists)#

python

res = client.upsert(collection_name="my_collection", data=data)
# Returns: {"upsert_count": 2}

Search (vector similarity)#

python

results = client.search(
    collection_name="my_collection",
    data=[[0.1, 0.2, ...]],           # List of query vectors
    anns_field="embedding",             # Vector field name
    limit=10,                           # Top-K
    output_fields=["text", "id"],       # Fields to return
    filter='age > 20 and status == "active"',  # Optional scalar filter
    search_params={
        "metric_type": "COSINE",
        "params": {"nprobe": 10}        # Index-specific params
    }
)
# Returns: List[List[dict]]
# Each hit: {"id": ..., "distance": ..., "entity": {"text": ...}}

Hybrid Search (multi-vector with reranking)#

python

from pymilvus import AnnSearchRequest, RRFRanker, WeightedRanker

req1 = AnnSearchRequest(
    data=[[0.1, 0.2, ...]],
    anns_field="dense_embedding",
    param={"metric_type": "COSINE", "params": {"nprobe": 10}},
    limit=10
)
req2 = AnnSearchRequest(
    data=[{1: 0.5, 100: 0.3}],          # Sparse vector
    anns_field="sparse_embedding",
    param={"metric_type": "IP"},
    limit=10
)

# RRF reranking
results = client.hybrid_search(
    collection_name="my_collection",
    reqs=[req1, req2],
    ranker=RRFRanker(k=60),
    limit=10,
    output_fields=["text"]
)

# Or weighted reranking
results = client.hybrid_search(
    collection_name="my_collection",
    reqs=[req1, req2],
    ranker=WeightedRanker(0.7, 0.3),
    limit=10
)

Query (filter-based retrieval)#

python

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

Get (by primary key)#

python

results = client.get(
    collection_name="my_collection",
    ids=[1, 2, 3],
    output_fields=["text"]
)

Delete#

python

# By primary keys
client.delete(collection_name="my_collection", ids=[1, 2, 3])

# By filter expression
client.delete(collection_name="my_collection", filter='status == "obsolete"')

Filter Expression Syntax#

ExpressionExampleComparisonage > 20Equalitystatus == "active"IN listid in [1, 2, 3]AND/ORage > 20 and status == "active"String matchtext like "hello%"Array containsARRAY_CONTAINS(tags, "ml")JSON fieldjson_field["key"] > 100Match allid > 0

Vector Guidance#

• The data parameter in search must match the collection's vector dimension exactly.
• For text-to-vector search, convert text to vectors using an embedding model first.
• For large inserts, batch data into chunks (e.g., 1000 rows per batch).
• Always specify output_fields to control which fields are returned.

---

Index Management#

Create Index#

python

index_params = client.prepare_index_params()

# Vector index
index_params.add_index(
    field_name="embedding",
    index_type="HNSW",               # See index types table below
    metric_type="COSINE",            # "COSINE", "L2", "IP"
    params={"M": 16, "efConstruction": 256}
)

# Optional: scalar index
index_params.add_index(
    field_name="text",
    index_type=""                    # Auto-select for scalars
)

client.create_index(
    collection_name="my_collection",
    index_params=index_params
)

Common Index Types#

Index TypeForKey ParamsNotesAUTOINDEXDense vectorsAuto-tunedRecommended for most casesFLATDense vectorsNoneBrute force, 100% recallIVF_FLATDense vectorsnlistGood balanceIVF_SQ8Dense vectorsnlistCompressed, less memoryHNSWDense vectorsM, efConstructionHigh recall, more memoryDISKANNDense vectorsNoneDisk-based, large datasetsSPARSE_INVERTED_INDEXSparse vectorsdrop_ratio_buildFor sparse vectorsSPARSE_WANDSparse vectorsdrop_ratio_buildFaster sparse search

Metric Types#

MetricDescriptionUse With"COSINE"Cosine similarity (larger = more similar)Dense vectors"L2"Euclidean distance (smaller = more similar)Dense vectors"IP"Inner product (larger = more similar)Dense & Sparse vectors

Other Index Operations#

python

# List indexes
indexes = client.list_indexes(collection_name="my_collection")

# Describe an index
info = client.describe_index(collection_name="my_collection", index_name="my_index")

# Drop an index
client.drop_index(collection_name="my_collection", index_name="my_index")

Index Guidance#

• AUTOINDEX is recommended for most use cases.
• An index is required before loading a collection.
• After creating an index, load the collection before searching.
• Sparse vectors only support "IP" metric type.

---

Partition Management#

python

# Create a partition
client.create_partition(collection_name="my_collection", partition_name="partition_A")

# List partitions
partitions = client.list_partitions(collection_name="my_collection")
# Returns: ["_default", "partition_A"]

# Check if partition exists
exists = client.has_partition(collection_name="my_collection", partition_name="partition_A")

# Load specific partitions
client.load_partitions(collection_name="my_collection", partition_names=["partition_A"])

# Release specific partitions
client.release_partitions(collection_name="my_collection", partition_names=["partition_A"])

# Drop a partition
client.drop_partition(collection_name="my_collection", partition_name="partition_A")

Partition Guidance#

• Every collection has a _default partition.
• Use is_partition_key=True on a field to enable automatic partitioning by field value.
• A partition must be loaded before search.
• Before dropping a partition, confirm with the user — all data in it will be deleted.

---

Database Management#

python

# Create a database
client.create_database(db_name="my_database")

# List all databases
databases = client.list_databases()
# Returns: ["default", "my_database"]

# Switch to a database
client.using_database(db_name="my_database")

# Drop a database (must drop all collections first)
client.drop_database(db_name="my_database")

# Or connect to a specific database at init
client = MilvusClient(uri="http://localhost:19530", db_name="my_database")

Database Guidance#

• Every Milvus instance has a "default" database.
• Before dropping a database, all collections in it must be dropped first.

---

User & Role Management (RBAC)#

User Operations#

python

# Create a user
client.create_user(user_name="analyst", password="SecureP@ss123")

# List users
users = client.list_users()

# Describe a user (shows assigned roles)
info = client.describe_user(user_name="analyst")

# Update password
client.update_password(user_name="analyst", old_password="SecureP@ss123", new_password="NewP@ss456")

# Grant role to user
client.grant_role(user_name="analyst", role_name="read_only")

# Revoke role from user
client.revoke_role(user_name="analyst", role_name="read_only")

# Drop a user
client.drop_user(user_name="analyst")

Role Operations#

python

# Create a role
client.create_role(role_name="read_only")

# List roles
roles = client.list_roles()

# Grant privilege (v2 API — recommended)
client.grant_privilege_v2(
    role_name="read_only",
    privilege="Search",                 # e.g., "Search", "Insert", "Query", "Delete"
    collection_name="my_collection",    # Use "*" for all collections
    db_name="default"                   # Use "*" for all databases
)

# Built-in privilege groups
client.grant_privilege_v2(
    role_name="admin_role",
    privilege="ClusterAdmin",           # See privilege groups below
    collection_name="*",
    db_name="*"
)

# Revoke privilege
client.revoke_privilege_v2(
    role_name="read_only",
    privilege="Search",
    collection_name="my_collection",
    db_name="default"
)

# Describe role (see granted privileges)
info = client.describe_role(role_name="read_only")

# Drop a role
client.drop_role(role_name="read_only")

Built-in Privilege Groups#

GroupScopeClusterAdminFull cluster accessClusterReadOnlyRead-only cluster accessClusterReadWriteRead-write cluster accessDatabaseAdminFull database accessDatabaseReadOnlyRead-only database accessDatabaseReadWriteRead-write database accessCollectionAdminFull collection accessCollectionReadOnlyRead-only collection accessCollectionReadWriteRead-write collection access

Common Individual Privileges#

Search, Query, Insert, Delete, Upsert, CreateIndex, DropIndex, CreateCollection, DropCollection, Load, Release, CreatePartition, DropPartition

RBAC Guidance#

• Recommended workflow: create role → grant privileges → create user → assign role.
• Use "*" for collection_name/db_name to grant on all resources.
• Before dropping a user or role, confirm with the user.

---

Common Patterns#

RAG Pipeline Pattern#

python

from pymilvus import MilvusClient, DataType

# 1. Connect
client = MilvusClient(uri="http://localhost:19530")

# 2. Create collection
schema = client.create_schema(auto_id=True, enable_dynamic_field=True)
schema.add_field("id", DataType.INT64, is_primary=True)
schema.add_field("text", DataType.VARCHAR, max_length=2048)
schema.add_field("embedding", DataType.FLOAT_VECTOR, dim=768)
schema.add_field("source", DataType.VARCHAR, max_length=256)

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

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

# 3. Insert documents (after embedding with your model)
client.insert("knowledge_base", data=[
    {"text": "chunk text...", "embedding": [...], "source": "doc1.pdf"},
])

# 4. Retrieve relevant context
results = client.search(
    collection_name="knowledge_base",
    data=[query_embedding],
    limit=5,
    output_fields=["text", "source"],
    search_params={"metric_type": "COSINE"}
)

Quick Semantic Search Pattern#

python

# Simplest possible setup
client = MilvusClient(uri="./search.db")
client.create_collection(collection_name="docs", dimension=768)
client.insert("docs", data=[{"id": i, "vector": emb, "text": txt} for i, (emb, txt) in enumerate(zip(embeddings, texts))])
results = client.search("docs", data=[query_vector], limit=10, output_fields=["text"])

---

General Guidance#

• Always check if pymilvus is installed: pip install pymilvus.
• For quick prototyping, use Milvus Lite ( uri="./file.db") — no server needed.
• A collection must be loaded into memory before search/query.
• The vector dimension in search data must exactly match the collection schema.
• For text queries, users need an embedding model to convert text to vectors first. Suggest pymilvus[model] for built-in embedding support.
• Before any destructive operation (drop collection, drop database, delete vectors), always confirm with the user.
• Use enable_dynamic_field=True when the schema may evolve.
• For large-scale inserts, batch data into chunks of ~1000 rows.
• Prefer AUTOINDEX unless the user has specific performance requirements.