""" Reliability Analyzer for AegisLM Scoring System. Analyzes confidence reliability, overconfidence, underconfidence, and calibration gaps to ensure trustworthy confidence scores. """ import numpy as np import statistics from typing import List, Dict, Any, Tuple, Optional from dataclasses import dataclass from datetime import datetime, timedelta from enum import Enum import logging logger = logging.getLogger(__name__) class ReliabilityIssue(str, Enum): """Types of reliability issues.""" OVERCONFIDENCE = "overconfidence" UNDERCONFIDENCE = "underconfidence" POOR_CALIBRATION = "poor_calibration" HIGH_VARIANCE = "high_variance" INCONSISTENT_RESPONSES = "inconsistent_responses" @dataclass class ReliabilityMetric: """Single reliability metric.""" name: str value: float threshold: float status: str # "good", "warning", "critical" description: str @dataclass class ReliabilityReport: """Comprehensive reliability analysis report.""" overall_reliability_score: float reliability_grade: str # "A", "B", "C", "D", "F" metrics: List[ReliabilityMetric] issues: List[ReliabilityIssue] recommendations: List[str] analysis_timestamp: datetime sample_size: int confidence_range: Tuple[float, float] class ReliabilityAnalyzer: """ Advanced reliability analyzer for confidence scores. Analyzes overconfidence, underconfidence, calibration gaps, and provides actionable recommendations for improvement. """ def __init__(self, overconfidence_threshold: float = 0.2, underconfidence_threshold: float = 0.3, variance_threshold: float = 0.1, consistency_threshold: float = 0.8): """ Initialize reliability analyzer. Args: overconfidence_threshold: Threshold for detecting overconfidence underconfidence_threshold: Threshold for detecting underconfidence variance_threshold: Threshold for acceptable variance consistency_threshold: Threshold for response consistency """ self.overconfidence_threshold = overconfidence_threshold self.underconfidence_threshold = underconfidence_threshold self.variance_threshold = variance_threshold self.consistency_threshold = consistency_threshold def analyze_reliability( self, confidences: List[float], correctness: List[bool], predictions: Optional[List[Any]] = None, contexts: Optional[List[Dict[str, Any]]] = None ) -> ReliabilityReport: """ Analyze reliability of confidence scores. Args: confidences: List of confidence scores correctness: List of correctness indicators predictions: Optional list of predictions for consistency analysis contexts: Optional list of contexts for analysis Returns: ReliabilityReport: Comprehensive reliability analysis """ if len(confidences) != len(correctness): raise ValueError("Confidences and correctness must have same length") if not confidences: raise ValueError("No data provided for analysis") # Calculate basic statistics mean_confidence = statistics.mean(confidences) accuracy = sum(correctness) / len(correctness) confidence_variance = statistics.variance(confidences) if len(confidences) > 1 else 0.0 # Analyze calibration calibration_metrics = self._analyze_calibration(confidences, correctness) # Analyze overconfidence/underconfidence confidence_analysis = self._analyze_confidence_distribution( confidences, correctness, mean_confidence, accuracy ) # Analyze variance variance_analysis = self._analyze_variance(confidences, confidence_variance) # Analyze consistency (if predictions provided) consistency_analysis = self._analyze_consistency( predictions, contexts, confidences ) if predictions else None # Calculate overall reliability score overall_score = self._calculate_overall_reliability_score( calibration_metrics, confidence_analysis, variance_analysis, consistency_analysis ) # Determine grade grade = self._determine_reliability_grade(overall_score) # Identify issues issues = self._identify_reliability_issues( calibration_metrics, confidence_analysis, variance_analysis, consistency_analysis ) # Generate recommendations recommendations = self._generate_recommendations(issues, overall_score) # Create metrics list metrics = self._create_metrics_list( calibration_metrics, confidence_analysis, variance_analysis, consistency_analysis ) return ReliabilityReport( overall_reliability_score=overall_score, reliability_grade=grade, metrics=metrics, issues=issues, recommendations=recommendations, analysis_timestamp=datetime.utcnow(), sample_size=len(confidences), confidence_range=(min(confidences), max(confidences)) ) def _analyze_calibration( self, confidences: List[float], correctness: List[bool] ) -> Dict[str, Any]: """ Analyze calibration quality. Args: confidences: List of confidence scores correctness: List of correctness indicators Returns: Dict[str, Any]: Calibration analysis results """ # Calculate Expected Calibration Error (ECE) ece = self._calculate_ece(confidences, correctness) # Calculate Brier score brier_score = self._calculate_brier_score(confidences, correctness) # Calculate calibration curve slope calibration_slope = self._calculate_calibration_slope(confidences, correctness) # Calculate reliability diagram data reliability_data = self._calculate_reliability_diagram(confidences, correctness) return { "ece": ece, "brier_score": brier_score, "calibration_slope": calibration_slope, "reliability_data": reliability_data, "is_well_calibrated": ece < 0.1 } def _analyze_confidence_distribution( self, confidences: List[float], correctness: List[bool], mean_confidence: float, accuracy: float ) -> Dict[str, Any]: """ Analyze confidence distribution for over/underconfidence. Args: confidences: List of confidence scores correctness: List of correctness indicators mean_confidence: Mean confidence score accuracy: Overall accuracy Returns: Dict[str, Any]: Confidence distribution analysis """ # Calculate confidence-accuracy gap confidence_accuracy_gap = mean_confidence - accuracy # Determine overconfidence/underconfidence if confidence_accuracy_gap > self.overconfidence_threshold: confidence_bias = "overconfident" bias_severity = min(confidence_accuracy_gap / 0.5, 1.0) elif confidence_accuracy_gap < -self.underconfidence_threshold: confidence_bias = "underconfident" bias_severity = min(abs(confidence_accuracy_gap) / 0.5, 1.0) else: confidence_bias = "well_calibrated" bias_severity = 0.0 # Calculate confidence distribution statistics confidence_std = statistics.stdev(confidences) if len(confidences) > 1 else 0.0 confidence_range = max(confidences) - min(confidences) # Calculate confidence by correctness correct_confidences = [c for c, correct in zip(confidences, correctness) if correct] incorrect_confidences = [c for c, correct in zip(confidences, correctness) if not correct] mean_correct_confidence = statistics.mean(correct_confidences) if correct_confidences else 0.0 mean_incorrect_confidence = statistics.mean(incorrect_confidences) if incorrect_confidences else 0.0 # Calculate discrimination (difference between correct and incorrect confidences) discrimination = mean_correct_confidence - mean_incorrect_confidence return { "confidence_bias": confidence_bias, "bias_severity": bias_severity, "confidence_accuracy_gap": confidence_accuracy_gap, "confidence_std": confidence_std, "confidence_range": confidence_range, "mean_correct_confidence": mean_correct_confidence, "mean_incorrect_confidence": mean_incorrect_confidence, "discrimination": discrimination, "is_biased": confidence_bias != "well_calibrated" } def _analyze_variance( self, confidences: List[float], variance: float ) -> Dict[str, Any]: """ Analyze confidence variance. Args: confidences: List of confidence scores variance: Calculated variance Returns: Dict[str, Any]: Variance analysis results """ # Calculate coefficient of variation mean_confidence = statistics.mean(confidences) cv = (variance ** 0.5) / mean_confidence if mean_confidence > 0 else 0.0 # Determine variance level if variance > self.variance_threshold * 2: variance_level = "high" elif variance > self.variance_threshold: variance_level = "moderate" else: variance_level = "low" # Calculate confidence stability stability = 1.0 - min(variance / self.variance_threshold, 1.0) return { "variance": variance, "std_deviation": variance ** 0.5, "coefficient_of_variation": cv, "variance_level": variance_level, "stability": stability, "is_stable": variance <= self.variance_threshold } def _analyze_consistency( self, predictions: List[Any], contexts: Optional[List[Dict[str, Any]]], confidences: List[float] ) -> Dict[str, Any]: """ Analyze response consistency. Args: predictions: List of predictions contexts: List of contexts (for grouping similar inputs) confidences: List of confidence scores Returns: Dict[str, Any]: Consistency analysis results """ if not contexts: return { "consistency_score": 1.0, # No context to compare "is_consistent": True, "inconsistency_rate": 0.0 } # Group predictions by similar contexts context_groups = self._group_by_context(contexts, predictions) # Calculate consistency within groups consistency_scores = [] total_comparisons = 0 inconsistent_comparisons = 0 for group_predictions in context_groups.values(): if len(group_predictions) > 1: # Calculate pairwise consistency group_consistency = self._calculate_group_consistency(group_predictions) consistency_scores.append(group_consistency) # Count inconsistencies for i in range(len(group_predictions)): for j in range(i + 1, len(group_predictions)): total_comparisons += 1 if not self._are_predictions_consistent( group_predictions[i], group_predictions[j] ): inconsistent_comparisons += 1 # Calculate overall consistency overall_consistency = statistics.mean(consistency_scores) if consistency_scores else 1.0 inconsistency_rate = inconsistent_comparisons / total_comparisons if total_comparisons > 0 else 0.0 return { "consistency_score": overall_consistency, "is_consistent": overall_consistency >= self.consistency_threshold, "inconsistency_rate": inconsistency_rate, "total_comparisons": total_comparisons, "inconsistent_comparisons": inconsistent_comparisons, "context_groups": len(context_groups) } def _calculate_ece( self, confidences: List[float], correctness: List[bool] ) -> float: """ Calculate Expected Calibration Error (ECE). Args: confidences: List of confidence scores correctness: List of correctness indicators Returns: float: Expected Calibration Error (0-1) """ n_bins = 10 bin_boundaries = np.linspace(0, 1, n_bins + 1) bin_lowers = bin_boundaries[:-1] bin_uppers = bin_boundaries[1:] ece = 0.0 total_samples = len(confidences) for i in range(n_bins): bin_mask = (np.array(confidences) > bin_lowers[i]) & \ (np.array(confidences) <= bin_uppers[i]) bin_samples = np.sum(bin_mask) if bin_samples > 0: bin_correctness = np.array(correctness)[bin_mask] bin_confidences = np.array(confidences)[bin_mask] accuracy = np.mean(bin_correctness) avg_confidence = np.mean(bin_confidences) ece += (bin_samples / total_samples) * abs(accuracy - avg_confidence) return ece def _calculate_brier_score( self, confidences: List[float], correctness: List[bool] ) -> float: """ Calculate Brier score for confidence predictions. Args: confidences: List of confidence scores correctness: List of correctness indicators Returns: float: Brier score (0-1, lower is better) """ # Convert correctness to numeric (1 for correct, 0 for incorrect) numeric_correctness = [1.0 if c else 0.0 for c in correctness] # Calculate Brier score: mean((confidence - correctness)^2) brier_score = statistics.mean([ (c - r) ** 2 for c, r in zip(confidences, numeric_correctness) ]) return brier_score def _calculate_calibration_slope( self, confidences: List[float], correctness: List[bool] ) -> float: """ Calculate calibration curve slope. Args: confidences: List of confidence scores correctness: List of correctness indicators Returns: float: Calibration slope (ideal is 1.0) """ # Create confidence bins n_bins = 10 bin_boundaries = np.linspace(0, 1, n_bins + 1) bin_lowers = bin_boundaries[:-1] bin_uppers = bin_boundaries[1:] bin_centers = [] bin_accuracies = [] for i in range(n_bins): bin_mask = (np.array(confidences) > bin_lowers[i]) & \ (np.array(confidences) <= bin_uppers[i]) bin_samples = np.sum(bin_mask) if bin_samples > 0: bin_correctness = np.array(correctness)[bin_mask] bin_confidences = np.array(confidences)[bin_mask] accuracy = np.mean(bin_correctness) avg_confidence = np.mean(bin_confidences) bin_centers.append(avg_confidence) bin_accuracies.append(accuracy) if len(bin_centers) < 2: return 1.0 # Perfect calibration if insufficient data # Calculate linear regression slope x = np.array(bin_centers) y = np.array(bin_accuracies) # Simple linear regression slope = np.corrcoef(x, y)[0, 1] if len(x) > 1 else 1.0 return max(0.0, slope) # Slope should be non-negative def _calculate_reliability_diagram( self, confidences: List[float], correctness: List[bool] ) -> List[Dict[str, float]]: """ Calculate reliability diagram data. Args: confidences: List of confidence scores correctness: List of correctness indicators Returns: List[Dict[str, float]]: Reliability diagram data points """ n_bins = 10 bin_boundaries = np.linspace(0, 1, n_bins + 1) bin_lowers = bin_boundaries[:-1] bin_uppers = bin_boundaries[1:] diagram_data = [] for i in range(n_bins): bin_mask = (np.array(confidences) > bin_lowers[i]) & \ (np.array(confidences) <= bin_uppers[i]) bin_samples = np.sum(bin_mask) if bin_samples > 0: bin_correctness = np.array(correctness)[bin_mask] bin_confidences = np.array(confidences)[bin_mask] accuracy = np.mean(bin_correctness) avg_confidence = np.mean(bin_confidences) diagram_data.append({ "bin": i, "confidence_avg": avg_confidence, "accuracy": accuracy, "sample_count": int(bin_samples), "bin_lower": float(bin_lowers[i]), "bin_upper": float(bin_uppers[i]) }) return diagram_data def _group_by_context( self, contexts: List[Dict[str, Any]], predictions: List[Any] ) -> Dict[str, List[Any]]: """ Group predictions by similar contexts. Args: contexts: List of contexts predictions: List of predictions Returns: Dict[str, List[Any]]: Grouped predictions """ groups = {} for i, (context, prediction) in enumerate(zip(contexts, predictions)): # Create a simple context key (can be enhanced for more sophisticated grouping) context_key = self._create_context_key(context) if context_key not in groups: groups[context_key] = [] groups[context_key].append(prediction) return groups def _create_context_key(self, context: Dict[str, Any]) -> str: """ Create a key for grouping similar contexts. Args: context: Context dictionary Returns: str: Context key """ # Simple implementation - can be enhanced key_parts = [] # Add important context fields if "prompt_type" in context: key_parts.append(f"type:{context['prompt_type']}") if "attack_type" in context: key_parts.append(f"attack:{context['attack_type']}") if "model_config" in context: model_config = context["model_config"] if isinstance(model_config, dict): key_parts.append(f"model:{model_config.get('model_name', 'unknown')}") return "|".join(key_parts) if key_parts else "default" def _calculate_group_consistency(self, predictions: List[Any]) -> float: """ Calculate consistency within a group of predictions. Args: predictions: List of predictions Returns: float: Consistency score (0-1) """ if len(predictions) <= 1: return 1.0 # Simple consistency based on prediction similarity # This is a placeholder - actual implementation depends on prediction type consistent_pairs = 0 total_pairs = 0 for i in range(len(predictions)): for j in range(i + 1, len(predictions)): total_pairs += 1 if self._are_predictions_consistent(predictions[i], predictions[j]): consistent_pairs += 1 return consistent_pairs / total_pairs if total_pairs > 0 else 1.0 def _are_predictions_consistent(self, pred1: Any, pred2: Any) -> bool: """ Check if two predictions are consistent. Args: pred1: First prediction pred2: Second prediction Returns: bool: Whether predictions are consistent """ # Simple implementation - can be enhanced based on prediction structure if isinstance(pred1, dict) and isinstance(pred2, dict): # Compare key fields key_fields = ["success", "response_type", "attack_type"] for field in key_fields: if field in pred1 and field in pred2: if pred1[field] != pred2[field]: return False return True else: # For simple types, check equality return pred1 == pred2 def _calculate_overall_reliability_score( self, calibration_metrics: Dict[str, Any], confidence_analysis: Dict[str, Any], variance_analysis: Dict[str, Any], consistency_analysis: Optional[Dict[str, Any]] ) -> float: """ Calculate overall reliability score. Args: calibration_metrics: Calibration analysis results confidence_analysis: Confidence distribution analysis variance_analysis: Variance analysis results consistency_analysis: Consistency analysis results Returns: float: Overall reliability score (0-1) """ scores = [] # Calibration score (40% weight) calibration_score = 1.0 - calibration_metrics["ece"] scores.append(("calibration", calibration_score, 0.4)) # Confidence bias score (25% weight) bias_score = 1.0 - confidence_analysis["bias_severity"] scores.append(("bias", bias_score, 0.25)) # Variance score (20% weight) variance_score = variance_analysis["stability"] scores.append(("variance", variance_score, 0.2)) # Consistency score (15% weight) if consistency_analysis: consistency_score = consistency_analysis["consistency_score"] scores.append(("consistency", consistency_score, 0.15)) else: scores.append(("consistency", 1.0, 0.15)) # Default to perfect if no data # Calculate weighted average weighted_score = sum(score * weight for name, score, weight in scores) return max(0.0, min(1.0, weighted_score)) def _determine_reliability_grade(self, score: float) -> str: """ Determine reliability grade from score. Args: score: Reliability score (0-1) Returns: str: Reliability grade (A-F) """ if score >= 0.9: return "A" elif score >= 0.8: return "B" elif score >= 0.7: return "C" elif score >= 0.6: return "D" else: return "F" def _identify_reliability_issues( self, calibration_metrics: Dict[str, Any], confidence_analysis: Dict[str, Any], variance_analysis: Dict[str, Any], consistency_analysis: Optional[Dict[str, Any]] ) -> List[ReliabilityIssue]: """ Identify reliability issues. Args: calibration_metrics: Calibration analysis results confidence_analysis: Confidence distribution analysis variance_analysis: Variance analysis results consistency_analysis: Consistency analysis results Returns: List[ReliabilityIssue]: Identified issues """ issues = [] # Check calibration if not calibration_metrics["is_well_calibrated"]: issues.append(ReliabilityIssue.POOR_CALIBRATION) # Check confidence bias if confidence_analysis["is_biased"]: if confidence_analysis["confidence_bias"] == "overconfident": issues.append(ReliabilityIssue.OVERCONFIDENCE) else: issues.append(ReliabilityIssue.UNDERCONFIDENCE) # Check variance if not variance_analysis["is_stable"]: issues.append(ReliabilityIssue.HIGH_VARIANCE) # Check consistency if consistency_analysis and not consistency_analysis["is_consistent"]: issues.append(ReliabilityIssue.INCONSISTENT_RESPONSES) return issues def _generate_recommendations( self, issues: List[ReliabilityIssue], score: float ) -> List[str]: """ Generate recommendations based on issues and score. Args: issues: Identified reliability issues score: Overall reliability score Returns: List[str]: Recommendations """ recommendations = [] # General recommendations based on score if score < 0.6: recommendations.append("Overall reliability is poor - consider comprehensive model retraining") elif score < 0.8: recommendations.append("Reliability needs improvement - focus on identified issues") # Specific recommendations based on issues for issue in issues: if issue == ReliabilityIssue.OVERCONFIDENCE: recommendations.append("Reduce overconfidence by implementing temperature scaling") recommendations.append("Add confidence penalty during training") elif issue == ReliabilityIssue.UNDERCONFIDENCE: recommendations.append("Boost confidence through Platt scaling") recommendations.append("Review confidence calculation methodology") elif issue == ReliabilityIssue.POOR_CALIBRATION: recommendations.append("Implement confidence calibration using reliability diagrams") recommendations.append("Use Expected Calibration Error (ECE) as training objective") elif issue == ReliabilityIssue.HIGH_VARIANCE: recommendations.append("Reduce response variance through ensemble methods") recommendations.append("Implement consistency regularization") elif issue == ReliabilityIssue.INCONSISTENT_RESPONSES: recommendations.append("Improve prompt standardization") recommendations.append("Add consistency checks in evaluation pipeline") return recommendations def _create_metrics_list( self, calibration_metrics: Dict[str, Any], confidence_analysis: Dict[str, Any], variance_analysis: Dict[str, Any], consistency_analysis: Optional[Dict[str, Any]] ) -> List[ReliabilityMetric]: """ Create list of reliability metrics. Args: calibration_metrics: Calibration analysis results confidence_analysis: Confidence distribution analysis variance_analysis: Variance analysis results consistency_analysis: Consistency analysis results Returns: List[ReliabilityMetric]: List of metrics """ metrics = [] # Calibration metrics metrics.append(ReliabilityMetric( name="Expected Calibration Error", value=calibration_metrics["ece"], threshold=0.1, status="good" if calibration_metrics["ece"] < 0.1 else "warning", description="Measures calibration quality (lower is better)" )) metrics.append(ReliabilityMetric( name="Brier Score", value=calibration_metrics["brier_score"], threshold=0.25, status="good" if calibration_metrics["brier_score"] < 0.25 else "warning", description="Overall confidence prediction quality (lower is better)" )) # Confidence bias metrics metrics.append(ReliabilityMetric( name="Confidence-Accuracy Gap", value=abs(confidence_analysis["confidence_accuracy_gap"]), threshold=0.1, status="good" if abs(confidence_analysis["confidence_accuracy_gap"]) < 0.1 else "warning", description="Difference between mean confidence and accuracy" )) # Variance metrics metrics.append(ReliabilityMetric( name="Confidence Variance", value=variance_analysis["variance"], threshold=self.variance_threshold, status="good" if variance_analysis["variance"] <= self.variance_threshold else "warning", description="Stability of confidence scores (lower is better)" )) # Consistency metrics if consistency_analysis: metrics.append(ReliabilityMetric( name="Response Consistency", value=consistency_analysis["consistency_score"], threshold=self.consistency_threshold, status="good" if consistency_analysis["consistency_score"] >= self.consistency_threshold else "warning", description="Consistency of responses to similar inputs" )) return metrics # Global analyzer instance reliability_analyzer = ReliabilityAnalyzer() def get_reliability_analyzer() -> ReliabilityAnalyzer: """ Get the global reliability analyzer instance. Returns: ReliabilityAnalyzer: Global instance """ return reliability_analyzer