Responsible AI: Building Ethical Machine Learning Systems

import pandas as pd
from fairlearn.metrics import MetricFrame
from sklearn.metrics import accuracy_score, precision_score, recall_score
 
def audit_model_fairness(
    y_true: pd.Series,
    y_pred: pd.Series,
    sensitive_features: pd.DataFrame
) -> dict:
    """
    Comprehensive fairness audit across demographic groups.
    """
 
    # Calculate metrics across groups
    metrics = {
        'accuracy': accuracy_score,
        'precision': precision_score,
        'recall': recall_score,
        'selection_rate': lambda y_t, y_p: y_p.mean(),  # Positive prediction rate
    }
 
    results = {}
 
    for feature_name in sensitive_features.columns:
        metric_frame = MetricFrame(
            metrics=metrics,
            y_true=y_true,
            y_pred=y_pred,
            sensitive_features=sensitive_features[feature_name]
        )
 
        results[feature_name] = {
            'by_group': metric_frame.by_group.to_dict(),
            'differences': metric_frame.difference().to_dict(),
            'ratios': metric_frame.ratio().to_dict(),
            'overall': metric_frame.overall.to_dict()
        }
 
        # Flag significant disparities
        for metric_name, ratio in metric_frame.ratio().items():
            if ratio < 0.8:  # 80% rule commonly used in employment
                print(f"WARNING: {feature_name} - {metric_name} ratio = {ratio:.2f}")
 
    return results
 
# Example usage
audit_results = audit_model_fairness(
    y_true=test_df['outcome'],
    y_pred=predictions,
    sensitive_features=test_df[['gender', 'race', 'age_group']]
)

from fairlearn.reductions import ExponentiatedGradient, DemographicParity
from fairlearn.postprocessing import ThresholdOptimizer
 
class FairClassifier:
    """Wrapper that applies fairness constraints during training."""
 
    def __init__(self, base_estimator, fairness_constraint="demographic_parity"):
        self.base_estimator = base_estimator
 
        if fairness_constraint == "demographic_parity":
            self.constraint = DemographicParity()
        elif fairness_constraint == "equalized_odds":
            from fairlearn.reductions import EqualizedOdds
            self.constraint = EqualizedOdds()
 
        self.mitigator = ExponentiatedGradient(
            estimator=base_estimator,
            constraints=self.constraint
        )
 
    def fit(self, X, y, sensitive_features):
        """Train with fairness constraints."""
        self.mitigator.fit(X, y, sensitive_features=sensitive_features)
        return self
 
    def predict(self, X):
        return self.mitigator.predict(X)
 
# Post-processing approach (adjust thresholds per group)
def calibrate_thresholds(model, X_val, y_val, sensitive_features):
    """Find optimal thresholds for each group to equalize metrics."""
    optimizer = ThresholdOptimizer(
        estimator=model,
        constraints="equalized_odds",
        prefit=True
    )
    optimizer.fit(X_val, y_val, sensitive_features=sensitive_features)
    return optimizer

import shap
import matplotlib.pyplot as plt
 
def explain_prediction(model, instance, feature_names, background_data):
    """Generate SHAP explanation for a single prediction."""
 
    # Create explainer
    explainer = shap.Explainer(model, background_data)
 
    # Calculate SHAP values for this instance
    shap_values = explainer(instance)
 
    # Create explanation dictionary
    explanation = {
        'prediction': model.predict(instance)[0],
        'base_value': explainer.expected_value,
        'feature_contributions': dict(zip(
            feature_names,
            shap_values.values[0]
        ))
    }
 
    # Sort by absolute importance
    explanation['top_factors'] = sorted(
        explanation['feature_contributions'].items(),
        key=lambda x: abs(x[1]),
        reverse=True
    )[:5]
 
    return explanation
 
def generate_explanation_text(explanation: dict) -> str:
    """Convert SHAP explanation to human-readable text."""
    text = f"Prediction: {'Approved' if explanation['prediction'] == 1 else 'Denied'}\n\n"
    text += "Key factors:\n"
 
    for feature, contribution in explanation['top_factors']:
        direction = "increased" if contribution > 0 else "decreased"
        text += f"  • {feature}: {direction} likelihood by {abs(contribution):.2f}\n"
 
    return text

from dataclasses import dataclass
from typing import Optional
from datetime import date
 
@dataclass
class ModelCard:
    """Documentation template for ML models."""
 
    # Model Details
    model_name: str
    model_version: str
    model_type: str
    training_date: date
    developers: list[str]
 
    # Intended Use
    primary_intended_uses: list[str]
    primary_intended_users: list[str]
    out_of_scope_uses: list[str]
 
    # Training Data
    training_data_description: str
    training_data_size: int
    preprocessing_steps: list[str]
 
    # Evaluation Data
    evaluation_data_description: str
    evaluation_data_size: int
 
    # Metrics
    overall_performance: dict[str, float]
    performance_by_group: dict[str, dict[str, float]]
 
    # Fairness Considerations
    sensitive_attributes_tested: list[str]
    fairness_metrics: dict[str, float]
    known_biases: list[str]
 
    # Limitations
    known_limitations: list[str]
    recommendations: list[str]
 
    # Ethical Considerations
    potential_harms: list[str]
    mitigation_strategies: list[str]
 
    def to_markdown(self) -> str:
        """Generate readable model card documentation."""
        # Implementation to render as markdown
        pass
 
# Example
loan_model_card = ModelCard(
    model_name="Loan Approval Classifier",
    model_version="2.1.0",
    model_type="Gradient Boosted Trees",
    training_date=date(2024, 1, 15),
    developers=["ML Team"],
 
    primary_intended_uses=[
        "Pre-screening loan applications",
        "Flagging applications for human review"
    ],
    primary_intended_users=["Loan officers", "Credit analysts"],
    out_of_scope_uses=[
        "Automated final decisions without human review",
        "Applications from markets not in training data"
    ],
 
    known_limitations=[
        "Lower accuracy for applicants under 25",
        "Limited data for self-employed individuals"
    ],
 
    potential_harms=[
        "False denials may disproportionately affect minority groups",
        "Over-reliance may reduce human judgment in edge cases"
    ],
    mitigation_strategies=[
        "Mandatory human review for all denials",
        "Quarterly fairness audits",
        "Appeal process for denied applications"
    ]
)

from dataclasses import dataclass
from enum import Enum
 
class DecisionPath(Enum):
    AUTOMATIC = "automatic"
    HUMAN_REVIEW = "human_review"
    ESCALATION = "escalation"
 
@dataclass
class PredictionWithConfidence:
    prediction: int
    confidence: float
    explanation: dict
    decision_path: DecisionPath
    review_reason: Optional[str] = None
 
def route_decision(
    prediction: int,
    confidence: float,
    explanation: dict,
    is_high_stakes: bool = False
) -> PredictionWithConfidence:
    """
    Route predictions based on confidence and stakes.
    """
 
    # High-stakes decisions always get human review
    if is_high_stakes:
        return PredictionWithConfidence(
            prediction=prediction,
            confidence=confidence,
            explanation=explanation,
            decision_path=DecisionPath.HUMAN_REVIEW,
            review_reason="High-stakes decision requires human approval"
        )
 
    # Low confidence predictions need review
    if confidence < 0.7:
        return PredictionWithConfidence(
            prediction=prediction,
            confidence=confidence,
            explanation=explanation,
            decision_path=DecisionPath.HUMAN_REVIEW,
            review_reason=f"Low confidence ({confidence:.2f})"
        )
 
    # Check for unusual feature patterns
    if has_unusual_patterns(explanation):
        return PredictionWithConfidence(
            prediction=prediction,
            confidence=confidence,
            explanation=explanation,
            decision_path=DecisionPath.ESCALATION,
            review_reason="Unusual feature patterns detected"
        )
 
    # High confidence, normal case - can proceed automatically
    return PredictionWithConfidence(
        prediction=prediction,
        confidence=confidence,
        explanation=explanation,
        decision_path=DecisionPath.AUTOMATIC
    )

from datetime import datetime
from typing import Optional
import json
 
@dataclass
class AIDecisionLog:
    """Immutable record of AI-assisted decisions."""
 
    decision_id: str
    timestamp: datetime
    model_version: str
 
    # Input
    input_features: dict
    sensitive_attributes: dict  # Stored separately for audit
 
    # Output
    prediction: int
    confidence: float
    explanation: dict
 
    # Routing
    decision_path: str
    review_reason: Optional[str]
 
    # Human involvement
    human_reviewer_id: Optional[str]
    human_decision: Optional[int]
    human_override: bool
    human_notes: Optional[str]
 
    # Outcome (filled later)
    final_decision: int
    actual_outcome: Optional[int]  # Ground truth when available
 
class DecisionAuditLog:
    """Audit log for AI decisions - immutable append-only."""
 
    def __init__(self, storage):
        self.storage = storage
 
    def log_decision(self, log: AIDecisionLog) -> str:
        """Record decision with all context."""
        record = {
            **log.__dict__,
            'timestamp': log.timestamp.isoformat(),
            'logged_at': datetime.utcnow().isoformat()
        }
 
        # Append to immutable storage
        self.storage.append(record)
 
        return log.decision_id
 
    def get_decisions_for_audit(
        self,
        start_date: datetime,
        end_date: datetime,
        filters: Optional[dict] = None
    ) -> list[dict]:
        """Retrieve decisions for fairness auditing."""
        return self.storage.query(
            start_date=start_date,
            end_date=end_date,
            filters=filters
        )