| """Configuration recommendation module for MLPerf benchmarks.""" |
|
|
| import logging |
|
|
| import pandas as pd |
| from utils import get_feature_type |
|
|
| logger = logging.getLogger(__name__) |
|
|
|
|
| class ConfigurationFinder: |
| """Finds optimal hardware configurations based on user requirements.""" |
|
|
| def __init__(self, dataset: pd.DataFrame): |
| """Initialize with benchmark dataset.""" |
| self.df = dataset |
| self.perf_metric = "metrics.result_per_accelerator" |
| self.cost_metric = "cost_per_million_tokens" |
| self.total_perf_metric = "metrics.result" |
|
|
| def is_within_tolerance( |
| self, value1: float, value2: float, tolerance: float = 0.1 |
| ) -> bool: |
| """Check if two values are within a specified percentage tolerance.""" |
| if value1 is None or value2 is None: |
| return False |
|
|
| try: |
| if value1 == 0 or value2 == 0: |
| return value1 == value2 |
| percentage_diff = abs(value1 - value2) / max(abs(value1), abs(value2)) |
| return percentage_diff <= tolerance |
| except: |
| return False |
|
|
| def find_configurations( |
| self, constraints: dict, tolerance: float = 0.1 |
| ) -> pd.DataFrame: |
| """Find configurations matching the given constraints.""" |
| if self.df.empty: |
| return pd.DataFrame() |
|
|
| filtered_df = self.df.copy() |
|
|
| for feature, value in constraints.items(): |
| if feature not in filtered_df.columns or value is None or value == "Any": |
| continue |
|
|
| if get_feature_type(feature) == "continuous": |
| try: |
| target_value = float(value) |
| lower_bound = target_value * (1 - tolerance) |
| upper_bound = target_value * (1 + tolerance) |
| filtered_df = filtered_df[ |
| (filtered_df[feature] >= lower_bound) |
| & (filtered_df[feature] <= upper_bound) |
| ] |
| except: |
| filtered_df = filtered_df[filtered_df[feature] == value] |
| else: |
| filtered_df = filtered_df[filtered_df[feature] == value] |
|
|
| if "min_accelerators" in constraints and constraints["min_accelerators"]: |
| min_acc = constraints["min_accelerators"] |
| filtered_df = filtered_df[ |
| filtered_df["system.accelerator.total_count"] >= min_acc |
| ] |
|
|
| if "max_accelerators" in constraints and constraints["max_accelerators"]: |
| max_acc = constraints["max_accelerators"] |
| filtered_df = filtered_df[ |
| filtered_df["system.accelerator.total_count"] <= max_acc |
| ] |
|
|
| return filtered_df |
|
|
| def rank_configurations( |
| self, |
| df: pd.DataFrame, |
| metric: str = "metrics.result_per_accelerator", |
| ascending: bool = False, |
| ) -> pd.DataFrame: |
| """Rank configurations by the specified metric.""" |
| if df.empty or metric not in df.columns: |
| return df |
| return df.sort_values(by=metric, ascending=ascending) |
|
|
| def recommend(self, constraints: dict, top_n: int = 10) -> pd.DataFrame: |
| """Find and rank configurations based on constraints.""" |
| filtered_configs = self.find_configurations(constraints) |
| ranked_configs = self.rank_configurations( |
| filtered_configs, metric=self.perf_metric, ascending=False |
| ) |
|
|
| if len(ranked_configs) > top_n: |
| return ranked_configs.head(top_n) |
| return ranked_configs |
|
|
