Imported from LeoYeAI/openclaw-master-skills (skills/milvus/SKILL.md). Install upstream with npx skills add LeoYeAI/openclaw-master-skills --skill milvus. Copyright stays with the author.

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

Connection

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:

Async Client

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)

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

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:

Vector types:

add_field Parameters

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

# 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

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)

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

Search (vector similarity)

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)

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)

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

Get (by primary key)

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

Delete

# 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

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

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

Metric Types

Other Index Operations

# 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

# 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

# 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

# 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

# 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

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

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

# 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.