Milvus for Computer Vision: An In-Depth Guide

SELECT * FROM users WHERE name = 'Alice';

[0.023, -0.412, 0.881, 0.003, -0.667, ..., 0.142]  # 512 values total

import numpy as np

def normalize(vector: np.ndarray) -> np.ndarray:
    """Normalize a vector to unit length (L2 norm = 1)."""
    norm = np.linalg.norm(vector)
    if norm == 0:
        return vector
    return vector / norm

# For a batch of vectors (shape: [N, D])
def normalize_batch(vectors: np.ndarray) -> np.ndarray:
    norms = np.linalg.norm(vectors, axis=1, keepdims=True)
    norms = np.where(norms == 0, 1, norms)  # avoid division by zero
    return vectors / norms

# Create and activate a virtual environment (recommended)
python -m venv milvus-cv-env
source milvus-cv-env/bin/activate  # Linux/Mac
# milvus-cv-env\Scripts\activate   # Windows

# Install the Milvus Python SDK
pip install pymilvus

# Install pymilvus with MilvusClient support (recommended, includes model utilities)
pip install "pymilvus[model]"

# Common CV libraries
pip install numpy pillow
pip install torch torchvision  # if using PyTorch models

import pymilvus
print(pymilvus.__version__)  # Should print e.g. "2.4.x"

from pymilvus import MilvusClient
print("PyMilvus installed correctly")

pip install milvus-lite  # already included in pymilvus >= 2.4.2

from pymilvus import MilvusClient

# Pass a file path — Milvus Lite creates/opens a local database file
client = MilvusClient("./my_cv_database.db")

print("Connected to Milvus Lite")

# List all collections in this database
collections = client.list_collections()
print(collections)

from pymilvus import MilvusClient

# Milvus Lite
client = MilvusClient("./cv_database.db")

# Standalone (Docker)
client = MilvusClient(uri="http://localhost:19530")

# With authentication (if enabled)
client = MilvusClient(
    uri="http://localhost:19530",
    token="root:Milvus"  # format: "username:password"
)

# Distributed (Kubernetes)
client = MilvusClient(uri="http://EXTERNAL_IP:19530")

from pymilvus import MilvusClient, DataType

client = MilvusClient("./cv_database.db")

# Create a schema
schema = client.create_schema(
    auto_id=True,           # Milvus auto-generates the primary key
    enable_dynamic_field=True,  # Allow inserting extra fields not in schema
)

# Add the primary key field
schema.add_field(
    field_name="id",
    datatype=DataType.INT64,
    is_primary=True,
)

# Add the vector field — CRITICAL: dim must match your embedding model's output size
schema.add_field(
    field_name="embedding",
    datatype=DataType.FLOAT_VECTOR,
    dim=512,  # Change this to match your model (e.g., 768, 1536, 2048)
)

# Add scalar metadata fields
schema.add_field(
    field_name="image_path",
    datatype=DataType.VARCHAR,
    max_length=1024,
)

schema.add_field(
    field_name="label",
    datatype=DataType.VARCHAR,
    max_length=128,
)

schema.add_field(
    field_name="confidence",
    datatype=DataType.FLOAT,
)

schema.add_field(
    field_name="timestamp",
    datatype=DataType.INT64,  # store as Unix epoch milliseconds
)

from pymilvus import MilvusClient

# Define index parameters for the vector field
index_params = client.prepare_index_params()

index_params.add_index(
    field_name="embedding",      # must match your vector field name
    index_type="HNSW",           # index algorithm (see Section 9 for all options)
    metric_type="COSINE",        # distance metric: L2, IP, or COSINE
    params={
        "M": 16,                 # HNSW: number of neighbors per node (8–64; higher = better recall, more memory)
        "efConstruction": 200,   # HNSW: build-time search depth (higher = better quality index, slower build)
    }
)

# Also create an index on a scalar field for fast filtering
index_params.add_index(
    field_name="label",
    index_type="Trie",           # inverted index for VARCHAR fields
)

# Create the collection with the schema and index parameters
client.create_collection(
    collection_name="image_embeddings",
    schema=schema,
    index_params=index_params,
)

print("Collection created successfully")

# Verify it exists
collections = client.list_collections()
print(f"Collections: {collections}")

# Get collection info
info = client.describe_collection("image_embeddings")
print(info)

# Creates a collection with auto schema: id (INT64 PK) + vector (FLOAT_VECTOR)
client.create_collection(
    collection_name="quick_test",
    dimension=512,
    metric_type="COSINE",
)
# This is great for testing but you cannot add custom metadata fields this way

# WARNING: This permanently deletes all data in the collection
client.drop_collection("image_embeddings")

import numpy as np
import time

# Simulate embedding extraction
def mock_embed(n: int, dim: int = 512) -> np.ndarray:
    """Generate random normalized vectors to simulate embeddings."""
    vectors = np.random.randn(n, dim).astype(np.float32)
    norms = np.linalg.norm(vectors, axis=1, keepdims=True)
    return (vectors / norms).tolist()

# Prepare data as a list of dicts (one dict per entity)
data = [
    {
        # "id" is omitted because auto_id=True
        "embedding": mock_embed(1, dim=512)[0],
        "image_path": "/dataset/images/dog_001.jpg",
        "label": "dog",
        "confidence": 0.97,
        "timestamp": int(time.time() * 1000),
    },
    {
        "embedding": mock_embed(1, dim=512)[0],
        "image_path": "/dataset/images/cat_002.jpg",
        "label": "cat",
        "confidence": 0.92,
        "timestamp": int(time.time() * 1000),
    },
]

# Insert the data
result = client.insert(
    collection_name="image_embeddings",
    data=data,
)

print(f"Inserted {result['insert_count']} entities")
print(f"Primary keys: {result['ids']}")

import numpy as np
import time

def embed_batch(image_paths: list, dim: int = 512) -> list:
    """
    Placeholder function — replace with your actual embedding model call.
    Should return a list of normalized float vectors.
    """
    n = len(image_paths)
    vectors = np.random.randn(n, dim).astype(np.float32)
    norms = np.linalg.norm(vectors, axis=1, keepdims=True)
    return (vectors / norms).tolist()


def insert_images_in_batches(
    client: MilvusClient,
    collection_name: str,
    image_paths: list,
    labels: list,
    batch_size: int = 2000,
    embedding_dim: int = 512,
):
    """
    Extracts embeddings from images and inserts them into Milvus in batches.
    """
    total = len(image_paths)
    inserted = 0

    for start in range(0, total, batch_size):
        end = min(start + batch_size, total)
        batch_paths = image_paths[start:end]
        batch_labels = labels[start:end]

        # Extract embeddings for this batch
        batch_embeddings = embed_batch(batch_paths, dim=embedding_dim)

        # Build the data list
        batch_data = [
            {
                "embedding": batch_embeddings[i],
                "image_path": batch_paths[i],
                "label": batch_labels[i],
                "confidence": 1.0,  # placeholder
                "timestamp": int(time.time() * 1000),
            }
            for i in range(len(batch_paths))
        ]

        # Insert
        result = client.insert(
            collection_name=collection_name,
            data=batch_data,
        )

        inserted += result["insert_count"]
        print(f"Progress: {inserted}/{total} ({100*inserted/total:.1f}%)")

    print(f" Done! Inserted {inserted} entities total.")
    return inserted


# Example usage
image_paths = [f"/data/images/img_{i:06d}.jpg" for i in range(100_000)]
labels = ["dog" if i % 2 == 0 else "cat" for i in range(100_000)]

insert_images_in_batches(
    client=client,
    collection_name="image_embeddings",
    image_paths=image_paths,
    labels=labels,
    batch_size=2000,
    embedding_dim=512,
)

result = client.upsert(
    collection_name="image_embeddings",
    data=[
        {
            "id": 42,                  # specify the existing ID to update
            "embedding": new_vector,
            "image_path": "/updated/path/image.jpg",
            "label": "updated_label",
            "confidence": 0.99,
            "timestamp": int(time.time() * 1000),
        }
    ],
)

# Delete by primary key
client.delete(
    collection_name="image_embeddings",
    ids=[1, 2, 3, 42],
)

# Delete by filter expression
client.delete(
    collection_name="image_embeddings",
    filter="label == 'cat'",
)

# Force flush — seals all growing segments and ensures data is persisted
client.flush(collection_name="image_embeddings")

results = client.search(
    collection_name="image_embeddings",
    data=[query_vector],
    limit=10,
    consistency_level="Strong",  # waits for latest data to be visible
)

index_params.add_index(
    field_name="embedding",
    index_type="FLAT",
    metric_type="COSINE",
    params={},
)

index_params.add_index(
    field_name="embedding",
    index_type="IVF_FLAT",
    metric_type="L2",
    params={
        "nlist": 1024,  # number of clusters. Rule of thumb: sqrt(N) where N = dataset size
    }
)

search_params = {
    "nprobe": 16,  # number of clusters to search (higher = better recall, slower query)
}

index_params.add_index(
    field_name="embedding",
    index_type="IVF_SQ8",
    metric_type="L2",
    params={"nlist": 1024},
)

index_params.add_index(
    field_name="embedding",
    index_type="IVF_PQ",
    metric_type="L2",
    params={
        "nlist": 1024,
        "m": 8,       # number of sub-quantizers (must divide evenly into dim)
        "nbits": 8,   # bits per sub-quantizer code (typically 8)
    }
)

index_params.add_index(
    field_name="embedding",
    index_type="HNSW",
    metric_type="COSINE",
    params={
        "M": 16,              # max connections per node per layer
                              # Range: 4–64. Higher = better recall, more memory, slower build
        "efConstruction": 200, # search width during index construction
                              # Range: 8–512. Higher = better quality, slower build
    }
)

search_params = {
    "ef": 100,  # search-time expansion factor (must be >= limit/top_k)
                # Higher = better recall, slower queries
}

index_params.add_index(
    field_name="embedding",
    index_type="SCANN",
    metric_type="COSINE",
    params={
        "nlist": 1024,
        "with_raw_data": True,
    }
)

index_params.add_index(
    field_name="embedding",
    index_type="DISKANN",
    metric_type="L2",
    params={},
)

# GPU-accelerated IVF_FLAT
index_params.add_index(
    field_name="embedding",
    index_type="GPU_IVF_FLAT",
    metric_type="L2",
    params={"nlist": 1024},
)

# GPU-accelerated CAGRA (graph-based, state of the art for GPU)
index_params.add_index(
    field_name="embedding",
    index_type="GPU_CAGRA",
    metric_type="L2",
    params={
        "intermediate_graph_degree": 64,
        "graph_degree": 32,
    }
)

import numpy as np

# Simulate a query embedding (in practice, this comes from embedding your query image)
query_vector = np.random.randn(512).astype(np.float32)
query_vector = (query_vector / np.linalg.norm(query_vector)).tolist()

# Perform the search
results = client.search(
    collection_name="image_embeddings",
    data=[query_vector],          # list of query vectors (supports batch queries)
    limit=10,                     # return top 10 most similar
    output_fields=["image_path", "label", "confidence"],
    search_params={"ef": 100},    # HNSW-specific params (omit for FLAT)
)

# Results is a list of lists (one inner list per query vector)
for hit in results[0]:
    print(f"ID: {hit['id']}")
    print(f"Distance: {hit['distance']:.4f}")
    print(f"Image: {hit['entity']['image_path']}")
    print(f"Label: {hit['entity']['label']}")
    print()

query_vectors = [
    np.random.randn(512).astype(np.float32).tolist()
    for _ in range(5)
]

results = client.search(
    collection_name="image_embeddings",
    data=query_vectors,
    limit=10,
    output_fields=["image_path", "label"],
)

for query_idx, query_results in enumerate(results):
    print(f"Query {query_idx} top results:")
    for hit in query_results:
        print(f"  {hit['entity']['image_path']} (distance: {hit['distance']:.4f})")

results = client.search(
    collection_name="image_embeddings",
    data=[query_vector],
    limit=10,
    filter="label == 'dog'",
    output_fields=["image_path", "label", "confidence"],
)

# Comparison operators
"confidence > 0.9"
"timestamp >= 1700000000000"
"label != 'cat'"

# Logical operators
"label == 'dog' AND confidence > 0.8"
"label in ['dog', 'cat']"
"NOT (label in ['background', 'unknown'])"

# String operations
"image_path like '/dataset/train/%'"

# Range
"confidence > 0.7 AND confidence < 0.95"

# JSON field access
"metadata['camera_id'] == 'cam_01'"

results = client.query(
    collection_name="image_embeddings",
    filter="label == 'dog' AND confidence > 0.9",
    output_fields=["id", "image_path", "label", "confidence"],
    limit=100,
)

entities = client.get(
    collection_name="image_embeddings",
    ids=[1, 2, 42],
    output_fields=["image_path", "label"],
)

from pymilvus import MilvusClient, DataType
import numpy as np
import time

# ─── Configuration ────────────────────────────────────────────────────────────
COLLECTION_NAME = "image_similarity"
EMBEDDING_DIM = 512
MILVUS_URI = "./image_similarity.db"

client = MilvusClient(MILVUS_URI)

# ─── Create Collection ────────────────────────────────────────────────────────
def create_image_similarity_collection():
    if client.has_collection(COLLECTION_NAME):
        print(f"Collection '{COLLECTION_NAME}' already exists.")
        return

    schema = client.create_schema(auto_id=True, enable_dynamic_field=False)
    schema.add_field("id", DataType.INT64, is_primary=True)
    schema.add_field("embedding", DataType.FLOAT_VECTOR, dim=EMBEDDING_DIM)
    schema.add_field("image_path", DataType.VARCHAR, max_length=1024)
    schema.add_field("category", DataType.VARCHAR, max_length=128)
    schema.add_field("dataset_split", DataType.VARCHAR, max_length=16)
    schema.add_field("width", DataType.INT32)
    schema.add_field("height", DataType.INT32)
    schema.add_field("file_size_bytes", DataType.INT64)
    schema.add_field("inserted_at", DataType.INT64)

    index_params = client.prepare_index_params()
    index_params.add_index(
        field_name="embedding",
        index_type="HNSW",
        metric_type="COSINE",
        params={"M": 16, "efConstruction": 200},
    )
    index_params.add_index(field_name="category", index_type="Trie")
    index_params.add_index(field_name="dataset_split", index_type="Trie")

    client.create_collection(
        collection_name=COLLECTION_NAME,
        schema=schema,
        index_params=index_params,
    )
    print(f"Created collection '{COLLECTION_NAME}'")


# ─── Embedding Function (Model-Agnostic Placeholder) ─────────────────────────
def extract_embedding(image_path: str) -> np.ndarray:
    """
    Replace this function with your actual embedding model call.

    Example with torchvision (ResNet-50):
        from torchvision import models, transforms
        from PIL import Image
        import torch

        model = models.resnet50(pretrained=True)
        model.eval()
        embedding_model = torch.nn.Sequential(*list(model.children())[:-1])

        transform = transforms.Compose([
            transforms.Resize(256),
            transforms.CenterCrop(224),
            transforms.ToTensor(),
            transforms.Normalize(mean=[0.485, 0.456, 0.406],
                                 std=[0.229, 0.224, 0.225]),
        ])

        img = Image.open(image_path).convert("RGB")
        tensor = transform(img).unsqueeze(0)
        with torch.no_grad():
            embedding = embedding_model(tensor).squeeze().numpy()
        embedding = embedding / np.linalg.norm(embedding)
        return embedding
    """
    vec = np.random.randn(EMBEDDING_DIM).astype(np.float32)
    return vec / np.linalg.norm(vec)


# ─── Ingest Images ────────────────────────────────────────────────────────────
def ingest_images(image_records: list, batch_size: int = 2000):
    total = len(image_records)
    inserted = 0

    for start in range(0, total, batch_size):
        batch = image_records[start : start + batch_size]

        data = []
        for record in batch:
            embedding = extract_embedding(record["path"])
            data.append({
                "embedding": embedding.tolist(),
                "image_path": record["path"],
                "category": record["category"],
                "dataset_split": record["split"],
                "width": record["width"],
                "height": record["height"],
                "file_size_bytes": record["size"],
                "inserted_at": int(time.time() * 1000),
            })

        result = client.insert(collection_name=COLLECTION_NAME, data=data)
        inserted += result["insert_count"]
        print(f"Ingested {inserted}/{total} images")

    return inserted


# ─── Search ───────────────────────────────────────────────────────────────────
def find_similar_images(
    query_image_path: str,
    top_k: int = 10,
    category_filter: str = None,
    min_dimension: int = None,
) -> list:
    query_embedding = extract_embedding(query_image_path)

    filters = []
    if category_filter:
        filters.append(f"category == '{category_filter}'")
    if min_dimension:
        filters.append(f"width >= {min_dimension} AND height >= {min_dimension}")

    filter_expr = " AND ".join(filters) if filters else None

    results = client.search(
        collection_name=COLLECTION_NAME,
        data=[query_embedding.tolist()],
        limit=top_k,
        filter=filter_expr,
        search_params={"ef": max(top_k * 10, 100)},
        output_fields=["image_path", "category", "width", "height"],
    )

    return [
        {
            "id": hit["id"],
            "image_path": hit["entity"]["image_path"],
            "similarity": hit["distance"],
            "category": hit["entity"]["category"],
            "width": hit["entity"]["width"],
            "height": hit["entity"]["height"],
        }
        for hit in results[0]
    ]


# ─── Duplicate Detection ──────────────────────────────────────────────────────
def find_near_duplicates(similarity_threshold: float = 0.98, batch_size: int = 100):
    duplicates = []
    offset = 0

    while True:
        entities = client.query(
            collection_name=COLLECTION_NAME,
            filter="id > 0",
            output_fields=["id", "embedding", "image_path"],
            limit=batch_size,
            offset=offset,
        )

        if not entities:
            break

        for entity in entities:
            results = client.search(
                collection_name=COLLECTION_NAME,
                data=[entity["embedding"]],
                limit=5,
                search_params={"ef": 50},
                output_fields=["image_path"],
            )

            for hit in results[0][1:]:
                if hit["distance"] >= similarity_threshold:
                    pair = tuple(sorted([entity["id"], hit["id"]]))
                    entry = (pair[0], pair[1], hit["distance"])
                    if entry not in duplicates:
                        duplicates.append(entry)

        offset += batch_size

    return duplicates

from pymilvus import MilvusClient, DataType

COLLECTION_NAME = "face_identities"
FACE_EMBEDDING_DIM = 512

client = MilvusClient("./face_db.db")

schema = client.create_schema(auto_id=True, enable_dynamic_field=True)
schema.add_field("id", DataType.INT64, is_primary=True)
schema.add_field("embedding", DataType.FLOAT_VECTOR, dim=FACE_EMBEDDING_DIM)
schema.add_field("person_id", DataType.VARCHAR, max_length=64)
schema.add_field("person_name", DataType.VARCHAR, max_length=128)
schema.add_field("confidence_score", DataType.FLOAT)
schema.add_field("source_image", DataType.VARCHAR, max_length=1024)
schema.add_field("enrolled_at", DataType.INT64)
schema.add_field("is_active", DataType.BOOL)

index_params = client.prepare_index_params()
index_params.add_index(
    field_name="embedding",
    index_type="HNSW",
    metric_type="IP",
    params={"M": 16, "efConstruction": 200},
)
index_params.add_index(field_name="person_id", index_type="Trie")
index_params.add_index(field_name="is_active", index_type="BITMAP")

client.create_collection(
    collection_name=COLLECTION_NAME,
    schema=schema,
    index_params=index_params,
)

import numpy as np
import time

def extract_face_embedding(aligned_face_image_path: str) -> np.ndarray:
    """
    Placeholder — replace with your actual face recognition model.

    Example frameworks:
    - InsightFace (ArcFace): pip install insightface
    - deepface: pip install deepface
    - facenet-pytorch: pip install facenet-pytorch
    """
    vec = np.random.randn(FACE_EMBEDDING_DIM).astype(np.float32)
    return vec / np.linalg.norm(vec)


def assess_face_quality(image_path: str) -> float:
    """
    Estimate the quality of a face image for enrollment (0.0–1.0).
    In practice, use a dedicated face quality assessment model.
    """
    return 0.95  # placeholder


def enroll_person(
    person_id: str,
    person_name: str,
    face_image_paths: list,
    min_quality_threshold: float = 0.7,
):
    enrolled_count = 0

    for image_path in face_image_paths:
        quality = assess_face_quality(image_path)

        if quality < min_quality_threshold:
            print(f"Skipping {image_path} — quality {quality:.2f} below threshold")
            continue

        embedding = extract_face_embedding(image_path)

        client.insert(
            collection_name=COLLECTION_NAME,
            data=[{
                "embedding": embedding.tolist(),
                "person_id": person_id,
                "person_name": person_name,
                "confidence_score": quality,
                "source_image": image_path,
                "enrolled_at": int(time.time() * 1000),
                "is_active": True,
            }]
        )
        enrolled_count += 1

    print(f"Enrolled {enrolled_count} faces for {person_name} ({person_id})")
    return enrolled_count

def recognize_face(
    query_face_path: str,
    top_k: int = 5,
    similarity_threshold: float = 0.7,
) -> dict:
    query_embedding = extract_face_embedding(query_face_path)

    results = client.search(
        collection_name=COLLECTION_NAME,
        data=[query_embedding.tolist()],
        limit=top_k,
        filter="is_active == True",
        search_params={"ef": 200},
        output_fields=["person_id", "person_name", "confidence_score"],
    )

    if not results or not results[0]:
        return {"status": "unknown", "reason": "no results"}

    top_hit = results[0][0]
    top_similarity = top_hit["distance"]

    if top_similarity < similarity_threshold:
        return {
            "status": "unknown",
            "best_match": {"person_id": top_hit["entity"]["person_id"], "similarity": top_similarity},
            "reason": f"similarity {top_similarity:.4f} below threshold {similarity_threshold}",
        }

    person_votes = {}
    for hit in results[0]:
        if hit["distance"] >= similarity_threshold:
            pid = hit["entity"]["person_id"]
            person_votes.setdefault(pid, []).append(hit["distance"])

    if not person_votes:
        return {"status": "unknown", "reason": "no votes above threshold"}

    best_person = max(person_votes, key=lambda pid: sum(person_votes[pid]) / len(person_votes[pid]))
    avg_similarity = sum(person_votes[best_person]) / len(person_votes[best_person])

    return {
        "status": "recognized",
        "person_id": best_person,
        "person_name": results[0][0]["entity"]["person_name"],
        "similarity": avg_similarity,
        "num_matching_embeddings": len(person_votes[best_person]),
    }


# ─── Verification (1:1) ───────────────────────────────────────────────────────
def verify_identity(image_path_1: str, image_path_2: str, threshold: float = 0.7) -> dict:
    emb1 = extract_face_embedding(image_path_1)
    emb2 = extract_face_embedding(image_path_2)
    similarity = float(np.dot(emb1, emb2))
    return {"same_person": similarity >= threshold, "similarity": similarity, "threshold": threshold}


# ─── Removing an Identity ─────────────────────────────────────────────────────
def deactivate_person(person_id: str):
    entities = client.query(
        collection_name=COLLECTION_NAME,
        filter=f"person_id == '{person_id}'",
        output_fields=["id"],
    )
    if not entities:
        print(f"No enrollments found for person_id: {person_id}")
        return
    client.delete(collection_name=COLLECTION_NAME, ids=[e["id"] for e in entities])
    print(f"Deleted {len(entities)} enrollments for person {person_id}")

from pymilvus import MilvusClient, DataType

COLLECTION_NAME = "object_detections"
REGION_EMBEDDING_DIM = 512

client = MilvusClient("./object_detection.db")

schema = client.create_schema(auto_id=True, enable_dynamic_field=True)
schema.add_field("id", DataType.INT64, is_primary=True)
schema.add_field("embedding", DataType.FLOAT_VECTOR, dim=REGION_EMBEDDING_DIM)
schema.add_field("source_image_path", DataType.VARCHAR, max_length=1024)
schema.add_field("source_image_id", DataType.VARCHAR, max_length=64)
schema.add_field("bbox_x1", DataType.FLOAT)
schema.add_field("bbox_y1", DataType.FLOAT)
schema.add_field("bbox_x2", DataType.FLOAT)
schema.add_field("bbox_y2", DataType.FLOAT)
schema.add_field("class_name", DataType.VARCHAR, max_length=64)
schema.add_field("class_id", DataType.INT32)
schema.add_field("detection_score", DataType.FLOAT)
schema.add_field("area_fraction", DataType.FLOAT)
schema.add_field("detected_at", DataType.INT64)

index_params = client.prepare_index_params()
index_params.add_index(
    field_name="embedding",
    index_type="HNSW",
    metric_type="COSINE",
    params={"M": 16, "efConstruction": 200},
)
index_params.add_index(field_name="class_name", index_type="Trie")
index_params.add_index(field_name="class_id", index_type="STL_SORT")
index_params.add_index(field_name="detection_score", index_type="STL_SORT")

client.create_collection(
    collection_name=COLLECTION_NAME,
    schema=schema,
    index_params=index_params,
)

import numpy as np
import time
from dataclasses import dataclass

@dataclass
class Detection:
    class_name: str
    class_id: int
    score: float
    x1: float
    y1: float
    x2: float
    y2: float


def detect_objects(image_path: str) -> list[Detection]:
    """
    Placeholder for your object detection model.

    Example with Ultralytics YOLO:
        from ultralytics import YOLO
        model = YOLO("yolov8n.pt")
        results = model(image_path)
        detections = []
        for box in results[0].boxes:
            x1, y1, x2, y2 = box.xyxyn[0].tolist()
            detections.append(Detection(
                class_name=model.names[int(box.cls)],
                class_id=int(box.cls),
                score=float(box.conf),
                x1=x1, y1=y1, x2=x2, y2=y2,
            ))
        return detections
    """
    return [
        Detection("car", 2, 0.95, 0.1, 0.2, 0.4, 0.8),
        Detection("person", 0, 0.87, 0.5, 0.1, 0.7, 0.9),
    ]


def extract_region_embedding(image_path: str, detection: Detection) -> np.ndarray:
    """
    Crop the detected region and extract its embedding.

    Example with PIL:
        from PIL import Image
        img = Image.open(image_path).convert("RGB")
        w, h = img.size
        box = (int(detection.x1*w), int(detection.y1*h),
               int(detection.x2*w), int(detection.y2*h))
        region = img.crop(box)
        # Pass through embedding model...
    """
    vec = np.random.randn(REGION_EMBEDDING_DIM).astype(np.float32)
    return vec / np.linalg.norm(vec)


def process_and_ingest_image(image_path: str, image_id: str):
    detections = detect_objects(image_path)
    if not detections:
        return 0

    data = []
    for det in detections:
        if det.score < 0.5:
            continue
        embedding = extract_region_embedding(image_path, det)
        area = (det.x2 - det.x1) * (det.y2 - det.y1)
        data.append({
            "embedding": embedding.tolist(),
            "source_image_path": image_path,
            "source_image_id": image_id,
            "bbox_x1": det.x1, "bbox_y1": det.y1,
            "bbox_x2": det.x2, "bbox_y2": det.y2,
            "class_name": det.class_name,
            "class_id": det.class_id,
            "detection_score": det.score,
            "area_fraction": area,
            "detected_at": int(time.time() * 1000),
        })

    result = client.insert(collection_name=COLLECTION_NAME, data=data)
    return result["insert_count"]


def find_similar_objects(
    query_image_path: str,
    query_detection: Detection,
    top_k: int = 10,
    same_class_only: bool = True,
    min_score: float = 0.7,
) -> list:
    query_embedding = extract_region_embedding(query_image_path, query_detection)

    filters = [f"detection_score >= {min_score}"]
    if same_class_only:
        filters.append(f"class_name == '{query_detection.class_name}'")

    results = client.search(
        collection_name=COLLECTION_NAME,
        data=[query_embedding.tolist()],
        limit=top_k,
        filter=" AND ".join(filters),
        search_params={"ef": 150},
        output_fields=[
            "source_image_path", "class_name", "detection_score",
            "bbox_x1", "bbox_y1", "bbox_x2", "bbox_y2"
        ],
    )

    return [
        {
            "image_path": hit["entity"]["source_image_path"],
            "similarity": hit["distance"],
            "class_name": hit["entity"]["class_name"],
            "detection_score": hit["entity"]["detection_score"],
            "bbox": {
                "x1": hit["entity"]["bbox_x1"], "y1": hit["entity"]["bbox_y1"],
                "x2": hit["entity"]["bbox_x2"], "y2": hit["entity"]["bbox_y2"],
            }
        }
        for hit in results[0]
    ]

def get_keyframe_indices(total_frames: int, fps: float, strategy: str = "every_n_seconds", interval: float = 1.0):
    """
    Returns frame indices to sample based on the chosen strategy.

    Strategies:
    - "every_n_seconds": sample one frame every N seconds
    - "every_n_frames": sample every Nth frame
    - "uniform": uniformly sample a fixed number of frames
    """
    if strategy == "every_n_seconds":
        step = max(1, int(fps * interval))
        return list(range(0, total_frames, step))
    elif strategy == "every_n_frames":
        step = max(1, int(interval))
        return list(range(0, total_frames, step))
    elif strategy == "uniform":
        n_samples = int(interval)
        if n_samples >= total_frames:
            return list(range(total_frames))
        step = total_frames / n_samples
        return [int(i * step) for i in range(n_samples)]
    else:
        raise ValueError(f"Unknown strategy: {strategy}")

from pymilvus import MilvusClient, DataType

COLLECTION_NAME = "video_frames"
FRAME_EMBEDDING_DIM = 512

client = MilvusClient("./video_search.db")

schema = client.create_schema(auto_id=True, enable_dynamic_field=True)
schema.add_field("id", DataType.INT64, is_primary=True)
schema.add_field("embedding", DataType.FLOAT_VECTOR, dim=FRAME_EMBEDDING_DIM)
schema.add_field("video_id", DataType.VARCHAR, max_length=64)
schema.add_field("video_path", DataType.VARCHAR, max_length=1024)
schema.add_field("video_title", DataType.VARCHAR, max_length=256)
schema.add_field("channel", DataType.VARCHAR, max_length=128)
schema.add_field("frame_index", DataType.INT64)
schema.add_field("timestamp_ms", DataType.INT64)
schema.add_field("fps", DataType.FLOAT)
schema.add_field("scene_tag", DataType.VARCHAR, max_length=64)
schema.add_field("has_faces", DataType.BOOL)
schema.add_field("has_text", DataType.BOOL)

index_params = client.prepare_index_params()
index_params.add_index(
    field_name="embedding",
    index_type="HNSW",
    metric_type="COSINE",
    params={"M": 16, "efConstruction": 200},
)
index_params.add_index(field_name="video_id", index_type="Trie")
index_params.add_index(field_name="channel", index_type="Trie")
index_params.add_index(field_name="timestamp_ms", index_type="STL_SORT")

client.create_collection(
    collection_name=COLLECTION_NAME,
    schema=schema,
    index_params=index_params,
)

import numpy as np
import time

def extract_frame(video_path: str, frame_index: int) -> np.ndarray:
    """
    Extract a single frame from a video.

    Example with OpenCV:
        import cv2
        cap = cv2.VideoCapture(video_path)
        cap.set(cv2.CAP_PROP_POS_FRAMES, frame_index)
        ret, frame = cap.read()
        cap.release()
        if ret:
            return cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
        return None
    """
    return np.random.randint(0, 255, (480, 640, 3), dtype=np.uint8)


def embed_frame(frame: np.ndarray) -> np.ndarray:
    vec = np.random.randn(FRAME_EMBEDDING_DIM).astype(np.float32)
    return vec / np.linalg.norm(vec)


def get_video_metadata(video_path: str) -> dict:
    """
    Example with OpenCV:
        import cv2
        cap = cv2.VideoCapture(video_path)
        total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
        fps = cap.get(cv2.CAP_PROP_FPS)
        cap.release()
        return {"total_frames": total_frames, "fps": fps}
    """
    return {"total_frames": 3000, "fps": 30.0}


def process_video(
    video_path: str,
    video_id: str,
    video_title: str = "",
    channel: str = "",
    sample_every_n_seconds: float = 1.0,
    batch_size: int = 256,
):
    meta = get_video_metadata(video_path)
    total_frames = meta["total_frames"]
    fps = meta["fps"]

    frame_indices = get_keyframe_indices(
        total_frames, fps, strategy="every_n_seconds", interval=sample_every_n_seconds
    )

    print(f"Processing {video_path} — sampling {len(frame_indices)} frames")

    inserted = 0
    data_buffer = []

    for frame_idx in frame_indices:
        frame = extract_frame(video_path, frame_idx)
        if frame is None:
            continue

        embedding = embed_frame(frame)
        timestamp_ms = int((frame_idx / fps) * 1000)

        data_buffer.append({
            "embedding": embedding.tolist(),
            "video_id": video_id,
            "video_path": video_path,
            "video_title": video_title,
            "channel": channel,
            "frame_index": frame_idx,
            "timestamp_ms": timestamp_ms,
            "fps": fps,
            "has_faces": False,
            "has_text": False,
            "scene_tag": "unknown",
        })

        if len(data_buffer) >= batch_size:
            result = client.insert(collection_name=COLLECTION_NAME, data=data_buffer)
            inserted += result["insert_count"]
            data_buffer = []
            print(f"  Inserted {inserted} frames so far...")

    if data_buffer:
        result = client.insert(collection_name=COLLECTION_NAME, data=data_buffer)
        inserted += result["insert_count"]

    print(f"Done: inserted {inserted} frames")
    return inserted


def find_similar_frames(
    query_image_path: str = None,
    query_video_path: str = None,
    query_timestamp_ms: int = None,
    top_k: int = 20,
    channel_filter: str = None,
    time_range_ms: tuple = None,
) -> list:
    if query_image_path:
        frame = np.random.randint(0, 255, (480, 640, 3), dtype=np.uint8)
        query_embedding = embed_frame(frame)
    elif query_video_path and query_timestamp_ms is not None:
        meta = get_video_metadata(query_video_path)
        frame_idx = int((query_timestamp_ms / 1000) * meta["fps"])
        frame = extract_frame(query_video_path, frame_idx)
        query_embedding = embed_frame(frame)
    else:
        raise ValueError("Must provide query_image_path or (query_video_path + query_timestamp_ms)")

    filters = []
    if channel_filter:
        filters.append(f"channel == '{channel_filter}'")
    if time_range_ms:
        start_ms, end_ms = time_range_ms
        filters.append(f"timestamp_ms >= {start_ms} AND timestamp_ms <= {end_ms}")

    results = client.search(
        collection_name=COLLECTION_NAME,
        data=[query_embedding.tolist()],
        limit=top_k,
        filter=" AND ".join(filters) if filters else None,
        search_params={"ef": 200},
        output_fields=["video_id", "video_title", "video_path", "frame_index", "timestamp_ms", "channel"],
    )

    hits = []
    for hit in results[0]:
        ts = hit["entity"]["timestamp_ms"]
        hours = ts // 3_600_000
        minutes = (ts % 3_600_000) // 60_000
        seconds = (ts % 60_000) / 1000

        hits.append({
            "video_id": hit["entity"]["video_id"],
            "video_title": hit["entity"]["video_title"],
            "frame_index": hit["entity"]["frame_index"],
            "timestamp_ms": ts,
            "timestamp_str": f"{hours:02d}:{minutes:02d}:{seconds:05.2f}",
            "similarity": hit["distance"],
            "channel": hit["entity"]["channel"],
        })

    return hits

# Create partitions (e.g., by year for a video archive)
client.create_partition(collection_name="video_frames", partition_name="2023")
client.create_partition(collection_name="video_frames", partition_name="2024")
client.create_partition(collection_name="video_frames", partition_name="2025")

# Insert into a specific partition
client.insert(
    collection_name="video_frames",
    data=my_data_2024,
    partition_name="2024",
)

# Search only in the "2024" partition
results = client.search(
    collection_name="video_frames",
    data=[query_vector],
    limit=10,
    partition_names=["2024"],
    search_params={"ef": 100},
)

# Search across multiple partitions
results = client.search(
    collection_name="video_frames",
    data=[query_vector],
    limit=10,
    partition_names=["2024", "2025"],
)

# String operations
filter="label in ['dog', 'cat', 'bird']"
filter="image_path like '/dataset/train/%'"
filter="NOT (label in ['background', 'unknown'])"

# Numeric comparisons
filter="confidence > 0.85 AND detection_score < 0.99"
filter="width >= 1920 AND height >= 1080"

# JSON field access
filter="metadata['camera_id'] == 'cam_01'"

# Combining conditions
filter="(label == 'dog' OR label == 'cat') AND confidence > 0.9 AND dataset_split == 'train'"

# Array containment
filter="ARRAY_CONTAINS(tags, 'outdoor')"

from pymilvus import MilvusClient, AnnSearchRequest, RRFRanker, WeightedRanker

dense_request = AnnSearchRequest(
    data=[dense_query_vector],
    anns_field="dense_embedding",
    param={"metric_type": "COSINE", "params": {"ef": 100}},
    limit=20,
)

sparse_request = AnnSearchRequest(
    data=[sparse_query_vector],
    anns_field="sparse_embedding",
    param={"metric_type": "IP", "params": {}},
    limit=20,
)

results = client.hybrid_search(
    collection_name="multimodal_index",
    reqs=[dense_request, sparse_request],
    ranker=RRFRanker(k=60),
    limit=10,
    output_fields=["image_path", "caption"],
)

# ef must be >= limit (top_k)
search_params = {"ef": 50}    # Fast, lower recall
search_params = {"ef": 100}   # Balanced (recommended starting point)
search_params = {"ef": 500}   # High recall
search_params = {"ef": 2000}  # Maximum recall (approaching FLAT accuracy)

# Bad: insert one at a time
for record in all_records:
    client.insert(collection_name="...", data=[record])  # Very slow!

# Good: insert in batches
for batch in chunked(all_records, batch_size=2000):
    client.insert(collection_name="...", data=batch)

# Even better: use multiple workers
from concurrent.futures import ThreadPoolExecutor

def embed_and_insert(batch):
    embeddings = embed_batch([r["path"] for r in batch])
    data = [{"embedding": emb, **meta} for emb, meta in zip(embeddings, batch)]
    return client.insert(collection_name="...", data=data)

with ThreadPoolExecutor(max_workers=4) as executor:
    futures = [executor.submit(embed_and_insert, batch) for batch in batches]

# Load collection into memory before querying
client.load_collection("image_embeddings")

# Release collection from memory when not actively querying
client.release_collection("image_embeddings")

# Load only specific partitions into memory
client.load_partitions("image_embeddings", partition_names=["2024"])

import hashlib
import json

_search_cache = {}

def cached_search(collection_name, vector, limit, filter=None, ttl_seconds=300):
    vec_bytes = json.dumps([round(v, 6) for v in vector]).encode()
    cache_key = f"{collection_name}:{hashlib.sha256(vec_bytes).hexdigest()}:{limit}:{filter}"

    if cache_key in _search_cache:
        cached_result, cached_at = _search_cache[cache_key]
        if time.time() - cached_at < ttl_seconds:
            return cached_result

    result = client.search(
        collection_name=collection_name,
        data=[vector],
        limit=limit,
        filter=filter,
    )
    _search_cache[cache_key] = (result, time.time())
    return result

import time

def timed_search(client, collection_name, query_vector, limit=10, **kwargs):
    start = time.perf_counter()
    results = client.search(
        collection_name=collection_name,
        data=[query_vector],
        limit=limit,
        **kwargs,
    )
    latency_ms = (time.perf_counter() - start) * 1000
    print(f"Search latency: {latency_ms:.2f}ms | Results: {len(results[0])}")
    return results, latency_ms

standalone:
  environment:
    COMMON_SECURITY_AUTHORIZATIONENABLED: "true"

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

# Create a new user
client.create_user(user_name="cv_app_user", password="StrongP@ssword123")

# Grant a role
client.grant_role(user_name="cv_app_user", role_name="db_ro")

# List users
client.list_users()

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

# Grant specific privileges
client.grant_privilege(
    role_name="cv_readonly",
    object_type="Collection",
    object_name="image_embeddings",
    privilege="Query",
)
client.grant_privilege(
    role_name="cv_readonly",
    object_type="Collection",
    object_name="image_embeddings",
    privilege="Search",
)

# Assign role to user
client.grant_role(user_name="cv_app_user", role_name="cv_readonly")

client = MilvusClient(
    uri="https://milvus.example.com:19530",
    token="username:password",
    server_pem_path="/path/to/server.pem",
)

# Import the official Milvus dashboard (ID: 19120 on grafana.com)
wget https://raw.githubusercontent.com/milvus-io/milvus/master/deployments/monitoring/grafana/milvus-dashboard.json

import time
from collections import defaultdict
from statistics import mean, quantiles

class MilvusMonitor:
    def __init__(self):
        self.latencies = defaultdict(list)
        self.error_counts = defaultdict(int)

    def record_search(self, collection: str, latency_ms: float, success: bool):
        if success:
            self.latencies[collection].append(latency_ms)
        else:
            self.error_counts[collection] += 1

    def report(self):
        for collection, lats in self.latencies.items():
            if not lats:
                continue
            p50 = quantiles(lats, n=100)[49]
            p95 = quantiles(lats, n=100)[94]
            p99 = quantiles(lats, n=100)[98]
            print(f"Collection: {collection}")
            print(f"  Searches: {len(lats)}, Errors: {self.error_counts[collection]}")
            print(f"  Latency — mean: {mean(lats):.1f}ms, p50: {p50:.1f}ms, "
                  f"p95: {p95:.1f}ms, p99: {p99:.1f}ms")

monitor = MilvusMonitor()

def monitored_search(collection_name, query_vector, limit=10, **kwargs):
    start = time.perf_counter()
    success = True
    try:
        return client.search(collection_name=collection_name, data=[query_vector], limit=limit, **kwargs)
    except Exception:
        success = False
        raise
    finally:
        monitor.record_search(collection_name, (time.perf_counter() - start) * 1000, success)

def safe_insert(client, collection_name, data, expected_dim):
    for entity in data:
        vec = entity.get("embedding", [])
        assert len(vec) == expected_dim, f"Expected dim {expected_dim}, got {len(vec)}"
    return client.insert(collection_name=collection_name, data=data)

from pymilvus import Collection
col = Collection("image_embeddings")
col.load()

import numpy as np
vec = np.array(raw_embedding)
vec = vec / np.linalg.norm(vec)

client.flush("image_embeddings")
# Then wait for index building to complete before running benchmarks

index_params.add_index(field_name="label", index_type="Trie")
index_params.add_index(field_name="score", index_type="STL_SORT")
index_params.add_index(field_name="flags", index_type="BITMAP")