Module neuralops.evaluation

Evaluation module for model performance, fairness, robustness, label quality, and A/B testing.

Sub-modules

neuralops.evaluation.ab_testing

neuralops.evaluation.eval_all_models_datasets

neuralops.evaluation.eval_perplexity

Evaluate Module Perplexity (WikiText-2) on GPT-2 …

neuralops.evaluation.fairness

Comprehensive fairness evaluation – inspired by AIF360 (IBM) and Fairlearn (Microsoft). Supports 15+ group‑fairness metrics and 6 bias mitigation …

neuralops.evaluation.full_mnli_validation

full_mnli_validation.py

Re-evaluates the top PASS modules from the local sweep against the full 9815-sample MNLI validation_matched set …

neuralops.evaluation.label_quality

Label quality detection – inspired by Cleanlab, the industry standard for data‑centric AI. Finds label errors, outliers, and near‑duplicates …

neuralops.evaluation.metrics

evaluation/metrics.py Forwarding shim — canonical: framework.trainer.metrics

neuralops.evaluation.retraining

Automated retraining pipeline – triggers retraining based on drift, performance degradation, time schedule, or data volume thresholds …

neuralops.evaluation.run_baseline_comparison

Baseline Comparison Runner: Benchmarks key literature baselines against Proposed method …

neuralops.evaluation.run_lm_eval_harness

LM Evaluation Harness Runner for Injected Attention and Compression Modules …

neuralops.evaluation.run_targeted_b2_recovery

Targeted B2 Recovery Evaluation for high zero-shot logit difference modules. Evaluates recovery trajectories on GPT-2 with standard recovery …

Functions

def get_task_metrics(task_name: str) ‑> Any

Expand source code

def get_task_metrics(task_name: str) -> Any:
    """Return a compute_metrics function for a GLUE task."""
    task_name = task_name.lower()
    if task_name == "stsb":

        def fn(p) -> Any:
            """Fn."""
            preds = np.squeeze(p.predictions)
            labels = p.label_ids
            return compute_pearson_spearman(preds, labels)

    elif task_name == "cola":

        def fn(p) -> Any:
            """Fn."""
            preds = np.argmax(p.predictions, axis=1)
            labels = p.label_ids
            return {
                "matthews_correlation": matthews_corrcoef(labels, preds),
                "accuracy": accuracy_score(labels, preds),
            }

    elif task_name in ("mrpc", "qqp"):

        def fn(p) -> Any:
            """Fn."""
            preds = np.argmax(p.predictions, axis=1)
            labels = p.label_ids
            acc = accuracy_score(labels, preds)
            f1 = f1_score(labels, preds)
            return {"accuracy": acc, "f1": f1, "accuracy_f1": (acc + f1) / 2}

    else:

        def fn(p) -> Any:
            """Fn."""
            preds = np.argmax(p.predictions, axis=1)
            labels = p.label_ids
            return {
                "accuracy": accuracy_score(labels, preds),
                "f1_macro": f1_score(labels, preds, average="macro", zero_division=0),
                "f1_weighted": f1_score(labels, preds, average="weighted", zero_division=0),
            }

    return fn

Return a compute_metrics function for a GLUE task.

Classes

class ABTestResult (metric: str,
mean_a: float,
mean_b: float,
difference: float,
p_value: float,
significant: bool,
better_model: str,
confidence_interval: tuple[float, float])

Expand source code

@dataclass
class ABTestResult:
    metric: str
    mean_a: float
    mean_b: float
    difference: float
    p_value: float
    significant: bool
    better_model: str
    confidence_interval: tuple[float, float]

ABTestResult(metric: str, mean_a: float, mean_b: float, difference: float, p_value: float, significant: bool, better_model: str, confidence_interval: tuple[float, float])

Instance variables

var better_model : str

The type of the None singleton.

var confidence_interval : tuple[float, float]

The type of the None singleton.

var difference : float

The type of the None singleton.

var mean_a : float

The type of the None singleton.

var mean_b : float

The type of the None singleton.

var metric : str

The type of the None singleton.

var p_value : float

The type of the None singleton.

var significant : bool

The type of the None singleton.

class ABTester (alpha: float = 0.05)

Expand source code

class ABTester:
    def __init__(self, alpha: float = 0.05) -> None:
        self.alpha = alpha

    def compare(
        self, results_a: dict[str, list[float]], results_b: dict[str, list[float]]
    ) -> list[ABTestResult]:
        """Compare."""
        outcomes = []
        for metric in results_a:
            if metric not in results_b:
                continue
            a = np.array(results_a[metric])
            b = np.array(results_b[metric])
            t_stat, p_val = stats.ttest_ind(a, b, equal_var=False)
            mean_a, mean_b = np.mean(a), np.mean(b)
            diff = mean_b - mean_a
            se = np.sqrt(np.var(a) / len(a) + np.var(b) / len(b))
            ci = (diff - 1.96 * se, diff + 1.96 * se)
            outcomes.append(
                ABTestResult(
                    metric=metric,
                    mean_a=mean_a,
                    mean_b=mean_b,
                    difference=diff,
                    p_value=p_val,
                    significant=p_val < self.alpha,
                    better_model="B" if diff > 0 else "A",
                    confidence_interval=ci,
                )
            )
        return outcomes

Methods

def compare(self, results_a: dict[str, list[float]], results_b: dict[str, list[float]]) ‑> list[ABTestResult]

Expand source code

def compare(
    self, results_a: dict[str, list[float]], results_b: dict[str, list[float]]
) -> list[ABTestResult]:
    """Compare."""
    outcomes = []
    for metric in results_a:
        if metric not in results_b:
            continue
        a = np.array(results_a[metric])
        b = np.array(results_b[metric])
        t_stat, p_val = stats.ttest_ind(a, b, equal_var=False)
        mean_a, mean_b = np.mean(a), np.mean(b)
        diff = mean_b - mean_a
        se = np.sqrt(np.var(a) / len(a) + np.var(b) / len(b))
        ci = (diff - 1.96 * se, diff + 1.96 * se)
        outcomes.append(
            ABTestResult(
                metric=metric,
                mean_a=mean_a,
                mean_b=mean_b,
                difference=diff,
                p_value=p_val,
                significant=p_val < self.alpha,
                better_model="B" if diff > 0 else "A",
                confidence_interval=ci,
            )
        )
    return outcomes

Compare.

class FairnessEvaluator (threshold: float = 0.05)

Expand source code

class FairnessEvaluator:
    """
    Evaluate model fairness across protected groups.

    Usage:
        evaluator = FairnessEvaluator(threshold=0.05)
        report = evaluator.evaluate(predictions, labels, groups={"gender": gender_array})
    """

    def __init__(self, threshold: float = 0.05) -> None:
        self.threshold = threshold

    # ── Main API ────────────────────────────────────────────────────
    def evaluate(
        self,
        predictions: np.ndarray,
        labels: np.ndarray,
        groups: dict[str, np.ndarray],
        metrics: list[str] | None = None,
        favorable_label: int = 1,
    ) -> FairnessReport:
        """
        Args:
            predictions: predicted labels (or probabilities for the positive class)
            labels: ground truth
            groups: dict of attribute_name → array of group values per sample
            metrics: list of metric names (default: all)
            favorable_label: which label is considered "positive" (default 1)
        """
        if metrics is None:
            metrics = [
                "demographic_parity",
                "equalized_odds",
                "equal_opportunity",
                "predictive_parity",
                "treatment_equality",
                "overall_accuracy_equality",
                "conditional_demographic_parity",
            ]

        all_metric_results: dict[str, dict[str, float]] = {}
        all_group_sizes: dict[str, int] = {}
        disparities: dict[str, float] = {}

        # Determine which groups exist (union across all attributes)
        for attr_name, attr_values in groups.items():
            unique_vals = np.unique(attr_values)
            for val in unique_vals:
                key = f"{attr_name}={val}"
                all_group_sizes[key] = int(np.sum(attr_values == val))

        # Compute each metric
        for metric in metrics:
            result = self._compute_metric(predictions, labels, groups, metric, favorable_label)
            all_metric_results[metric] = result
            if result:
                values = list(result.values())
                disparities[metric] = max(values) - min(values)

        # Overall fairness score
        all_disps = list(disparities.values())
        overall = 1.0 - np.mean(all_disps) if all_disps else 1.0
        is_fair = max(all_disps) <= self.threshold if all_disps else True

        # Best / worst
        avg_by_group = self._average_metric_by_group(all_metric_results)
        worst = min(avg_by_group, key=avg_by_group.get) if avg_by_group else "N/A"
        best = max(avg_by_group, key=avg_by_group.get) if avg_by_group else "N/A"

        # Suggestions
        suggestions = []
        if not is_fair:
            suggestions = self._generate_suggestions(all_metric_results, self.threshold)

        return FairnessReport(
            metrics=all_metric_results,
            group_sizes=all_group_sizes,
            disparities=disparities,
            overall_score=round(overall, 4),
            worst_off_group=worst,
            best_off_group=best,
            is_fair=is_fair,
            threshold=self.threshold,
            mitigation_suggestions=suggestions,
        )

    # ── Individual metric computers ─────────────────────────────────
    def _compute_metric(self, preds, labels, groups, metric, favorable) -> dict[str, float]:
        """Route to the correct metric function."""
        methods = {
            "demographic_parity": self._demographic_parity,
            "equalized_odds": self._equalized_odds,
            "equal_opportunity": self._equal_opportunity,
            "predictive_parity": self._predictive_parity,
            "treatment_equality": self._treatment_equality,
            "overall_accuracy_equality": self._overall_accuracy_equality,
            "conditional_demographic_parity": self._conditional_demographic_parity,
            "disparate_impact": self._disparate_impact,
            "statistical_parity_difference": self._statistical_parity_diff,
        }
        fn = methods.get(metric)
        if fn is None:
            return {}
        return fn(preds, labels, groups, favorable)

    def _demographic_parity(self, preds, labels, groups, favorable) -> Any:
        """Selection rate parity: P(Ŷ=1|A=a) should be equal for all groups."""
        result = {}
        for attr, vals in groups.items():
            for val in np.unique(vals):
                mask = vals == val
                result[f"{attr}={val}"] = np.mean(preds[mask] == favorable)
        return result

    def _equalized_odds(self, preds, labels, groups, favorable) -> Any:
        """TPR and FPR should be equal across groups. We report TPR."""
        result = {}
        for attr, vals in groups.items():
            for val in np.unique(vals):
                mask = vals == val
                pos_mask = labels[mask] == favorable
                if pos_mask.sum() > 0:
                    result[f"{attr}={val}"] = np.mean(preds[mask][pos_mask] == favorable)
                else:
                    result[f"{attr}={val}"] = 0.0
        return result

    def _equal_opportunity(self, preds, labels, groups, favorable) -> Any:
        """TPR equality (a stricter form of equalized odds)."""
        return self._equalized_odds(preds, labels, groups, favorable)

    def _predictive_parity(self, preds, labels, groups, favorable) -> Any:
        """PPV equality: P(Y=1|Ŷ=1,A=a) equal across groups."""
        result = {}
        for attr, vals in groups.items():
            for val in np.unique(vals):
                mask = vals == val
                pred_pos = preds[mask] == favorable
                if pred_pos.sum() > 0:
                    result[f"{attr}={val}"] = np.mean(labels[mask][pred_pos] == favorable)
                else:
                    result[f"{attr}={val}"] = 0.0
        return result

    def _treatment_equality(self, preds, labels, groups, favorable) -> Any:
        """Ratio of FN to FP should be equal."""
        result = {}
        for attr, vals in groups.items():
            for val in np.unique(vals):
                mask = vals == val
                fn = np.sum((labels[mask] == favorable) & (preds[mask] != favorable))
                fp = np.sum((labels[mask] != favorable) & (preds[mask] == favorable))
                result[f"{attr}={val}"] = fn / max(fp, 1)
        return result

    def _overall_accuracy_equality(self, preds, labels, groups, favorable) -> Any:
        """Accuracy equality across groups."""
        result = {}
        for attr, vals in groups.items():
            for val in np.unique(vals):
                mask = vals == val
                result[f"{attr}={val}"] = np.mean(preds[mask] == labels[mask])
        return result

    def _conditional_demographic_parity(self, preds, labels, groups, favorable) -> Any:
        """Demographic parity conditioned on a legitimate feature (proxy)."""
        return self._demographic_parity(preds, labels, groups, favorable)

    def _disparate_impact(self, preds, labels, groups, favorable) -> Any:
        """Ratio of selection rates between unprivileged and privileged groups."""
        result = self._demographic_parity(preds, labels, groups, favorable)
        if len(result) >= 2:
            vals = list(result.values())
            ratio = min(vals) / max(vals) if max(vals) > 0 else 1.0
            return {"disparate_impact_ratio": ratio}
        return {"disparate_impact_ratio": 1.0}

    def _statistical_parity_diff(self, preds, labels, groups, favorable) -> Any:
        """Max difference in selection rates."""
        result = self._demographic_parity(preds, labels, groups, favorable)
        if result:
            vals = list(result.values())
            return {"statistical_parity_difference": max(vals) - min(vals)}
        return {"statistical_parity_difference": 0.0}

    # ── Helpers ─────────────────────────────────────────────────────
    def _average_metric_by_group(
        self, all_metrics: dict[str, dict[str, float]]
    ) -> dict[str, float]:
        """Compute the average of all metrics for each group."""
        group_scores: dict[str, list[float]] = {}
        for metric_results in all_metrics.values():
            for group, value in metric_results.items():
                group_scores.setdefault(group, []).append(value)
        return {g: np.mean(v) for g, v in group_scores.items()}

    def _generate_suggestions(
        self, metrics: dict[str, dict[str, float]], threshold: float
    ) -> list[str]:
        """generate suggestions."""
        suggestions = []
        # Check demographic parity
        dp = metrics.get("demographic_parity", {})
        if dp:
            vals = list(dp.values())
            if max(vals) - min(vals) > threshold:
                suggestions.append(
                    "Consider applying a post‑processing calibration method "
                    "(e.g. AIF360 RejectOptionClassification or Fairlearn ThresholdOptimizer)."
                )
        # Check equalized odds
        eo = metrics.get("equalized_odds", {})
        if eo and max(eo.values()) - min(eo.values()) > threshold:
            suggestions.append(
                "Try in‑processing fairness constraints during training "
                "(e.g. Fairlearn ExponentiatedGradient or GridSearch with fairness constraints)."
            )
        if len(suggestions) == 0:
            suggestions.append("All metrics within acceptable thresholds – no mitigation needed.")
        return suggestions

Evaluate model fairness across protected groups.

Usage -----= evaluator = FairnessEvaluator(threshold=0.05) report = evaluator.evaluate(predictions, labels, groups={"gender": gender_array})

Methods

def evaluate(self,
predictions: numpy.ndarray,
labels: numpy.ndarray,
groups: dict[str, numpy.ndarray],
metrics: list[str] | None = None,
favorable_label: int = 1) ‑> FairnessReport

Expand source code

def evaluate(
    self,
    predictions: np.ndarray,
    labels: np.ndarray,
    groups: dict[str, np.ndarray],
    metrics: list[str] | None = None,
    favorable_label: int = 1,
) -> FairnessReport:
    """
    Args:
        predictions: predicted labels (or probabilities for the positive class)
        labels: ground truth
        groups: dict of attribute_name → array of group values per sample
        metrics: list of metric names (default: all)
        favorable_label: which label is considered "positive" (default 1)
    """
    if metrics is None:
        metrics = [
            "demographic_parity",
            "equalized_odds",
            "equal_opportunity",
            "predictive_parity",
            "treatment_equality",
            "overall_accuracy_equality",
            "conditional_demographic_parity",
        ]

    all_metric_results: dict[str, dict[str, float]] = {}
    all_group_sizes: dict[str, int] = {}
    disparities: dict[str, float] = {}

    # Determine which groups exist (union across all attributes)
    for attr_name, attr_values in groups.items():
        unique_vals = np.unique(attr_values)
        for val in unique_vals:
            key = f"{attr_name}={val}"
            all_group_sizes[key] = int(np.sum(attr_values == val))

    # Compute each metric
    for metric in metrics:
        result = self._compute_metric(predictions, labels, groups, metric, favorable_label)
        all_metric_results[metric] = result
        if result:
            values = list(result.values())
            disparities[metric] = max(values) - min(values)

    # Overall fairness score
    all_disps = list(disparities.values())
    overall = 1.0 - np.mean(all_disps) if all_disps else 1.0
    is_fair = max(all_disps) <= self.threshold if all_disps else True

    # Best / worst
    avg_by_group = self._average_metric_by_group(all_metric_results)
    worst = min(avg_by_group, key=avg_by_group.get) if avg_by_group else "N/A"
    best = max(avg_by_group, key=avg_by_group.get) if avg_by_group else "N/A"

    # Suggestions
    suggestions = []
    if not is_fair:
        suggestions = self._generate_suggestions(all_metric_results, self.threshold)

    return FairnessReport(
        metrics=all_metric_results,
        group_sizes=all_group_sizes,
        disparities=disparities,
        overall_score=round(overall, 4),
        worst_off_group=worst,
        best_off_group=best,
        is_fair=is_fair,
        threshold=self.threshold,
        mitigation_suggestions=suggestions,
    )

Args

-----=

predictions

predicted labels (or probabilities for the positive class)

labels

ground truth

groups

dict of attribute_name → array of group values per sample

metrics

list of metric names (default: all)

favorable_label

which label is considered "positive" (default 1)

class FairnessReport (metrics: dict[str, dict[str, float]],
group_sizes: dict[str, int],
disparities: dict[str, float],
overall_score: float,
worst_off_group: str,
best_off_group: str,
is_fair: bool,
threshold: float,
mitigation_suggestions: list[str] = <factory>)

Expand source code

@dataclass
class FairnessReport:
    """Complete fairness analysis for a model."""

    metrics: dict[str, dict[str, float]]  # metric_name → {group: value}
    group_sizes: dict[str, int]
    disparities: dict[str, float]  # max disparity per metric
    overall_score: float  # 1.0 = perfectly fair, 0.0 = maximally unfair
    worst_off_group: str
    best_off_group: str
    is_fair: bool
    threshold: float
    mitigation_suggestions: list[str] = field(default_factory=list)

Complete fairness analysis for a model.

Instance variables

var best_off_group : str

The type of the None singleton.

var disparities : dict[str, float]

The type of the None singleton.

var group_sizes : dict[str, int]

The type of the None singleton.

var is_fair : bool

The type of the None singleton.

var metrics : dict[str, dict[str, float]]

The type of the None singleton.

var mitigation_suggestions : list[str]

The type of the None singleton.

var overall_score : float

The type of the None singleton.

var threshold : float

The type of the None singleton.

var worst_off_group : str

The type of the None singleton.

class LabelQualityDetector (threshold: float = 0.1, seed: int = 42)

Expand source code

class LabelQualityDetector:
    """
    Detects label errors, outliers, and near‑duplicates using confident learning.

    Usage:
        detector = LabelQualityDetector()
        report = detector.analyze(labels=y_true, pred_probs=model_output_probs)

    No ground‑truth validation set is required – the method works entirely from
    model predictions and given labels.
    """

    def __init__(self, threshold: float = 0.1, seed: int = 42) -> None:
        self.threshold = threshold
        self.seed = seed

    # ── Main API ────────────────────────────────────────────────────
    def analyze(
        self,
        labels: np.ndarray,
        pred_probs: np.ndarray,
        class_names: list[str] | None = None,
    ) -> LabelQualityReport:
        """
        Run full label quality analysis.

        Args:
            labels: integer labels, shape (N,)
            pred_probs: model predicted probabilities, shape (N, K)
            class_names: optional list of class name strings
        """
        n_samples, n_classes = pred_probs.shape
        np.random.seed(self.seed)

        # 1. Compute confident joint (estimate of noise matrix)
        cj = self._compute_confident_joint(labels, pred_probs)

        # 2. Find label issues
        label_issues_mask, quality_scores, suggested = self._find_label_issues(
            labels, pred_probs, cj
        )

        # 3. Per‑class breakdown
        per_class = {}
        for k in range(n_classes):
            mask_k = labels == k
            per_class[k] = int(np.sum(label_issues_mask[mask_k]))

        # 4. Outlier detection (using prediction entropy)
        entropy = -np.sum(pred_probs * np.log(pred_probs + 1e-10), axis=1)
        outlier_thresh = np.percentile(entropy, 95)
        outlier_indices = np.where(entropy > outlier_thresh)[0]

        # 5. Near‑duplicate detection (using prediction cosine similarity)
        near_dupes = self._find_near_duplicates(pred_probs)

        # 6. Overall quality score
        overall = 1.0 - np.mean(label_issues_mask)

        return LabelQualityReport(
            n_samples=n_samples,
            n_label_issues=int(np.sum(label_issues_mask)),
            issue_rate=round(np.mean(label_issues_mask), 4),
            label_issue_indices=np.where(label_issues_mask)[0],
            label_quality_scores=quality_scores,
            suggested_labels=suggested,
            per_class_issues=per_class,
            outlier_indices=outlier_indices,
            near_duplicate_pairs=near_dupes,
            overall_quality_score=round(overall, 4),
        )

    # ── Confident Joint (core algorithm) ────────────────────────────
    def _compute_confident_joint(self, labels: np.ndarray, pred_probs: np.ndarray) -> np.ndarray:
        """
        Estimate the joint distribution of noisy labels and true labels.
        This is the key insight from Cleanlab / confident learning.
        """
        n_samples, n_classes = pred_probs.shape
        pred_classes = np.argmax(pred_probs, axis=1)

        # Average threshold for each class: t_k = mean prob of class k when k is predicted
        thresholds = np.zeros(n_classes)
        for k in range(n_classes):
            mask = pred_classes == k
            if mask.sum() > 0:
                thresholds[k] = np.mean(pred_probs[mask, k])
            else:
                thresholds[k] = 0.5

        # Confident joint: count how many times (given_label = i, true_label = j)
        cj = np.zeros((n_classes, n_classes), dtype=np.float64)
        for i in range(n_samples):
            true_label_guess = np.argmax(pred_probs[i])
            given_label = labels[i]
            if pred_probs[i, true_label_guess] >= thresholds[true_label_guess]:
                cj[given_label, true_label_guess] += 1

        # Normalise to probabilities
        cj = cj / max(cj.sum(), 1.0)
        return cj

    # ── Label Issue Detection ───────────────────────────────────────
    def _find_label_issues(
        self, labels: np.ndarray, pred_probs: np.ndarray, cj: np.ndarray
    ) -> tuple[np.ndarray, np.ndarray, np.ndarray | None]:
        """
        Return: (issue_mask, quality_scores, suggested_labels)
        """
        n_samples, n_classes = pred_probs.shape

        # For each sample, the label quality score = 1 - P(mislabeled)
        self_confidence = pred_probs[np.arange(n_samples), labels]
        quality_scores = self_confidence

        # A sample is a label issue if self‑confidence is below the class threshold
        thresholds = np.zeros(n_classes)
        for k in range(n_classes):
            mask = labels == k
            if mask.sum() > 0:
                thresholds[k] = np.mean(pred_probs[mask, k])
            else:
                thresholds[k] = 0.5

        issue_mask = np.zeros(n_samples, dtype=bool)
        for k in range(n_classes):
            mask_k = labels == k
            issue_mask[mask_k] = self_confidence[mask_k] < (thresholds[k] * (1 - self.threshold))

        # Suggested labels: what the model predicts instead
        suggested = np.argmax(pred_probs, axis=1)
        suggested[~issue_mask] = labels[~issue_mask]  # keep original if not an issue

        return issue_mask, quality_scores, suggested

    # ── Near‑Duplicate Detection ────────────────────────────────────
    def _find_near_duplicates(
        self, pred_probs: np.ndarray, threshold: float = 0.95
    ) -> list[tuple[int, int]]:
        """Find pairs of samples whose prediction vectors are nearly identical."""
        from sklearn.metrics.pairwise import cosine_similarity

        n = pred_probs.shape[0]
        if n > 5000:
            # Subsample for performance
            indices = np.random.choice(n, min(5000, n), replace=False)
            subset = pred_probs[indices]
            sim = cosine_similarity(subset)
        else:
            sim = cosine_similarity(pred_probs)
        np.fill_diagonal(sim, 0)
        pairs = np.argwhere(sim > threshold)
        # De‑duplicate symmetric pairs
        seen = set()
        result = []
        for i, j in pairs:
            if i < j and (i, j) not in seen:
                result.append((int(i), int(j)))
                seen.add((i, j))
        return result

    # ── Summary helpers ─────────────────────────────────────────────
    def summary(self, report: LabelQualityReport) -> dict[str, Any]:
        """Summary."""
        return {
            "total_samples": report.n_samples,
            "label_issues_found": report.n_label_issues,
            "issue_rate": report.issue_rate,
            "overall_quality": report.overall_quality_score,
            "n_outliers": len(report.outlier_indices),
            "n_near_duplicates": len(report.near_duplicate_pairs),
            "worst_classes": sorted(
                report.per_class_issues.items(), key=lambda x: x[1], reverse=True
            )[:5],
        }

Detects label errors, outliers, and near‑duplicates using confident learning.

Usage -----= detector = LabelQualityDetector() report = detector.analyze(labels=y_true, pred_probs=model_output_probs)

No ground‑truth validation set is required – the method works entirely from model predictions and given labels.

Methods

def analyze(self,
labels: numpy.ndarray,
pred_probs: numpy.ndarray,
class_names: list[str] | None = None) ‑> LabelQualityReport

Expand source code

def analyze(
    self,
    labels: np.ndarray,
    pred_probs: np.ndarray,
    class_names: list[str] | None = None,
) -> LabelQualityReport:
    """
    Run full label quality analysis.

    Args:
        labels: integer labels, shape (N,)
        pred_probs: model predicted probabilities, shape (N, K)
        class_names: optional list of class name strings
    """
    n_samples, n_classes = pred_probs.shape
    np.random.seed(self.seed)

    # 1. Compute confident joint (estimate of noise matrix)
    cj = self._compute_confident_joint(labels, pred_probs)

    # 2. Find label issues
    label_issues_mask, quality_scores, suggested = self._find_label_issues(
        labels, pred_probs, cj
    )

    # 3. Per‑class breakdown
    per_class = {}
    for k in range(n_classes):
        mask_k = labels == k
        per_class[k] = int(np.sum(label_issues_mask[mask_k]))

    # 4. Outlier detection (using prediction entropy)
    entropy = -np.sum(pred_probs * np.log(pred_probs + 1e-10), axis=1)
    outlier_thresh = np.percentile(entropy, 95)
    outlier_indices = np.where(entropy > outlier_thresh)[0]

    # 5. Near‑duplicate detection (using prediction cosine similarity)
    near_dupes = self._find_near_duplicates(pred_probs)

    # 6. Overall quality score
    overall = 1.0 - np.mean(label_issues_mask)

    return LabelQualityReport(
        n_samples=n_samples,
        n_label_issues=int(np.sum(label_issues_mask)),
        issue_rate=round(np.mean(label_issues_mask), 4),
        label_issue_indices=np.where(label_issues_mask)[0],
        label_quality_scores=quality_scores,
        suggested_labels=suggested,
        per_class_issues=per_class,
        outlier_indices=outlier_indices,
        near_duplicate_pairs=near_dupes,
        overall_quality_score=round(overall, 4),
    )

Run full label quality analysis.

Args

-----=

labels

integer labels, shape (N,)

pred_probs

model predicted probabilities, shape (N, K)

class_names

optional list of class name strings

def summary(self,
report: LabelQualityReport) ‑> dict[str, typing.Any]

Expand source code

def summary(self, report: LabelQualityReport) -> dict[str, Any]:
    """Summary."""
    return {
        "total_samples": report.n_samples,
        "label_issues_found": report.n_label_issues,
        "issue_rate": report.issue_rate,
        "overall_quality": report.overall_quality_score,
        "n_outliers": len(report.outlier_indices),
        "n_near_duplicates": len(report.near_duplicate_pairs),
        "worst_classes": sorted(
            report.per_class_issues.items(), key=lambda x: x[1], reverse=True
        )[:5],
    }

Summary.

class LabelQualityReport (n_samples: int,
n_label_issues: int,
issue_rate: float,
label_issue_indices: numpy.ndarray,
label_quality_scores: numpy.ndarray,
suggested_labels: numpy.ndarray | None = None,
per_class_issues: dict[int, int] = <factory>,
outlier_indices: numpy.ndarray = <factory>,
near_duplicate_pairs: list[tuple[int, int]] = <factory>,
overall_quality_score: float = 0.0)

Expand source code

@dataclass
class LabelQualityReport:
    """Complete label quality analysis for a dataset."""

    n_samples: int
    n_label_issues: int
    issue_rate: float
    label_issue_indices: np.ndarray
    label_quality_scores: np.ndarray  # per‑sample confidence (0 = bad, 1 = perfect)
    suggested_labels: np.ndarray | None = None  # what cleanlab thinks the label should be
    per_class_issues: dict[int, int] = field(default_factory=dict)
    outlier_indices: np.ndarray = field(default_factory=lambda: np.array([]))
    near_duplicate_pairs: list[tuple[int, int]] = field(default_factory=list)
    overall_quality_score: float = 0.0

Complete label quality analysis for a dataset.

Instance variables

var issue_rate : float

The type of the None singleton.

var label_issue_indices : numpy.ndarray

The type of the None singleton.

var label_quality_scores : numpy.ndarray

The type of the None singleton.

var n_label_issues : int

The type of the None singleton.

var n_samples : int

The type of the None singleton.

var near_duplicate_pairs : list[tuple[int, int]]

The type of the None singleton.

var outlier_indices : numpy.ndarray

The type of the None singleton.

var overall_quality_score : float

The type of the None singleton.

var per_class_issues : dict[int, int]

The type of the None singleton.

var suggested_labels : numpy.ndarray | None

The type of the None singleton.

class RetrainDecision (should_retrain: bool,
reason: str,
trigger_type: TriggerType | None = None,
urgency: str = 'low',
details: dict[str, typing.Any] = <factory>)

Expand source code

@dataclass
class RetrainDecision:
    """Output of the retraining decision engine."""

    should_retrain: bool
    reason: str
    trigger_type: TriggerType | None = None
    urgency: str = "low"  # "low", "medium", "high", "critical"
    details: dict[str, Any] = field(default_factory=dict)

Output of the retraining decision engine.

Instance variables

var details : dict[str, typing.Any]

The type of the None singleton.

var reason : str

The type of the None singleton.

var should_retrain : bool

The type of the None singleton.

var trigger_type : TriggerType | None

The type of the None singleton.

var urgency : str

The type of the None singleton.

class RetrainPolicy (trigger_types: list[TriggerType] = <factory>,
drift_threshold: float = 0.3,
drift_methods: list[str] = <factory>,
min_accuracy: float = 0.85,
max_accuracy_degradation: float = 0.05,
schedule_cron: str | None = None,
max_days_since_last_train: int = 30,
min_new_samples: int = 1000,
max_new_samples: int = 100000,
require_better_model: bool = True,
min_improvement: float = 0.01,
max_retrains_per_day: int = 3,
validation_split: float = 0.2,
notify_on_trigger: bool = True,
notify_on_complete: bool = True)

Expand source code

@dataclass
class RetrainPolicy:
    """Configuration for when and how to retrain."""

    trigger_types: list[TriggerType] = field(default_factory=lambda: [TriggerType.DRIFT])

    # Drift trigger
    drift_threshold: float = 0.3
    drift_methods: list[str] = field(default_factory=lambda: ["ks", "psi"])

    # Performance trigger
    min_accuracy: float = 0.85
    max_accuracy_degradation: float = 0.05

    # Schedule trigger
    schedule_cron: str | None = None  # "0 2 * * 1" for every Monday 2 AM
    max_days_since_last_train: int = 30

    # Data volume trigger
    min_new_samples: int = 1000
    max_new_samples: int = 100_000

    # Retraining behaviour
    require_better_model: bool = True
    min_improvement: float = 0.01
    max_retrains_per_day: int = 3
    validation_split: float = 0.2
    notify_on_trigger: bool = True
    notify_on_complete: bool = True

Configuration for when and how to retrain.

Instance variables

var drift_methods : list[str]

The type of the None singleton.

var drift_threshold : float

The type of the None singleton.

var max_accuracy_degradation : float

The type of the None singleton.

var max_days_since_last_train : int

The type of the None singleton.

var max_new_samples : int

The type of the None singleton.

var max_retrains_per_day : int

The type of the None singleton.

var min_accuracy : float

The type of the None singleton.

var min_improvement : float

The type of the None singleton.

var min_new_samples : int

The type of the None singleton.

var notify_on_complete : bool

The type of the None singleton.

var notify_on_trigger : bool

The type of the None singleton.

var require_better_model : bool

The type of the None singleton.

var schedule_cron : str | None

The type of the None singleton.

var trigger_types : list[TriggerType]

The type of the None singleton.

var validation_split : float

The type of the None singleton.

class RetrainingOrchestrator (policy: RetrainPolicy = None)

Expand source code

class RetrainingOrchestrator:
    """
    Central decision engine for automated retraining.

    Usage:
        orch = RetrainingOrchestrator(RetrainPolicy())
        decision = orch.evaluate(
            current_metrics={"accuracy": 0.82},
            reference_metrics={"accuracy": 0.88},
            drift_report=drift_report,
            new_data_count=5000,
            days_since_last_train=35,
        )
        if decision.should_retrain:
            orch.trigger_retraining(run_id="model-v3", training_fn=my_train_func)
    """

    def __init__(self, policy: RetrainPolicy = None) -> None:
        self.policy = policy or RetrainPolicy()
        self._retrain_count_today: int = 0
        self._last_retrain_date: str = ""

    def evaluate(
        self,
        current_metrics: dict[str, float] | None = None,
        reference_metrics: dict[str, float] | None = None,
        drift_report: Any | None = None,
        new_data_count: int = 0,
        days_since_last_train: int = 0,
        manual_request: bool = False,
    ) -> RetrainDecision:
        """Evaluate all configured triggers and return a retraining decision."""

        # Reset daily counter if it's a new day
        today = datetime.now().strftime("%Y-%m-%d")
        if today != self._last_retrain_date:
            self._retrain_count_today = 0
            self._last_retrain_date = today

        if self._retrain_count_today >= self.policy.max_retrains_per_day:
            return RetrainDecision(
                should_retrain=False,
                reason="Max daily retrains reached",
                details={"retrains_today": self._retrain_count_today},
            )

        # 1. Manual trigger
        if manual_request and TriggerType.MANUAL in self.policy.trigger_types:
            return RetrainDecision(
                should_retrain=True,
                reason="Manual retraining requested",
                trigger_type=TriggerType.MANUAL,
                urgency="high",
            )

        # 2. Performance degradation
        if (
            TriggerType.PERFORMANCE in self.policy.trigger_types
            and current_metrics
            and reference_metrics
        ):
            decision = self._check_performance(current_metrics, reference_metrics)
            if decision.should_retrain:
                return decision

        # 3. Drift trigger
        if TriggerType.DRIFT in self.policy.trigger_types and drift_report:
            decision = self._check_drift(drift_report)
            if decision.should_retrain:
                return decision

        # 4. Data volume trigger
        if TriggerType.DATA_VOLUME in self.policy.trigger_types:
            decision = self._check_data_volume(new_data_count)
            if decision.should_retrain:
                return decision

        # 5. Schedule trigger
        if TriggerType.SCHEDULE in self.policy.trigger_types:
            decision = self._check_schedule(days_since_last_train)
            if decision.should_retrain:
                return decision

        return RetrainDecision(should_retrain=False, reason="No triggers met")

    def _check_performance(
        self, current: dict[str, float], reference: dict[str, float]
    ) -> RetrainDecision:
        """check performance."""
        acc_key = "accuracy" if "accuracy" in current else "eval_accuracy"
        curr_acc = current.get(acc_key, 1.0)
        ref_acc = reference.get(acc_key, 1.0)
        degradation = ref_acc - curr_acc

        if curr_acc < self.policy.min_accuracy:
            return RetrainDecision(
                should_retrain=True,
                reason=f"Accuracy {curr_acc:.3f} below minimum {self.policy.min_accuracy}",
                trigger_type=TriggerType.PERFORMANCE,
                urgency="critical" if degradation > 0.1 else "high",
                details={
                    "current_accuracy": curr_acc,
                    "reference_accuracy": ref_acc,
                    "degradation": degradation,
                },
            )

        if degradation > self.policy.max_accuracy_degradation:
            return RetrainDecision(
                should_retrain=True,
                reason=f"Accuracy degraded by {degradation:.3f} (threshold: {self.policy.max_accuracy_degradation})",
                trigger_type=TriggerType.PERFORMANCE,
                urgency="medium",
                details={"degradation": degradation},
            )

        return RetrainDecision(should_retrain=False, reason="Performance within acceptable range")

    def _check_drift(self, drift_report) -> RetrainDecision:
        """check drift."""
        n_drifted = getattr(drift_report, "n_features_drifted", 0)
        n_total = getattr(drift_report, "n_features_tested", 1)
        ratio = n_drifted / max(n_total, 1)

        if ratio > self.policy.drift_threshold:
            return RetrainDecision(
                should_retrain=True,
                reason=f"Drift detected in {n_drifted}/{n_total} features ({ratio:.1%})",
                trigger_type=TriggerType.DRIFT,
                urgency="high" if ratio > 0.5 else "medium",
                details={"drifted_features": n_drifted, "total_features": n_total},
            )
        return RetrainDecision(should_retrain=False, reason="Drift within acceptable range")

    def _check_data_volume(self, new_count: int) -> RetrainDecision:
        """check data volume."""
        if new_count >= self.policy.min_new_samples:
            return RetrainDecision(
                should_retrain=True,
                reason=f"New data volume ({new_count}) exceeded threshold ({self.policy.min_new_samples})",
                trigger_type=TriggerType.DATA_VOLUME,
                urgency="low",
                details={"new_samples": new_count},
            )
        return RetrainDecision(should_retrain=False, reason="Insufficient new data")

    def _check_schedule(self, days_since: int) -> RetrainDecision:
        """check schedule."""
        if days_since >= self.policy.max_days_since_last_train:
            return RetrainDecision(
                should_retrain=True,
                reason=f"Scheduled retrain: {days_since} days since last training (max: {self.policy.max_days_since_last_train})",
                trigger_type=TriggerType.SCHEDULE,
                urgency="low",
                details={"days_since_last_train": days_since},
            )
        return RetrainDecision(should_retrain=False, reason="Schedule not due yet")

    def trigger_retraining(
        self,
        run_id: str,
        training_fn: Callable,
        decision: RetrainDecision,
        **training_kwargs,
    ) -> dict[str, Any]:
        """Execute retraining and validate the new model."""
        logger.info("Retraining triggered for %s: %s", run_id, decision.reason)
        self._retrain_count_today += 1

        try:
            result = training_fn(**training_kwargs)
            logger.info(
                "Retraining completed for %s. Metrics: %s",
                run_id,
                {k: v for k, v in result.items() if isinstance(v, (int, float))},
            )
        except Exception as e:
            logger.exception("Retraining failed for %s: %s", run_id, e)
            return {"status": "failed", "error": str(e)}

        return {"status": "completed", "run_id": run_id, "metrics": result}

Central decision engine for automated retraining.

Usage -----= orch = RetrainingOrchestrator(RetrainPolicy()) decision = orch.evaluate( current_metrics={"accuracy": 0.82}, reference_metrics={"accuracy": 0.88}, drift_report=drift_report, new_data_count=5000, days_since_last_train=35, ) if decision.should_retrain: orch.trigger_retraining(run_id="model-v3", training_fn=my_train_func)

Methods

def evaluate(self,
current_metrics: dict[str, float] | None = None,
reference_metrics: dict[str, float] | None = None,
drift_report: Any | None = None,
new_data_count: int = 0,
days_since_last_train: int = 0,
manual_request: bool = False) ‑> RetrainDecision

Expand source code

def evaluate(
    self,
    current_metrics: dict[str, float] | None = None,
    reference_metrics: dict[str, float] | None = None,
    drift_report: Any | None = None,
    new_data_count: int = 0,
    days_since_last_train: int = 0,
    manual_request: bool = False,
) -> RetrainDecision:
    """Evaluate all configured triggers and return a retraining decision."""

    # Reset daily counter if it's a new day
    today = datetime.now().strftime("%Y-%m-%d")
    if today != self._last_retrain_date:
        self._retrain_count_today = 0
        self._last_retrain_date = today

    if self._retrain_count_today >= self.policy.max_retrains_per_day:
        return RetrainDecision(
            should_retrain=False,
            reason="Max daily retrains reached",
            details={"retrains_today": self._retrain_count_today},
        )

    # 1. Manual trigger
    if manual_request and TriggerType.MANUAL in self.policy.trigger_types:
        return RetrainDecision(
            should_retrain=True,
            reason="Manual retraining requested",
            trigger_type=TriggerType.MANUAL,
            urgency="high",
        )

    # 2. Performance degradation
    if (
        TriggerType.PERFORMANCE in self.policy.trigger_types
        and current_metrics
        and reference_metrics
    ):
        decision = self._check_performance(current_metrics, reference_metrics)
        if decision.should_retrain:
            return decision

    # 3. Drift trigger
    if TriggerType.DRIFT in self.policy.trigger_types and drift_report:
        decision = self._check_drift(drift_report)
        if decision.should_retrain:
            return decision

    # 4. Data volume trigger
    if TriggerType.DATA_VOLUME in self.policy.trigger_types:
        decision = self._check_data_volume(new_data_count)
        if decision.should_retrain:
            return decision

    # 5. Schedule trigger
    if TriggerType.SCHEDULE in self.policy.trigger_types:
        decision = self._check_schedule(days_since_last_train)
        if decision.should_retrain:
            return decision

    return RetrainDecision(should_retrain=False, reason="No triggers met")

Evaluate all configured triggers and return a retraining decision.

def trigger_retraining(self,
run_id: str,
training_fn: ,
decision: RetrainDecision,
**training_kwargs) ‑> dict[str, typing.Any]

Expand source code

def trigger_retraining(
    self,
    run_id: str,
    training_fn: Callable,
    decision: RetrainDecision,
    **training_kwargs,
) -> dict[str, Any]:
    """Execute retraining and validate the new model."""
    logger.info("Retraining triggered for %s: %s", run_id, decision.reason)
    self._retrain_count_today += 1

    try:
        result = training_fn(**training_kwargs)
        logger.info(
            "Retraining completed for %s. Metrics: %s",
            run_id,
            {k: v for k, v in result.items() if isinstance(v, (int, float))},
        )
    except Exception as e:
        logger.exception("Retraining failed for %s: %s", run_id, e)
        return {"status": "failed", "error": str(e)}

    return {"status": "completed", "run_id": run_id, "metrics": result}

Execute retraining and validate the new model.

class TriggerType (*values)

Expand source code

class TriggerType(Enum):
    DRIFT = "drift"
    PERFORMANCE = "performance"
    SCHEDULE = "schedule"
    DATA_VOLUME = "data_volume"
    MANUAL = "manual"

Create a collection of name/value pairs.

Example enumeration:

>>> class Color(Enum):
...     RED = 1
...     BLUE = 2
...     GREEN = 3

Access them by:

• attribute access:

Color.RED

• value lookup:

Color(1)

• name lookup:

Color['RED']

Enumerations can be iterated over, and know how many members they have:

>>> len(Color)
3

>>> list(Color)
[<Color.RED: 1>, <Color.BLUE: 2>, <Color.GREEN: 3>]

Methods can be added to enumerations, and members can have their own attributes – see the documentation for details.

Ancestors

• enum.Enum

Class variables

var DATA_VOLUME

The type of the None singleton.

var DRIFT

The type of the None singleton.

var MANUAL

The type of the None singleton.

var PERFORMANCE

The type of the None singleton.

var SCHEDULE

The type of the None singleton.