The Complete Production SOP: How to Deploy and Run Milvus Distributed Vector Indexing & Search in Autonomous AI Agents

---
name: "milvus-distributed-vector-indexing"
description: "Scale billion-scale similarity search with Milvus distributed vector database clusters."
metadata:
  author: "Milvus"
  version: "1.0.0"
  runtime_compatibility:
    - "codex"
    - "claude-code"
    - "cursor"
    - "windsurf"
triggers:
  - "create milvus collection"
  - "build milvus hnsw index"
  - "query milvus vector similarity"
permissions:
  fs_read: true
  fs_write: "workspace"
  shell_exec: false
  net_access: false
---

# Milvus Distributed Vector Indexing & Search

## Execution SOP & Rules
1. Connect using pymilvus: connections.connect('default', host=os.getenv('MILVUS_HOST', 'localhost'), port=os.getenv('MILVUS_PORT', '19530')).
2. Create CollectionSchema with primary key, auto_id=True, and DataType.FLOAT_VECTOR field.
3. Create index: collection.create_index(field_name='vector', index_params={'metric_type': 'COSINE', 'index_type': 'HNSW', 'params': {'M': 16, 'efConstruction': 200}}).
4. Call collection.load() into memory before executing any collection.search() queries.
5. Match the metric_type in search_params identically with the index metric_type.

import os
from pymilvus import connections, utility

def init_milvus_connection():
    host = os.getenv("MILVUS_HOST", "localhost")
    port = os.getenv("MILVUS_PORT", "19530")
    token = os.getenv("MILVUS_TOKEN", None)
    
    connect_kwargs = {"alias": "default", "host": host, "port": port}
    if token:
        connect_kwargs["token"] = token
        
    connections.connect(**connect_kwargs)
    return utility.get_server_version()

from pymilvus import Collection, CollectionSchema, FieldSchema, DataType

def build_knowledge_collection(collection_name: str, dim: int = 1536) -> Collection:
    fields = [
        FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
        FieldSchema(name="doc_id", dtype=DataType.VARCHAR, max_length=64),
        FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=dim)
    ]
    schema = CollectionSchema(fields=fields, description="Knowledge base embeddings", enable_dynamic_field=True)
    collection = Collection(name=collection_name, schema=schema)
    
    index_params = {
        "metric_type": "COSINE",
        "index_type": "HNSW",
        "params": {"M": 16, "efConstruction": 200}
    }
    collection.create_index(field_name="vector", index_params=index_params)
    collection.load()
    return collection

import pytest
from pymilvus import utility, Collection
from vector_store import build_knowledge_collection
from milvus_client import init_milvus_connection

def test_milvus_collection_and_index():
    init_milvus_connection()
    test_coll_name = "test_verification_coll"
    if utility.has_collection(test_coll_name):
        utility.drop_collection(test_coll_name)
        
    collection = build_knowledge_collection(test_coll_name, dim=128)
    assert utility.has_collection(test_coll_name) is True
    assert len(collection.indexes) > 0
    assert collection.indexes[0].params["index_type"] == "HNSW"
    assert utility.load_state(test_coll_name).name == "Loaded"
    
    utility.drop_collection(test_coll_name)