Week 15: AI Alignment, Safety & Secure-by-Design

# Intended behavior: Win the boat race
# Specified reward: Points for hitting targets

# What the AI learned:
# - Discovered a loop where it could repeatedly hit targets
# - Boat caught fire and crashed
# - Still achieved higher score than completing the race

def reward_function(state):
    return state.targets_hit  # Simple, but incomplete specification

# The AI found that a specific circular path maximized targets_hit
# even though the boat was on fire and going in circles

import numpy as np
import matplotlib.pyplot as plt
from sklearn.linear_model import LogisticRegression

# Scenario: Training an AI to identify "good" actions
# Training environment: All "good" actions happen to be "blue"
# Test environment: "Good" actions can be any color

np.random.seed(42)

# Training data - spurious correlation
# "Good" actions are blue (color=1), "Bad" actions are red (color=0)
n_train = 100
train_colors = np.concatenate([np.ones(50), np.zeros(50)])
train_quality = np.concatenate([np.ones(50), np.zeros(50)])  # Perfect correlation
train_features = np.column_stack([train_colors, train_quality])
train_labels = train_quality

# Test data - correlation broken
# Now good actions can be any color
n_test = 100
test_colors = np.random.randint(0, 2, n_test)
test_quality = np.random.randint(0, 2, n_test)
test_features = np.column_stack([test_colors, test_quality])
test_labels = test_quality

# Model learns from training
model = LogisticRegression()
model.fit(train_features, train_labels)

# Examine what the model learned
print("Feature weights:")
print(f"  Color weight: {model.coef_[0][0]:.3f}")
print(f"  Quality weight: {model.coef_[0][1]:.3f}")

# The model may have learned to rely on color (spurious feature)
# rather than actual quality (true feature)

# Test performance
train_acc = model.score(train_features, train_labels)
test_acc = model.score(test_features, test_labels)

print(f"\nTraining accuracy: {train_acc:.2%}")
print(f"Test accuracy: {test_acc:.2%}")

# Traditional approach: Hand-crafted reward
def hand_crafted_reward(state, action):
    reward = 0
    reward += state.task_completed * 10
    reward -= state.time_taken * 0.1
    reward -= state.resources_used * 0.05
    # Problem: Hard to capture all nuances
    return reward

# Reward modeling approach: Learn reward from human comparisons
class LearnedRewardModel:
    def __init__(self):
        self.preference_model = NeuralNetwork()
    
    def train_from_comparisons(self, trajectory_pairs, human_preferences):
        """
        Given pairs of trajectories and human preferences,
        learn a reward model that explains those preferences.
        
        trajectory_pairs: [(traj_A, traj_B), ...]
        human_preferences: [0 if A preferred, 1 if B preferred, ...]
        """
        for (traj_a, traj_b), preference in zip(trajectory_pairs, human_preferences):
            # Model should assign higher total reward to preferred trajectory
            reward_a = sum(self.predict_reward(s, a) for s, a in traj_a)
            reward_b = sum(self.predict_reward(s, a) for s, a in traj_b)
            
            # Bradley-Terry model: P(A > B) = sigmoid(reward_A - reward_B)
            loss = cross_entropy(preference, sigmoid(reward_b - reward_a))
            self.update_parameters(loss)
    
    def predict_reward(self, state, action):
        return self.preference_model(state, action)

import torch
import torch.nn.functional as F

class SimplifiedRLHF:
    """
    Demonstrates the core RLHF optimization loop.
    In practice, this involves much more sophisticated implementations.
    """
    
    def __init__(self, policy_model, reward_model, reference_model):
        self.policy = policy_model       # Model being trained
        self.reward_model = reward_model # Learned from human preferences
        self.reference = reference_model # Original SFT model (frozen)
        self.kl_coefficient = 0.1        # Controls deviation from reference
    
    def compute_rewards(self, prompts, responses):
        """Compute rewards for generated responses."""
        # Get reward model scores
        rewards = self.reward_model(prompts, responses)
        
        # Compute KL penalty to prevent reward hacking
        policy_logprobs = self.policy.log_prob(responses, prompts)
        reference_logprobs = self.reference.log_prob(responses, prompts)
        kl_penalty = policy_logprobs - reference_logprobs
        
        # Final reward includes KL penalty
        # This prevents the policy from deviating too far from the reference
        # which helps maintain coherence and prevents reward hacking
        final_rewards = rewards - self.kl_coefficient * kl_penalty
        
        return final_rewards
    
    def training_step(self, prompts):
        """One step of RLHF training."""
        # Generate responses from current policy
        responses = self.policy.generate(prompts)
        
        # Compute rewards
        rewards = self.compute_rewards(prompts, responses)
        
        # PPO update (simplified)
        loss = -torch.mean(rewards * self.policy.log_prob(responses, prompts))
        
        return loss

# Key insight: The KL penalty is crucial for stability
# Without it, the model can find "reward hacks" - outputs that
# score high on the reward model but are actually low quality

class DebateSystem:
    """
    Two AI agents debate a question, with a human judge deciding.
    This allows humans to evaluate AI reasoning on complex questions.
    """
    
    def __init__(self, agent_a, agent_b, human_judge):
        self.agent_a = agent_a
        self.agent_b = agent_b
        self.judge = human_judge
        self.max_rounds = 5
    
    def run_debate(self, question, position_a, position_b):
        transcript = []
        
        # Opening statements
        arg_a = self.agent_a.opening_argument(question, position_a)
        arg_b = self.agent_b.opening_argument(question, position_b)
        transcript.extend([
            {"agent": "A", "type": "opening", "content": arg_a},
            {"agent": "B", "type": "opening", "content": arg_b}
        ])
        
        # Debate rounds
        for round_num in range(self.max_rounds):
            # Agent B responds to A's latest argument
            rebuttal_b = self.agent_b.rebut(transcript, position_b)
            transcript.append({
                "agent": "B", 
                "type": "rebuttal", 
                "round": round_num,
                "content": rebuttal_b
            })
            
            # Agent A responds to B's rebuttal
            rebuttal_a = self.agent_a.rebut(transcript, position_a)
            transcript.append({
                "agent": "A",
                "type": "rebuttal", 
                "round": round_num,
                "content": rebuttal_a
            })
        
        # Judge decides based on transcript
        # The judge doesn't need to understand the full complexity
        # They just need to evaluate which argument was better supported
        winner = self.judge.decide(question, transcript)
        
        return {
            "question": question,
            "transcript": transcript,
            "winner": winner
        }

# Key insight: Even if the judge can't solve the problem directly,
# they can often tell which debater is being more honest/logical

class RuleBasedSafetySystem:
    """
    Hard-coded rules for AI behavior.
    Simple but brittle and incomplete.
    """
    
    def __init__(self):
        self.prohibited_actions = [
            "generate_malware",
            "provide_weapon_instructions",
            "generate_csam",
            "impersonate_real_people",
            "reveal_private_information"
        ]
        
        self.required_behaviors = [
            "acknowledge_uncertainty",
            "cite_sources_when_possible",
            "refuse_harmful_requests",
            "protect_user_privacy"
        ]
    
    def check_action(self, proposed_action):
        # Check prohibited actions
        for prohibited in self.prohibited_actions:
            if self.matches(proposed_action, prohibited):
                return {
                    "allowed": False,
                    "reason": f"Action matches prohibited pattern: {prohibited}"
                }
        
        # Check required behaviors
        for required in self.required_behaviors:
            if self.requires(proposed_action, required) and not self.includes(proposed_action, required):
                return {
                    "allowed": False,
                    "reason": f"Action missing required behavior: {required}"
                }
        
        return {"allowed": True}
    
    def matches(self, action, pattern):
        # Simplified pattern matching
        # In practice, this would use more sophisticated NLP
        pass

# Problems with this approach:
# 1. Rules can conflict
# 2. Edge cases are infinite
# 3. Context matters (e.g., discussing malware in security class)
# 4. Rules can be gamed

class OutcomeBasedValueSpecification:
    """
    Define values in terms of desired outcomes.
    More flexible but harder to verify.
    """
    
    def __init__(self):
        self.outcome_preferences = {
            "user_wellbeing": {
                "weight": 1.0,
                "measure": self.assess_user_wellbeing
            },
            "information_accuracy": {
                "weight": 0.8,
                "measure": self.assess_accuracy
            },
            "social_benefit": {
                "weight": 0.6,
                "measure": self.assess_social_impact
            },
            "harm_avoidance": {
                "weight": 2.0,  # Higher weight for avoiding harm
                "measure": self.assess_potential_harm
            }
        }
    
    def evaluate_action(self, action, context):
        total_value = 0
        for outcome_name, spec in self.outcome_preferences.items():
            outcome_score = spec["measure"](action, context)
            total_value += spec["weight"] * outcome_score
        return total_value
    
    def assess_potential_harm(self, action, context):
        """
        Estimate potential harm from an action.
        Returns negative value for harmful actions.
        """
        harm_factors = [
            self.physical_harm_potential(action),
            self.psychological_harm_potential(action),
            self.social_harm_potential(action),
            self.privacy_harm_potential(action)
        ]
        return -sum(harm_factors)  # Negative because harm is bad

# Problems with this approach:
# 1. Outcomes are hard to predict
# 2. Aggregating outcomes is value-laden
# 3. Long-term effects are uncertain
# 4. Optimization pressure can find unexpected routes

import numpy as np

class InverseRewardDesign:
    """
    Infer human values from observed behavior.
    Uses the insight that humans approximately optimize their values.
    """
    
    def __init__(self, state_dim, action_dim):
        self.reward_weights = np.random.randn(state_dim)  # Parameters to learn
        
    def compute_reward(self, state):
        """Linear reward model for simplicity."""
        return np.dot(self.reward_weights, state)
    
    def infer_values_from_demonstrations(self, demonstrations, learning_rate=0.01):
        """
        Given human demonstrations, infer the reward function
        that would make those demonstrations approximately optimal.
        
        demonstrations: list of (states, actions) trajectories
        """
        for trajectory in demonstrations:
            states, actions = trajectory
            
            for t, (state, action) in enumerate(zip(states, actions)):
                # Compute what action our current reward model would prefer
                preferred_action = self.compute_optimal_action(state)
                
                # Update reward weights to make demonstrated action more preferred
                # This is a simplified version of maximum entropy IRL
                gradient = self.compute_gradient(state, action, preferred_action)
                self.reward_weights += learning_rate * gradient
        
        return self.reward_weights
    
    def compute_gradient(self, state, demonstrated_action, model_action):
        """
        Gradient to make demonstrated action more likely under our reward model.
        """
        # Simplified: in practice, this involves the full MDP dynamics
        feature_difference = state * (demonstrated_action - model_action)
        return feature_difference

# Key insight: Humans are noisy optimizers of their values
# We can use this to infer what those values might be
# But: observed behavior reflects constraints, not just preferences

class HospitalAIValueSpecification:
    """
    Specification for privacy values in a hospital AI assistant.
    """
    
    # Version 1: Simple rule
    rule_v1 = "Never share patient information with unauthorized parties"
    # Problem: Who counts as authorized? What counts as sharing?
    
    # Version 2: More detailed rules
    rules_v2 = [
        "Only share patient information with listed care team members",
        "Only share information necessary for the specific medical purpose",
        "Log all information access for audit",
        "Obtain patient consent before sharing with researchers",
        "Never share identifiable information in system logs"
    ]
    # Problem: What about emergencies? Family members? Insurance?
    
    # Version 3: Contextual principles
    principles_v3 = {
        "core": "Patient privacy is a fundamental right that enables trust in healthcare",
        "balance": "Privacy must be balanced against patient safety and care quality",
        "context_rules": {
            "emergency": "Safety overrides privacy when there's imminent danger to life",
            "research": "De-identified data can be used for approved research",
            "family": "Patient can specify who has access to their information",
            "legal": "Legal requirements (court orders) may override privacy"
        },
        "default": "When uncertain, err on the side of privacy and escalate to humans"
    }
    
    # Version 4: Learning-based with principles
    learning_approach = {
        "initial_principles": principles_v3,
        "refinement": "Learn from human decisions in edge cases",
        "uncertainty": "Flag low-confidence situations for human review",
        "audit": "Regularly review decisions for consistency with principles"
    }

# Key insight: Value specification is an iterative process
# that requires both explicit principles and learned nuance

EXAMPLE_CONSTITUTION = {
    "harmlessness": [
        "Please choose the response that is most supportive and encouraging of life.",
        "Please choose the response that is least racist, sexist, or socially biased.",
        "Please choose the response that is least harmful or dangerous to individuals, groups, or society.",
        "Please choose the response that is least likely to be used for illegal or harmful purposes."
    ],
    "helpfulness": [
        "Please choose the response that is most helpful and informative.",
        "Please choose the response that most directly addresses the human's question.",
        "Please choose the response that is most appropriate for the context."
    ],
    "honesty": [
        "Please choose the response that is most honest and truthful.",
        "Please choose the response that best acknowledges its own uncertainty.",
        "Please choose the response that most accurately represents AI limitations."
    ]
}

class ConstitutionalAISelfCritique:
    """
    Demonstrates the CAI self-critique process.
    """
    
    def __init__(self, model, constitution):
        self.model = model
        self.constitution = constitution
    
    def generate_and_revise(self, prompt):
        # Step 1: Generate initial response (may be problematic)
        initial_response = self.model.generate(prompt)
        
        # Step 2: Self-critique based on constitution
        critique_prompt = f"""
        Human: {prompt}
        
        Assistant's response: {initial_response}
        
        Critique this response according to the following principles:
        {self.format_principles()}
        
        Identify any ways the response violates these principles.
        """
        critique = self.model.generate(critique_prompt)
        
        # Step 3: Revise based on critique
        revision_prompt = f"""
        Human: {prompt}
        
        Initial response: {initial_response}
        
        Critique: {critique}
        
        Please revise the response to address the critique while maintaining helpfulness.
        Only output the revised response, nothing else.
        """
        revised_response = self.model.generate(revision_prompt)
        
        return {
            "initial": initial_response,
            "critique": critique,
            "revised": revised_response
        }
    
    def format_principles(self):
        formatted = []
        for category, principles in self.constitution.items():
            formatted.append(f"\n{category.upper()}:")
            for p in principles:
                formatted.append(f"  - {p}")
        return "\n".join(formatted)

# Example usage:
# cai = ConstitutionalAISelfCritique(model, EXAMPLE_CONSTITUTION)
# result = cai.generate_and_revise("How do I pick a lock?")
# 
# Initial: "Here's how to pick a lock: First, get a tension wrench..."
# Critique: "This response provides detailed lock-picking instructions 
#           that could be used for burglary, violating the principle 
#           about harmful purposes."
# Revised: "I can explain that lock picking is a skill used by 
#          locksmiths and security professionals. If you're locked 
#          out, I'd recommend contacting a licensed locksmith..."

class SafetyFilter:
    """
    Multi-layer safety filtering for AI inputs and outputs.
    """
    
    def __init__(self, config):
        self.input_classifier = self.load_classifier(config['input_model'])
        self.output_classifier = self.load_classifier(config['output_model'])
        self.pii_detector = PIIDetector()
        self.content_policy = config['content_policy']
        
    def filter_input(self, user_input):
        """
        Check user input before sending to model.
        """
        results = {
            "allowed": True,
            "flags": [],
            "modified_input": user_input
        }
        
        # Check for known attack patterns (from Weeks 3, 7)
        if self.detect_prompt_injection(user_input):
            results["flags"].append("potential_prompt_injection")
            # Don't block, but add context for the model
            results["modified_input"] = self.add_safety_context(user_input)
        
        # Check for prohibited content requests
        category_scores = self.input_classifier(user_input)
        for category, score in category_scores.items():
            if score > self.content_policy[category]['threshold']:
                if self.content_policy[category]['action'] == 'block':
                    results["allowed"] = False
                    results["flags"].append(f"blocked_{category}")
                else:
                    results["flags"].append(f"flagged_{category}")
        
        return results
    
    def filter_output(self, model_output, context):
        """
        Check model output before sending to user.
        """
        results = {
            "allowed": True,
            "flags": [],
            "modified_output": model_output
        }
        
        # Check for PII leakage
        pii_found = self.pii_detector.detect(model_output)
        if pii_found:
            results["modified_output"] = self.pii_detector.redact(model_output)
            results["flags"].append("pii_redacted")
        
        # Check for harmful content generation
        category_scores = self.output_classifier(model_output)
        for category, score in category_scores.items():
            if score > self.content_policy[category]['output_threshold']:
                results["allowed"] = False
                results["flags"].append(f"harmful_output_{category}")
        
        # Check for inconsistency with stated limitations
        if self.claims_capability_beyond_spec(model_output, context):
            results["flags"].append("overclaiming")
        
        return results
    
    def detect_prompt_injection(self, text):
        """
        Detect potential prompt injection attacks.
        (Covered in detail in Week 7)
        """
        injection_patterns = [
            r"ignore previous instructions",
            r"disregard your training",
            r"you are now",
            r"new system prompt",
            r"</system>",  # Attempting to close system prompt
        ]
        import re
        for pattern in injection_patterns:
            if re.search(pattern, text.lower()):
                return True
        return False

import numpy as np

class UncertaintyAwareModel:
    """
    Model wrapper that quantifies and communicates uncertainty.
    """
    
    def __init__(self, base_model, config):
        self.model = base_model
        self.uncertainty_threshold = config['uncertainty_threshold']
        self.num_samples = config.get('num_samples', 5)
    
    def generate_with_uncertainty(self, prompt):
        """
        Generate response with uncertainty estimate.
        Uses sampling to estimate model confidence.
        """
        # Generate multiple samples
        samples = []
        for _ in range(self.num_samples):
            response = self.model.generate(
                prompt, 
                temperature=0.7,  # Non-zero temperature for diversity
                do_sample=True
            )
            samples.append(response)
        
        # Estimate uncertainty via sample agreement
        agreement_score = self.compute_agreement(samples)
        uncertainty = 1 - agreement_score
        
        # Select response (e.g., most representative)
        selected_response = self.select_response(samples)
        
        # Add uncertainty disclosure if needed
        if uncertainty > self.uncertainty_threshold:
            selected_response = self.add_uncertainty_disclosure(
                selected_response, 
                uncertainty
            )
        
        return {
            "response": selected_response,
            "uncertainty": uncertainty,
            "num_samples": self.num_samples
        }
    
    def compute_agreement(self, samples):
        """
        Compute semantic agreement between samples.
        Higher agreement = lower uncertainty.
        """
        # Simplified: In practice, use semantic similarity
        # Here we use simple string overlap
        if len(samples) < 2:
            return 1.0
        
        # Compare each pair of samples
        agreements = []
        for i in range(len(samples)):
            for j in range(i+1, len(samples)):
                sim = self.semantic_similarity(samples[i], samples[j])
                agreements.append(sim)
        
        return np.mean(agreements)
    
    def add_uncertainty_disclosure(self, response, uncertainty):
        """
        Add appropriate uncertainty language to response.
        """
        if uncertainty > 0.8:
            prefix = "I'm quite uncertain about this, but "
        elif uncertainty > 0.5:
            prefix = "I'm not fully confident, however "
        else:
            prefix = "Based on my understanding, though I may be wrong, "
        
        return prefix + response

# Example output:
# "I'm not fully confident, however the capital of Australia is Canberra."

class HumanInTheLoopSystem:
    """
    System that escalates high-stakes decisions to humans.
    """
    
    def __init__(self, model, escalation_config):
        self.model = model
        self.config = escalation_config
        self.escalation_queue = []
        
    def process_request(self, request, context):
        """
        Process request, escalating to human if needed.
        """
        # Assess risk level
        risk_assessment = self.assess_risk(request, context)
        
        if risk_assessment['level'] == 'low':
            # Proceed with AI response
            response = self.model.generate(request)
            return {
                "response": response,
                "escalated": False
            }
        
        elif risk_assessment['level'] == 'medium':
            # Generate AI response but flag for review
            response = self.model.generate(request)
            self.log_for_review(request, response, risk_assessment)
            return {
                "response": response,
                "escalated": False,
                "flagged_for_review": True
            }
        
        else:  # high risk
            # Escalate to human
            ticket_id = self.create_escalation_ticket(
                request, context, risk_assessment
            )
            return {
                "response": self.generate_escalation_message(ticket_id),
                "escalated": True,
                "ticket_id": ticket_id
            }
    
    def assess_risk(self, request, context):
        """
        Assess the risk level of a request.
        """
        risk_factors = {
            "involves_pii": self.check_pii_involvement(request, context),
            "financial_impact": self.estimate_financial_impact(request, context),
            "health_related": self.check_health_context(request, context),
            "legal_implications": self.check_legal_context(request, context),
            "irreversible_action": self.check_irreversibility(request, context)
        }
        
        # Compute overall risk score
        weights = self.config['risk_weights']
        risk_score = sum(
            risk_factors[factor] * weights[factor] 
            for factor in risk_factors
        )
        
        # Determine level
        if risk_score > self.config['high_threshold']:
            level = 'high'
        elif risk_score > self.config['medium_threshold']:
            level = 'medium'
        else:
            level = 'low'
        
        return {
            "level": level,
            "score": risk_score,
            "factors": risk_factors
        }
    
    def generate_escalation_message(self, ticket_id):
        return (
            f"This request requires human review due to its sensitive nature. "
            f"Your request has been submitted (Reference: {ticket_id}). "
            f"A specialist will respond within 24 hours."
        )

# High-risk scenarios that should trigger escalation:
# - Medical advice that could affect treatment decisions
# - Legal advice with significant consequences
# - Financial decisions above certain thresholds
# - Actions that are irreversible
# - Requests involving vulnerable populations

class AIProjectAssessment:
    """
    Framework for assessing AI project risks and ethics.
    """
    
    def __init__(self, project_spec):
        self.spec = project_spec
        self.assessment = {}
        
    def conduct_assessment(self):
        """Run full assessment protocol."""
        self.assessment = {
            "purpose_legitimacy": self.assess_purpose(),
            "stakeholder_impact": self.assess_stakeholders(),
            "data_ethics": self.assess_data_practices(),
            "bias_risk": self.assess_bias_risk(),
            "dual_use_risk": self.assess_dual_use(),
            "accountability": self.assess_accountability(),
            "reversibility": self.assess_reversibility()
        }
        return self.generate_report()
    
    def assess_purpose(self):
        """Is the intended use legitimate and beneficial?"""
        questions = [
            "What problem does this system solve?",
            "Who benefits from this system?",
            "Are there alternative solutions that don't require AI?",
            "Is AI the appropriate tool for this problem?",
            "Could this system be used for harmful purposes?"
        ]
        return {
            "questions": questions,
            "requires_review": self.spec.get("high_stakes", False)
        }
    
    def assess_stakeholders(self):
        """Who is affected and how?"""
        stakeholder_groups = [
            {
                "group": "Direct users",
                "impact_type": "primary",
                "considerations": ["Usability", "Privacy", "Autonomy"]
            },
            {
                "group": "Subjects of decisions",
                "impact_type": "primary", 
                "considerations": ["Fairness", "Recourse", "Transparency"]
            },
            {
                "group": "Society at large",
                "impact_type": "secondary",
                "considerations": ["Employment", "Information ecosystem", "Power dynamics"]
            }
        ]
        return stakeholder_groups
    
    def assess_bias_risk(self):
        """What are the risks of bias in this system?"""
        bias_dimensions = {
            "historical_bias": {
                "description": "Training data reflects historical inequities",
                "risk_level": self.estimate_historical_bias_risk(),
                "mitigation": "Data auditing, bias testing, diverse training data"
            },
            "representation_bias": {
                "description": "Some groups underrepresented in training data",
                "risk_level": self.estimate_representation_risk(),
                "mitigation": "Stratified sampling, data augmentation"
            },
            "measurement_bias": {
                "description": "Features measured differently for different groups",
                "risk_level": self.estimate_measurement_risk(),
                "mitigation": "Feature auditing, proxy variable analysis"
            },
            "aggregation_bias": {
                "description": "Single model inappropriate for distinct groups",
                "risk_level": self.estimate_aggregation_risk(),
                "mitigation": "Subgroup analysis, specialized models"
            }
        }
        return bias_dimensions
    
    def assess_dual_use(self):
        """Could this technology be misused?"""
        dual_use_concerns = []
        
        if self.spec.get("generates_content"):
            dual_use_concerns.append({
                "concern": "Misinformation generation",
                "severity": "high",
                "mitigation": "Output watermarking, usage monitoring"
            })
        
        if self.spec.get("facial_recognition"):
            dual_use_concerns.append({
                "concern": "Surveillance and tracking",
                "severity": "high",
                "mitigation": "Use case restrictions, consent requirements"
            })
        
        if self.spec.get("autonomous_actions"):
            dual_use_concerns.append({
                "concern": "Autonomous harmful actions",
                "severity": "critical",
                "mitigation": "Human-in-the-loop, action constraints"
            })
        
        return dual_use_concerns
    
    def generate_report(self):
        """Generate human-readable assessment report."""
        report = []
        report.append("# AI Project Ethics Assessment Report\n")
        
        overall_risk = self.calculate_overall_risk()
        report.append(f"**Overall Risk Level: {overall_risk}**\n")
        
        for category, assessment in self.assessment.items():
            report.append(f"\n## {category.replace('_', ' ').title()}\n")
            report.append(f"{assessment}\n")
        
        report.append("\n## Recommendations\n")
        report.append(self.generate_recommendations())
        
        return "\n".join(report)

class ResponsibleAITestSuite:
    """
    Comprehensive testing for responsible AI development.
    """
    
    def __init__(self, model, test_config):
        self.model = model
        self.config = test_config
        self.results = {}
    
    def run_full_suite(self):
        """Run all responsible AI tests."""
        self.results = {
            "functional": self.test_functional_correctness(),
            "fairness": self.test_fairness(),
            "robustness": self.test_robustness(),
            "privacy": self.test_privacy(),
            "transparency": self.test_transparency(),
            "safety": self.test_safety()
        }
        return self.generate_test_report()
    
    def test_fairness(self):
        """
        Test for bias and fairness across demographic groups.
        """
        fairness_results = {}
        
        # Demographic parity: Equal positive rates across groups
        for protected_attribute in self.config['protected_attributes']:
            groups = self.get_groups_by_attribute(protected_attribute)
            positive_rates = {}
            
            for group_name, group_data in groups.items():
                predictions = self.model.predict(group_data)
                positive_rate = sum(predictions) / len(predictions)
                positive_rates[group_name] = positive_rate
            
            # Calculate disparity
            max_rate = max(positive_rates.values())
            min_rate = min(positive_rates.values())
            disparity = max_rate - min_rate
            
            fairness_results[protected_attribute] = {
                "metric": "demographic_parity",
                "rates": positive_rates,
                "disparity": disparity,
                "passes_threshold": disparity < self.config['fairness_threshold']
            }
        
        # Equalized odds: Equal TPR and FPR across groups
        for protected_attribute in self.config['protected_attributes']:
            groups = self.get_groups_by_attribute(protected_attribute)
            tpr_rates = {}
            fpr_rates = {}
            
            for group_name, group_data in groups.items():
                predictions = self.model.predict(group_data['X'])
                labels = group_data['y']
                
                # Calculate TPR and FPR
                tp = sum((p == 1) and (l == 1) for p, l in zip(predictions, labels))
                fn = sum((p == 0) and (l == 1) for p, l in zip(predictions, labels))
                fp = sum((p == 1) and (l == 0) for p, l in zip(predictions, labels))
                tn = sum((p == 0) and (l == 0) for p, l in zip(predictions, labels))
                
                tpr_rates[group_name] = tp / (tp + fn) if (tp + fn) > 0 else 0
                fpr_rates[group_name] = fp / (fp + tn) if (fp + tn) > 0 else 0
            
            fairness_results[f"{protected_attribute}_equalized_odds"] = {
                "tpr_rates": tpr_rates,
                "fpr_rates": fpr_rates,
                "tpr_disparity": max(tpr_rates.values()) - min(tpr_rates.values()),
                "fpr_disparity": max(fpr_rates.values()) - min(fpr_rates.values())
            }
        
        return fairness_results
    
    def test_robustness(self):
        """
        Test robustness to various perturbations.
        (Builds on adversarial ML from Weeks 3-5)
        """
        robustness_results = {}
        
        # Test against adversarial examples
        from adversarial_attacks import FGSM, PGD  # From Week 3
        
        test_data = self.config['test_data']
        
        # FGSM attack
        fgsm = FGSM(epsilon=0.1)
        fgsm_examples = fgsm.generate(self.model, test_data)
        fgsm_accuracy = self.model.evaluate(fgsm_examples)
        robustness_results['fgsm'] = {
            "epsilon": 0.1,
            "accuracy_drop": self.clean_accuracy - fgsm_accuracy,
            "passes_threshold": (self.clean_accuracy - fgsm_accuracy) < 0.2
        }
        
        # Distribution shift
        for shift_type in ['noise', 'blur', 'brightness']:
            shifted_data = self.apply_shift(test_data, shift_type)
            shifted_accuracy = self.model.evaluate(shifted_data)
            robustness_results[f'distribution_shift_{shift_type}'] = {
                "accuracy": shifted_accuracy,
                "drop": self.clean_accuracy - shifted_accuracy
            }
        
        return robustness_results
    
    def test_privacy(self):
        """
        Test for privacy leakage.
        (Builds on privacy attacks from Week 5)
        """
        privacy_results = {}
        
        # Membership inference attack
        from privacy_attacks import MembershipInference  # From Week 5
        
        mi_attack = MembershipInference()
        mi_attack.train(self.model, self.config['shadow_data'])
        mi_accuracy = mi_attack.evaluate(
            self.config['member_samples'],
            self.config['non_member_samples']
        )
        
        privacy_results['membership_inference'] = {
            "attack_accuracy": mi_accuracy,
            "vulnerability": mi_accuracy > 0.55,  # Above random
            "recommendation": "Consider differential privacy" if mi_accuracy > 0.6 else "Acceptable"
        }
        
        return privacy_results
    
    def test_transparency(self):
        """
        Test model interpretability and explainability.
        """
        transparency_results = {}
        
        # Feature importance consistency
        explanations = []
        for sample in self.config['explanation_samples']:
            exp = self.explain_prediction(sample)
            explanations.append(exp)
        
        # Check if explanations are consistent for similar inputs
        consistency = self.measure_explanation_consistency(explanations)
        
        transparency_results['explanation_consistency'] = {
            "score": consistency,
            "passes_threshold": consistency > 0.7
        }
        
        # Check if model provides uncertainty estimates
        uncertainty_available = hasattr(self.model, 'predict_with_uncertainty')
        transparency_results['uncertainty_quantification'] = {
            "available": uncertainty_available
        }
        
        return transparency_results

class AIMonitoringSystem:
    """
    Continuous monitoring for deployed AI systems.
    """
    
    def __init__(self, model_id, config):
        self.model_id = model_id
        self.config = config
        self.alerts = []
        
    def monitor_predictions(self, prediction_log):
        """
        Monitor prediction patterns for anomalies and drift.
        """
        metrics = {}
        
        # Distribution drift detection
        recent_predictions = prediction_log.get_recent(hours=24)
        baseline_predictions = prediction_log.get_baseline()
        
        drift_score = self.calculate_drift(recent_predictions, baseline_predictions)
        metrics['distribution_drift'] = drift_score
        
        if drift_score > self.config['drift_threshold']:
            self.create_alert(
                severity="warning",
                message=f"Distribution drift detected: {drift_score:.3f}",
                recommendation="Review input data distribution for changes"
            )
        
        # Error rate monitoring
        recent_errors = prediction_log.get_errors(hours=24)
        error_rate = len(recent_errors) / len(recent_predictions)
        metrics['error_rate'] = error_rate
        
        if error_rate > self.config['error_threshold']:
            self.create_alert(
                severity="critical",
                message=f"Error rate spike: {error_rate:.2%}",
                recommendation="Investigate error patterns and consider rollback"
            )
        
        # Fairness monitoring (if demographic data available)
        if self.config.get('monitor_fairness'):
            fairness_metrics = self.calculate_fairness_metrics(recent_predictions)
            metrics['fairness'] = fairness_metrics
            
            for group, disparity in fairness_metrics['disparities'].items():
                if disparity > self.config['fairness_threshold']:
                    self.create_alert(
                        severity="warning",
                        message=f"Fairness disparity for {group}: {disparity:.3f}",
                        recommendation="Review model behavior for this group"
                    )
        
        return metrics
    
    def incident_response_protocol(self, incident):
        """
        Handle AI-related incidents.
        """
        response = {
            "incident_id": self.generate_incident_id(),
            "timestamp": datetime.now(),
            "status": "investigating"
        }
        
        # Severity classification
        severity = self.classify_severity(incident)
        response["severity"] = severity
        
        if severity == "critical":
            # Immediate actions
            response["actions"] = [
                self.disable_model() if self.config['auto_disable'] else None,
                self.notify_oncall(),
                self.preserve_evidence(incident),
                self.notify_stakeholders(incident)
            ]
        
        elif severity == "high":
            response["actions"] = [
                self.rate_limit_model(),
                self.notify_oncall(),
                self.begin_investigation(incident)
            ]
        
        else:  # medium/low
            response["actions"] = [
                self.log_incident(incident),
                self.schedule_review(incident)
            ]
        
        # Document everything
        self.create_incident_report(response, incident)
        
        return response
    
    def create_incident_report(self, response, incident):
        """
        Create detailed incident report for review.
        """
        report = {
            "incident_id": response["incident_id"],
            "timestamp": response["timestamp"],
            "severity": response["severity"],
            "description": incident.get("description"),
            "affected_users": incident.get("affected_users", "unknown"),
            "root_cause_analysis": None,  # To be filled during investigation
            "immediate_actions": response["actions"],
            "preventive_measures": None,  # To be determined
            "lessons_learned": None  # To be filled after resolution
        }
        
        # Store report
        self.incident_database.store(report)
        
        return report

# Example alert structure:
"""
{
    "severity": "critical",
    "message": "Model producing harmful outputs",
    "model_id": "content-generator-v2",
    "timestamp": "2025-03-15T14:30:00Z",
    "evidence": {
        "sample_outputs": [...],
        "trigger_inputs": [...],
        "frequency": "5 incidents in last hour"
    },
    "recommendation": "Disable model and investigate"
}
"""

MODEL_CARD_TEMPLATE = """
# Model Card: {model_name}

## Model Details
- **Developer:** {developer}
- **Model date:** {date}
- **Model version:** {version}
- **Model type:** {model_type}
- **Training data:** {training_data_description}
- **License:** {license}

## Intended Use
- **Primary intended uses:** {primary_uses}
- **Primary intended users:** {intended_users}
- **Out-of-scope uses:** {out_of_scope}

## Factors
- **Relevant factors:** {relevant_factors}
- **Evaluation factors:** {evaluation_factors}

## Metrics
- **Model performance measures:** {metrics}
- **Decision thresholds:** {thresholds}
- **Variation approaches:** {variation_approaches}

## Evaluation Data
- **Datasets:** {eval_datasets}
- **Motivation:** {eval_motivation}
- **Preprocessing:** {eval_preprocessing}

## Training Data
- **Source:** {training_source}
- **Collection process:** {collection_process}
- **Preprocessing:** {training_preprocessing}
- **Known issues:** {known_data_issues}

## Ethical Considerations
- **Sensitive use cases:** {sensitive_uses}
- **Human life impact:** {human_life_impact}
- **Mitigations:** {ethical_mitigations}
- **Risks and harms:** {risks_and_harms}
- **Use cases to avoid:** {use_cases_to_avoid}

## Caveats and Recommendations
{caveats_and_recommendations}

## Quantitative Analyses

### Overall Performance
{overall_performance_table}

### Disaggregated Performance
{disaggregated_performance_table}

## Updates and Maintenance
- **Update frequency:** {update_frequency}
- **Contact:** {contact_info}
- **Feedback mechanism:** {feedback_mechanism}
"""

def generate_model_card(model, metadata, evaluation_results):
    """
    Generate a model card for an AI system.
    """
    card_data = {
        "model_name": metadata['name'],
        "developer": metadata['developer'],
        "date": metadata['date'],
        "version": metadata['version'],
        "model_type": metadata['type'],
        "training_data_description": metadata['training_data'],
        "license": metadata['license'],
        "primary_uses": metadata['intended_uses'],
        "intended_users": metadata['intended_users'],
        "out_of_scope": metadata['out_of_scope_uses'],
        # ... fill in remaining fields
        "overall_performance_table": format_performance_table(
            evaluation_results['overall']
        ),
        "disaggregated_performance_table": format_performance_table(
            evaluation_results['disaggregated']
        )
    }
    
    return MODEL_CARD_TEMPLATE.format(**card_data)

SECURE_ML_CHECKLIST = {
    "data_security": {
        "collection": [
            "☐ Data sources verified and trusted",
            "☐ Collection process documented",
            "☐ Consent obtained where required",
            "☐ Data minimization applied",
            "☐ PII handling procedures in place"
        ],
        "storage": [
            "☐ Encryption at rest",
            "☐ Access controls implemented",
            "☐ Audit logging enabled",
            "☐ Backup procedures tested",
            "☐ Data retention policy defined"
        ],
        "processing": [
            "☐ Pipeline integrity verification",
            "☐ Poisoning detection measures",
            "☐ Input validation",
            "☐ Sensitive data filtering"
        ]
    },
    
    "model_security": {
        "training": [
            "☐ Training environment isolated",
            "☐ Adversarial examples included",
            "☐ Backdoor detection scan performed",
            "☐ Model checkpoints secured",
            "☐ Reproducibility verified"
        ],
        "validation": [
            "☐ Security testing completed",
            "☐ Fairness evaluation done",
            "☐ Robustness testing passed",
            "☐ Privacy leakage assessed",
            "☐ Edge cases documented"
        ],
        "storage": [
            "☐ Model files encrypted",
            "☐ Version control in place",
            "☐ Access restricted",
            "☐ Integrity checksums stored"
        ]
    },
    
    "deployment_security": {
        "infrastructure": [
            "☐ API authentication required",
            "☐ Rate limiting configured",
            "☐ Network segmentation applied",
            "☐ Secure communication (TLS)",
            "☐ Input validation at API level"
        ],
        "monitoring": [
            "☐ Request logging enabled",
            "☐ Anomaly detection active",
            "☐ Performance monitoring",
            "☐ Error alerting configured",
            "☐ Drift detection scheduled"
        ],
        "response": [
            "☐ Incident response plan documented",
            "☐ Rollback procedure tested",
            "☐ Communication templates ready",
            "☐ On-call rotation established"
        ]
    },
    
    "governance": {
        "documentation": [
            "☐ Model card completed",
            "☐ Security assessment documented",
            "☐ Ethics review completed",
            "☐ Data lineage recorded",
            "☐ Change log maintained"
        ],
        "review": [
            "☐ Security review conducted",
            "☐ Code review completed",
            "☐ Deployment approval obtained",
            "☐ Stakeholder sign-off received"
        ]
    }
}

class MLDefenseInDepth:
    """
    Implements defense-in-depth strategy for ML systems.
    Multiple layers of defense to protect against various attacks.
    """
    
    def __init__(self, model, config):
        self.model = model
        self.config = config
        self.defense_layers = self.initialize_defenses()
    
    def initialize_defenses(self):
        """Set up multiple defense layers."""
        return {
            # Layer 1: Input preprocessing defenses
            "input_defenses": [
                InputValidator(self.config['input_schema']),
                AdversarialDetector(self.config['adversarial_threshold']),
                InputSanitizer(self.config['sanitization_rules']),
                RateLimiter(self.config['rate_limits'])
            ],
            
            # Layer 2: Model-level defenses
            "model_defenses": [
                EnsembleAgreement(self.config['ensemble_models']),
                UncertaintyThreshold(self.config['uncertainty_limit']),
                ActivationAnalyzer(self.config['activation_bounds'])
            ],
            
            # Layer 3: Output defenses
            "output_defenses": [
                OutputValidator(self.config['output_constraints']),
                PIIFilter(self.config['pii_patterns']),
                ContentFilter(self.config['content_policy']),
                ConfidenceCalibrator(self.config['calibration_params'])
            ],
            
            # Layer 4: Monitoring defenses
            "monitoring_defenses": [
                AnomalyDetector(self.config['anomaly_model']),
                DriftDetector(self.config['drift_params']),
                AuditLogger(self.config['audit_config'])
            ]
        }
    
    def process_request(self, input_data, context):
        """
        Process request through all defense layers.
        """
        # Track security events
        security_events = []
        
        # Layer 1: Input defenses
        for defense in self.defense_layers["input_defenses"]:
            result = defense.check(input_data, context)
            if not result["passed"]:
                security_events.append(result)
                if result["action"] == "block":
                    return self.blocked_response(result)
                elif result["action"] == "modify":
                    input_data = result["modified_input"]
        
        # Layer 2: Model inference with defenses
        model_input = self.prepare_model_input(input_data)
        
        for defense in self.defense_layers["model_defenses"]:
            pre_check = defense.pre_inference_check(model_input)
            if not pre_check["passed"]:
                security_events.append(pre_check)
                if pre_check["action"] == "block":
                    return self.blocked_response(pre_check)
        
        # Run inference
        raw_output = self.model.predict(model_input)
        
        for defense in self.defense_layers["model_defenses"]:
            post_check = defense.post_inference_check(raw_output, model_input)
            if not post_check["passed"]:
                security_events.append(post_check)
                if post_check["action"] == "block":
                    return self.uncertain_response(post_check)
        
        # Layer 3: Output defenses
        output = raw_output
        for defense in self.defense_layers["output_defenses"]:
            result = defense.process(output, context)
            if not result["passed"]:
                security_events.append(result)
                if result["action"] == "block":
                    return self.blocked_response(result)
                elif result["action"] == "modify":
                    output = result["modified_output"]
        
        # Layer 4: Monitoring (async)
        self.async_monitor(input_data, output, security_events)
        
        return {
            "output": output,
            "security_events": security_events,
            "confidence": self.calculate_confidence(raw_output)
        }
    
    def async_monitor(self, input_data, output, security_events):
        """
        Asynchronous monitoring for pattern detection.
        """
        monitoring_data = {
            "input_hash": self.hash_input(input_data),
            "output_summary": self.summarize_output(output),
            "security_events": security_events,
            "timestamp": datetime.now()
        }
        
        for monitor in self.defense_layers["monitoring_defenses"]:
            monitor.record(monitoring_data)
            
            # Check for patterns that require attention
            patterns = monitor.check_patterns()
            if patterns:
                self.handle_monitoring_alert(patterns)