欧盟人工智能法案合规实操：开发者安全清单

// risk-classifier.ts - AI系统风险分类器
interface AISystemProfile {
  purpose: string;
  domain: string;
  interactsWithNaturalPersons: boolean;
  usesPersonalData: boolean;
  outputAffectsRights: boolean;
  deploymentRegions: string[];
}

enum RiskLevel {
  UNACCEPTABLE = 'UNACCEPTABLE',
  HIGH = 'HIGH',
  LIMITED = 'LIMITED',
  MINIMAL = 'MINIMAL'
}

const PROHIBITED_PRACTICES = [
  'social_scoring_by_public_authorities',
  'real_time_remote_biometric_identification_public_spaces',
  'subliminal_manipulation_causing_harm',
  'exploitation_of_vulnerable_groups',
  'emotion_inference_in_workplace_education',
  'untargeted_facial_image_scraping',
  'biometric_categorization_sensitive_attributes'
];

const ANNEX_III_DOMAINS = [
  'biometric_identification',
  'critical_infrastructure_management',
  'education_vocational_training',
  'employment_worker_management',
  'essential_services_access', // credit scoring, insurance
  'law_enforcement',
  'migration_asylum_border_control',
  'administration_of_justice'
];

function classifyRisk(system: AISystemProfile): RiskLevel {
  // Check prohibited practices first
  if (PROHIBITED_PRACTICES.some(p => system.purpose.includes(p))) {
    return RiskLevel.UNACCEPTABLE;
  }
  
  // Check Annex III high-risk domains
  if (ANNEX_III_DOMAINS.some(d => system.domain.includes(d))) {
    return RiskLevel.HIGH;
  }
  
  // Transparency obligations for direct interaction
  if (system.interactsWithNaturalPersons) {
    return RiskLevel.LIMITED;
  }
  
  return RiskLevel.MINIMAL;
}

# model_card_generator.py - 自动化模型卡片生成
import json
import hashlib
from datetime import datetime
from dataclasses import dataclass, field, asdict
from typing import List, Dict, Optional

@dataclass
class TrainingDataInfo:
    dataset_name: str
    source: str
    size: int
    collection_date: str
    preprocessing_steps: List[str]
    personal_data_categories: List[str]
    geographic_coverage: List[str]
    demographic_representation: Dict[str, float]

@dataclass
class PerformanceMetrics:
    metric_name: str
    value: float
    confidence_interval: tuple
    evaluation_dataset: str
    disaggregated_results: Dict[str, float]  # by demographic group

@dataclass
class RiskAssessment:
    risk_id: str
    description: str
    severity: str  # critical, high, medium, low
    likelihood: str
    mitigation_measures: List[str]
    residual_risk_level: str
    monitoring_indicators: List[str]

@dataclass
class ModelCard:
    """EU AI Act Annex IV compliant model card"""
    # Section 1: General Description
    model_name: str
    version: str
    provider: str
    intended_purpose: str
    intended_users: List[str]
    prohibited_uses: List[str]
    deployment_regions: List[str]
    
    # Section 2: Technical Architecture
    model_type: str
    architecture_description: str
    input_format: str
    output_format: str
    computational_requirements: Dict[str, str]
    
    # Section 3: Training Information
    training_data: List[TrainingDataInfo] = field(default_factory=list)
    training_methodology: str = ""
    hyperparameters: Dict[str, any] = field(default_factory=dict)
    training_duration: str = ""
    
    # Section 4: Evaluation
    performance_metrics: List[PerformanceMetrics] = field(default_factory=list)
    bias_evaluation: Dict[str, any] = field(default_factory=dict)
    robustness_tests: List[str] = field(default_factory=list)
    
    # Section 5: Risk Management
    risk_assessments: List[RiskAssessment] = field(default_factory=list)
    
    # Metadata
    generation_timestamp: str = field(
        default_factory=lambda: datetime.utcnow().isoformat()
    )
    document_hash: str = ""
    
    def generate_hash(self) -> str:
        content = json.dumps(asdict(self), sort_keys=True, default=str)
        return hashlib.sha256(content.encode()).hexdigest()
    
    def validate_completeness(self) -> List[str]:
        """Validate all Annex IV required fields are populated"""
        missing = []
        if not self.intended_purpose:
            missing.append("intended_purpose (Annex IV, Section 1)")
        if not self.training_data:
            missing.append("training_data (Annex IV, Section 3)")
        if not self.performance_metrics:
            missing.append("performance_metrics (Annex IV, Section 4)")
        if not self.risk_assessments:
            missing.append("risk_assessments (Annex IV, Section 5)")
        if not self.bias_evaluation:
            missing.append("bias_evaluation (Annex IV, Section 4)")
        return missing
    
    def export_for_authority(self) -> Dict:
        """Export format suitable for regulatory authority review"""
        card = asdict(self)
        card['document_hash'] = self.generate_hash()
        card['compliance_standard'] = 'EU_AI_ACT_2024_ANNEX_IV'
        card['export_timestamp'] = datetime.utcnow().isoformat()
        return card


def generate_model_card_from_mlflow(run_id: str) -> ModelCard:
    """Generate model card from MLflow experiment tracking"""
    import mlflow
    
    run = mlflow.get_run(run_id)
    params = run.data.params
    metrics = run.data.metrics
    
    card = ModelCard(
        model_name=params.get('model_name', ''),
        version=params.get('version', '1.0.0'),
        provider="Your Company Name",
        intended_purpose=params.get('intended_purpose', ''),
        intended_users=json.loads(params.get('intended_users', '[]')),
        prohibited_uses=json.loads(params.get('prohibited_uses', '[]')),
        deployment_regions=json.loads(params.get('deployment_regions', '[]')),
        model_type=params.get('model_type', ''),
        architecture_description=params.get('architecture', ''),
        input_format=params.get('input_format', ''),
        output_format=params.get('output_format', ''),
        computational_requirements={
            'gpu': params.get('gpu_type', ''),
            'memory': params.get('memory_gb', ''),
            'inference_latency_p99': str(metrics.get('inference_p99', ''))
        }
    )
    
    # Validate completeness before export
    missing = card.validate_completeness()
    if missing:
        raise ValueError(
            f"Model card incomplete. Missing fields: {missing}"
        )
    
    return card

// audit-middleware.ts - EU AI Act Article 12 compliant audit logging
import { v4 as uuidv4 } from 'uuid';
import crypto from 'crypto';

interface AuditEvent {
  eventId: string;
  timestamp: string;
  systemId: string;
  systemVersion: string;
  operationType: 'inference' | 'training' | 'human_override' | 'system_event';
  
  // Input tracking
  inputHash: string;
  inputMetadata: Record<string, any>;
  
  // Output tracking
  outputHash: string;
  outputMetadata: Record<string, any>;
  confidenceScore?: number;
  
  // Decision context
  modelVersion: string;
  featureImportance?: Record<string, number>;
  
  // Human oversight
  humanOversightTriggered: boolean;
  humanDecision?: string;
  humanOperatorId?: string;
  
  // Compliance metadata
  riskLevel: string;
  dataSubjectCategories: string[];
  retentionPeriodDays: number;
  
  // Integrity
  previousEventHash: string;
  eventChainHash: string;
}

class EUAIActAuditLogger {
  private systemId: string;
  private systemVersion: string;
  private lastEventHash: string = '';
  private storageBackend: AuditStorageBackend;
  
  constructor(config: {
    systemId: string;
    systemVersion: string;
    storageBackend: AuditStorageBackend;
  }) {
    this.systemId = config.systemId;
    this.systemVersion = config.systemVersion;
    this.storageBackend = config.storageBackend;
  }
  
  async logInference(params: {
    input: any;
    output: any;
    modelVersion: string;
    confidenceScore: number;
    featureImportance?: Record<string, number>;
    dataSubjectCategories: string[];
  }): Promise<string> {
    const inputHash = this.hashContent(params.input);
    const outputHash = this.hashContent(params.output);
    
    const event: AuditEvent = {
      eventId: uuidv4(),
      timestamp: new Date().toISOString(),
      systemId: this.systemId,
      systemVersion: this.systemVersion,
      operationType: 'inference',
      inputHash,
      inputMetadata: {
        contentType: typeof params.input,
        size: JSON.stringify(params.input).length
      },
      outputHash,
      outputMetadata: {
        contentType: typeof params.output,
        size: JSON.stringify(params.output).length
      },
      confidenceScore: params.confidenceScore,
      modelVersion: params.modelVersion,
      featureImportance: params.featureImportance,
      humanOversightTriggered: false,
      riskLevel: 'HIGH',
      dataSubjectCategories: params.dataSubjectCategories,
      retentionPeriodDays: this.calculateRetentionPeriod(),
      previousEventHash: this.lastEventHash,
      eventChainHash: ''
    };
    
    // Chain integrity - tamper-evident log chain
    event.eventChainHash = this.computeChainHash(event);
    this.lastEventHash = event.eventChainHash;
    
    await this.storageBackend.store(event);
    return event.eventId;
  }
  
  async logHumanOverride(params: {
    originalEventId: string;
    operatorId: string;
    decision: string;
    reason: string;
  }): Promise<string> {
    const event: AuditEvent = {
      eventId: uuidv4(),
      timestamp: new Date().toISOString(),
      systemId: this.systemId,
      systemVersion: this.systemVersion,
      operationType: 'human_override',
      inputHash: params.originalEventId,
      inputMetadata: { originalEvent: params.originalEventId },
      outputHash: this.hashContent(params.decision),
      outputMetadata: { reason: params.reason },
      modelVersion: '',
      humanOversightTriggered: true,
      humanDecision: params.decision,
      humanOperatorId: params.operatorId,
      riskLevel: 'HIGH',
      dataSubjectCategories: [],
      retentionPeriodDays: this.calculateRetentionPeriod(),
      previousEventHash: this.lastEventHash,
      eventChainHash: ''
    };
    
    event.eventChainHash = this.computeChainHash(event);
    this.lastEventHash = event.eventChainHash;
    
    await this.storageBackend.store(event);
    return event.eventId;
  }
  
  private hashContent(content: any): string {
    return crypto
      .createHash('sha256')
      .update(JSON.stringify(content))
      .digest('hex');
  }
  
  private computeChainHash(event: AuditEvent): string {
    const payload = `${event.eventId}:${event.timestamp}:${event.previousEventHash}`;
    return crypto.createHash('sha256').update(payload).digest('hex');
  }
  
  private calculateRetentionPeriod(): number {
    // Article 12(3): logs retained for period appropriate to intended purpose
    // Retention is a policy decision tied to purpose, applicable law,
    // incident handling, and deletion requirements; it is not a universal
    // EU AI Act five-year default.
    throw new Error('Configure retention from the system purpose, applicable law, and deletion policy');
  }
}

interface AuditStorageBackend {
  store(event: AuditEvent): Promise<void>;
  retrieve(eventId: string): Promise<AuditEvent | null>;
  queryByTimeRange(start: Date, end: Date): Promise<AuditEvent[]>;
  verifyChainIntegrity(startEventId: string): Promise<boolean>;
}

// fria-assessment.ts - Fundamental Rights Impact Assessment (Article 27)
interface FRIAReport {
  assessmentId: string;
  assessmentDate: string;
  assessor: string;
  systemDescription: string;
  
  // Affected fundamental rights
  rightsImpacted: FundamentalRight[];
  
  // Risk analysis per right
  riskAnalysis: RightRiskAnalysis[];
  
  // Mitigation measures
  mitigations: MitigationMeasure[];
  
  // Stakeholder consultation
  consultationRecords: ConsultationRecord[];
  
  // Review schedule
  nextReviewDate: string;
  reviewFrequency: string;
}

interface FundamentalRight {
  rightId: string;
  rightName: string;
  charterArticle: string; // EU Charter reference
  impactLevel: 'none' | 'minimal' | 'moderate' | 'significant' | 'severe';
  justification: string;
}

interface RightRiskAnalysis {
  rightId: string;
  riskDescription: string;
  affectedGroups: string[];
  likelihoodScore: number; // 1-5
  severityScore: number;   // 1-5
  riskScore: number;       // likelihood * severity
  existingControls: string[];
  residualRiskLevel: string;
}

interface MitigationMeasure {
  measureId: string;
  targetRightId: string;
  description: string;
  implementationType: 'technical' | 'organizational' | 'legal';
  status: 'planned' | 'implemented' | 'verified';
  effectivenessMetric: string;
  responsiblePerson: string;
  deadline: string;
}

const EU_CHARTER_RIGHTS_MAPPING = {
  'human_dignity': { article: 'Article 1', relevantDomains: ['all'] },
  'right_to_life': { article: 'Article 2', relevantDomains: ['healthcare', 'autonomous_vehicles'] },
  'integrity_of_person': { article: 'Article 3', relevantDomains: ['healthcare', 'biometrics'] },
  'prohibition_of_torture': { article: 'Article 4', relevantDomains: ['law_enforcement'] },
  'right_to_liberty': { article: 'Article 6', relevantDomains: ['law_enforcement', 'migration'] },
  'private_life': { article: 'Article 7', relevantDomains: ['surveillance', 'social_media'] },
  'data_protection': { article: 'Article 8', relevantDomains: ['all_processing_personal_data'] },
  'non_discrimination': { article: 'Article 21', relevantDomains: ['employment', 'credit', 'education'] },
  'rights_of_child': { article: 'Article 24', relevantDomains: ['education', 'social_media'] },
  'rights_of_elderly': { article: 'Article 25', relevantDomains: ['healthcare', 'social_services'] },
  'fair_working_conditions': { article: 'Article 31', relevantDomains: ['employment'] },
  'consumer_protection': { article: 'Article 38', relevantDomains: ['commerce', 'financial_services'] },
  'right_to_good_administration': { article: 'Article 41', relevantDomains: ['public_services'] },
  'effective_remedy': { article: 'Article 47', relevantDomains: ['all'] }
};

class FRIAAssessmentEngine {
  assessSystem(systemProfile: AISystemProfile): FRIAReport {
    const affectedRights = this.identifyAffectedRights(systemProfile);
    const riskAnalysis = this.analyzeRisks(affectedRights, systemProfile);
    const mitigations = this.proposeMitigations(riskAnalysis);
    
    return {
      assessmentId: crypto.randomUUID(),
      assessmentDate: new Date().toISOString(),
      assessor: 'AI Governance Team',
      systemDescription: systemProfile.purpose,
      rightsImpacted: affectedRights,
      riskAnalysis,
      mitigations,
      consultationRecords: [],
      nextReviewDate: this.calculateNextReview(),
      reviewFrequency: 'quarterly'
    };
  }
  
  private identifyAffectedRights(system: AISystemProfile): FundamentalRight[] {
    const rights: FundamentalRight[] = [];
    
    for (const [rightKey, mapping] of Object.entries(EU_CHARTER_RIGHTS_MAPPING)) {
      const isRelevant = mapping.relevantDomains.includes('all') ||
        mapping.relevantDomains.some(d => system.domain.includes(d));
      
      if (isRelevant) {
        rights.push({
          rightId: rightKey,
          rightName: rightKey.replace(/_/g, ' '),
          charterArticle: mapping.article,
          impactLevel: this.assessImpactLevel(system, rightKey),
          justification: ''
        });
      }
    }
    
    return rights;
  }
  
  private assessImpactLevel(
    system: AISystemProfile, 
    rightKey: string
  ): FundamentalRight['impactLevel'] {
    // Scoring based on data sensitivity and decision autonomy
    if (system.outputAffectsRights && system.usesPersonalData) {
      return 'significant';
    }
    if (system.outputAffectsRights || system.usesPersonalData) {
      return 'moderate';
    }
    return 'minimal';
  }
  
  private analyzeRisks(
    rights: FundamentalRight[], 
    system: AISystemProfile
  ): RightRiskAnalysis[] {
    return rights
      .filter(r => r.impactLevel !== 'none')
      .map(right => ({
        rightId: right.rightId,
        riskDescription: `Potential impact on ${right.rightName} through ${system.purpose}`,
        affectedGroups: this.identifyAffectedGroups(system),
        likelihoodScore: this.scoreLikelihood(right.impactLevel),
        severityScore: this.scoreSeverity(right.impactLevel),
        riskScore: this.scoreLikelihood(right.impactLevel) * this.scoreSeverity(right.impactLevel),
        existingControls: [],
        residualRiskLevel: ''
      }));
  }
  
  private proposeMitigations(risks: RightRiskAnalysis[]): MitigationMeasure[] {
    return risks
      .filter(r => r.riskScore >= 9) // High risk threshold
      .map(risk => ({
        measureId: crypto.randomUUID(),
        targetRightId: risk.rightId,
        description: `Implement safeguards for ${risk.rightId}`,
        implementationType: 'technical' as const,
        status: 'planned' as const,
        effectivenessMetric: 'Risk score reduction below threshold',
        responsiblePerson: 'AI Safety Lead',
        deadline: '2026-06-01'
      }));
  }
  
  private identifyAffectedGroups(system: AISystemProfile): string[] {
    return ['end_users', 'data_subjects'];
  }
  
  private scoreLikelihood(level: string): number {
    const map: Record<string, number> = {
      'minimal': 1, 'moderate': 3, 'significant': 4, 'severe': 5
    };
    return map[level] || 2;
  }
  
  private scoreSeverity(level: string): number {
    const map: Record<string, number> = {
      'minimal': 1, 'moderate': 2, 'significant': 4, 'severe': 5
    };
    return map[level] || 2;
  }
  
  private calculateNextReview(): string {
    const next = new Date();
    next.setMonth(next.getMonth() + 3);
    return next.toISOString().split('T')[0];
  }
}

# .github/workflows/bias-testing.yml
name: EU AI Act Bias Testing Pipeline

on:
  push:
    paths:
      - 'models/**'
      - 'data/**'
  schedule:
    - cron: '0 6 * * 1'  # Weekly Monday 6AM

env:
  BIAS_THRESHOLD_DEMOGRAPHIC_PARITY: 0.1
  BIAS_THRESHOLD_EQUALIZED_ODDS: 0.05
  BIAS_THRESHOLD_DISPARATE_IMPACT: 0.8

jobs:
  bias-audit:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      
      - name: Install dependencies
        run: |
          pip install fairlearn aequitas scikit-learn pandas numpy
      
      - name: Run demographic parity test
        run: python tests/bias/test_demographic_parity.py
      
      - name: Run equalized odds test
        run: python tests/bias/test_equalized_odds.py
      
      - name: Run disparate impact analysis
        run: python tests/bias/test_disparate_impact.py
      
      - name: Generate bias report
        run: python tests/bias/generate_report.py
      
      - name: Upload bias report artifact
        uses: actions/upload-artifact@v4
        with:
          name: bias-report-${{ github.sha }}
          path: reports/bias/
      
      - name: Fail if bias exceeds threshold
        run: python tests/bias/check_thresholds.py

# tests/bias/test_demographic_parity.py
import numpy as np
import pandas as pd
from dataclasses import dataclass
from typing import Dict, List, Tuple
from fairlearn.metrics import (
    demographic_parity_difference,
    equalized_odds_difference,
    MetricFrame
)
from sklearn.metrics import accuracy_score, precision_score, recall_score

@dataclass
class BiasTestResult:
    test_name: str
    metric_name: str
    overall_value: float
    group_values: Dict[str, float]
    threshold: float
    passed: bool
    protected_attribute: str
    sample_size: int
    confidence_level: float
    recommendations: List[str]

class EUAIActBiasTestSuite:
    """Comprehensive bias testing per Article 10 requirements"""
    
    def __init__(self, config: Dict):
        self.demographic_parity_threshold = config.get(
            'demographic_parity_threshold', 0.1
        )
        self.equalized_odds_threshold = config.get(
            'equalized_odds_threshold', 0.05
        )
        self.disparate_impact_threshold = config.get(
            'disparate_impact_threshold', 0.8
        )
        self.protected_attributes = config.get(
            'protected_attributes', 
            ['gender', 'age_group', 'ethnicity', 'disability_status']
        )
    
    def run_full_audit(
        self, 
        y_true: np.ndarray, 
        y_pred: np.ndarray,
        sensitive_features: pd.DataFrame
    ) -> List[BiasTestResult]:
        results = []
        
        for attr in self.protected_attributes:
            if attr not in sensitive_features.columns:
                continue
            
            sf = sensitive_features[attr]
            
            # Test 1: Demographic Parity
            dp_result = self._test_demographic_parity(y_pred, sf, attr)
            results.append(dp_result)
            
            # Test 2: Equalized Odds
            eo_result = self._test_equalized_odds(y_true, y_pred, sf, attr)
            results.append(eo_result)
            
            # Test 3: Disparate Impact Ratio
            di_result = self._test_disparate_impact(y_pred, sf, attr)
            results.append(di_result)
            
            # Test 4: Calibration across groups
            cal_result = self._test_calibration(y_true, y_pred, sf, attr)
            results.append(cal_result)
        
        return results
    
    def _test_demographic_parity(
        self, 
        y_pred: np.ndarray, 
        sensitive_feature: pd.Series,
        attr_name: str
    ) -> BiasTestResult:
        dpd = demographic_parity_difference(
            y_true=y_pred,  # Using predictions as "true" for selection rate
            y_pred=y_pred,
            sensitive_features=sensitive_feature
        )
        
        # Per-group selection rates
        groups = sensitive_feature.unique()
        group_rates = {}
        for group in groups:
            mask = sensitive_feature == group
            group_rates[str(group)] = float(y_pred[mask].mean())
        
        passed = abs(dpd) <= self.demographic_parity_threshold
        
        recommendations = []
        if not passed:
            recommendations.append(
                f"Demographic parity violation detected for {attr_name}. "
                f"Consider resampling training data or applying post-processing calibration."
            )
            recommendations.append(
                "Review training data representation per Article 10(3)."
            )
        
        return BiasTestResult(
            test_name=f'demographic_parity_{attr_name}',
            metric_name='Demographic Parity Difference',
            overall_value=abs(dpd),
            group_values=group_rates,
            threshold=self.demographic_parity_threshold,
            passed=passed,
            protected_attribute=attr_name,
            sample_size=len(y_pred),
            confidence_level=0.95,
            recommendations=recommendations
        )
    
    def _test_equalized_odds(
        self,
        y_true: np.ndarray,
        y_pred: np.ndarray,
        sensitive_feature: pd.Series,
        attr_name: str
    ) -> BiasTestResult:
        eod = equalized_odds_difference(
            y_true=y_true,
            y_pred=y_pred,
            sensitive_features=sensitive_feature
        )
        
        # Per-group TPR and FPR
        groups = sensitive_feature.unique()
        group_values = {}
        for group in groups:
            mask = sensitive_feature == group
            tpr = recall_score(y_true[mask], y_pred[mask], zero_division=0)
            group_values[f'{group}_TPR'] = float(tpr)
        
        passed = abs(eod) <= self.equalized_odds_threshold
        
        recommendations = []
        if not passed:
            recommendations.append(
                f"Equalized odds violation for {attr_name}. "
                f"Model has different error rates across groups."
            )
        
        return BiasTestResult(
            test_name=f'equalized_odds_{attr_name}',
            metric_name='Equalized Odds Difference',
            overall_value=abs(eod),
            group_values=group_values,
            threshold=self.equalized_odds_threshold,
            passed=passed,
            protected_attribute=attr_name,
            sample_size=len(y_pred),
            confidence_level=0.95,
            recommendations=recommendations
        )
    
    def _test_disparate_impact(
        self,
        y_pred: np.ndarray,
        sensitive_feature: pd.Series,
        attr_name: str
    ) -> BiasTestResult:
        groups = sensitive_feature.unique()
        selection_rates = {}
        for group in groups:
            mask = sensitive_feature == group
            selection_rates[str(group)] = float(y_pred[mask].mean())
        
        max_rate = max(selection_rates.values())
        min_rate = min(selection_rates.values())
        
        # 这是一个可选的政策指标，不是 EU AI Act 的通用法律门槛。
        di_ratio = min_rate / max_rate if max_rate > 0 else 1.0
        passed = di_ratio >= self.disparate_impact_threshold
        
        recommendations = []
        if not passed:
            recommendations.append(
                f"Disparate impact ratio {di_ratio:.3f} below the configured policy threshold. "
                f"Requires legal and domain review; it is not a legal conclusion."
            )
        
        return BiasTestResult(
            test_name=f'disparate_impact_{attr_name}',
            metric_name='Disparate Impact Ratio',
            overall_value=di_ratio,
            group_values=selection_rates,
            threshold=self.disparate_impact_threshold,
            passed=passed,
            protected_attribute=attr_name,
            sample_size=len(y_pred),
            confidence_level=0.95,
            recommendations=recommendations
        )
    
    def _test_calibration(
        self,
        y_true: np.ndarray,
        y_pred: np.ndarray,
        sensitive_feature: pd.Series,
        attr_name: str
    ) -> BiasTestResult:
        groups = sensitive_feature.unique()
        group_accuracy = {}
        for group in groups:
            mask = sensitive_feature == group
            acc = accuracy_score(y_true[mask], y_pred[mask])
            group_accuracy[str(group)] = float(acc)
        
        max_acc = max(group_accuracy.values())
        min_acc = min(group_accuracy.values())
        calibration_gap = max_acc - min_acc
        
        passed = calibration_gap <= 0.05
        
        return BiasTestResult(
            test_name=f'calibration_{attr_name}',
            metric_name='Calibration Gap',
            overall_value=calibration_gap,
            group_values=group_accuracy,
            threshold=0.05,
            passed=passed,
            protected_attribute=attr_name,
            sample_size=len(y_pred),
            confidence_level=0.95,
            recommendations=[]
        )

// human-oversight.ts - Article 14 Human Oversight System
interface HumanOversightConfig {
  confidenceThreshold: number;
  sensitiveDecisionTypes: string[];
  maxAutoDecisionsBeforeReview: number;
  escalationTimeoutMs: number;
  requiredApproverRole: string;
}

interface AIDecision {
  decisionId: string;
  systemOutput: any;
  confidenceScore: number;
  decisionType: string;
  affectedPersonIds: string[];
  explanation: string;
  featureContributions: Record<string, number>;
  timestamp: string;
}

interface HumanReview {
  reviewId: string;
  decisionId: string;
  reviewerId: string;
  reviewerRole: string;
  action: 'approve' | 'modify' | 'reject' | 'escalate';
  modifiedOutput?: any;
  justification: string;
  reviewTimestamp: string;
  timeToDecisionMs: number;
}

class HumanOversightGateway {
  private config: HumanOversightConfig;
  private autoDecisionCounter: Map<string, number> = new Map();
  private auditLogger: EUAIActAuditLogger;
  
  constructor(config: HumanOversightConfig, auditLogger: EUAIActAuditLogger) {
    this.config = config;
    this.auditLogger = auditLogger;
  }
  
  async processDecision(decision: AIDecision): Promise<{
    finalOutput: any;
    requiresHumanReview: boolean;
    reviewResult?: HumanReview;
  }> {
    const requiresReview = this.requiresHumanReview(decision);
    
    if (!requiresReview) {
      // Auto-approve but still log
      await this.auditLogger.logInference({
        input: decision.decisionId,
        output: decision.systemOutput,
        modelVersion: 'current',
        confidenceScore: decision.confidenceScore,
        featureImportance: decision.featureContributions,
        dataSubjectCategories: decision.affectedPersonIds.length > 0 
          ? ['natural_persons'] 
          : []
      });
      
      this.incrementAutoCounter(decision.decisionType);
      
      return {
        finalOutput: decision.systemOutput,
        requiresHumanReview: false
      };
    }
    
    // Route to human review
    const review = await this.requestHumanReview(decision);
    
    // Log the human override
    await this.auditLogger.logHumanOverride({
      originalEventId: decision.decisionId,
      operatorId: review.reviewerId,
      decision: review.action,
      reason: review.justification
    });
    
    // Reset auto-decision counter after human review
    this.autoDecisionCounter.set(decision.decisionType, 0);
    
    const finalOutput = review.action === 'modify' 
      ? review.modifiedOutput 
      : review.action === 'approve' 
        ? decision.systemOutput 
        : null;
    
    return {
      finalOutput,
      requiresHumanReview: true,
      reviewResult: review
    };
  }
  
  private requiresHumanReview(decision: AIDecision): boolean {
    // Rule 1: Below confidence threshold
    if (decision.confidenceScore < this.config.confidenceThreshold) {
      return true;
    }
    
    // Rule 2: Sensitive decision type
    if (this.config.sensitiveDecisionTypes.includes(decision.decisionType)) {
      return true;
    }
    
    // Rule 3: Max auto-decisions reached (periodic review)
    const counter = this.autoDecisionCounter.get(decision.decisionType) || 0;
    if (counter >= this.config.maxAutoDecisionsBeforeReview) {
      return true;
    }
    
    // Rule 4: Affects vulnerable persons
    if (decision.affectedPersonIds.length > 10) {
      return true;
    }
    
    return false;
  }
  
  private async requestHumanReview(decision: AIDecision): Promise<HumanReview> {
    // In production, this would integrate with a review queue system
    // (e.g., internal dashboard, Slack workflow, etc.)
    throw new Error(
      `Human review required for decision ${decision.decisionId}. ` +
      `Route to review queue with timeout ${this.config.escalationTimeoutMs}ms.`
    );
  }
  
  private incrementAutoCounter(decisionType: string): void {
    const current = this.autoDecisionCounter.get(decisionType) || 0;
    this.autoDecisionCounter.set(decisionType, current + 1);
  }
  
  // Article 14(4)(d) - "stop button" capability
  async emergencyStop(reason: string, operatorId: string): Promise<void> {
    await this.auditLogger.logHumanOverride({
      originalEventId: 'EMERGENCY_STOP',
      operatorId,
      decision: 'system_halt',
      reason
    });
    
    // Halt all pending decisions
    // Notify all affected operators
    // Preserve system state for investigation
  }
}

// compliance-assessment.ts - Article 43 Conformity Assessment
interface ConformityChecklist {
  systemId: string;
  assessmentDate: string;
  assessmentType: 'internal' | 'third_party';
  
  // Article 9: Risk Management System
  riskManagement: {
    riskIdentified: boolean;
    mitigationImplemented: boolean;
    residualRiskAcceptable: boolean;
    continuousMonitoring: boolean;
  };
  
  // Article 10: Data Governance
  dataGovernance: {
    trainingDataDocumented: boolean;
    biasExamined: boolean;
    dataQualityMetrics: boolean;
    representativenessVerified: boolean;
  };
  
  // Article 11: Technical Documentation
  technicalDocumentation: {
    annexIVComplete: boolean;
    modelCardGenerated: boolean;
    architectureDocumented: boolean;
    changeLogMaintained: boolean;
  };
  
  // Article 12: Record Keeping
  recordKeeping: {
    auditLogsEnabled: boolean;
    tamperEvident: boolean;
    retentionPolicySet: boolean;
    accessControlImplemented: boolean;
  };
  
  // Article 13: Transparency
  transparency: {
    instructionsForUse: boolean;
    capabilitiesDocumented: boolean;
    limitationsDocumented: boolean;
    humanReadableExplanations: boolean;
  };
  
  // Article 14: Human Oversight
  humanOversight: {
    oversightMechanismDesigned: boolean;
    stopMechanismAvailable: boolean;
    overridePossible: boolean;
    operatorTraining: boolean;
  };
  
  // Article 15: Accuracy, Robustness, Cybersecurity
  technicalRobustness: {
    accuracyMetricsDeclared: boolean;
    robustnessTestsPassed: boolean;
    cybersecurityAssessed: boolean;
    adversarialTestingDone: boolean;
  };
}

class ConformityAssessmentEngine {
  async runAssessment(systemId: string): Promise<{
    checklist: ConformityChecklist;
    overallResult: 'PASS' | 'FAIL' | 'PARTIAL';
    failedItems: string[];
    recommendations: string[];
  }> {
    const checklist = await this.gatherEvidence(systemId);
    const failedItems = this.identifyFailures(checklist);
    
    return {
      checklist,
      overallResult: failedItems.length === 0 ? 'PASS' : 
                     failedItems.length <= 3 ? 'PARTIAL' : 'FAIL',
      failedItems,
      recommendations: this.generateRecommendations(failedItems)
    };
  }
  
  private identifyFailures(checklist: ConformityChecklist): string[] {
    const failures: string[] = [];
    
    // Check each category
    const categories = [
      'riskManagement', 'dataGovernance', 'technicalDocumentation',
      'recordKeeping', 'transparency', 'humanOversight', 'technicalRobustness'
    ] as const;
    
    for (const category of categories) {
      const section = checklist[category];
      for (const [key, value] of Object.entries(section)) {
        if (value === false) {
          failures.push(`${category}.${key}`);
        }
      }
    }
    
    return failures;
  }
  
  private generateRecommendations(failures: string[]): string[] {
    const recommendations: string[] = [];
    
    if (failures.some(f => f.startsWith('recordKeeping'))) {
      recommendations.push(
        'Implement tamper-evident audit logging with chain hashing. ' +
        'See Article 12 middleware implementation.'
      );
    }
    
    if (failures.some(f => f.startsWith('dataGovernance'))) {
      recommendations.push(
        'Run bias testing pipeline and document training data provenance. ' +
        'Ensure demographic representation analysis per Article 10(3).'
      );
    }
    
    if (failures.some(f => f.startsWith('humanOversight'))) {
      recommendations.push(
        'Deploy human-in-the-loop gateway with confidence thresholds. ' +
        'Implement emergency stop mechanism per Article 14(4)(d).'
      );
    }
    
    return recommendations;
  }
  
  private async gatherEvidence(systemId: string): Promise<ConformityChecklist> {
    // In production, this queries various system components
    // to automatically verify compliance status
    throw new Error('Implementation depends on system architecture');
  }
}

# ai-literacy-program.yml - Article 4 AI Literacy Program
program:
  name: "EU AI Act Literacy Program"
  version: "2.0"
  effective_date: "2025-02-02"
  review_cycle: "quarterly"

target_audiences:
  - role: "developers"
    required_modules:
      - ai_fundamentals
      - bias_and_fairness
      - eu_ai_act_technical_requirements
      - secure_ai_development
      - testing_and_validation
    assessment_passing_score: 80
    recertification_months: 12
    
  - role: "product_managers"
    required_modules:
      - ai_fundamentals
      - risk_classification
      - eu_ai_act_business_impact
      - responsible_ai_governance
    assessment_passing_score: 75
    recertification_months: 12
    
  - role: "executives"
    required_modules:
      - ai_strategic_overview
      - eu_ai_act_business_impact
      - liability_and_penalties
      - governance_frameworks
    assessment_passing_score: 70
    recertification_months: 24
    
  - role: "human_reviewers"
    required_modules:
      - ai_fundamentals
      - understanding_ai_outputs
      - override_procedures
      - bias_recognition
      - documentation_requirements
    assessment_passing_score: 85
    recertification_months: 6

modules:
  ai_fundamentals:
    title: "AI基础知识"
    duration_hours: 4
    topics:
      - "机器学习基本概念"
      - "深度学习与神经网络"
      - "LLM工作原理"
      - "AI系统局限性"
    
  bias_and_fairness:
    title: "偏差与公平性"
    duration_hours: 6
    topics:
      - "算法偏差的来源"
      - "公平性度量标准"
      - "偏差检测方法"
      - "缓解策略"
    
  eu_ai_act_technical_requirements:
    title: "欧盟AI法案技术要求"
    duration_hours: 8
    topics:
      - "风险分类实操"
      - "附件IV文档要求"
      - "审计日志实现"
      - "人工监督设计模式"
      - "合规性评估流程"
    
  risk_classification:
    title: "风险分类决策"
    duration_hours: 3
    topics:
      - "四级风险体系"
      - "附件III高风险清单"
      - "例外条件判断"
      - "跨境适用规则"

compliance_tracking:
  database: "compliance_db"
  table: "ai_literacy_records"
  required_fields:
    - employee_id
    - module_completed
    - completion_date
    - assessment_score
    - certificate_expiry
  
  reporting:
    frequency: "monthly"
    recipients: ["compliance_officer", "cto", "hr_director"]
    include_metrics:
      - completion_rate_by_department
      - average_scores_by_role
      - overdue_recertifications
      - training_gap_analysis

// post-market-monitoring.ts - Article 72 Post-Market Monitoring
interface MonitoringAlert {
  alertId: string;
  severity: 'critical' | 'high' | 'medium' | 'low';
  category: 'performance_degradation' | 'bias_drift' | 'security_incident' 
    | 'user_complaint' | 'adverse_event';
  description: string;
  detectedAt: string;
  affectedUsers: number;
  automaticActions: string[];
  requiredHumanAction: string;
}

interface MonitoringMetrics {
  // Performance metrics
  accuracyScore: number;
  precisionScore: number;
  recallScore: number;
  latencyP99Ms: number;
  
  // Fairness metrics
  demographicParityGap: number;
  equalizedOddsGap: number;
  
  // Operational metrics
  humanOverrideRate: number;
  userComplaintRate: number;
  systemAvailability: number;
  
  // Drift metrics
  dataDistributionDrift: number;
  predictionDistributionDrift: number;
  featureImportanceDrift: number;
}

class PostMarketMonitor {
  private alertThresholds: Record<string, number>;
  private baselineMetrics: MonitoringMetrics;
  
  constructor(baseline: MonitoringMetrics) {
    this.baselineMetrics = baseline;
    this.alertThresholds = {
      accuracy_drop: 0.05,
      bias_increase: 0.03,
      drift_threshold: 0.1,
      complaint_rate: 0.01,
      override_rate_increase: 0.15
    };
  }
  
  async evaluateMetrics(current: MonitoringMetrics): Promise<MonitoringAlert[]> {
    const alerts: MonitoringAlert[] = [];
    
    // Check accuracy degradation
    const accuracyDrop = this.baselineMetrics.accuracyScore - current.accuracyScore;
    if (accuracyDrop > this.alertThresholds.accuracy_drop) {
      alerts.push({
        alertId: crypto.randomUUID(),
        severity: accuracyDrop > 0.1 ? 'critical' : 'high',
        category: 'performance_degradation',
        description: `Accuracy dropped by ${(accuracyDrop * 100).toFixed(1)}% from baseline`,
        detectedAt: new Date().toISOString(),
        affectedUsers: 0,
        automaticActions: ['pause_auto_decisions', 'increase_human_review_rate'],
        requiredHumanAction: 'Investigate root cause and retrain if necessary'
      });
    }
    
    // Check bias drift
    const biasIncrease = current.demographicParityGap - this.baselineMetrics.demographicParityGap;
    if (biasIncrease > this.alertThresholds.bias_increase) {
      alerts.push({
        alertId: crypto.randomUUID(),
        severity: 'high',
        category: 'bias_drift',
        description: `Demographic parity gap increased by ${(biasIncrease * 100).toFixed(1)}%`,
        detectedAt: new Date().toISOString(),
        affectedUsers: 0,
        automaticActions: ['trigger_bias_audit', 'notify_dpo'],
        requiredHumanAction: 'Run full bias test suite and document findings'
      });
    }
    
    // Check data distribution drift
    if (current.dataDistributionDrift > this.alertThresholds.drift_threshold) {
      alerts.push({
        alertId: crypto.randomUUID(),
        severity: 'medium',
        category: 'performance_degradation',
        description: `Data distribution drift detected: ${current.dataDistributionDrift.toFixed(3)}`,
        detectedAt: new Date().toISOString(),
        affectedUsers: 0,
        automaticActions: ['log_drift_event', 'schedule_retrain_evaluation'],
        requiredHumanAction: 'Evaluate whether model retraining is required'
      });
    }
    
    // Check human override rate spike
    const overrideIncrease = current.humanOverrideRate - this.baselineMetrics.humanOverrideRate;
    if (overrideIncrease > this.alertThresholds.override_rate_increase) {
      alerts.push({
        alertId: crypto.randomUUID(),
        severity: 'medium',
        category: 'performance_degradation',
        description: `Human override rate increased by ${(overrideIncrease * 100).toFixed(1)}%`,
        detectedAt: new Date().toISOString(),
        affectedUsers: 0,
        automaticActions: ['analyze_override_patterns'],
        requiredHumanAction: 'Review override reasons and assess model fitness'
      });
    }
    
    return alerts;
  }
  
  // Article 72(2): Serious incident reporting to authorities
  async reportSeriousIncident(incident: {
    description: string;
    affectedPersons: number;
    harmType: string;
    immediateActions: string[];
  }): Promise<string> {
    // Must report within 15 days of becoming aware
    const report = {
      reportId: crypto.randomUUID(),
      reportDate: new Date().toISOString(),
      deadline: this.calculateReportingDeadline(),
      ...incident
    };
    
    // Submit to EU AI database
    // Notify national market surveillance authority
    return report.reportId;
  }
  
  private calculateReportingDeadline(): string {
    const deadline = new Date();
    deadline.setDate(deadline.getDate() + 15);
    return deadline.toISOString();
  }
}

// gpai-compliance.ts - Articles 51-56 GPAI Model Obligations
interface GPAIModelCompliance {
  // Article 53: General obligations
  technicalDocumentation: {
    modelArchitecture: boolean;        // Detailed architecture description
    trainingProcess: boolean;          // Training methodology and resources
    evaluationResults: boolean;        // Benchmark results
    knownLimitations: boolean;         // Documented limitations
    energyConsumption: boolean;        // Training energy usage reported
  };
  
  // Copyright compliance
  copyrightPolicy: {
    trainingDataCopyrightAnalysis: boolean;
    optOutMechanismImplemented: boolean;  // Article 53(1)(c)
    detailedSummaryOfTrainingData: boolean;
    euCopyrightDirectiveCompliance: boolean;
  };
  
  // Downstream provider support
  downstreamSupport: {
    sufficientInfoProvided: boolean;
    integrationGuidelines: boolean;
    complianceDocumentation: boolean;
  };
  
  // Systemic risk (for models with high impact)
  systemicRisk?: {
    modelEvaluationPerformed: boolean;  // Article 55
    adversarialTestingDone: boolean;
    riskMitigationPlan: boolean;
    incidentReportingProcess: boolean;
    cybersecurityMeasures: boolean;
  };
}

// Systemic-risk status is not determined by a universal user-count or
// training-compute threshold in this application code. Check the current
// GPAI provisions, Commission designation, and provider evidence instead.

# jurisdiction_check.py - Extraterritorial application check
from dataclasses import dataclass
from typing import List, Optional

@dataclass
class JurisdictionAnalysis:
    subject_to_eu_ai_act: Optional[bool]
    applicable_articles: List[str]
    reason: str
    recommended_actions: List[str]

def check_eu_ai_act_applicability(
    company_registration: str,      # Country of registration
    ai_output_used_in_eu: bool,     # Whether outputs affect EU persons
    eu_users_count: int,            # Number of EU-based users
    ai_system_type: str,            # Type of AI system
    data_subjects_in_eu: bool       # Whether processes EU personal data
) -> JurisdictionAnalysis:
    """Engineering triage only. Confirm Article 2 with legal counsel.
    The result depends on provider/deployer role, market placement,
    deployment location, output use, and applicable transitional rules.
    """
    
    reasons = []
    applicable_articles = []
    
    # Territorial scope check
    if ai_output_used_in_eu:
        reasons.append("Output-use fact requires Article 2 applicability review")
    
    if eu_users_count > 0:
        reasons.append(
            f"System has {eu_users_count} EU-based users"
        )
    
    if data_subjects_in_eu:
        reasons.append(
            "Processes personal data of EU data subjects"
        )
    
    # Presence of EU users or data is a review signal, not an automatic
    # legal conclusion. Resolve the role and Article 2 facts with counsel.
    subject_to_act = None
    
    recommended_actions = []
    if reasons:
        recommended_actions = [
            "Map provider/deployer role and Article 2 facts",
            "Check whether registration or an EU representative duty applies",
            "Classify intended purpose under Article 5, Article 6 and Annex III",
            "Record applicable dates and transition rules",
            "Escalate to legal and domain review before deployment"
        ]
    
    return JurisdictionAnalysis(
        subject_to_eu_ai_act=subject_to_act,
        applicable_articles=applicable_articles,
        reason=" | ".join(reasons) if reasons else "Insufficient facts; obtain an applicability review",
        recommended_actions=recommended_actions
    )


# Example: Chinese company with global AI product
result = check_eu_ai_act_applicability(
    company_registration="CN",
    ai_output_used_in_eu=True,  # EU users receive AI-generated content
    eu_users_count=50000,
    ai_system_type="recommendation_system",
    data_subjects_in_eu=True
)

print(f"EU AI Act applicability result: {result.subject_to_eu_ai_act}")
# Output: None means the engineering helper deliberately does not decide.

# compliance-checklist.yml - Pre-launch verification
pre_launch_checklist:
  risk_classification:
    - system_risk_level_determined: false
    - prohibited_practices_excluded: false
    - annex_iii_mapping_complete: false
    
  technical_documentation:
    - model_card_generated: false
    - architecture_documented: false
    - training_data_provenance: false
    - performance_benchmarks: false
    - risk_assessment_complete: false
    
  audit_logging:
    - logging_middleware_deployed: false
    - tamper_evident_chain: false
    - retention_policy_configured: false
    - access_control_verified: false
    
  bias_testing:
    - demographic_parity_tested: false
    - equalized_odds_tested: false
    - disparate_impact_checked: false
    - ci_cd_pipeline_integrated: false
    
  human_oversight:
    - confidence_threshold_set: false
    - override_mechanism_tested: false
    - emergency_stop_verified: false
    - operator_training_complete: false
    
  transparency:
    - user_notification_implemented: false
    - explanation_mechanism_available: false
    - limitations_documented: false
    
  post_market:
    - monitoring_system_deployed: false
    - alert_thresholds_configured: false
    - incident_reporting_process: false
    - periodic_review_scheduled: false
    
  organizational:
    - eu_representative_appointed: false
    - ai_literacy_program_launched: false
    - quality_management_system: false
    - fria_completed: false
    - ce_marking_documentation: false