Milvus Cheatsheet: Indexes, Hybrid Search and ANN Reference

# Milvus Lite — zero-ops, embedded in Python (file-based)
pip install "pymilvus>=2.4"
python -c "from pymilvus import MilvusClient; \
  print(MilvusClient('./milvus.db').list_collections())"

# Docker — standalone for dev (gRPC :19530, REST :9091)
curl -sf https://raw.githubusercontent.com/milvus-io/milvus/master/scripts/standalone_embed.sh | bash
bash standalone_embed.sh start

# Or production-grade compose
wget https://github.com/milvus-io/milvus/releases/latest/download/milvus-standalone-docker-compose.yml -O docker-compose.yml
docker compose up -d

# Helm chart (k8s, distributed mode)
helm repo add milvus https://zilliztech.github.io/milvus-helm/
helm install my-milvus milvus/milvus --set cluster.enabled=true

# JS / Go / Java clients
npm install @zilliz/milvus2-sdk-node

from pymilvus import MilvusClient, DataType

client = MilvusClient("./milvus.db")        # Milvus Lite
# or MilvusClient(uri="http://localhost:19530", token="root:Milvus")

# Quick-start path — fixed schema, dynamic fields enabled
client.create_collection(
    collection_name="kb_quick",
    dimension=384,
    metric_type="COSINE",                   # COSINE | L2 | IP
)

# Explicit schema — for production, multi-vector, FP16, etc.
schema = MilvusClient.create_schema(auto_id=False, enable_dynamic_field=True)
schema.add_field("id",      DataType.INT64, is_primary=True)
schema.add_field("vec",     DataType.FLOAT_VECTOR, dim=384)
schema.add_field("sparse",  DataType.SPARSE_FLOAT_VECTOR)        # BM25 / SPLADE
schema.add_field("title",   DataType.VARCHAR, max_length=256)
schema.add_field("tags",    DataType.ARRAY, element_type=DataType.VARCHAR,
                 max_capacity=16, max_length=64)
schema.add_field("year",    DataType.INT32)

# Per-vector index params — must be created before loading
idx = client.prepare_index_params()
idx.add_index(field_name="vec",    index_type="HNSW",
              metric_type="COSINE", params={"M": 16, "efConstruction": 200})
idx.add_index(field_name="sparse", index_type="SPARSE_INVERTED_INDEX",
              metric_type="IP")

client.create_collection("kb", schema=schema, index_params=idx)
client.load_collection("kb")                # must be loaded into memory to query

from pymilvus import MilvusClient, AnnSearchRequest, RRFRanker, WeightedRanker

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

# Two sub-requests — one dense, one sparse — same collection, different fields
dense_req = AnnSearchRequest(
    data=[[0.0]*384],
    anns_field="vec",
    param={"metric_type": "COSINE", "params": {"ef": 64}},
    limit=20,
)
sparse_req = AnnSearchRequest(
    data=[{7: 1.5, 42: 0.8}],               # sparse vector as {idx: val}
    anns_field="sparse",
    param={"metric_type": "IP", "params": {"drop_ratio_search": 0.2}},
    limit=20,
)

# Fuse — RRF or weighted
res = client.hybrid_search(
    collection_name="kb",
    reqs=[dense_req, sparse_req],
    ranker=RRFRanker(k=60),                 # or WeightedRanker(0.7, 0.3)
    limit=5,
    output_fields=["title", "tags"],
)
for hit in res[0]:
    print(f"{hit['distance']:.3f}  {hit['entity']['title']}")

# Minimal Milvus RAG store — Milvus Lite, insert, filtered search
from pymilvus import MilvusClient
from openai import OpenAI

client = MilvusClient("./milvus.db")
oai    = OpenAI()
NAME, DIM = "kb", 1536

if NAME not in client.list_collections():
    client.create_collection(
        collection_name=NAME,
        dimension=DIM,
        metric_type="COSINE",
        primary_field_name="id",
        id_type="int",
        vector_field_name="vector",
        auto_id=False,
    )

def embed(texts):
    return [d.embedding for d in
            oai.embeddings.create(model="text-embedding-3-small", input=texts).data]

docs = [
    {"id": 1, "text": "Milvus is an ANN vector DB.",     "source": "blog"},
    {"id": 2, "text": "RAG joins retrieval with gen.",   "source": "docs"},
    {"id": 3, "text": "DiskANN trades RAM for SSD.",     "source": "docs"},
]
vecs = embed([d["text"] for d in docs])
client.insert(NAME, [{**d, "vector": v} for d, v in zip(docs, vecs)])

q   = embed(["what is RAG?"])[0]
res = client.search(
    collection_name=NAME,
    data=[q], limit=2,
    filter="source == 'docs'",              # boolean expression on scalar fields
    output_fields=["text", "source"],
    search_params={"metric_type": "COSINE", "params": {"nprobe": 10}},
)
for hit in res[0]:
    e = hit["entity"]
    print(f"{hit['distance']:.3f}  [{e['source']}]  {e['text']}")