Model Evaluation & Quality Gates

In Continuous Delivery for Machine Learning (CD4ML), we treat code, data, and models as first-class citizens. Just as code is guarded by unit tests, models must be guarded by statistical Quality Gates to prevent regressions before deployment.

1. Deep Dive on Evaluation Metrics

We evaluate our model using three complementary regression metrics:

Mean Absolute Error (MAE)

MAE measures the average magnitude of prediction errors, without considering their direction:

MAE=1n∑i=1n∣yi−y^i∣\text{MAE} = \frac{1}{n} \sum_{i=1}^n |y_i - \hat{y}_i|

  • Interpretation: Expressed in the target's unit (number of taxi pickups). An MAE of 11.5 means that, on average, our forecasts miss the actual hourly pickup count by approximately 11.5 rides.
  • Robustness: Highly robust to statistical outliers. It is useful for representing typical error behaviors.

Root Mean Squared Error (RMSE)

RMSE is the square root of the average of squared prediction errors:

RMSE=1n∑i=1n(yi−y^i)2\text{RMSE} = \sqrt{\frac{1}{n} \sum_{i=1}^n (y_i - \hat{y}_i)^2}

  • Interpretation: By squaring the errors before averaging, RMSE penalizes larger errors more heavily.
  • Business Significance: Highly critical for fleet logistics. A huge forecasting miss (e.g. predicting 10 pickups when 200 occur) is operationally disastrous because it leaves hundreds of passengers stranded. Dispatchers prefer a model with a slightly higher MAE if it maintains a significantly lower RMSE (fewer catastrophic misses).

Coefficient of Determination (R2R^2)

R2R^2 represents the proportion of variance in the target variable that is predictable from the independent features:

$$R^2 = 1 - \frac{\sum_{i=1}^n (y_i - \hat{y}i)^2}{\sum{i=1}^n (y_i - \bar{y})^2}$$

  • Interpretation: Scale-free metric ranging from −∞-\infty to 11. An R2R^2 of 0.650.65 indicates that our model explains 65% of the demand variance compared to a naive historical baseline model (yˉ\bar{y}).

2. Automated Quality Gates

To automate deployment safely, our pipeline implements a gating mechanism:

metrics.require_within(config.evaluation)
  1. Retrieve Candidate Metrics: The pipeline pulls the candidate model version from MLflow and runs predictions on the test set.
  2. Evaluate Thresholds: The calculated metrics are compared against maximum error thresholds defined in configs/project.yaml:
    • max_mae: Max MAE permitted.
    • max_rmse: Max RMSE permitted.
  3. Deployment Block: If the candidate model's MAE or RMSE exceeds the thresholds, the evaluation script raises a validation exception and aborts. This stops downstream deployment stages, keeping the previous production model active.

3. Pipeline Implementation

Below is the code executing the model evaluation and gating:

Sourcesrc/nyc_taxi_demand_forecasting/pipelines/evaluate.py
import logging
from importlib import import_module
from pathlib import Path
from typing import Any

import mlflow
import numpy as np
from mlflow.client import MlflowClient
from mlops_shared.evaluation import (
    RegressionMetricCalculator,
    RegressionMetrics,
)

from nyc_taxi_demand_forecasting.configuration import ProjectConfigLoader
from nyc_taxi_demand_forecasting.features.hourly_demand import (
    FEATURE_COLUMNS,
    FEATURE_REFS,
    load_entity_df,
)
from nyc_taxi_demand_forecasting.models.registry import latest_model_version
from nyc_taxi_demand_forecasting.models.training import DemandDatasetSplitter


def _tag_evaluation_outcome(
    client: MlflowClient,
    model_name: str,
    model_version: str,
    outcome: str,
    metrics: RegressionMetrics,
) -> None:
    for key, value in (
        ("evaluated", outcome),
        ("mae", str(round(metrics.mae, 4))),
        ("rmse", str(round(metrics.rmse, 4))),
    ):
        client.set_model_version_tag(
            name=model_name,
            version=model_version,
            key=key,
            value=value,
        )


def run(config_path: Path) -> None:
    config = ProjectConfigLoader().load(config_path)

    # 1. Load entities dataframe and fetch test historical features from Feast
    entity_df = load_entity_df(config.features.training_dataset_path)

    feature_store_type = import_module("feast").FeatureStore
    store = feature_store_type(repo_path=str(config.feast.repo_path))
    training_data = store.get_historical_features(
        entity_df=entity_df,
        features=list(FEATURE_REFS),
    ).to_df()

    # Split dataset (only evaluate on test split)
    _, test_frame = DemandDatasetSplitter().split(
        training_data, config.training.test_size
    )

    # 2. Connect to MLflow and get the latest registered model version
    mlflow.set_tracking_uri(config.mlflow.tracking_uri)
    client = MlflowClient()

    model_name = config.mlflow.registered_model_name
    latest_version = latest_model_version(client, model_name)
    model_uri = f"models:/{model_name}/{latest_version}"

    # 3. Load the model and make predictions
    model = mlflow.pyfunc.load_model(model_uri)
    # pyfunc emits an ndarray, so normalize the output shape instead of
    # asserting a Series — same contract the api/monitor sites carry.
    raw_predictions: Any = model.predict(  # pyright: ignore[reportUnknownMemberType]
        test_frame.loc[:, list(FEATURE_COLUMNS)]
    )
    predictions = np.asarray(raw_predictions).ravel()

    # 4. Compute metrics
    metrics = RegressionMetricCalculator().calculate(
        test_frame[config.training.target_column], predictions
    )

    logging.getLogger(__name__).info(
        "model_evaluation_completed",
        extra={
            "mae": metrics.mae,
            "rmse": metrics.rmse,
            "r2": metrics.r2,
            "version": latest_version,
        },
    )

    # 5. Gate before tagging — only a pass marks the version deployable via
    # the candidate alias; a failure records its truthful outcome so a
    # gate-failed version is distinguishable from a never-evaluated one.
    try:
        metrics.require_within(config.evaluation)
    except ValueError:
        _tag_evaluation_outcome(
            client, model_name, latest_version, "failed", metrics
        )
        raise
    _tag_evaluation_outcome(
        client, model_name, latest_version, "true", metrics
    )
    client.set_registered_model_alias(
        name=model_name,
        alias="candidate",
        version=latest_version,
    )