Milvus

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"
)

from pymilvus import AsyncMilvusClient

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

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

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
)

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=""
)

# 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")

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]}

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

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": ...}}

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
)

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

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

# 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"')

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
)