Agent Skills 开发实战：轻松搞定 10 个 Milvus 集群的运维管理

def safe_operation(self, operation, *args, **kwargs):
    """安全执行操作，支持自动重连"""
    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

with MilvusConnection("http://localhost:19530") as conn:
    # 连接自动管理，不用担心资源泄漏
    collections = conn.safe_operation(client.list_collections)

def get_collection_info(self, collection_name):
    """获取集合详细信息（聚合多个 API）"""
    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),
    }

def check_collection_health(self, collection_name):
    """检查集合健康状态"""
    info = self.get_collection_info(collection_name)
    issues = []

    if info.get("row_count", 0) == 0:
        issues.append("集合为空")

    if info.get("load_state") != "Loaded":
        issues.append(f"集合未加载: {info.get('load_state')}")

    indexes = self.client.list_indexes(collection_name)
    for index_name in indexes:
        index_info = self.client.describe_index(collection_name, index_name)
        if index_info.get("state") != 3:
            issues.append(f"索引 {index_name} 状态异常")

    return {
        "collection": collection_name,
        "status": "healthy"ifnot issues else"warning",
        "issues": issues,
        "info": info,
    }

def check_index_status(self, index_info):
    """检查索引状态（三态判断）"""
    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"

def monitor_indexes(self, collection_name):
    """监控索引状态（含覆盖率计算）"""
    indexes = self.list_indexes(collection_name)
    report = []

    for index_name in indexes:
        index_info = self.get_index_info(collection_name, index_name)
        coverage = 0
        total_rows = index_info.get("total_rows", 0)
        indexed_rows = index_info.get("indexed_rows", 0)

        if total_rows > 0:
            coverage = (indexed_rows / total_rows) * 100

        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

def check_all_collections(self):
    """检查所有集合健康（聚合所有集合的状态）"""
    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

def performance_test(self, collection_name, test_count=10):
    """性能测试（执行 N 次查询，统计延迟）"""
    try:
        results = self.client.query(
            collection_name=collection_name,
            filter="",
            limit=1,
            output_fields=["vector"],
        )

        ifnot results or"vector"notin results[0]:
            return {"status": "failed", "error": "无法获取测试数据"}

        test_vector = results[0]["vector"]
        query_times = []

        for i in range(test_count):
            start_time = time.time()
            self.client.search(
                collection_name=collection_name, data=[test_vector], limit=10
            )
            query_time = (time.time() - start_time) * 1000
            query_times.append(query_time)

        avg_time = sum(query_times) / len(query_times)
        min_time = min(query_times)
        max_time = max(query_times)

        return {
            "status": "success",
            "avg_query_time_ms": round(avg_time, 2),
            "min_query_time_ms": round(min_time, 2),
            "max_query_time_ms": round(max_time, 2),
            "test_count": test_count,
        }
    except Exception as e:
        return {"status": "failed", "error": str(e)}

with MilvusConnection("http://localhost:19530") as conn:
    manager = CollectionManager(conn)
    manager.list_collections()
# 退出时自动断开连接，不用担心资源泄漏