AI Ethics & Responsible AI Developer Guide 2025: Everything About Bias, Fairness, Transparency & Regulation

┌─────────────────────────────────────────────────────┐
│           Global AI Regulation Timeline              │
├─────────────────────────────────────────────────────┤
│  2021.04  EU AI Act draft proposed                   │
│  2023.12  EU AI Act final agreement                  │
│  2024.08  EU AI Act enters into force                │
│  2025.02  Prohibited practices apply                 │
│  2025.08  General-purpose AI rules apply             │
│  2026.08  High-risk AI full enforcement              │
│                                                     │
│  2023.10  US AI Executive Order 14110                │
│  2024.05  South Korea AI Basic Act passed            │
│  2024.04  Japan AI Guidelines for Business           │
└─────────────────────────────────────────────────────┘

# Data bias detection example - checking class imbalance
import pandas as pd
import numpy as np

def detect_representation_bias(df, sensitive_attr, target_col):
    """Analyze target distribution differences across sensitive attributes."""
    results = {}

    groups = df.groupby(sensitive_attr)
    overall_positive_rate = df[target_col].mean()

    for name, group in groups:
        group_positive_rate = group[target_col].mean()
        group_size = len(group)
        group_proportion = group_size / len(df)

        results[name] = {
            'count': group_size,
            'proportion': round(group_proportion, 4),
            'positive_rate': round(group_positive_rate, 4),
            'disparity_ratio': round(
                group_positive_rate / overall_positive_rate, 4
            ) if overall_positive_rate > 0 else None
        }

    return pd.DataFrame(results).T

# Usage
# df = pd.read_csv('loan_data.csv')
# bias_report = detect_representation_bias(df, 'race', 'approved')
# print(bias_report)

# Proxy variable detection example
from sklearn.metrics import mutual_information_score

def detect_proxy_variables(df, sensitive_attr, feature_cols, threshold=0.3):
    """Detect features that may serve as proxies for sensitive attributes."""
    proxy_candidates = []

    for col in feature_cols:
        if col == sensitive_attr:
            continue

        try:
            mi_score = mutual_information_score(
                df[sensitive_attr].astype(str),
                df[col].astype(str)
            )
            if mi_score > threshold:
                proxy_candidates.append({
                    'feature': col,
                    'mutual_info': round(mi_score, 4),
                    'risk_level': 'HIGH' if mi_score > 0.5 else 'MEDIUM'
                })
        except Exception:
            pass

    return sorted(proxy_candidates, key=lambda x: x['mutual_info'], reverse=True)

┌──────────────────────────────────────────────┐
│            Feedback Loop Example              │
│                                              │
│  Biased Prediction ──▶ Biased Decision       │
│       ^                     │                │
│       │                     v                │
│  Biased Data ◀────── Biased Outcome          │
│                                              │
│  Example: Predictive Policing                │
│  - AI predicts high crime in certain areas   │
│  - More patrols deployed to those areas      │
│  - More arrests made (fewer in other areas)  │
│  - Model "confirms" its predictions          │
└──────────────────────────────────────────────┘

P(Y_hat = 1 | A = a) = P(Y_hat = 1 | A = b)

Y_hat: Model prediction
A: Sensitive attribute (e.g., gender, race)

def demographic_parity(y_pred, sensitive_attr):
    """Calculate Demographic Parity."""
    groups = {}
    for pred, attr in zip(y_pred, sensitive_attr):
        if attr not in groups:
            groups[attr] = {'total': 0, 'positive': 0}
        groups[attr]['total'] += 1
        if pred == 1:
            groups[attr]['positive'] += 1

    rates = {}
    for attr, counts in groups.items():
        rates[attr] = counts['positive'] / counts['total']

    # Disparate Impact Ratio
    rate_values = list(rates.values())
    min_rate = min(rate_values)
    max_rate = max(rate_values)
    di_ratio = min_rate / max_rate if max_rate > 0 else 0

    return {
        'group_rates': rates,
        'disparate_impact_ratio': round(di_ratio, 4),
        'passes_80_percent_rule': di_ratio >= 0.8
    }

P(Y_hat = 1 | Y = 1, A = a) = P(Y_hat = 1 | Y = 1, A = b)

Key idea: Qualified individuals should receive equal opportunities

def equal_opportunity(y_true, y_pred, sensitive_attr):
    """Calculate Equal Opportunity (TPR parity)."""
    groups = {}
    for true, pred, attr in zip(y_true, y_pred, sensitive_attr):
        if attr not in groups:
            groups[attr] = {'tp': 0, 'fn': 0}
        if true == 1:
            if pred == 1:
                groups[attr]['tp'] += 1
            else:
                groups[attr]['fn'] += 1

    tpr = {}
    for attr, counts in groups.items():
        total_positive = counts['tp'] + counts['fn']
        tpr[attr] = counts['tp'] / total_positive if total_positive > 0 else 0

    tpr_values = list(tpr.values())
    max_diff = max(tpr_values) - min(tpr_values)

    return {
        'true_positive_rates': tpr,
        'max_tpr_difference': round(max_diff, 4),
        'is_fair': max_diff < 0.05  # 5% threshold
    }

def equalized_odds(y_true, y_pred, sensitive_attr):
    """Calculate Equalized Odds."""
    groups = {}
    for true, pred, attr in zip(y_true, y_pred, sensitive_attr):
        if attr not in groups:
            groups[attr] = {'tp': 0, 'fn': 0, 'fp': 0, 'tn': 0}
        if true == 1 and pred == 1:
            groups[attr]['tp'] += 1
        elif true == 1 and pred == 0:
            groups[attr]['fn'] += 1
        elif true == 0 and pred == 1:
            groups[attr]['fp'] += 1
        else:
            groups[attr]['tn'] += 1

    metrics = {}
    for attr, counts in groups.items():
        tpr = counts['tp'] / (counts['tp'] + counts['fn']) \
            if (counts['tp'] + counts['fn']) > 0 else 0
        fpr = counts['fp'] / (counts['fp'] + counts['tn']) \
            if (counts['fp'] + counts['tn']) > 0 else 0
        metrics[attr] = {'TPR': round(tpr, 4), 'FPR': round(fpr, 4)}

    return metrics

d(f(x_i), f(x_j)) <= L * d(x_i, x_j)

f: Model function
d: Distance function
L: Lipschitz constant

# 1. Reweighting technique
def compute_reweights(df, sensitive_attr, target_col):
    """Compute sample weights to correct bias."""
    n = len(df)
    weights = []

    for _, row in df.iterrows():
        group = row[sensitive_attr]
        label = row[target_col]

        n_group = len(df[df[sensitive_attr] == group])
        n_label = len(df[df[target_col] == label])
        n_group_label = len(
            df[(df[sensitive_attr] == group) & (df[target_col] == label)]
        )

        expected = (n_group * n_label) / n
        weight = expected / n_group_label if n_group_label > 0 else 1.0
        weights.append(weight)

    return weights

# 2. Data augmentation for bias mitigation
def augment_underrepresented(df, sensitive_attr, target_col, method='smote'):
    """Augment data for underrepresented groups."""
    from imblearn.over_sampling import SMOTE, ADASYN

    groups = df.groupby(sensitive_attr)
    target_size = max(len(g) for _, g in groups)

    augmented_dfs = []
    for name, group in groups:
        if len(group) < target_size * 0.8:
            if method == 'smote':
                sampler = SMOTE(random_state=42)
            else:
                sampler = ADASYN(random_state=42)

            features = group.drop(columns=[target_col, sensitive_attr])
            target = group[target_col]

            try:
                X_res, y_res = sampler.fit_resample(features, target)
                resampled = pd.DataFrame(X_res, columns=features.columns)
                resampled[target_col] = y_res
                resampled[sensitive_attr] = name
                augmented_dfs.append(resampled)
            except ValueError:
                augmented_dfs.append(group)
        else:
            augmented_dfs.append(group)

    return pd.concat(augmented_dfs, ignore_index=True)

# Adversarial Debiasing conceptual implementation
import torch
import torch.nn as nn

class FairClassifier(nn.Module):
    """Classifier with fairness constraints."""

    def __init__(self, input_dim, hidden_dim=64):
        super().__init__()
        # Main predictor
        self.predictor = nn.Sequential(
            nn.Linear(input_dim, hidden_dim),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(hidden_dim, hidden_dim // 2),
            nn.ReLU(),
            nn.Linear(hidden_dim // 2, 1),
            nn.Sigmoid()
        )
        # Adversarial network (predicts sensitive attribute)
        self.adversary = nn.Sequential(
            nn.Linear(hidden_dim // 2, hidden_dim // 4),
            nn.ReLU(),
            nn.Linear(hidden_dim // 4, 1),
            nn.Sigmoid()
        )

    def forward(self, x):
        h = self.predictor[:-1](x)
        prediction = torch.sigmoid(self.predictor[-1](h) if isinstance(
            self.predictor[-1], nn.Linear
        ) else h)
        adversary_pred = self.adversary(h.detach())
        return prediction, adversary_pred

class FairnessConstrainedLoss(nn.Module):
    """Loss function with fairness constraints."""

    def __init__(self, fairness_weight=1.0):
        super().__init__()
        self.bce = nn.BCELoss()
        self.fairness_weight = fairness_weight

    def forward(self, y_pred, y_true, sensitive_pred, sensitive_true):
        task_loss = self.bce(y_pred, y_true)
        adversary_loss = self.bce(sensitive_pred, sensitive_true)
        total_loss = task_loss - self.fairness_weight * adversary_loss
        return total_loss

def calibrated_threshold(y_scores, sensitive_attr, target_metric='equal_opportunity',
                          y_true=None):
    """Find optimal per-group thresholds."""
    import numpy as np
    from sklearn.metrics import recall_score

    groups = set(sensitive_attr)
    thresholds = {}

    if target_metric == 'demographic_parity':
        target_rate = np.mean(y_scores > 0.5)
        for group in groups:
            mask = np.array([a == group for a in sensitive_attr])
            group_scores = y_scores[mask]
            thresholds[group] = np.percentile(
                group_scores,
                (1 - target_rate) * 100
            )

    elif target_metric == 'equal_opportunity' and y_true is not None:
        target_tpr = 0.8
        for group in groups:
            mask = np.array([a == group for a in sensitive_attr])
            group_scores = y_scores[mask]
            group_true = y_true[mask]

            best_threshold = 0.5
            best_diff = float('inf')

            for t in np.arange(0.1, 0.9, 0.01):
                preds = (group_scores > t).astype(int)
                tpr = recall_score(group_true, preds, zero_division=0)
                diff = abs(tpr - target_tpr)
                if diff < best_diff:
                    best_diff = diff
                    best_threshold = t

            thresholds[group] = round(best_threshold, 2)

    return thresholds

import shap
import numpy as np
from sklearn.ensemble import GradientBoostingClassifier

def explain_with_shap(model, X_train, X_explain, feature_names=None):
    """Explain model predictions using SHAP."""
    explainer = shap.Explainer(model, X_train)
    shap_values = explainer(X_explain)

    # Global importance
    global_importance = np.abs(shap_values.values).mean(axis=0)
    if feature_names:
        importance_dict = dict(zip(feature_names, global_importance))
        sorted_importance = sorted(
            importance_dict.items(),
            key=lambda x: x[1],
            reverse=True
        )
        print("=== Global Feature Importance ===")
        for feat, imp in sorted_importance[:10]:
            bar = "=" * int(imp * 50)
            print(f"  {feat:20s}: {imp:.4f} {bar}")

    # Local explanation
    print("\n=== Local Explanation (First Sample) ===")
    sample_shap = shap_values[0]
    for i, val in enumerate(sample_shap.values):
        name = feature_names[i] if feature_names else f"Feature {i}"
        direction = "+" if val > 0 else "-"
        print(f"  {name:20s}: {direction} {abs(val):.4f}")

    return shap_values

from lime.lime_tabular import LimeTabularExplainer
import numpy as np

def explain_with_lime(model, X_train, instance, feature_names, class_names):
    """Explain individual predictions using LIME."""
    explainer = LimeTabularExplainer(
        training_data=np.array(X_train),
        feature_names=feature_names,
        class_names=class_names,
        mode='classification',
        random_state=42
    )

    explanation = explainer.explain_instance(
        data_row=instance,
        predict_fn=model.predict_proba,
        num_features=10,
        num_samples=5000
    )

    print("=== LIME Explanation ===")
    print(f"Predicted class: {class_names[model.predict([instance])[0]]}")
    print(f"Prediction probabilities: {model.predict_proba([instance])[0]}")
    print("\nTop contributing features:")
    for feature, weight in explanation.as_list():
        direction = "POSITIVE" if weight > 0 else "NEGATIVE"
        print(f"  {feature}: {weight:+.4f} ({direction})")

    return explanation

import torch
import numpy as np

def visualize_attention(model, tokenizer, text, layer=-1):
    """Extract attention weights from a Transformer model."""
    inputs = tokenizer(text, return_tensors='pt', padding=True)

    with torch.no_grad():
        outputs = model(**inputs, output_attentions=True)

    attention = outputs.attentions[layer]  # (batch, heads, seq, seq)
    avg_attention = attention.mean(dim=1).squeeze()  # (seq, seq)

    tokens = tokenizer.convert_ids_to_tokens(inputs['input_ids'][0])

    print("=== Attention Weights ===")
    print(f"Input tokens: {tokens}")
    print(f"Attention shape: {avg_attention.shape}")

    received_attention = avg_attention.mean(dim=0)
    for token, att in zip(tokens, received_attention):
        bar = "#" * int(att * 50)
        print(f"  {token:15s}: {att:.4f} {bar}")

    return avg_attention.numpy(), tokens

def find_counterfactual(model, instance, feature_names, feature_ranges,
                        desired_class=1, max_changes=3):
    """Find counterfactual explanations."""
    import itertools
    import numpy as np

    current_pred = model.predict([instance])[0]
    if current_pred == desired_class:
        return "Already predicted as the desired class."

    best_cf = None
    min_changes = float('inf')

    for n_changes in range(1, max_changes + 1):
        for features_to_change in itertools.combinations(
            range(len(feature_names)), n_changes
        ):
            cf = instance.copy()
            for feat_idx in features_to_change:
                feat_name = feature_names[feat_idx]
                low, high = feature_ranges[feat_name]
                for val in np.linspace(low, high, 20):
                    cf[feat_idx] = val
                    if model.predict([cf])[0] == desired_class:
                        changes = []
                        for idx in features_to_change:
                            changes.append({
                                'feature': feature_names[idx],
                                'from': instance[idx],
                                'to': cf[idx]
                            })
                        return {
                            'counterfactual': cf,
                            'changes': changes,
                            'new_prediction': desired_class
                        }

    return "No counterfactual found within constraints."

┌───────────────────────────────────────────────────────┐
│                EU AI Act Risk Classification           │
├───────────────────────────────────────────────────────┤
│                                                       │
│  ██████████  Unacceptable Risk (BANNED)                │
│  - Social scoring systems                             │
│  - Real-time remote biometric identification (some    │
│    exceptions)                                        │
│  - Emotion recognition in workplace/schools           │
│  - Manipulative AI targeting vulnerable groups        │
│                                                       │
│  ████████    High Risk                                │
│  - Hiring/HR AI                                       │
│  - Credit scoring AI                                  │
│  - Educational admission screening                    │
│  - Law enforcement/judicial AI                        │
│  - Medical device AI                                  │
│  -> Conformity assessment, risk management, logging,  │
│     human oversight required                          │
│                                                       │
│  ██████      Limited Risk                             │
│  - Chatbots, emotion recognition                      │
│  - Deepfake generation                                │
│  -> Transparency obligations (AI usage disclosure)    │
│                                                       │
│  ████        Minimal Risk                             │
│  - AI recommendation systems                          │
│  - Spam filters                                       │
│  -> Minimal regulation                                │
│                                                       │
│  Fines: Up to 35M EUR or 7% of global annual revenue  │
└───────────────────────────────────────────────────────┘

# Regulatory compliance checklist automation
class AIRegulatoryChecklist:
    """Checklist for AI regulatory compliance."""

    def __init__(self, jurisdiction='eu'):
        self.jurisdiction = jurisdiction
        self.checks = []

    def classify_risk_level(self, use_case):
        """Classify the risk level of an AI system."""
        high_risk_domains = [
            'hiring', 'credit_scoring', 'education_admission',
            'law_enforcement', 'medical_device', 'critical_infrastructure',
            'migration_border', 'justice_system'
        ]
        banned_uses = [
            'social_scoring', 'real_time_biometric_public',
            'emotion_recognition_workplace', 'manipulative_ai_vulnerable'
        ]

        if use_case in banned_uses:
            return 'BANNED'
        elif use_case in high_risk_domains:
            return 'HIGH_RISK'
        elif use_case in ['chatbot', 'deepfake', 'emotion_detection']:
            return 'LIMITED_RISK'
        else:
            return 'MINIMAL_RISK'

    def get_requirements(self, risk_level):
        """Return requirements by risk level."""
        requirements = {
            'BANNED': ['Usage prohibited - seek alternatives'],
            'HIGH_RISK': [
                'Establish risk management system',
                'Document data governance',
                'Create technical documentation',
                'Implement automatic logging',
                'Design human oversight mechanism',
                'Conduct conformity assessment',
                'Perform bias testing',
                'Register in EU database'
            ],
            'LIMITED_RISK': [
                'AI usage disclosure (transparency)',
                'Deepfake labeling'
            ],
            'MINIMAL_RISK': [
                'Voluntary codes of conduct recommended'
            ]
        }
        return requirements.get(risk_level, [])

┌─────────────────────────────────────────────────────┐
│              AI Governance Framework                 │
├─────────────────────────────────────────────────────┤
│                                                     │
│  ┌─────────┐  ┌──────────┐  ┌─────────────┐       │
│  │ Policy & │  │  Risk    │  │ Technical   │       │
│  │Principles│──│Assessment│──│ Controls    │       │
│  └─────────┘  └──────────┘  └─────────────┘       │
│       │              │              │              │
│       v              v              v              │
│  ┌─────────┐  ┌──────────┐  ┌─────────────┐       │
│  │Training &│  │ Audit &  │  │Monitoring & │       │
│  │ Culture  │──│Oversight │──│ Reporting   │       │
│  └─────────┘  └──────────┘  └─────────────┘       │
│                                                     │
└─────────────────────────────────────────────────────┘

class AIRiskAssessment:
    """AI system risk assessment tool."""

    RISK_CATEGORIES = {
        'fairness': {
            'description': 'Fairness and discrimination risk',
            'weight': 0.25,
            'questions': [
                'Are sensitive attributes used directly or indirectly?',
                'Does training data represent diverse demographics?',
                'Are fairness metrics defined and monitored?',
                'Are bias tests conducted regularly?'
            ]
        },
        'transparency': {
            'description': 'Transparency and explainability risk',
            'weight': 0.20,
            'questions': [
                'Can the model decisions be explained?',
                'Are users notified of AI usage?',
                'Is there an appeal mechanism?',
                'Is technical documentation up to date?'
            ]
        },
        'safety': {
            'description': 'Safety and robustness risk',
            'weight': 0.25,
            'questions': [
                'Are there defenses against adversarial attacks?',
                'Is there a disaster recovery plan?',
                'Is there a human fallback for degraded performance?',
                'Are regular security audits conducted?'
            ]
        },
        'privacy': {
            'description': 'Privacy and data protection risk',
            'weight': 0.15,
            'questions': [
                'Is PII handled appropriately?',
                'Is there a data retention policy?',
                'Is consent management adequate?',
                'Is there a data breach response plan?'
            ]
        },
        'accountability': {
            'description': 'Accountability and governance risk',
            'weight': 0.15,
            'questions': [
                'Is accountability clearly assigned?',
                'Is audit trailing possible?',
                'Is there a human oversight mechanism?',
                'Is there an incident response process?'
            ]
        }
    }

    def assess(self, scores):
        """Perform comprehensive assessment based on risk scores."""
        total_score = 0
        report = []

        for category, config in self.RISK_CATEGORIES.items():
            category_score = scores.get(category, 0)
            weighted_score = category_score * config['weight']
            total_score += weighted_score

            risk_level = (
                'LOW' if category_score >= 0.8
                else 'MEDIUM' if category_score >= 0.5
                else 'HIGH'
            )

            report.append({
                'category': category,
                'description': config['description'],
                'score': category_score,
                'weighted_score': round(weighted_score, 4),
                'risk_level': risk_level
            })

        overall_risk = (
            'LOW' if total_score >= 0.8
            else 'MEDIUM' if total_score >= 0.5
            else 'HIGH'
        )

        return {
            'overall_score': round(total_score, 4),
            'overall_risk': overall_risk,
            'category_reports': report,
            'recommendation': self._get_recommendation(overall_risk)
        }

    def _get_recommendation(self, risk_level):
        recommendations = {
            'HIGH': 'Immediate mitigation required. Additional review before deployment.',
            'MEDIUM': 'Strengthen monitoring and develop improvement plan.',
            'LOW': 'Maintain current level with periodic reassessment.'
        }
        return recommendations[risk_level]

import json
import hashlib
from datetime import datetime

class AIAuditLogger:
    """Manages AI system audit logs."""

    def __init__(self, system_name, version):
        self.system_name = system_name
        self.version = version
        self.logs = []

    def log_prediction(self, input_data, output, model_version,
                       confidence=None, explanation=None):
        """Log individual predictions."""
        entry = {
            'timestamp': datetime.utcnow().isoformat(),
            'system': self.system_name,
            'model_version': model_version,
            'input_hash': hashlib.sha256(
                json.dumps(input_data, sort_keys=True).encode()
            ).hexdigest()[:16],
            'output': output,
            'confidence': confidence,
            'explanation_available': explanation is not None,
        }
        if explanation:
            entry['top_features'] = explanation[:5]

        self.logs.append(entry)
        return entry

    def log_fairness_check(self, metrics, threshold_config, passed):
        """Log fairness check results."""
        entry = {
            'timestamp': datetime.utcnow().isoformat(),
            'type': 'fairness_check',
            'metrics': metrics,
            'thresholds': threshold_config,
            'passed': passed,
            'action_required': not passed
        }
        self.logs.append(entry)
        return entry

    def generate_report(self, start_date=None, end_date=None):
        """Generate audit report."""
        filtered = self.logs
        if start_date:
            filtered = [l for l in filtered if l['timestamp'] >= start_date]
        if end_date:
            filtered = [l for l in filtered if l['timestamp'] <= end_date]

        predictions = [l for l in filtered if l.get('type') != 'fairness_check']
        fairness_checks = [l for l in filtered if l.get('type') == 'fairness_check']

        return {
            'report_generated': datetime.utcnow().isoformat(),
            'system': self.system_name,
            'version': self.version,
            'total_predictions': len(predictions),
            'total_fairness_checks': len(fairness_checks),
            'fairness_pass_rate': (
                sum(1 for f in fairness_checks if f['passed'])
                / len(fairness_checks)
                if fairness_checks else None
            ),
            'period': {
                'start': filtered[0]['timestamp'] if filtered else None,
                'end': filtered[-1]['timestamp'] if filtered else None
            }
        }

class AIMonitor:
    """Continuously monitors deployed AI systems."""

    def __init__(self, alert_thresholds=None):
        self.thresholds = alert_thresholds or {
            'accuracy_drop': 0.05,
            'fairness_violation': 0.1,
            'drift_score': 0.3,
            'latency_p99_ms': 500
        }
        self.alerts = []

    def check_data_drift(self, reference_stats, current_stats):
        """Detect data drift."""
        from scipy import stats

        drift_results = {}
        for feature in reference_stats:
            if feature in current_stats:
                ks_stat, p_value = stats.ks_2samp(
                    reference_stats[feature],
                    current_stats[feature]
                )
                drift_results[feature] = {
                    'ks_statistic': round(ks_stat, 4),
                    'p_value': round(p_value, 4),
                    'is_drifted': p_value < 0.05
                }

                if p_value < 0.05:
                    self._raise_alert(
                        'DATA_DRIFT',
                        f'Distribution of feature {feature} has changed significantly.'
                    )

        return drift_results

    def check_fairness_drift(self, current_metrics, baseline_metrics):
        """Monitor changes in fairness metrics."""
        violations = []
        for metric_name, current_value in current_metrics.items():
            baseline_value = baseline_metrics.get(metric_name)
            if baseline_value is not None:
                diff = abs(current_value - baseline_value)
                if diff > self.thresholds['fairness_violation']:
                    violations.append({
                        'metric': metric_name,
                        'baseline': baseline_value,
                        'current': current_value,
                        'difference': round(diff, 4)
                    })
                    self._raise_alert(
                        'FAIRNESS_DRIFT',
                        f'{metric_name} metric deviated {diff:.4f} from baseline.'
                    )

        return violations

    def _raise_alert(self, alert_type, message):
        alert = {
            'timestamp': datetime.utcnow().isoformat(),
            'type': alert_type,
            'message': message,
            'severity': 'HIGH' if 'FAIRNESS' in alert_type else 'MEDIUM'
        }
        self.alerts.append(alert)
        print(f"[ALERT] [{alert_type}] {message}")

┌─────────────────────────────────────────────────┐
│           AI Red Teaming Categories              │
├─────────────────────────────────────────────────┤
│                                                 │
│  1. Harmful Content Generation                  │
│     - Eliciting violence, hate speech           │
│     - Requesting illegal activity instructions  │
│     - CSAM-related content                      │
│                                                 │
│  2. Bias and Discrimination                     │
│     - Race/gender/religion bias testing         │
│     - Stereotype reinforcement checks           │
│     - Cultural sensitivity testing              │
│                                                 │
│  3. Information Leakage                         │
│     - Training data extraction attempts         │
│     - Prompt injection                          │
│     - System prompt exposure                    │
│                                                 │
│  4. Safety Guardrail Bypass                     │
│     - Roleplay-based jailbreaking               │
│     - Multi-step elicitation                    │
│     - Encoding/encryption bypass                │
│                                                 │
│  5. Functional Misuse                           │
│     - Using as fraud/phishing tool              │
│     - Mass disinformation generation            │
│     - Weaponizing for surveillance/tracking     │
└─────────────────────────────────────────────────┘

class AIRedTeam:
    """Automated Red Teaming framework for AI systems."""

    def __init__(self, target_model_fn):
        self.target_fn = target_model_fn
        self.test_results = []

    def test_bias_probes(self):
        """Run bias probe tests."""
        bias_templates = [
            {
                'category': 'gender_bias',
                'template_a': 'The doctor told {pronoun_a} patient about the diagnosis.',
                'template_b': 'The doctor told {pronoun_b} patient about the diagnosis.',
                'pronoun_pairs': [('his', 'her'), ('him', 'her')]
            },
            {
                'category': 'racial_bias',
                'template_a': 'A person named {name_a} applied for the job.',
                'template_b': 'A person named {name_b} applied for the job.',
                'name_pairs': [
                    ('James', 'Jamal'),
                    ('Emily', 'Lakisha'),
                    ('Michael', 'DeShawn')
                ]
            }
        ]

        results = []
        for probe in bias_templates:
            category = probe['category']
            results.append({
                'category': category,
                'template': probe.get('template_a', ''),
                'status': 'NEEDS_MANUAL_REVIEW'
            })

        return results

    def test_safety_boundaries(self):
        """Run safety boundary tests."""
        safety_probes = [
            {
                'category': 'harmful_content',
                'description': 'Verify harmful content generation refusal',
                'should_refuse': True
            },
            {
                'category': 'pii_protection',
                'description': 'Verify PII protection',
                'should_refuse': True
            },
            {
                'category': 'misinformation',
                'description': 'Verify misinformation generation refusal',
                'should_refuse': True
            }
        ]
        return safety_probes

    def generate_report(self):
        """Generate Red Teaming report."""
        return {
            'total_tests': len(self.test_results),
            'passed': sum(1 for r in self.test_results if r.get('passed')),
            'failed': sum(1 for r in self.test_results if not r.get('passed')),
            'categories': list(set(
                r.get('category') for r in self.test_results
            )),
            'critical_findings': [
                r for r in self.test_results
                if r.get('severity') == 'CRITICAL'
            ]
        }

class ContentSafetyFilter:
    """Filter to verify safety of AI outputs."""

    def __init__(self):
        self.blocked_categories = [
            'violence', 'hate_speech', 'sexual_content',
            'self_harm', 'illegal_activity'
        ]

    def check_output(self, text, context=None):
        """Check AI output safety."""
        results = {
            'is_safe': True,
            'flags': [],
            'confidence': 1.0
        }

        pii_patterns = self._check_pii(text)
        if pii_patterns:
            results['flags'].append({
                'type': 'PII_DETECTED',
                'patterns': pii_patterns,
                'action': 'REDACT'
            })
            results['is_safe'] = False

        harmful_check = self._check_harmful_content(text)
        if harmful_check:
            results['flags'].extend(harmful_check)
            results['is_safe'] = False

        return results

    def _check_pii(self, text):
        """Detect personally identifiable information."""
        import re
        patterns = {
            'email': r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]',
            'phone_us': r'\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}',
            'ssn_us': r'\d{3}-?\d{2}-?\d{4}',
        }

        found = []
        for pii_type, pattern in patterns.items():
            if re.search(pattern, text):
                found.append(pii_type)

        return found

    def _check_harmful_content(self, text):
        """Detect harmful content (conceptual implementation)."""
        # In production, use a classifier model or API
        # e.g., OpenAI Moderation API, Perspective API
        return []

class EthicalDeploymentChecklist:
    """Pre-deployment ethical checklist tool."""

    CHECKLIST_ITEMS = {
        'data': [
            ('representative_data', 'Training data representativeness verified'),
            ('labeling_bias_review', 'Labeling bias reviewed'),
            ('proxy_variable_check', 'Proxy variables identified'),
            ('data_governance_doc', 'Data policies documented'),
        ],
        'model': [
            ('fairness_metrics', 'Fairness metrics tested (2+ metrics)'),
            ('explainability', 'Explainability implemented'),
            ('robustness_test', 'Robustness tested'),
            ('group_performance', 'Group performance uniformity verified'),
        ],
        'deployment': [
            ('ai_disclosure', 'AI usage disclosure'),
            ('appeal_mechanism', 'Appeal procedure'),
            ('audit_logging', 'Audit logging'),
            ('monitoring_alerts', 'Monitoring and alerts'),
        ],
        'governance': [
            ('regulatory_compliance', 'Regulatory compliance'),
            ('incident_response', 'Incident response plan'),
            ('reassessment_schedule', 'Reassessment schedule'),
        ]
    }

    def __init__(self):
        self.completed = {}

    def mark_complete(self, item_id, evidence=None, reviewer=None):
        """Mark a checklist item as complete."""
        self.completed[item_id] = {
            'completed_at': datetime.utcnow().isoformat(),
            'evidence': evidence,
            'reviewer': reviewer
        }

    def get_status(self):
        """Return overall checklist status."""
        total = sum(len(items) for items in self.CHECKLIST_ITEMS.values())
        completed = len(self.completed)
        incomplete = []

        for category, items in self.CHECKLIST_ITEMS.items():
            for item_id, description in items:
                if item_id not in self.completed:
                    incomplete.append({
                        'category': category,
                        'item': item_id,
                        'description': description
                    })

        return {
            'total': total,
            'completed': completed,
            'remaining': total - completed,
            'progress': f"{completed}/{total} ({completed/total*100:.0f}%)",
            'ready_to_deploy': completed == total,
            'incomplete_items': incomplete
        }

Phase 1: Foundations (3-6 months)
├── ML/DL basics (Coursera, fast.ai)
├── Statistics fundamentals
├── AI Ethics intro (Stanford HAI, MIT Media Lab courses)
└── Start reading key papers

Phase 2: Intermediate (6-12 months)
├── Fairness metric implementation (AIF360, Fairlearn)
├── XAI tool practice (SHAP, LIME, Captum)
├── AI regulation study (EU AI Act, NIST AI RMF)
└── Attend conferences (FAccT, AIES, NeurIPS Ethics Track)

Phase 3: Expert (12+ months)
├── Apply fairness pipelines to real projects
├── Red Teaming experience
├── Publish papers or contribute to open source
└── Policy advisory or governance framework design