EU AI Act in Practice

"""EU AI Act risk classifier with Pydantic validation and cited reasoning.

The classifier examines an LLM application's domain, data-processing pattern,
and user-facing properties, then returns the EU AI Act risk tier together
with the specific Article / Annex paragraph that justifies the verdict.
Legal teams can audit the reasoning trail; engineering teams can wire the
output into a compliance dashboard.
"""
from enum import Enum
from pydantic import BaseModel, Field, field_validator

class RiskTier(str, Enum):
    PROHIBITED   = "prohibited"      # Art.5
    HIGH         = "high"            # Art.6 + Annex III
    LIMITED      = "limited"         # Art.50 transparency obligations
    MINIMAL      = "minimal"         # No specific obligations

# Annex III enumerates the eight high-risk domains. The sub-domains within
# each are the concrete use cases the regulation calls out. This dict is a
# faithful (abbreviated) transcription, NOT an exhaustive legal list.
ANNEX_III_DOMAINS = {
    "employment":          ["recruitment", "hiring", "performance_evaluation",
                            "task_allocation", "termination"],
    "education":           ["admissions", "grading", "learning_assessment",
                            "student_monitoring"],
    "essential_services":  ["credit_scoring", "insurance_underwriting",
                            "emergency_dispatch", "utility_access"],
    "law_enforcement":     ["crime_prediction", "evidence_evaluation",
                            "profiling", "polygraph_replacement"],
    "migration":           ["visa_processing", "asylum_assessment",
                            "border_control", "document_verification"],
    "justice":             ["sentencing_support", "legal_research_for_judges",
                            "dispute_resolution"],
    "biometrics":          ["identity_verification", "emotion_recognition",
                            "categorization"],
    "critical_infrastructure": ["traffic_management", "power_grid_control",
                                "water_supply_control"],
}

# Article 5 prohibits these regardless of domain.
PROHIBITED_SUBDOMAINS = {
    "social_scoring",
    "subliminal_manipulation",
    "real_time_biometric_identification",
    "predictive_policing_individuals",
    "emotion_inference_workplace",
}

class LLMApplication(BaseModel):
    name:                       str
    description:                str
    domain:                     str
    sub_domain:                 str
    affects_individuals_rights: bool
    user_facing:                bool
    makes_automated_decisions:  bool

    @field_validator("name", "description", "domain", "sub_domain")
    @classmethod
    def non_empty(cls, v):
        if not v.strip():
            raise ValueError("must be non-empty")
        return v

class Verdict(BaseModel):
    tier:      RiskTier
    article:   str               # e.g. "Art.5(1)(c)" or "Art.6 / Annex III"
    reasoning: str               # one-sentence justification
    next_steps: list[str]        # concrete obligations the team must fulfill

def classify(app: LLMApplication) -> Verdict:
    """Return the risk tier, the cited Article, and concrete obligations."""
    if app.sub_domain in PROHIBITED_SUBDOMAINS:
        return Verdict(
            tier=RiskTier.PROHIBITED,
            article="Art.5(1)",
            reasoning=f"{app.sub_domain!r} is enumerated in Article 5 prohibited practices.",
            next_steps=["Do not deploy. Redesign the use case to remove the prohibited capability."],
        )
    annex_iii_subdomains = ANNEX_III_DOMAINS.get(app.domain, [])
    in_annex_iii = app.sub_domain in annex_iii_subdomains
    if in_annex_iii and app.makes_automated_decisions:
        return Verdict(
            tier=RiskTier.HIGH,
            article="Art.6 / Annex III",
            reasoning=(f"{app.domain}/{app.sub_domain} is an Annex III high-risk use case "
                       f"AND the system makes automated decisions affecting individuals."),
            next_steps=[
                "Complete a conformity assessment (Art.43)",
                "Maintain technical documentation (Annex IV)",
                "Implement human oversight (Art.14)",
                "Register in the EU AI Act database (Art.49)",
                "Maintain post-market monitoring (Art.72)",
            ],
        )
    if app.user_facing and not app.makes_automated_decisions:
        return Verdict(
            tier=RiskTier.LIMITED,
            article="Art.50",
            reasoning="User-facing AI without automated decision-making is subject to transparency obligations.",
            next_steps=[
                "Inform users they are interacting with an AI system",
                "Label AI-generated content (deepfakes, synthetic media)",
            ],
        )
    return Verdict(
        tier=RiskTier.MINIMAL,
        article="(none)",
        reasoning="No specific AI Act obligations beyond existing horizontal regulations (GDPR, sector law).",
        next_steps=["Continue normal product development; document risk position for the AI act register."],
    )

# Demo: four applications covering every tier
apps = [
    LLMApplication(name="ResumeScreener", description="Filters job applications",
                   domain="employment", sub_domain="recruitment",
                   affects_individuals_rights=True, user_facing=False,
                   makes_automated_decisions=True),
    LLMApplication(name="DeepfakeBot", description="Generates synthetic videos of real people",
                   domain="media", sub_domain="subliminal_manipulation",
                   affects_individuals_rights=True, user_facing=True,
                   makes_automated_decisions=False),
    LLMApplication(name="SupportBot", description="Answers product questions",
                   domain="retail", sub_domain="customer_support",
                   affects_individuals_rights=False, user_facing=True,
                   makes_automated_decisions=False),
    LLMApplication(name="CodeCompletion", description="Suggests code completions in an IDE",
                   domain="engineering", sub_domain="development_tools",
                   affects_individuals_rights=False, user_facing=True,
                   makes_automated_decisions=False),
]

for app in apps:
    v = classify(app)
    print(f"{app.name:18s} -> {v.tier.value:10s} ({v.article})")
    print(f"  Why: {v.reasoning}")
    for step in v.next_steps[:2]:
        print(f"  TODO: {step}")
    print()

# Automated EU AI Act compliance checker
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
import json
class ComplianceStatus(Enum):
    COMPLIANT = "compliant"
    NON_COMPLIANT = "non_compliant"
    NEEDS_REVIEW = "needs_review"
@dataclass
class ComplianceCheck:
    article: str
    requirement: str
    status: ComplianceStatus
    evidence: str
    remediation: str = ""
def check_transparency_obligations(app_config: dict) -> ComplianceCheck:
    """Article 13: Check that AI disclosure is present."""
    has_disclosure = app_config.get("ai_disclosure_enabled", False)
    disclosure_text = app_config.get("ai_disclosure_text", "")
    if has_disclosure and len(disclosure_text) > 20:
        return ComplianceCheck(
            article="Article 13",
            requirement="Users informed of AI interaction",
            status=ComplianceStatus.COMPLIANT,
            evidence=f"Disclosure text: '{disclosure_text[:80]}...'",
            )
    return ComplianceCheck(
        article="Article 13",
        requirement="Users informed of AI interaction",
        status=ComplianceStatus.NON_COMPLIANT,
        evidence="AI disclosure not configured or text too short",
        remediation="Add ai_disclosure_enabled=true and meaningful "
        "disclosure text to application config",
        )
def check_logging_requirements(
    log_config: dict,
    ) -> ComplianceCheck:
    """Article 12: Verify automatic logging is configured."""
    required_fields = [
        "input_logging", "output_logging",
        "model_version_logging", "guardrail_event_logging",
        ]
    missing = [f for f in required_fields if not log_config.get(f)]
    if not missing:
        return ComplianceCheck(
            article="Article 12",
            requirement="Automatic event logging",
            status=ComplianceStatus.COMPLIANT,
            evidence=f"All {len(required_fields)} required log "
            f"streams configured",
            )
    return ComplianceCheck(
        article="Article 12",
        requirement="Automatic event logging",
        status=ComplianceStatus.NON_COMPLIANT,
        evidence=f"Missing log streams: {', '.join(missing)}",
        remediation=f"Enable logging for: {', '.join(missing)}",
        )
def check_human_oversight(app_config: dict) -> ComplianceCheck:
    """Article 14: Verify human oversight mechanisms."""
    has_kill_switch = app_config.get("kill_switch_enabled", False)
    has_escalation = app_config.get("escalation_path", "")
    has_override = app_config.get("human_override_enabled", False)
    mechanisms = sum([has_kill_switch, bool(has_escalation), has_override])
    if mechanisms >= 2:
        return ComplianceCheck(
            article="Article 14",
            requirement="Human oversight mechanisms",
            status=ComplianceStatus.COMPLIANT,
            evidence=f"{mechanisms}/3 oversight mechanisms active",
            )
    return ComplianceCheck(
        article="Article 14",
        requirement="Human oversight mechanisms",
        status=ComplianceStatus.NON_COMPLIANT,
        evidence=f"Only {mechanisms}/3 oversight mechanisms active",
        remediation="Enable kill_switch, escalation_path, and "
        "human_override in application config",
        )
def run_compliance_suite(
    app_config: dict,
    log_config: dict,
    ) -> list[ComplianceCheck]:
    """Run all compliance checks and return results."""
    checks = [
        check_transparency_obligations(app_config),
        check_logging_requirements(log_config),
        check_human_oversight(app_config),
        ]
    # Print report
    for check in checks:
        icon = {
            ComplianceStatus.COMPLIANT: "PASS",
            ComplianceStatus.NON_COMPLIANT: "FAIL",
            ComplianceStatus.NEEDS_REVIEW: "REVIEW",
            }[check.status]
        print(f"[{icon}] {check.article}: {check.requirement}")
        print(f" Evidence: {check.evidence}")
        if check.remediation:
            print(f" Fix: {check.remediation}")
            print()
            return checks

# Example: Compliance-as-code checks in a CI/CD pipeline
# This runs as a GitHub Actions step or similar CI job
from dataclasses import dataclass
@dataclass
class ComplianceCheck:
    name: str
    framework: str # "EU_AI_ACT", "NIST_AI_RMF", "ISO_42001"
    requirement_id: str # e.g., "Art.9", "MAP-1.1", "8.3"
    passed: bool
    evidence_path: str # Path to the generated artifact
    def run_compliance_gate(model_version: str) -> list[ComplianceCheck]:
        """Run all compliance checks and return results."""
        checks = []
        # Check 1: Model card exists and is current
        checks.append(ComplianceCheck(
            name="Model card generated",
            framework="EU_AI_ACT", requirement_id="Art.11",
            passed=model_card_exists(model_version),
            evidence_path=f"artifacts/{model_version}/model_card.json",
            ))
        # Check 2: Bias evaluation completed with passing thresholds
        bias_results = load_bias_results(model_version)
        checks.append(ComplianceCheck(
            name="Bias evaluation within thresholds",
            framework="NIST_AI_RMF", requirement_id="MEASURE-2.6",
            passed=all(r["disparity"] < 0.1 for r in bias_results),
            evidence_path=f"artifacts/{model_version}/bias_report.json",
            ))
        # Check 3: Red team evaluation completed
        checks.append(ComplianceCheck(
            name="Red team evaluation completed",
            framework="EU_AI_ACT", requirement_id="Art.9",
            passed=red_team_report_exists(model_version),
            evidence_path=f"artifacts/{model_version}/red_team.json",
            ))
        # Check 4: Artifact signatures verified
        checks.append(ComplianceCheck(
            name="Model artifact signatures valid",
            framework="NIST_SSDF", requirement_id="PW.6",
            passed=verify_signatures(model_version),
            evidence_path=f"artifacts/{model_version}/signatures.json",
            ))
        return checks
    def enforce_gate(checks: list[ComplianceCheck]) -> bool:
        """Block deployment if any required check fails."""
        failed = [c for c in checks if not c.passed]
        if failed:
            for c in failed:
                print(f"FAILED: [{c.framework} {c.requirement_id}] {c.name}")
                return False
                print(f"All {len(checks)} compliance checks passed.")
                return True