How to Easily Manage Operations of 10 Milvus Clusters with an Agent Skill

Why Build This Skill?

One night the author received an alert that query latency on a production Milvus cluster spiked above 5 seconds because an index was still building while new data kept arriving. Managing over ten clusters, each with dozens of collections, made manual checks impossible, prompting the need for an automated health‑monitoring solution.

What Is Milvus?

Milvus is a high‑performance vector database used for similarity search on large‑scale embeddings. Typical use cases include searching for similar recipes, visually similar products, or recommending items based on user‑history vectors. Unlike traditional databases, Milvus has complex concepts such as collection loading, index building (HNSW, IVF, etc.), and query performance monitoring.

Operational Pain Points

Connection Management : Network glitches cause pymilvus connections to drop, and the library does not auto‑reconnect, forcing repetitive manual checks.

Scattered Information : Determining a collection’s status requires calling describe_collection, get_collection_stats, get_load_state, and list_partitions separately.

Index State Ambiguity : The state field alone ("Building", "Finished") does not reflect whether all data is indexed; total_rows and indexed_rows must be considered.

Lack of Unified Monitoring : No simple way exists to batch‑check health across all collections, leading to ad‑hoc scripts.

Design Goals

Simple Usage : One command should return health for all collections.

Robust Fault Tolerance : Automatic reconnection on network issues.

Information Aggregation : Merge data from multiple APIs into a single view.

Extensibility : Modular design for future feature additions.

Technical Choices

Language: Python 3.14 (pymilvus provides the best official support).

Execution: uv run -p 3.14 --no-project --with pymilvus scripts/xxx.py for quick script runs without a full project.

Core Library: pymilvus.

Core Modules

Connection Management : MilvusConnection handles connection lifecycle and auto‑reconnect.

Collection Management : CollectionManager wraps creation, deletion, loading, and releasing of collections.

Index Management : IndexManager creates, deletes, and monitors indexes.

Monitoring & Alerting : Monitor performs unified health checks and aggregates results.

Key Implementation Details

Connection Management – safe_operation

The MilvusConnection class defines a safe_operation method that ensures a live connection before invoking any API call. If a Connection error is caught, it disconnects, re‑establishes the link, and retries the operation.

def safe_operation(self, operation, *args, **kwargs):
    """Safely execute an operation with automatic reconnection."""
    try:
        self._ensure_connected()
        return operation(*args, **kwargs)
    except Exception as e:
        if "Connection" in str(e):
            self.disconnect()
            self._ensure_connected()
            return operation(*args, **kwargs)
        raise

It also implements __enter__ and __exit__ so the connection can be used with a with statement, guaranteeing proper cleanup.

with MilvusConnection("http://localhost:19530") as conn:
    collections = conn.safe_operation(client.list_collections)

Collection Management – Information Aggregation

CollectionManager.get_collection_info

calls four Milvus APIs and returns a consolidated dictionary containing name, description, fields, row count, load state, and partition count.

def get_collection_info(self, collection_name):
    """Gather detailed collection info by aggregating multiple APIs."""
    info = self.client.describe_collection(collection_name)
    stats = self.client.get_collection_stats(collection_name)
    load_state = self.client.get_load_state(collection_name)
    partitions = self.client.list_partitions(collection_name)
    return {
        "name": info.get("name"),
        "description": info.get("description", ""),
        "fields": info.get("fields", []),
        "row_count": stats.get("row_count", 0),
        "load_state": load_state.get("state", "Unknown"),
        "partition_count": len(partitions),
    }

The helper check_collection_health evaluates the aggregated info and returns a health status (healthy, warning, error) together with any detected issues.

Index Management – Three‑State Judgment

Milvus returns a state field that alone is insufficient. The check_index_status method uses three dimensions— state, total_rows, and indexed_rows —to classify an index as building, waiting, or completed.

def check_index_status(self, index_info):
    """Determine index status using a three‑state logic."""
    state = index_info.get("state")
    total_rows = index_info.get("total_rows", 0)
    indexed_rows = index_info.get("indexed_rows", 0)
    if state != 3:
        return "building"
    if total_rows == 0:
        return "waiting"
    coverage = indexed_rows / total_rows
    if coverage < 1.0:
        return "building"
    return "completed"

The monitor_indexes method iterates over all indexes of a collection, applies the three‑state check, and records coverage percentages.

def monitor_indexes(self, collection_name):
    """Monitor index status and compute coverage for a collection."""
    indexes = self.list_indexes(collection_name)
    report = []
    for index_name in indexes:
        index_info = self.get_index_info(collection_name, index_name)
        total_rows = index_info.get("total_rows", 0)
        indexed_rows = index_info.get("indexed_rows", 0)
        coverage = (indexed_rows / total_rows) * 100 if total_rows > 0 else 0
        report.append({
            "index_name": index_name,
            "index_type": index_info.get("index_type"),
            "metric_type": index_info.get("metric_type"),
            "state": index_info.get("state"),
            "status": self.check_index_status(index_info),
            "total_rows": total_rows,
            "indexed_rows": indexed_rows,
            "coverage": round(coverage, 2),
        })
    return report

Unified Monitoring – check_all_collections

The Monitor.check_all_collections method loops through every collection, invokes CollectionManager.check_collection_health, and aggregates counts of healthy, warning, and error collections along with a timestamp.

def check_all_collections(self):
    """Check health of all collections and aggregate results."""
    collections = self.collection_manager.list_collections()
    report = {
        "timestamp": datetime.now().isoformat(),
        "total_collections": len(collections),
        "healthy": 0,
        "warning": 0,
        "error": 0,
        "details": [],
    }
    for collection in collections:
        health = self.collection_manager.check_collection_health(collection)
        report["details"].append(health)
        if health.get("status") == "healthy":
            report["healthy"] += 1
        elif health.get("status") == "warning":
            report["warning"] += 1
        else:
            report["error"] += 1
    return report

Performance Testing

The performance_test method runs a configurable number of queries against a collection, measures each latency, and returns average, min, max, and the test count.

def performance_test(self, collection_name, test_count=10):
    """Run N queries and report latency statistics."""
    try:
        results = self.client.query(collection_name=collection_name, filter="", limit=1, output_fields=["vector"])
        if not results or "vector" not in results[0]:
            return {"status": "failed", "error": "Unable to fetch test data"}
        test_vector = results[0]["vector"]
        query_times = []
        for _ in range(test_count):
            start = time.time()
            self.client.search(collection_name=collection_name, data=[test_vector], limit=10)
            query_times.append((time.time() - start) * 1000)
        return {
            "status": "success",
            "avg_query_time_ms": round(sum(query_times) / len(query_times), 2),
            "min_query_time_ms": round(min(query_times), 2),
            "max_query_time_ms": round(max(query_times), 2),
            "test_count": test_count,
        }
    except Exception as e:
        return {"status": "failed", "error": str(e)}

Capacity Analysis

The analyze_capacity function (not shown in code) estimates storage size for vectors, scalar fields, and total data, helping with capacity planning.

Design Highlights & Pitfalls

Highlight 1 – Context Manager

Implementing __enter__ and __exit__ in MilvusConnection lets users write concise with blocks that automatically open and close connections, eliminating manual resource handling.

Highlight 2 – Fault Tolerance

All core methods wrap operations in safe_operation, catching exceptions and attempting reconnection, so a single failure does not abort the whole batch process.

Pitfall 1 – Misleading Index State

Relying solely on the state field can hide partially built indexes; the three‑state logic solves this by also checking indexed_rows vs total_rows.

Pitfall 2 – Silent Connection Drops

pymilvus only raises an exception when a dropped connection is used, making failures hard to detect. The explicit _ensure_connected check before each operation mitigates this hidden issue.

Pitfall 3 – Performance Test Cache Effects

Running repeated queries on the same vector can hit cache and produce unrealistically low latency. Using different vectors or setting a random seed avoids this bias.

Takeaways & Methodology

Address Pain Points First : Focus on the most critical issues (connection handling, data aggregation) before adding extra features.

Aggregate Over Scatter : Consolidating multiple API responses into a single view provides far more value than thin wrappers.

Prioritize Fault Tolerance : Production environments encounter network glitches and node failures; robust error handling prevents script crashes.

Tests Serve as Documentation : Writing test cases uncovers design flaws and produces clear usage examples.

Avoid Over‑Engineering : Simple, maintainable designs often outperform complex patterns that increase maintenance cost.

The Skill remains open‑source and extensible; developers can adapt it to their own Milvus operational needs.