7.1 Workforce Enablement

7.1.1 Training for Developers: Coding for Fairness & Security

Development teams are at the frontline of responsible AI implementation. Their technical decisions—from data preprocessing to model architecture to deployment configuration—directly determine whether AI systems operate fairly, securely, and reliably. Developer training must combine theoretical understanding with practical skills that integrate into existing workflows.

Developer Competency Framework

Fairness Engineering Curriculum

"""
Fairness Engineering Lab 1: Training Data Bias Detection
Learning Objective: Identify and quantify representation bias
"""
import pandas as pd
import numpy as np
from aif360.datasets import StandardDataset
from aif360.metrics import BinaryLabelDatasetMetric

class TrainingDataBiasAnalyzer:
    """
    Tool for detecting representation and historical bias in training data.
    Developers should use this during data curation phase.
    """
    
    def __init__(self, df: pd.DataFrame, label_col: str, 
                 protected_attrs: list, favorable_label: int = 1):
        self.df = df
        self.label_col = label_col
        self.protected_attrs = protected_attrs
        self.favorable_label = favorable_label
        
    def analyze_representation(self) -> dict:
        """
        Analyze representation of protected groups in training data.
        Compare to population benchmarks (e.g., census data).
        """
        results = {}
        
        for attr in self.protected_attrs:
            group_counts = self.df[attr].value_counts(normalize=True)
            results[attr] = {
                'distribution': group_counts.to_dict(),
                'representation_ratio': group_counts.min() / group_counts.max(),
                'underrepresented_groups': group_counts[
                    group_counts < 0.1  # Flag groups with <10% representation
                ].index.tolist()
            }
            
        return results
    
    def analyze_label_distribution(self) -> dict:
        """
        Analyze positive/negative label distribution across groups.
        Historical bias manifests as different base rates.
        """
        results = {}
        
        overall_positive_rate = (
            self.df[self.label_col] == self.favorable_label
        ).mean()
        
        for attr in self.protected_attrs:
            group_rates = self.df.groupby(attr)[self.label_col].apply(
                lambda x: (x == self.favorable_label).mean()
            )
            
            # Calculate disparate impact ratio
            min_rate = group_rates.min()
            max_rate = group_rates.max()
            di_ratio = min_rate / max_rate if max_rate > 0 else 0
            
            results[attr] = {
                'group_positive_rates': group_rates.to_dict(),
                'overall_positive_rate': overall_positive_rate,
                'disparate_impact_ratio': di_ratio,
                'four_fifths_rule_violated': di_ratio < 0.8,
                'disadvantaged_groups': group_rates[
                    group_rates < overall_positive_rate * 0.8
                ].index.tolist()
            }
            
        return results
    
    def detect_proxy_variables(self, threshold: float = 0.7) -> dict:
        """
        Identify variables that may serve as proxies for protected attributes.
        High correlation suggests potential indirect discrimination pathway.
        """
        proxies = {}
        
        for attr in self.protected_attrs:
            # Convert to numeric for correlation analysis
            attr_numeric = pd.factorize(self.df[attr])[0]
            
            correlations = {}
            for col in self.df.columns:
                if col not in self.protected_attrs and col != self.label_col:
                    try:
                        col_numeric = pd.to_numeric(
                            self.df[col], errors='coerce'
                        ).fillna(0)
                        corr = np.corrcoef(attr_numeric, col_numeric)[0, 1]
                        if abs(corr) >= threshold:
                            correlations[col] = round(corr, 3)
                    except:
                        continue
                        
            if correlations:
                proxies[attr] = {
                    'high_correlation_variables': correlations,
                    'recommendation': 'Review these variables for potential removal or mitigation'
                }
                
        return proxies
    
    def generate_bias_report(self) -> dict:
        """
        Generate comprehensive bias assessment report for model card.
        """
        return {
            'representation_analysis': self.analyze_representation(),
            'label_distribution_analysis': self.analyze_label_distribution(),
            'proxy_variable_analysis': self.detect_proxy_variables(),
            'overall_risk_assessment': self._calculate_overall_risk(),
            'recommended_mitigations': self._generate_recommendations()
        }
    
    def _calculate_overall_risk(self) -> str:
        label_analysis = self.analyze_label_distribution()
        
        violations = sum(
            1 for attr in label_analysis.values() 
            if attr['four_fifths_rule_violated']
        )
        
        if violations == 0:
            return "LOW"
        elif violations < len(self.protected_attrs) / 2:
            return "MEDIUM"
        else:
            return "HIGH"
    
    def _generate_recommendations(self) -> list:
        recommendations = []
        
        rep_analysis = self.analyze_representation()
        for attr, data in rep_analysis.items():
            if data['underrepresented_groups']:
                recommendations.append(
                    f"Consider oversampling or collecting more data for "
                    f"underrepresented {attr} groups: {data['underrepresented_groups']}"
                )
                
        label_analysis = self.analyze_label_distribution()
        for attr, data in label_analysis.items():
            if data['four_fifths_rule_violated']:
                recommendations.append(
                    f"Four-fifths rule violated for {attr}. Consider: "
                    f"1) Reweighting samples, 2) Applying bias mitigation during training, "
                    f"3) Post-processing calibration"
                )
                
        proxy_analysis = self.detect_proxy_variables()
        for attr, data in proxy_analysis.items():
            recommendations.append(
                f"Review proxy variables for {attr}: {list(data['high_correlation_variables'].keys())}. "
                f"Consider removal or fairness constraints during training."
            )
            
        return recommendations

# Example usage in development workflow
if __name__ == "__main__":
    # Load training data
    df = pd.read_csv("training_data.csv")
    
    # Initialize analyzer with protected attributes
    analyzer = TrainingDataBiasAnalyzer(
        df=df,
        label_col='approved',
        protected_attrs=['gender', 'race', 'age_group'],
        favorable_label=1
    )
    
    # Generate report for model card
    bias_report = analyzer.generate_bias_report()
    
    # Check if deployment should proceed
    if bias_report['overall_risk_assessment'] == "HIGH":
        print("⚠️ HIGH BIAS RISK: Do not proceed without mitigation")
        print("Required mitigations:")
        for rec in bias_report['recommended_mitigations']:
            print(f"  - {rec}")
    else:
        print("✓ Bias risk acceptable for next phase")

"""
AI Security Lab: Secure Model Serving Implementation
Learning Objective: Build defense-in-depth for model API
"""
from typing import Optional, Dict, Any
import hashlib
import time
import logging
from dataclasses import dataclass
from functools import wraps

@dataclass
class SecurityConfig:
    """Security configuration for model serving."""
    rate_limit_requests: int = 100  # Per minute
    rate_limit_window: int = 60  # Seconds
    max_input_length: int = 10000
    confidence_masking: bool = True
    confidence_precision: int = 2  # Decimal places
    query_logging: bool = True
    anomaly_detection: bool = True
    
class SecureModelServer:
    """
    Defense-in-depth implementation for secure model serving.
    Implements protections against common ML attacks.
    """
    
    def __init__(self, model, config: SecurityConfig):
        self.model = model
        self.config = config
        self.request_counts: Dict[str, list] = {}
        self.query_log: list = []
        
    def rate_limit(self, client_id: str) -> bool:
        """
        Protect against model extraction via query flooding.
        Defense: ML05 Model Theft
        """
        current_time = time.time()
        
        if client_id not in self.request_counts:
            self.request_counts[client_id] = []
            
        # Remove old requests outside window
        self.request_counts[client_id] = [
            t for t in self.request_counts[client_id]
            if current_time - t < self.config.rate_limit_window
        ]
        
        if len(self.request_counts[client_id]) >= self.config.rate_limit_requests:
            logging.warning(f"Rate limit exceeded for {client_id}")
            return False
            
        self.request_counts[client_id].append(current_time)
        return True
    
    def validate_input(self, input_data: Any) -> tuple[bool, str]:
        """
        Validate and sanitize input to prevent adversarial attacks.
        Defense: ML01 Input Manipulation
        """
        # Check input length
        if isinstance(input_data, str):
            if len(input_data) > self.config.max_input_length:
                return False, "Input exceeds maximum length"
                
        # Check for common attack patterns
        if isinstance(input_data, str):
            # Basic prompt injection detection for LLMs
            injection_patterns = [
                "ignore previous instructions",
                "disregard all prior",
                "you are now",
                "new instruction:",
            ]
            input_lower = input_data.lower()
            for pattern in injection_patterns:
                if pattern in input_lower:
                    logging.warning(f"Potential prompt injection detected")
                    return False, "Input contains prohibited patterns"
                    
        # Add domain-specific validation here
        return True, "Valid"
    
    def mask_confidence(self, prediction: Dict) -> Dict:
        """
        Reduce precision of confidence scores to hinder model extraction.
        Defense: ML03 Model Inversion, ML05 Model Theft
        """
        if not self.config.confidence_masking:
            return prediction
            
        masked = prediction.copy()
        
        if 'confidence' in masked:
            masked['confidence'] = round(
                masked['confidence'], 
                self.config.confidence_precision
            )
            
        if 'probabilities' in masked:
            masked['probabilities'] = {
                k: round(v, self.config.confidence_precision)
                for k, v in masked['probabilities'].items()
            }
            
        return masked
    
    def detect_anomalous_queries(self, client_id: str, input_data: Any) -> bool:
        """
        Detect query patterns indicative of model extraction.
        Defense: ML05 Model Theft
        """
        if not self.config.anomaly_detection:
            return False
            
        # Get recent queries from this client
        client_queries = [
            q for q in self.query_log 
            if q['client_id'] == client_id
        ][-100:]  # Last 100 queries
        
        if len(client_queries) < 10:
            return False
            
        # Check for systematic exploration patterns
        # (This is simplified - production would use ML-based detection)
        
        # Pattern 1: Rapid similar queries (gradient estimation)
        recent_times = [q['timestamp'] for q in client_queries[-10:]]
        if len(recent_times) > 1:
            avg_interval = (recent_times[-1] - recent_times[0]) / len(recent_times)
            if avg_interval < 0.1:  # Less than 100ms between queries
                logging.warning(f"Anomalous query rate from {client_id}")
                return True
                
        return False
    
    def log_query(self, client_id: str, input_hash: str, 
                  output_hash: str, latency: float):
        """
        Log queries for audit trail and anomaly detection.
        """
        if self.config.query_logging:
            self.query_log.append({
                'timestamp': time.time(),
                'client_id': client_id,
                'input_hash': input_hash,
                'output_hash': output_hash,
                'latency': latency
            })
            
            # Trim old logs
            if len(self.query_log) > 100000:
                self.query_log = self.query_log[-50000:]
    
    def predict(self, input_data: Any, client_id: str) -> Dict:
        """
        Secure prediction endpoint with defense-in-depth.
        """
        start_time = time.time()
        
        # Defense 1: Rate limiting
        if not self.rate_limit(client_id):
            return {
                'error': 'Rate limit exceeded',
                'retry_after': self.config.rate_limit_window
            }
            
        # Defense 2: Input validation
        valid, message = self.validate_input(input_data)
        if not valid:
            return {'error': message}
            
        # Defense 3: Anomaly detection
        if self.detect_anomalous_queries(client_id, input_data):
            logging.warning(f"Blocking anomalous client: {client_id}")
            return {'error': 'Unusual query pattern detected'}
            
        # Generate prediction
        try:
            raw_prediction = self.model.predict(input_data)
        except Exception as e:
            logging.error(f"Prediction error: {e}")
            return {'error': 'Prediction failed'}
            
        # Defense 4: Confidence masking
        masked_prediction = self.mask_confidence(raw_prediction)
        
        # Log for audit
        latency = time.time() - start_time
        input_hash = hashlib.sha256(str(input_data).encode()).hexdigest()[:16]
        output_hash = hashlib.sha256(str(masked_prediction).encode()).hexdigest()[:16]
        self.log_query(client_id, input_hash, output_hash, latency)
        
        return masked_prediction

# Decorator for adding security to existing endpoints
def secure_endpoint(config: SecurityConfig):
    """Decorator to add security controls to model endpoints."""
    def decorator(func):
        server = None
        
        @wraps(func)
        def wrapper(input_data, client_id: str = "anonymous", *args, **kwargs):
            nonlocal server
            
            # Lazy initialization
            if server is None:
                # Create server with the function as the model
                class FuncModel:
                    def predict(self, x):
                        return func(x, *args, **kwargs)
                server = SecureModelServer(FuncModel(), config)
                
            return server.predict(input_data, client_id)
        return wrapper
    return decorator

Developer Certification Program

7.1.2 Training for Executives: Interpreting AI Risk

Executive leadership sets the tone for responsible AI adoption. Board members, C-suite executives, and senior leaders must understand AI risk well enough to make informed strategic decisions, allocate appropriate resources, and hold the organization accountable. Executive training focuses on governance, risk interpretation, and strategic decision-making rather than technical implementation.

Executive AI Literacy Program

Module 1: AI Fundamentals for Leaders

Duration: 3 hours | Format: Executive briefing with discussion

Module 2: Regulatory Landscape & Compliance Obligations

Duration: 3 hours | Format: Expert briefing with case studies

Module 3: Reading AI Risk Reports

Duration: 2 hours | Format: Workshop with real reports

Module 4: Crisis Leadership for AI Incidents

Duration: 2 hours | Format: Tabletop exercise

Executive Certification & Ongoing Education

7.1.3 Training for General Staff: Safe Use of Generative AI Tools

The proliferation of generative AI tools like ChatGPT, Claude, Copilot, and Gemini means that virtually every employee now has access to powerful AI capabilities. Without proper training, well-intentioned employees may inadvertently expose confidential data, violate copyright, spread misinformation, or create compliance violations. General staff training focuses on practical, actionable guidance for safe and productive AI use.

The Shadow AI Challenge

General Staff Training Curriculum

Module 1: AI Basics & Organizational Policy

Module 2: Data Protection & Confidentiality

❌ UNSAFE: "Summarize this customer complaint from John Smith at Acme Corp 
   about our product failure that might lead to litigation"

✅ SAFE: "Provide a template for responding to a customer complaint 
   about a product quality issue"

---

❌ UNSAFE: "Here's our Q3 financial report, analyze the trends"

✅ SAFE: "What financial metrics should I analyze when reviewing 
   quarterly performance for a SaaS company?"

---

❌ UNSAFE: "Review this employment contract for our new VP of Sales 
   with $500k salary and explain the non-compete"

✅ SAFE: "What are typical components of an executive employment 
   agreement I should understand?"

Module 3: Effective & Safe Prompting

Module 4: Role-Specific Use Cases

Module 5: Quality Assurance & Verification

Training Delivery & Compliance

Implementation Roadmap