import copy import inspect import json import logging import pathlib import pickle import shutil import tempfile import traceback from abc import abstractmethod from typing import Any, Callable, NamedTuple, Optional import numpy as np import pandas as pd import mlflow from mlflow import MlflowClient, MlflowException from mlflow.data.evaluation_dataset import EvaluationDataset from mlflow.entities.metric import Metric from mlflow.metrics.base import MetricValue from mlflow.models.evaluation.artifacts import ( CsvEvaluationArtifact, ImageEvaluationArtifact, JsonEvaluationArtifact, NumpyEvaluationArtifact, _infer_artifact_type_and_ext, ) from mlflow.models.evaluation.base import EvaluationMetric, EvaluationResult, ModelEvaluator from mlflow.models.evaluation.utils.metric import MetricDefinition from mlflow.protos.databricks_pb2 import INVALID_PARAMETER_VALUE from mlflow.pyfunc import _ServedPyFuncModel from mlflow.utils.file_utils import TempDir from mlflow.utils.proto_json_utils import NumpyEncoder from mlflow.utils.time import get_current_time_millis _logger = logging.getLogger(__name__) _EVAL_TABLE_FILE_NAME = "eval_results_table.json" _TOKEN_COUNT_METRIC_NAME = "token_count" _LATENCY_METRIC_NAME = "latency" def _extract_raw_model(model): if not getattr(model, "metadata", None): return None, None model_loader_module = model.metadata.flavors["python_function"]["loader_module"] # If we load a model with mlflow.pyfunc.load_model, the model will be wrapped # with a pyfunc wrapper. We need to extract the raw model so that shap # explainer uses the raw model instead of the wrapper and skips data schema validation. if model_loader_module in [ "mlflow.catboost", "mlflow.sklearn", "mlflow.xgboost", ] and not isinstance(model, _ServedPyFuncModel): if hasattr(model._model_impl, "get_raw_model"): return model_loader_module, model._model_impl.get_raw_model() return model_loader_module, model._model_impl else: return model_loader_module, None def _extract_output_and_other_columns( model_predictions: list[Any] | dict[str, Any] | pd.DataFrame | pd.Series, output_column_name: str | None, ) -> tuple[pd.Series, pd.DataFrame | None, str]: y_pred = None other_output_columns = None ERROR_MISSING_OUTPUT_COLUMN_NAME = ( "Output column name is not specified for the multi-output model. " "Please set the correct output column name using the `predictions` parameter." ) if isinstance(model_predictions, list) and all(isinstance(p, dict) for p in model_predictions): # Extract 'y_pred' and 'other_output_columns' from list of dictionaries if output_column_name in model_predictions[0]: y_pred = pd.Series( [p.get(output_column_name) for p in model_predictions], name=output_column_name ) other_output_columns = pd.DataFrame([ {k: v for k, v in p.items() if k != output_column_name} for p in model_predictions ]) elif len(model_predictions[0]) == 1: # Set the only key as self.predictions and its value as self.y_pred key, value = list(model_predictions[0].items())[0] y_pred = pd.Series(value, name=key) output_column_name = key elif output_column_name is None: raise MlflowException( ERROR_MISSING_OUTPUT_COLUMN_NAME, error_code=INVALID_PARAMETER_VALUE, ) else: raise MlflowException( f"Output column name '{output_column_name}' is not found in the model " f"predictions list: {model_predictions}. Please set the correct output column " "name using the `predictions` parameter.", error_code=INVALID_PARAMETER_VALUE, ) elif isinstance(model_predictions, pd.DataFrame): if output_column_name in model_predictions.columns: y_pred = model_predictions[output_column_name] other_output_columns = model_predictions.drop(columns=output_column_name) elif len(model_predictions.columns) == 1: output_column_name = model_predictions.columns[0] y_pred = model_predictions[output_column_name] elif output_column_name is None: raise MlflowException( ERROR_MISSING_OUTPUT_COLUMN_NAME, error_code=INVALID_PARAMETER_VALUE, ) else: raise MlflowException( f"Output column name '{output_column_name}' is not found in the model " f"predictions dataframe {model_predictions.columns}. Please set the correct " "output column name using the `predictions` parameter.", error_code=INVALID_PARAMETER_VALUE, ) elif isinstance(model_predictions, dict): if output_column_name in model_predictions: y_pred = pd.Series(model_predictions[output_column_name], name=output_column_name) other_output_columns = pd.DataFrame({ k: v for k, v in model_predictions.items() if k != output_column_name }) elif len(model_predictions) == 1: key, value = list(model_predictions.items())[0] y_pred = pd.Series(value, name=key) output_column_name = key elif output_column_name is None: raise MlflowException( ERROR_MISSING_OUTPUT_COLUMN_NAME, error_code=INVALID_PARAMETER_VALUE, ) else: raise MlflowException( f"Output column name '{output_column_name}' is not found in the " f"model predictions dict {model_predictions}. Please set the correct " "output column name using the `predictions` parameter.", error_code=INVALID_PARAMETER_VALUE, ) return ( y_pred if y_pred is not None else model_predictions, other_output_columns, output_column_name, ) def _extract_predict_fn(model: Any) -> Callable[..., Any] | None: """ Extracts the predict function from the given model or raw_model. Precedence order: 1. If raw_model is specified, its predict function is used. 2. If model is specified, its predict function is used. 3. If none of the above, predict function is None. Args: model: A model object that has a predict method. raw_model: A raw model object that has a predict method. Returns: The predict function. """ _, raw_model = _extract_raw_model(model) predict_fn = None if raw_model is not None: predict_fn = raw_model.predict try: from mlflow.xgboost import _wrapped_xgboost_model_predict_fn # Because shap evaluation will pass evaluation data in ndarray format # (without feature names), if set validate_features=True it will raise error. predict_fn = _wrapped_xgboost_model_predict_fn(raw_model, validate_features=False) except ImportError: pass elif model is not None: predict_fn = model.predict return predict_fn def _get_dataframe_with_renamed_columns(x, new_column_names): """ Downstream inference functions may expect a pd.DataFrame to be created from x. However, if x is already a pd.DataFrame, and new_column_names != x.columns, we cannot simply call pd.DataFrame(x, columns=new_column_names) because the resulting pd.DataFrame will contain NaNs for every column in new_column_names that does not exist in x.columns. This function instead creates a new pd.DataFrame object from x, and then explicitly renames the columns to avoid NaNs. Args: x: A data object, such as a Pandas DataFrame, numPy array, or list new_column_names: Column names for the output Pandas DataFrame Returns: A pd.DataFrame with x as data, with columns new_column_names """ df = pd.DataFrame(x) return df.rename(columns=dict(zip(df.columns, new_column_names))) def _get_aggregate_metrics_values(metrics): return {name: MetricValue(aggregate_results={name: value}) for name, value in metrics.items()} _matplotlib_config = { "figure.dpi": 175, "figure.figsize": [6.0, 4.0], "figure.autolayout": True, "font.size": 8, } class _CustomArtifact(NamedTuple): """ A namedtuple representing a custom artifact function and its properties. function : the custom artifact function name : the name of the custom artifact function index : the index of the function in the ``custom_artifacts`` argument of mlflow.evaluate artifacts_dir : the path to a temporary directory to store produced artifacts of the function """ function: Callable[..., Any] name: str index: int artifacts_dir: str def _is_valid_artifacts(artifacts): return isinstance(artifacts, dict) and all(isinstance(k, str) for k in artifacts.keys()) def _evaluate_custom_artifacts(custom_artifact_tuple, eval_df, builtin_metrics): """ This function calls the `custom_artifact` function and performs validations on the returned result to ensure that they are in the expected format. It will raise a MlflowException if the result is not in the expected format. Args: custom_artifact_tuple: Containing a user provided function and its index in the ``custom_artifacts`` parameter of ``mlflow.evaluate`` eval_df: A Pandas dataframe object containing a prediction and a target column. builtin_metrics: A dictionary of metrics produced by the default evaluator. Returns: A dictionary of artifacts. """ exception_header = ( f"Custom artifact function '{custom_artifact_tuple.name}' " " at index {custom_artifact_tuple.index}" " in the `custom_artifacts` parameter" ) artifacts = custom_artifact_tuple.function( eval_df, builtin_metrics, custom_artifact_tuple.artifacts_dir ) if artifacts is None: _logger.warning(f"{exception_header} returned None.") return if not _is_valid_artifacts(artifacts): _logger.warning( f"{exception_header} did not return artifacts as a dictionary of string artifact " "names with their corresponding objects." ) return return artifacts # TODO: Move this to the /evaluators directory class BuiltInEvaluator(ModelEvaluator): """ The base class for all evaluators that are built-in to MLflow. Each evaluator is responsible for implementing the `_evaluate()` method, which is called by the `evaluate()` method of this base class. This class contains many helper methods used across built-in evaluators, such as logging metrics, artifacts, and ordering metrics. """ def __init__(self): self.client = MlflowClient() @abstractmethod def _evaluate( self, model: Optional["mlflow.pyfunc.PyFuncModel"], extra_metrics: list[EvaluationMetric], custom_artifacts=None, **kwargs, ) -> EvaluationResult | None: """Implement the evaluation logic for each evaluator.""" def log_metrics(self): """ Helper method to log metrics into specified run. """ self._add_prefix_to_metrics() timestamp = get_current_time_millis() self.client.log_batch( self.run_id, metrics=[ Metric( key=key, value=value, timestamp=timestamp, step=0, model_id=self.model_id, dataset_name=self.dataset.name, dataset_digest=self.dataset.digest, run_id=self.run_id, ) for key, value in self.aggregate_metrics.items() ], ) def _log_image_artifact( self, do_plot, artifact_name, ): from matplotlib import pyplot prefix = self.evaluator_config.get("metric_prefix", "") artifact_file_name = f"{prefix}{artifact_name}.png" artifact_file_local_path = self.temp_dir.path(artifact_file_name) try: pyplot.clf() do_plot() pyplot.savefig(artifact_file_local_path, bbox_inches="tight") except Exception as e: _logger.warning(f"Failed to log image artifact {artifact_name!r}: {e!r}") else: mlflow.log_artifact(artifact_file_local_path) artifact = ImageEvaluationArtifact(uri=mlflow.get_artifact_uri(artifact_file_name)) artifact._load(artifact_file_local_path) self.artifacts[artifact_name] = artifact finally: pyplot.close(pyplot.gcf()) def _evaluate_sklearn_model_score_if_scorable(self, model, y_true, sample_weights): model_loader_module, raw_model = _extract_raw_model(model) if model_loader_module == "mlflow.sklearn" and raw_model is not None: try: score = raw_model.score( self.X.copy_to_avoid_mutation(), y_true, sample_weight=sample_weights ) self.metrics_values.update(_get_aggregate_metrics_values({"score": score})) except Exception as e: _logger.warning( f"Computing sklearn model score failed: {e!r}. Set logging level to " "DEBUG to see the full traceback." ) _logger.debug("", exc_info=True) def _log_custom_metric_artifact(self, artifact_name, raw_artifact, custom_metric_tuple): """ This function logs and returns a custom metric artifact. Two cases: - The provided artifact is a path to a file, the function will make a copy of it with a formatted name in a temporary directory and call mlflow.log_artifact. - Otherwise: will attempt to save the artifact to an temporary path with an inferred type. Then call mlflow.log_artifact. Args: artifact_name: the name of the artifact raw_artifact: the object representing the artifact custom_metric_tuple: an instance of the _CustomMetric namedtuple Returns: EvaluationArtifact """ exception_and_warning_header = ( f"Custom artifact function '{custom_metric_tuple.name}' at index " f"{custom_metric_tuple.index} in the `custom_artifacts` parameter" ) inferred_from_path, inferred_type, inferred_ext = _infer_artifact_type_and_ext( artifact_name, raw_artifact, custom_metric_tuple ) artifact_file_local_path = self.temp_dir.path(artifact_name + inferred_ext) if pathlib.Path(artifact_file_local_path).exists(): raise MlflowException( f"{exception_and_warning_header} produced an artifact '{artifact_name}' that " "cannot be logged because there already exists an artifact with the same name." ) # ParquetEvaluationArtifact isn't explicitly stated here because such artifacts can only # be supplied through file. Which is handled by the first if clause. This is because # DataFrame objects default to be stored as CsvEvaluationArtifact. if inferred_from_path: shutil.copy2(raw_artifact, artifact_file_local_path) elif inferred_type is JsonEvaluationArtifact: with open(artifact_file_local_path, "w") as f: if isinstance(raw_artifact, str): f.write(raw_artifact) else: json.dump(raw_artifact, f, cls=NumpyEncoder) elif inferred_type is CsvEvaluationArtifact: raw_artifact.to_csv(artifact_file_local_path, index=False) elif inferred_type is NumpyEvaluationArtifact: np.save(artifact_file_local_path, raw_artifact, allow_pickle=False) elif inferred_type is ImageEvaluationArtifact: raw_artifact.savefig(artifact_file_local_path) else: # storing as pickle try: with open(artifact_file_local_path, "wb") as f: pickle.dump(raw_artifact, f) _logger.warning( f"{exception_and_warning_header} produced an artifact '{artifact_name}'" f" with type '{type(raw_artifact)}' that is logged as a pickle artifact." ) except pickle.PickleError: raise MlflowException( f"{exception_and_warning_header} produced an unsupported artifact " f"'{artifact_name}' with type '{type(raw_artifact)}' that cannot be pickled. " "Supported object types for artifacts are:\n" "- A string uri representing the file path to the artifact. MLflow" " will infer the type of the artifact based on the file extension.\n" "- A string representation of a JSON object. This will be saved as a " ".json artifact.\n" "- Pandas DataFrame. This will be saved as a .csv artifact." "- Numpy array. This will be saved as a .npy artifact." "- Matplotlib Figure. This will be saved as an .png image artifact." "- Other objects will be attempted to be pickled with default protocol." ) mlflow.log_artifact(artifact_file_local_path) artifact = inferred_type(uri=mlflow.get_artifact_uri(artifact_name + inferred_ext)) artifact._load(artifact_file_local_path) return artifact def _get_column_in_metrics_values(self, column): for metric_name, metric_value in self.metrics_values.items(): if metric_name.split("/")[0] == column: return metric_value def _get_args_for_metrics( self, metric: MetricDefinition, eval_df: pd.DataFrame, input_df: pd.DataFrame, other_output_df: pd.DataFrame | None, ) -> tuple[bool, list[str | pd.DataFrame]]: """ Given a metric_tuple, read the signature of the metric function and get the appropriate arguments from the input/output columns, other calculated metrics, and evaluator_config. Args: metric: The metric definition containing a user provided function and its index in the ``extra_metrics`` parameter of ``mlflow.evaluate``. eval_df: The evaluation dataframe containing the prediction and target columns. input_df: The input dataframe containing the features used to make predictions. other_output_df: A dataframe containing all model output columns but the predictions. Returns: tuple: A tuple of (bool, list) where the bool indicates if the given metric can be calculated with the given eval_df, metrics, and input_df. - If the user is missing "targets" or "predictions" parameters when needed, or we cannot find a column or metric for a parameter to the metric, return (False, list of missing parameters) - If all arguments to the metric function were found, return (True, list of arguments). """ # deepcopying eval_df and builtin_metrics for each custom metric function call, # in case the user modifies them inside their function(s). eval_df_copy = eval_df.copy() parameters = inspect.signature(metric.function).parameters eval_fn_args = [] params_not_found = [] if len(parameters) == 2: param_0_name, param_1_name = parameters.keys() # eval_fn has parameters (eval_df, builtin_metrics) for backwards compatibility if len(parameters) == 2 and param_0_name != "predictions" and param_1_name != "targets": eval_fn_args.append(eval_df_copy) self._update_aggregate_metrics() eval_fn_args.append(copy.deepcopy(self.aggregate_metrics)) # eval_fn can have parameters like (predictions, targets, metrics, random_col) else: for param_name, param in parameters.items(): column = self.col_mapping.get(param_name, param_name) if column in ("predictions", self.predictions, self.dataset.predictions_name): eval_fn_args.append(eval_df_copy["prediction"]) elif column in ("targets", self.dataset.targets_name): if "target" in eval_df_copy: eval_fn_args.append(eval_df_copy["target"]) else: if param.default == inspect.Parameter.empty: params_not_found.append(param_name) else: eval_fn_args.append(param.default) elif column == "metrics": eval_fn_args.append(copy.deepcopy(self.metrics_values)) else: # case when column passed in col_mapping contains the entire column if not isinstance(column, str): eval_fn_args.append(column) # case column in col_mapping is string and the column value # is part of the input_df elif column in input_df.columns: eval_fn_args.append(input_df[column]) # case column in col_mapping is string and the column value # is part of the output_df(other than predictions) elif other_output_df is not None and column in other_output_df.columns: self.other_output_columns_for_eval.add(column) eval_fn_args.append(other_output_df[column]) # case where the param is defined as part of the evaluator_config elif column in self.evaluator_config: eval_fn_args.append(self.evaluator_config.get(column)) # case where this is the name of another metric elif metric_value := self._get_column_in_metrics_values(column): eval_fn_args.append(metric_value) # in the case that: # the metric has not been calculated yet, but is scheduled to be calculated # "before" this metric in self.ordered_metrics, we append None to indicate # that there is not an error in the dependencies elif column in [metric_tuple.name for metric_tuple in self.ordered_metrics]: eval_fn_args.append(None) elif param.default == inspect.Parameter.empty: params_not_found.append(param_name) else: eval_fn_args.append(param.default) if len(params_not_found) > 0: return False, params_not_found return True, eval_fn_args def evaluate_and_log_custom_artifacts( self, custom_artifacts: list[_CustomArtifact], prediction: pd.Series, target: np.ndarray | None = None, ): """Evaluate custom artifacts provided by users.""" if not custom_artifacts: return eval_df = self._get_eval_df(prediction, target) for index, custom_artifact in enumerate(custom_artifacts): with tempfile.TemporaryDirectory() as artifacts_dir: # deepcopying eval_df and builtin_metrics for each custom artifact function call, # in case the user modifies them inside their function(s). custom_artifact_tuple = _CustomArtifact( function=custom_artifact, index=index, name=getattr(custom_artifact, "__name__", repr(custom_artifact)), artifacts_dir=artifacts_dir, ) artifact_results = _evaluate_custom_artifacts( custom_artifact_tuple, eval_df.copy(), copy.deepcopy(self.metrics_values), ) if artifact_results: for artifact_name, raw_artifact in artifact_results.items(): self.artifacts[artifact_name] = self._log_custom_metric_artifact( artifact_name, raw_artifact, custom_artifact_tuple, ) def _get_error_message_missing_columns(self, metric_name, param_names): error_message_parts = [f"Metric '{metric_name}' requires the following:"] special_params = ["targets", "predictions"] error_message_parts.extend( f" - the '{param}' parameter needs to be specified" for param in special_params if param in param_names ) if remaining_params := [param for param in param_names if param not in special_params]: error_message_parts.append( f" - missing columns {remaining_params} need to be defined or mapped" ) return "\n".join(error_message_parts) def _construct_error_message_for_malformed_metrics( self, malformed_results, input_columns, output_columns ): error_messages = [ self._get_error_message_missing_columns(metric_name, param_names) for metric_name, param_names in malformed_results ] joined_error_message = "\n".join(error_messages) full_message = f"""Error: Metric calculation failed for the following metrics: {joined_error_message} Below are the existing column names for the input/output data: Input Columns: {input_columns} Output Columns: {output_columns} To resolve this issue, you may need to: - specify any required parameters - if you are missing columns, check that there are no circular dependencies among your metrics, and you may want to map them to an existing column using the following configuration: evaluator_config={{'col_mapping': {{: }}}}""" return "\n".join(l.lstrip() for l in full_message.splitlines()) def _raise_exception_for_malformed_metrics(self, malformed_results, eval_df, other_output_df): output_columns = [] if other_output_df is None else list(other_output_df.columns) if self.predictions: output_columns.append(self.predictions) elif self.dataset.predictions_name: output_columns.append(self.dataset.predictions_name) else: output_columns.append("predictions") input_columns = list(self.X.copy_to_avoid_mutation().columns) if "target" in eval_df: if self.dataset.targets_name: input_columns.append(self.dataset.targets_name) else: input_columns.append("targets") error_message = self._construct_error_message_for_malformed_metrics( malformed_results, input_columns, output_columns ) raise MlflowException(error_message, error_code=INVALID_PARAMETER_VALUE) def _get_eval_df(self, prediction: pd.Series, target: np.ndarray | None = None): """ Create a DataFrame with "prediction" and "target" columns. This is a standard format that can be passed to the metric functions. """ eval_df = pd.DataFrame({"prediction": copy.deepcopy(prediction)}) if target is not None: eval_df["target"] = target return eval_df def _order_metrics( self, metrics: list[EvaluationMetric], eval_df: pd.DataFrame, other_output_df: pd.DataFrame | None, ): """ Order the list metrics so they can be computed in sequence. Some metrics might use the results of other metrics to compute their own results. This function iteratively resolve this dependency, by checking if each metric can be computed with the current available columns and metrics values. """ remaining_metrics = metrics input_df = self.X.copy_to_avoid_mutation() while len(remaining_metrics) > 0: pending_metrics = [] failed_results = [] did_append_metric = False for metric_tuple in remaining_metrics: can_calculate, eval_fn_args = self._get_args_for_metrics( metric_tuple, eval_df, input_df, other_output_df ) if can_calculate: self.ordered_metrics.append(metric_tuple) did_append_metric = True else: # cannot calculate the metric yet pending_metrics.append(metric_tuple) failed_results.append((metric_tuple.name, eval_fn_args)) # cant calculate any more metrics if not did_append_metric: self._raise_exception_for_malformed_metrics( failed_results, eval_df, other_output_df ) remaining_metrics = pending_metrics return self.ordered_metrics def _test_first_row( self, metrics: list[MetricDefinition], eval_df: pd.DataFrame, other_output_df: pd.DataFrame | None, ): # test calculations on first row of eval_df _logger.info("Testing metrics on first row...") exceptions = [] first_row_df = eval_df.iloc[[0]] first_row_input_df = self.X.copy_to_avoid_mutation().iloc[[0]] for metric in metrics: try: _, eval_fn_args = self._get_args_for_metrics( metric, first_row_df, first_row_input_df, other_output_df ) if metric_value := metric.evaluate(eval_fn_args): name = f"{metric.name}/{metric.version}" if metric.version else metric.name self.metrics_values.update({name: metric_value}) except Exception as e: stacktrace_str = traceback.format_exc() if isinstance(e, MlflowException): exceptions.append( f"Metric '{metric.name}': Error:\n{e.message}\n{stacktrace_str}" ) else: exceptions.append(f"Metric '{metric.name}': Error:\n{e!r}\n{stacktrace_str}") if len(exceptions) > 0: raise MlflowException("\n".join(exceptions)) def evaluate_metrics( self, metrics: list[EvaluationMetric], prediction: pd.Series, target: np.ndarray | None = None, other_output_df: pd.DataFrame | None = None, ): """ Evaluate the metrics on the given prediction and target data. Args: metrics: A list of metrics to evaluate. prediction: A Pandas Series containing the predictions. target: A numpy array containing the target values. other_output_df: A Pandas DataFrame containing other output columns from the model. Returns: None, the metrics values are recorded in the self.metrics_values dictionary. """ eval_df = self._get_eval_df(prediction, target) metrics = [ MetricDefinition.from_index_and_metric(i, metric) for i, metric in enumerate(metrics) ] metrics = self._order_metrics(metrics, eval_df, other_output_df) self._test_first_row(metrics, eval_df, other_output_df) # calculate metrics for the full eval_df input_df = self.X.copy_to_avoid_mutation() for metric in metrics: _, eval_fn_args = self._get_args_for_metrics(metric, eval_df, input_df, other_output_df) if metric_value := metric.evaluate(eval_fn_args): name = f"{metric.name}/{metric.version}" if metric.version else metric.name self.metrics_values.update({name: metric_value}) def log_eval_table(self, y_pred, other_output_columns=None): # only log eval table if there are per row metrics recorded if not any( metric_value.scores is not None or metric_value.justifications is not None for _, metric_value in self.metrics_values.items() ): return metric_prefix = self.evaluator_config.get("metric_prefix", "") if not isinstance(metric_prefix, str): metric_prefix = "" if isinstance(self.dataset.features_data, pd.DataFrame): # Handle DataFrame case if self.dataset.has_targets: data = self.dataset.features_data.assign(**{ self.dataset.targets_name or "target": self.dataset.labels_data, self.dataset.predictions_name or self.predictions or "outputs": y_pred, }) else: data = self.dataset.features_data.assign(outputs=y_pred) else: # Handle NumPy array case, converting it to a DataFrame data = pd.DataFrame(self.dataset.features_data, columns=self.dataset.feature_names) if self.dataset.has_targets: data = data.assign(**{ self.dataset.targets_name or "target": self.dataset.labels_data, self.dataset.predictions_name or self.predictions or "outputs": y_pred, }) else: data = data.assign(outputs=y_pred) # Include other_output_columns used in evaluation to the eval table if other_output_columns is not None and len(self.other_output_columns_for_eval) > 0: for column in self.other_output_columns_for_eval: data[column] = other_output_columns[column] columns = {} for metric_name, metric_value in self.metrics_values.items(): scores = metric_value.scores justifications = metric_value.justifications if scores: if metric_name.startswith(metric_prefix) and metric_name[len(metric_prefix) :] in [ _TOKEN_COUNT_METRIC_NAME, _LATENCY_METRIC_NAME, ]: columns[metric_name] = scores else: columns[f"{metric_name}/score"] = scores if justifications: columns[f"{metric_name}/justification"] = justifications data = data.assign(**columns) artifact_file_name = f"{metric_prefix}{_EVAL_TABLE_FILE_NAME}" mlflow.log_table(data, artifact_file=artifact_file_name) if self.eval_results_path: eval_table_spark = self.spark_session.createDataFrame(data) try: eval_table_spark.write.mode(self.eval_results_mode).option( "mergeSchema", "true" ).format("delta").saveAsTable(self.eval_results_path) except Exception as e: _logger.info(f"Saving eval table to delta table failed. Reason: {e}") name = _EVAL_TABLE_FILE_NAME.split(".", 1)[0] self.artifacts[name] = JsonEvaluationArtifact( uri=mlflow.get_artifact_uri(artifact_file_name) ) def _update_aggregate_metrics(self): self.aggregate_metrics = {} for metric_name, metric_value in self.metrics_values.items(): if metric_value.aggregate_results: for agg_name, agg_value in metric_value.aggregate_results.items(): if agg_value is not None: if agg_name == metric_name.split("/")[0]: self.aggregate_metrics[metric_name] = agg_value else: self.aggregate_metrics[f"{metric_name}/{agg_name}"] = agg_value def _add_prefix_to_metrics(self): def _prefix_value(value): aggregate = ( {f"{prefix}{k}": v for k, v in value.aggregate_results.items()} if value.aggregate_results else None ) return MetricValue(value.scores, value.justifications, aggregate) if prefix := self.evaluator_config.get("metric_prefix"): self.metrics_values = { f"{prefix}{k}": _prefix_value(v) for k, v in self.metrics_values.items() } self._update_aggregate_metrics() def evaluate( self, *, model_type, dataset, run_id, evaluator_config, model: "mlflow.pyfunc.PyFuncModel" = None, extra_metrics=None, custom_artifacts=None, predictions=None, model_id=None, **kwargs, ) -> EvaluationResult: if model is None and predictions is None and dataset.predictions_data is None: raise MlflowException( message=( "Either a model or set of predictions must be specified in order to use the" " default evaluator. Either specify the `model` parameter, the `predictions`" " parameter, an MLflow dataset containing the `predictions` column name" " (via the `data` parameter), or a different evaluator (via the `evaluators`" " parameter)." ), error_code=INVALID_PARAMETER_VALUE, ) self.artifacts = {} self.aggregate_metrics = {} self.metrics_values = {} self.ordered_metrics = [] self.other_output_columns_for_eval = set() self.dataset: EvaluationDataset = dataset self.run_id = run_id self.model_type = model_type self.model_id = model_id self.evaluator_config = evaluator_config self.predictions = predictions self.col_mapping = self.evaluator_config.get("col_mapping", {}) self.eval_results_path = self.evaluator_config.get("eval_results_path") self.eval_results_mode = self.evaluator_config.get("eval_results_mode", "overwrite") if self.eval_results_path: from mlflow.utils._spark_utils import _get_active_spark_session self.spark_session = _get_active_spark_session() if not self.spark_session: raise MlflowException( message="eval_results_path is only supported in Spark environment. ", error_code=INVALID_PARAMETER_VALUE, ) if self.eval_results_mode not in ["overwrite", "append"]: raise MlflowException( message="eval_results_mode can only be 'overwrite' or 'append'. ", error_code=INVALID_PARAMETER_VALUE, ) if extra_metrics is None: extra_metrics = [] bad_metrics = [ metric for metric in extra_metrics if not isinstance(metric, EvaluationMetric) ] if len(bad_metrics) > 0: message = "\n".join([ f"- Metric '{m}' has type '{type(m).__name__}'" for m in bad_metrics ]) raise MlflowException( f"In the 'extra_metrics' parameter, the following metrics have the wrong type:\n" f"{message}\n" f"Please ensure that all extra metrics are instances of " f"mlflow.metrics.EvaluationMetric." ) import matplotlib with TempDir() as temp_dir, matplotlib.rc_context(_matplotlib_config): self.temp_dir = temp_dir return self._evaluate(model, extra_metrics, custom_artifacts) @property def X(self) -> pd.DataFrame: """ The features (`X`) portion of the dataset, guarded against accidental mutations. """ return BuiltInEvaluator._MutationGuardedData( _get_dataframe_with_renamed_columns( self.dataset.features_data, self.dataset.feature_names ) ) class _MutationGuardedData: """ Wrapper around a data object that requires explicit API calls to obtain either a copy of the data object, or, in cases where the caller can guaranteed that the object will not be mutated, the original data object. """ def __init__(self, data): """ Args: data: A data object, such as a Pandas DataFrame, numPy array, or list. """ self._data = data def copy_to_avoid_mutation(self): """ Obtain a copy of the data. This method should be called every time the data needs to be used in a context where it may be subsequently mutated, guarding against accidental reuse after mutation. Returns: A copy of the data object. """ if isinstance(self._data, pd.DataFrame): return self._data.copy(deep=True) else: return copy.deepcopy(self._data) def get_original(self): """ Obtain the original data object. This method should only be called if the caller can guarantee that it will not mutate the data during subsequent operations. Returns: The original data object. """ return self._data