KI-Systeme konform zum EU AI Act entwickeln: Ein Implementierungsleitfaden fuer Entwickler

Die Durchsetzungsfrist des EU AI Act rueckt in weniger als fuenf Monaten naeher. Bis zum 2. August 2026 muessen alle Hochrisiko-KI-Systeme, die in der Europaeischen Union eingesetzt werden, strenge technische Anforderungen erfuellen. Das gilt fuer Ihr Unternehmen unabhaengig vom Firmensitz — sobald Ihr KI-System EU-Buergerinnen und -Buerger betrifft, greift das Gesetz.

Hochrisiko-KI-Systeme umfassen unter anderem Algorithmen fuer Personalauswahl, Kreditwuerdigkeitsbewertung, medizinische Diagnostik, Bildungsbewertungssysteme und Anwendungen in der Strafverfolgung. Diese Systeme erfordern die vollstaendige Einhaltung von Risikomanagement, Daten-Governance, technischer Dokumentation, Audit-Protokollierung und Mechanismen zur menschlichen Aufsicht. Bei Verstoessen drohen Bussgelder von bis zu 15 Millionen Euro oder 3 % des weltweiten Jahresumsatzes.

Dieser Leitfaden liefert konkrete Implementierungsmuster und Code-Beispiele fuer die Entwicklung EU-AI-Act-konformer KI-Systeme. Sie erfahren, wie Sie Risikoklassifizierung, Audit-Logging-Infrastruktur, Daten-Governance-Pipelines, Schnittstellen zur menschlichen Aufsicht und die automatische Generierung von Compliance-Dokumentation umsetzen.

Schnelltest: JuiceFactory API

Bevor wir in die Compliance-Implementierung einsteigen, pruefen wir den Zugang zur JuiceFactory API:

curl -X POST https://api.juicefactory.ai/v1/chat/completions \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -d '{
    "model": "juicefactory/qwen3-vl",
    "messages": [
      {"role": "user", "content": "Hello, JuiceFactory!"}
    ]
  }'

Erwartete Antwort:

{
  "id": "chatcmpl-1234567890",
  "object": "chat.completion",
  "created": 1709251200,
  "model": "juicefactory/qwen3-vl",
  "choices": [{
    "index": 0,
    "message": {
      "role": "assistant",
      "content": "Hello! I am Qwen3-VL from JuiceFactory."
    },
    "finish_reason": "stop"
  }],
  "usage": {
    "prompt_tokens": 15,
    "completion_tokens": 10,
    "total_tokens": 25
  }
}

Damit ist bestaetigt, dass die JuiceFactory API erreichbar ist. Ihren API-Key erhalten Sie unter https://juicefactory.ai/api-key.

Uebersicht der Systemarchitektur

┌─────────────────────────────────────────────────────────────────────────┐
│              SYSTEMARCHITEKTUR KONFORM ZUM EU AI ACT                    │
└─────────────────────────────────────────────────────────────────────────┘

┌──────────────┐     ┌──────────────────┐     ┌──────────────────────────┐
│   BENUTZER-  │────▶│  MENSCHLICHE     │────▶│  JUICEFACTORY API (EU)   │
│  OBERFLAECHE │     │  EINGRIFFS-      │     │   qwen3-vl Inferenz      │
└──────────────┘     │  SCHNITTSTELLE   │     └───────────┬──────────────┘
                     └──────────────────┘                  │
                                │                          │
                                │ NEIN                     │ JA
                                ▼                          ▼
                         ┌─────────────┐          ┌──────────────────┐
                         │  EINGRIFF   │          │  AUDIT-LOGGING  │
                         │  (OVERRIDE) │          │   MIDDLEWARE     │
                         └──────┬──────┘          └────────┬─────────┘
                                │                           │
                                ▼                           ▼
                         ┌─────────────────────────────────────────────────┐
                         │       SICHERER SPEICHER (EU-GEHOSTET)           │
                         │  ┌─────────────┐ ┌──────────────┐ ┌──────────┐ │
                         │  │ Audit-Logs  │ │ Daten-       │ │ Override-│ │
                         │  │ (Artikel 10)│ │ Governance   │ │ Eintraege│ │
                         │  └─────────────┘ │ (Artikel 10) │ │(Art. 14) │ │
                         │                  └──────────────┘ └──────────┘ │
                         └─────────────────────────────────────────────────┘
                                      │                 │
                                      ▼                 ▼
                         ┌─────────────────────────────────────────────────┐
                         │      UEBERWACHUNGS- & COMPLIANCE-SCHICHT        │
                         │  ┌─────────────┐ ┌──────────────┐ ┌──────────┐ │
                         │  │  Prometheus  │ │   Grafana    │ │ Dok-Gen  │ │
                         │  │   Metriken   │ │  Dashboard   │ │(Art. 11) │ │
                         │  └─────────────┘ └──────────────┘ └──────────┘ │
                         └─────────────────────────────────────────────────┘
                                      │
                                      ▼
                         ┌─────────────────────────────────────────────────┐
                         │          COMPLIANCE-BERICHTSWESEN                │
                         │    (Artikel 12: Automatische Aufzeichnung)       │
                         │    Verfuegbar fuer nationale Aufsichtsbehoerden  │
                         └─────────────────────────────────────────────────┘

LEGENDE:
─────────────────────────────────────────────────────────────────────────────
• JuiceFactory API: EU-gehostet, zustandslose Inferenz ohne Datenspeicherung
• Audit-Logging-Middleware: Erfasst alle Ein-/Ausgaben (Artikel 10)
• Menschliche Eingriffs-Schnittstelle: Interventionspunkt (Artikel 14)
• Sicherer Speicher: EU-gehostet, DSGVO-konform, Zugriffskontrollen
• Ueberwachungsschicht: Echtzeit-Metriken, Override-Raten-Tracking
• Compliance-Berichtswesen: Automatische Dokumentationserstellung (Artikel 11)
─────────────────────────────────────────────────────────────────────────────

Diese Architektur erfuellt die zentralen Anforderungen des EU AI Act:

  • Artikel 10: Automatische Protokollierung ueber Middleware
  • Artikel 11: Technische Dokumentation ueber den Dokumentationsgenerator
  • Artikel 12: Aufzeichnungen in sicherem EU-Speicher
  • Artikel 14: Schnittstelle zur menschlichen Aufsicht
  • Artikel 15: Robustheit ueber die Ueberwachungsschicht

JuiceFactory wird hier als konforme Inferenzschicht hervorgehoben — EU-gehostet, standardmaessig zustandslos, mit automatischer Audit-Logging-Integration.

Die vier Risikostufen verstehen und Ihr System einordnen

Jedes KI-System faellt unter eine von vier Risikostufen gemaess dem EU AI Act. Ihre Compliance-Pflichten haengen vollstaendig davon ab, in welcher Stufe sich Ihr System befindet.

Unannehmbares Risiko (Verboten): Systeme, die seit Februar 2025 vollstaendig verboten sind. Dazu gehoeren Social Scoring, unterschwellige Manipulation, Ausnutzung vulnerabler Gruppen, ungezieltes biometrisches Scraping, biometrische Kategorisierung nach geschuetzten Merkmalen, biometrische Echtzeit-Ueberwachung in oeffentlichen Raeumen, vorausschauende Polizeiarbeit rein auf Basis von Profiling sowie Emotionserkennung am Arbeitsplatz und in Bildungseinrichtungen.

Hohes Risiko: KI-Systeme in kritischen Bereichen wie Personalauswahl, Arbeitnehmerverwaltung, Kreditwuerdigkeitsbewertung, medizinische Triage, Bildung, kritische Infrastruktur, Strafverfolgung und Rechtspflege. Diese Systeme erfordern die vollstaendige Einhaltung von Artikel 10 (Audit-Logging), Risikomanagementsysteme, Daten-Governance, technische Dokumentation und menschliche Aufsicht. Die vollstaendige Durchsetzung beginnt am 2. August 2026.

Begrenztes Risiko: Chatbots, Deepfake-Generatoren und Tools zur Erstellung KI-generierter Inhalte. Muessen die KI-Interaktion gegenueber Nutzern offenlegen und KI-generierte Inhalte kennzeichnen. Die Transparenzpflichten gelten ab dem 2. August 2026.

Minimales Risiko: Spamfilter, Empfehlungsalgorithmen, Videospiel-KI. Weitgehend unreguliert, wobei sonstiges anwendbares EU-Recht weiterhin gilt.

Die meisten Enterprise- und SaaS-Entwickler muessen sich auf die Hochrisiko-Stufe konzentrieren. Die Implementierungsmuster in diesem Leitfaden richten sich an Hochrisiko-KI-Systeme, die den strengsten Anforderungen und dem hoechsten Bussgeldrisiko unterliegen.

JuiceFactory's standardmaessig zustandslose Inferenz ohne Datenspeicherung macht es ideal fuer Hochrisiko-KI-Systeme, die strikte DSGVO-Konformitaet und Datensouveraenitaet gemaess dem EU AI Act erfordern. Die gesamte Inferenz findet auf EU-gehosteter Infrastruktur statt, ohne Speicherung von Prompts oder Antworten — es sei denn, dies wird explizit konfiguriert.

Risikoklassifizierung fuer Ihr KI-System implementieren

Die Grundlage der Konformitaet mit dem EU AI Act ist die korrekte Einordnung Ihres KI-Systems in die entsprechende Risikostufe. Artikel 6 und Anhang III bilden den rechtlichen Rahmen fuer diese Klassifizierung, doch Entwickler benoetigen einen systematischen Ansatz zur programmatischen Bewertung ihrer Systeme.

Der Entscheidungsbaum zur Risikoklassifizierung

Die Klassifizierung folgt einem Entscheidungsbaum basierend auf dem Verwendungszweck Ihres KI-Systems, der Einsatzdomaene und dem potenziellen Schaden fuer Grundrechte. Die zentralen Fragen lauten:

  1. Faellt das System unter eine verbotene Kategorie? Falls ja, liegt ein unannehmbares Risiko vor und der Betrieb muss eingestellt werden.
  2. Ist das System in einem der in Anhang III aufgefuehrten Sektoren taetig (kritische Infrastruktur, Bildung, Beschaeftigung, wesentliche Dienstleistungen, Strafverfolgung, Migration, Justiz, demokratische Prozesse)? Falls ja, ist es voraussichtlich ein Hochrisiko-System.
  3. Manipuliert das System das Nutzerverhalten oder nutzt Schwachstellen aus? Falls ja, liegt ein unannehmbares Risiko vor.
  4. Handelt es sich um einen Chatbot oder Content-Generator? Falls ja, besteht ein begrenztes Risiko mit Transparenzpflichten.
  5. Trifft keiner der oben genannten Punkte zu, handelt es sich um ein System mit minimalem Risiko.

Code: API zur Risikoklassifizierung

Hier ist eine Python-FastAPI-Implementierung, die KI-Systeme anhand der Kriterien des EU AI Act klassifiziert:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
from enum import Enum

class RiskTier(str, Enum):
    UNACCEPTABLE = "unacceptable"
    HIGH = "high"
    LIMITED = "limited"
    MINIMAL = "minimal"

class AnnexIIICategory(str, Enum):
    CRITICAL_INFRASTRUCTURE = "critical_infrastructure"
    EDUCATION = "education"
    EMPLOYMENT = "employment"
    ESSENTIAL_SERVICES = "essential_services"
    LAW_ENFORCEMENT = "law_enforcement"
    MIGRATION = "migration"
    JUSTICE = "justice"
    DEMOCRATIC_PROCESSES = "democratic_processes"

class AISystemDescription(BaseModel):
    name: str
    primary_function: str
    sector: Optional[AnnexIIICategory] = None
    is_biometric: bool = False
    is_emotion_recognition: bool = False
    is_social_scoring: bool = False
    is_subliminal_manipulation: bool = False
    is_chatbot: bool = False
    is_content_generator: bool = False
    targets_vulnerable_groups: bool = False

app = FastAPI(title="EU AI Act Risk Classification API")

def classify_system(system: AISystemDescription) -> RiskTier:
    """Classify AI system according to EU AI Act risk tiers"""

    # Unacceptable risk checks (Article 5)
    if system.is_social_scoring:
        return RiskTier.UNACCEPTABLE
    if system.is_subliminal_manipulation:
        return RiskTier.UNACCEPTABLE
    if system.targets_vulnerable_groups and system.is_biometric:
        return RiskTier.UNACCEPTABLE
    if system.is_emotion_recognition and system.sector in [
        AnnexIIICategory.EMPLOYMENT, AnnexIIICategory.EDUCATION
    ]:
        return RiskTier.UNACCEPTABLE

    # High risk checks (Article 6 and Annex III)
    if system.sector is not None:
        return RiskTier.HIGH

    # Limited risk checks (Article 50)
    if system.is_chatbot or system.is_content_generator:
        return RiskTier.LIMITED

    # Default to minimal risk
    return RiskTier.MINIMAL

@app.post("/classify")
async def classify_ai_system(system: AISystemDescription) -> dict:
    """Classify an AI system and return risk tier with explanation"""

    risk_tier = classify_system(system)

    response = {
        "system_name": system.name,
        "risk_tier": risk_tier.value,
        "compliance_deadline": None,
        "key_requirements": [],
        "recommendations": []
    }

    if risk_tier == RiskTier.UNACCEPTABLE:
        response["compliance_deadline"] = "ALREADY BANNED (Feb 2025)"
        response["key_requirements"] = [
            "Immediate discontinuation required",
            "No compliance path exists",
            "Penalties: €35M or 7% global revenue"
        ]
        response["recommendations"] = [
            "Redesign system to remove prohibited functionality",
            "Consult legal counsel before proceeding"
        ]

    elif risk_tier == RiskTier.HIGH:
        response["compliance_deadline"] = "August 2, 2026"
        response["key_requirements"] = [
            "Risk management system (Article 9)",
            "Data governance (Article 10)",
            "Technical documentation (Article 11)",
            "Audit logging (Article 12)",
            "Human oversight (Article 14)",
            "Transparency (Article 13)",
            "Accuracy, robustness, cybersecurity (Article 15)"
        ]
        response["recommendations"] = [
            "Begin conformity assessment immediately (6-12 months)",
            "Engage a Notified Body for certification",
            "Implement technical controls covered in this guide"
        ]

    elif risk_tier == RiskTier.LIMITED:
        response["compliance_deadline"] = "August 2, 2026"
        response["key_requirements"] = [
            "User disclosure of AI interaction",
            "Labeling of AI-generated content"
        ]
        response["recommendations"] = [
            "Implement transparency notices in UI",
            "Add metadata to AI-generated outputs"
        ]

    else:  # Minimal risk
        response["compliance_deadline"] = "None (voluntary codes of practice)"
        response["key_requirements"] = [
            "Voluntary codes of conduct encouraged",
            "Other applicable EU law still applies (GDPR, etc.)"
        ]
        response["recommendations"] = [
            "Consider voluntary compliance for trust building",
            "Document system capabilities for transparency"
        ]

    return response

# Example usage
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Diese API bietet eine systematische Moeglichkeit, Ihre KI-Systeme zu klassifizieren. Nutzen Sie sie bereits beim Systementwurf, um von Anfang an konforme Architekturen zu bauen — anstatt Compliance nachtraeglich einzubauen.

Aufbau einer Audit-Logging-Infrastruktur (Artikel 10)

Artikel 10 des EU AI Act verlangt eine automatische Protokollierung fuer Hochrisiko-KI-Systeme. Die Logs muessen Eingaben, Ausgaben, Zeitstempel und Modellversionen erfassen und die Nachvollziehbarkeit einzelner Entscheidungen ermoeglichen. Dies ist fuer die Konformitaet nicht verhandelbar.

Die Anforderungen von Artikel 10 im Detail

Die Protokollierungsanforderungen umfassen:

  • Automatische Protokollierung der Operationen des KI-Systems
  • Nachvollziehbarkeit einzelner Entscheidungen
  • Aufbewahrung der Aufzeichnungen fuer mindestens 6 Monate nach Ausserbetriebnahme des Systems
  • Sichere Speicherung der Logs mit Zugriffskontrollen
  • Verfuegbarkeit der Logs fuer nationale Aufsichtsbehoerden auf Anfrage

Bei Hochrisiko-Systemen kann die Protokollierung nicht deaktiviert werden. Jede Inferenz, Entscheidung oder Vorhersage muss mit ausreichend Detailtiefe erfasst werden, um rekonstruieren zu koennen, was passiert ist und warum.

Code: JuiceFactory-Integration mit Audit-Logging

Die OpenAI-kompatible API von JuiceFactory macht die Integration von Audit-Logging unkompliziert. Hier ist ein Python-Middleware-Wrapper, der alle API-Aufrufe automatisch protokolliert:

import os
import json
import time
from datetime import datetime
from typing import Dict, Any, Optional
from openai import OpenAI
import hashlib

# JuiceFactory API configuration
JUICEFACTORY_API_KEY = os.getenv("JUICEFACTORY_API_KEY")
JUICEFACTORY_BASE_URL = "https://api.juicefactory.ai/v1"

# Secure storage configuration (EU-hosted recommended)
LOG_STORAGE_PATH = "/var/log/ai-act-audit/"

class JuiceFactoryAuditLogger:
    """Middleware wrapper for JuiceFactory API with Article 10 audit logging"""

    def __init__(self, api_key: str = JUICEFACTORY_API_KEY):
        self.client = OpenAI(
            api_key=api_key,
            base_url=JUICEFACTORY_BASE_URL
        )
        self.model_version = None
        self._ensure_log_directory()

    def _ensure_log_directory(self):
        """Ensure audit log directory exists with proper permissions"""
        os.makedirs(LOG_STORAGE_PATH, mode=0o700, exist_ok=True)

    def _generate_request_id(self, system_id: str) -> str:
        """Generate unique request identifier"""
        timestamp = datetime.utcnow().isoformat()
        hash_input = f"{system_id}-{timestamp}".encode()
        return hashlib.sha256(hash_input).hexdigest()[:16]

    def _log_request(self, request_id: str, log_entry: Dict[str, Any]):
        """Write audit log entry to secure storage"""
        log_filename = f"{LOG_STORAGE_PATH}{request_id}.json"
        with open(log_filename, 'w') as f:
            json.dump(log_entry, f, indent=2, default=str)

    def chat(
        self,
        messages: list,
        model: str = "juicefactory/qwen3-vl",
        system_id: str = "default",
        **kwargs
    ) -> Dict[str, Any]:
        """JuiceFactory chat completion with automatic audit logging"""

        request_id = self._generate_request_id(system_id)
        start_time = time.time()

        # Capture input for audit log
        audit_entry = {
            "request_id": request_id,
            "system_id": system_id,
            "model": model,
            "timestamp": datetime.utcnow().isoformat(),
            "input": {
                "messages": messages,
                "parameters": {k: v for k, v in kwargs.items() if k != 'api_key'}
            },
            "output": None,
            "metadata": {
                "start_time": start_time,
                "end_time": None,
                "latency_ms": None,
                "status": "started"
            }
        }

        try:
            # Make JuiceFactory API call
            response = self.client.chat.completions.create(
                model=model,
                messages=messages,
                **kwargs
            )

            end_time = time.time()
            latency_ms = (end_time - start_time) * 1000

            # Extract response data
            output_data = {
                "content": response.choices[0].message.content,
                "finish_reason": response.choices[0].finish_reason,
                "model_used": response.model
            }

            # Update audit entry with success
            audit_entry["output"] = output_data
            audit_entry["metadata"]["end_time"] = end_time
            audit_entry["metadata"]["latency_ms"] = latency_ms
            audit_entry["metadata"]["status"] = "completed"

            # Write audit log
            self._log_request(request_id, audit_entry)

            return {
                "request_id": request_id,
                "response": output_data,
                "latency_ms": latency_ms
            }

        except Exception as e:
            # Log failure with error details
            audit_entry["metadata"]["end_time"] = time.time()
            audit_entry["metadata"]["latency_ms"] = (time.time() - start_time) * 1000
            audit_entry["metadata"]["status"] = "failed"
            audit_entry["error"] = {
                "type": type(e).__name__,
                "message": str(e)
            }
            self._log_request(request_id, audit_entry)

            raise

# Usage example
if __name__ == "__main__":
    logger = JuiceFactoryAuditLogger()

    # Example: High-risk credit scoring system
    response = logger.chat(
        messages=[
            {"role": "system", "content": "You are a credit scoring assistant."},
            {"role": "user", "content": "Evaluate creditworthiness for applicant with income: €50,000, debt: €10,000"}
        ],
        model="juicefactory/qwen3-vl",
        system_id="credit-scoring-system-v1",
        temperature=0.3,
        max_tokens=500
    )

    print(f"Request ID: {response['request_id']}")
    print(f"Response: {response['response']['content']}")
    print(f"Latency: {response['latency_ms']:.2f}ms")

Diese Middleware protokolliert automatisch jeden JuiceFactory-API-Aufruf mit vollstaendigen Eingabe- und Ausgabedaten, Zeitstempeln und Fehlerbehandlung. Die Logs werden sicher mit eindeutigen Request-IDs fuer die Nachvollziehbarkeit gespeichert.

Code: Abfrage und Analyse von Audit-Logs

Fuer die Compliance ist es erforderlich, Audit-Logs fuer die regulatorische Berichterstattung abrufen und analysieren zu koennen. Hier ist ein Abfrage- und Analysemodul:

import json
import os
from datetime import datetime, timedelta
from typing import List, Dict, Any
from collections import defaultdict

class AuditLogAnalyzer:
    """Query and analyze audit logs for EU AI Act compliance reporting"""

    def __init__(self, log_path: str = LOG_STORAGE_PATH):
        self.log_path = log_path

    def get_logs_by_date_range(
        self,
        start_date: datetime,
        end_date: datetime,
        system_id: Optional[str] = None
    ) -> List[Dict[str, Any]]:
        """Retrieve logs within date range, optionally filtered by system"""

        logs = []
        for filename in os.listdir(self.log_path):
            if not filename.endswith('.json'):
                continue

            filepath = os.path.join(self.log_path, filename)
            with open(filepath, 'r') as f:
                log_entry = json.load(f)

            # Parse timestamp
            log_time = datetime.fromisoformat(log_entry['timestamp'])

            # Filter by date range
            if not (start_date <= log_time <= end_date):
                continue

            # Filter by system ID if provided
            if system_id and log_entry.get('system_id') != system_id:
                continue

            logs.append(log_entry)

        return logs

    def calculate_metrics(self, logs: List[Dict[str, Any]]) -> Dict[str, Any]:
        """Calculate compliance metrics from audit logs"""

        total_requests = len(logs)
        successful_requests = sum(1 for log in logs if log['metadata']['status'] == 'completed')
        failed_requests = sum(1 for log in logs if log['metadata']['status'] == 'failed')

        latency_values = [
            log['metadata']['latency_ms']
            for log in logs
            if log['metadata'].get('latency_ms') is not None
        ]

        avg_latency = sum(latency_values) / len(latency_values) if latency_values else 0
        p95_latency = sorted(latency_values)[int(len(latency_values) * 0.95)] if latency_values else 0

        # Count by system
        requests_by_system = defaultdict(int)
        for log in logs:
            requests_by_system[log['system_id']] += 1

        return {
            "reporting_period": {
                "total_requests": total_requests,
                "successful_requests": successful_requests,
                "failed_requests": failed_requests,
                "success_rate": (successful_requests / total_requests * 100) if total_requests > 0 else 0
            },
            "performance": {
                "average_latency_ms": round(avg_latency, 2),
                "p95_latency_ms": round(p95_latency, 2)
            },
            "systems": dict(requests_by_system)
        }

    def generate_compliance_report(
        self,
        system_id: str,
        days: int = 30
    ) -> str:
        """Generate human-readable compliance report for specific system"""

        end_date = datetime.utcnow()
        start_date = end_date - timedelta(days=days)

        logs = self.get_logs_by_date_range(start_date, end_date, system_id)
        metrics = self.calculate_metrics(logs)

        report = f"""
EU AI Act Compliance-Bericht
============================

System-ID: {system_id}
Berichtszeitraum: {start_date.date()} bis {end_date.date()}
Erstellt: {datetime.utcnow().isoformat()}

ZUSAMMENFASSUNG
---------------
Anfragen gesamt: {metrics['reporting_period']['total_requests']}
Erfolgreich: {metrics['reporting_period']['successful_requests']}
Fehlgeschlagen: {metrics['reporting_period']['failed_requests']}
Erfolgsquote: {metrics['reporting_period']['success_rate']:.2f}%

LEISTUNGSKENNZAHLEN
-------------------
Durchschnittliche Latenz: {metrics['performance']['average_latency_ms']}ms
P95-Latenz: {metrics['performance']['p95_latency_ms']}ms

COMPLIANCE-STATUS
-----------------
✓ Artikel 10 (Automatische Protokollierung): IMPLEMENTIERT
✓ Nachvollziehbarkeit: AKTIVIERT (eindeutige Request-IDs)
✓ Log-Aufbewahrung: KONFIGURIERT (Aufbewahrungsrichtlinie pruefen)
✓ Sichere Speicherung: IMPLEMENTIERT (Verzeichnisberechtigungen 0o700)

Hinweis: Dieser Bericht deckt die Verfuegbarkeit und Nachvollziehbarkeit
der Audit-Logs ab. Die vollstaendige Konformitaet mit Artikel 10 erfordert
zusaetzliche Kontrollen einschliesslich Risikomanagement, Daten-Governance
und menschlicher Aufsicht.
"""

        return report

# Usage example
if __name__ == "__main__":
    analyzer = AuditLogAnalyzer()

    # Generate last 30 days compliance report
    report = analyzer.generate_compliance_report("credit-scoring-system-v1", days=30)
    print(report)

Dieser Analyzer ermoeglicht es Ihnen, Compliance-Berichte auf Abruf fuer nationale Aufsichtsbehoerden oder interne Audits zu erstellen.

Daten-Governance-Pipelines fuer Hochrisiko-KI

Artikel 10 verlangt zudem eine Daten-Governance fuer Hochrisiko-KI-Systeme. Dazu gehoeren die Dokumentation von Trainingsdaten, die Verfolgung der Datenherkunft sowie die Sicherstellung von Datenqualitaet und Repraesentativitaet.

Verfolgung der Trainingsdaten-Herkunft

Bei Hochrisiko-Systemen muessen Sie alle Trainingsdatenquellen, Vorverarbeitungsschritte und angewandte Datenaugmentierungen dokumentieren. Diese Dokumentation muss fuer die Konformitaetsbewertung verfuegbar sein.

Code: Daten-Governance-Workflow

Hier ist ein Airflow-DAG, der die Datenprovenienz von der Quelle bis zum Modell-Deployment nachverfolgt:

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
import json
import hashlib

default_args = {
    'owner': 'ai-governance',
    'depends_on_past': False,
    'start_date': datetime(2026, 1, 1),
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5)
}

def document_data_source(**context):
    """Document training data source for EU AI Act compliance"""

    data_source = {
        "source_id": context['params']['source_id'],
        "source_type": context['params']['source_type'],  # e.g., "internal", "licensed", "synthetic"
        "collection_date": datetime.utcnow().isoformat(),
        "data_volume": context['params']['data_volume'],
        "retention_period": context['params'].get('retention_period', "7 years"),
        "consent_documented": context['params'].get('consent_documented', False),
        "gdpr_compliant": context['params'].get('gdpr_compliant', True),
        "data_categories": context['params'].get('data_categories', []),
        "metadata": {
            "collector": context['params'].get('collector', "system"),
            "collection_method": context['params'].get('collection_method', "api"),
            "quality_checks": context['params'].get('quality_checks', [])
        }
    }

    # Hash for data provenance
    data_hash = hashlib.sha256(
        json.dumps(data_source, sort_keys=True).encode()
    ).hexdigest()[:16]

    data_source['provenance_hash'] = data_hash

    # Save to governance database
    with open(f"/var/lib/ai-governance/sources/{data_source['source_id']}.json", 'w') as f:
        json.dump(data_source, f, indent=2)

    return data_source

def track_transformation(**context):
    """Document data preprocessing and transformation steps"""

    transformation = {
        "transformation_id": context['params']['transformation_id'],
        "input_hash": context['task_instance'].xcom_pull(task_ids='document_data_source')['provenance_hash'],
        "transformation_type": context['params']['type'],  # e.g., "normalization", "augmentation", "filtering"
        "parameters": context['params'].get('parameters', {}),
        "output_volume": context['params'].get('output_volume'),
        "timestamp": datetime.utcnow().isoformat(),
        "metadata": {
            "reproducible": context['params'].get('reproducible', True),
            "version": context['params'].get('version', "1.0")
        }
    }

    # Save transformation record
    with open(f"/var/lib/ai-governance/transformations/{transformation['transformation_id']}.json", 'w') as f:
        json.dump(transformation, f, indent=2)

    return transformation

def link_to_model(**context):
    """Link data lineage to deployed model version"""

    model_linkage = {
        "model_id": context['params']['model_id'],
        "model_version": context['params']['model_version'],
        "training_data_hash": context['task_instance'].xcom_pull(task_ids='track_transformation')['output_hash'],
        "deployment_date": datetime.utcnow().isoformat(),
        "compliance_metadata": {
            "article_10_compliant": True,
            "data_governance_documented": True,
            "last_audit_date": datetime.utcnow().isoformat()
        }
    }

    # Save model linkage
    with open(f"/var/lib/ai-governance/models/{model_linkage['model_id']}.json", 'w') as f:
        json.dump(model_linkage, f, indent=2)

    return model_linkage

dag = DAG(
    'ai_act_data_governance',
    default_args=default_args,
    description='EU AI Act Data Governance Pipeline',
    schedule_interval='@daily',
    catchup=False
)

# Example DAG run with parameters
t1 = PythonOperator(
    task_id='document_data_source',
    python_callable=document_data_source,
    op_kwargs={
        'source_id': 'credit-scoring-training-data-2026-01',
        'source_type': 'internal',
        'data_volume': 50000,
        'data_categories': ['financial', 'demographic'],
        'collector': 'automated_pipeline',
        'quality_checks': ['null_check', 'duplicate_removal', 'outlier_detection']
    },
    dag=dag
)

t2 = PythonOperator(
    task_id='track_transformation',
    python_callable=track_transformation,
    op_kwargs={
        'transformation_id': 'credit-scoring-preprocessing-v2',
        'type': 'normalization',
        'parameters': {'method': 'z-score', 'feature_wise': True},
        'output_volume': 50000,
        'version': '2.1'
    },
    dag=dag
)

t3 = PythonOperator(
    task_id='link_to_model',
    python_callable=link_to_model,
    op_kwargs={
        'model_id': 'credit-scoring-model',
        'model_version': '1.5.0'
    },
    dag=dag
)

t1 >> t2 >> t3

Dieser Airflow-DAG dokumentiert die vollstaendige Datenherkunft von der Quelle bis zum eingesetzten Modell und erfuellt damit die Daten-Governance-Anforderungen von Artikel 10.

Schnittstellen fuer menschliche Aufsicht und Kontrollschleifen

Artikel 14 verlangt eine menschliche Aufsicht fuer Hochrisiko-KI-Systeme. Dies ist keine optionale Anforderung — menschliche Operatoren muessen in der Lage sein, die Operationen des KI-Systems zu verstehen, zu ueberwachen und bei Bedarf einzugreifen.

Interventionspunkte fuer menschliche Eingriffe gestalten

Menschliche Aufsicht erfordert:

  • Verstaendliche Erklaerungen von KI-Entscheidungen
  • Die Moeglichkeit fuer Menschen, Entscheidungen zu ueberschreiben
  • Ueberwachung der Override-Raten
  • Schulung der menschlichen Operatoren

Ihre Systemarchitektur muss Interventionspunkte enthalten, an denen menschliche Operatoren KI-Entscheidungen pruefen und ueberschreiben koennen — mit vollstaendiger Audit-Protokollierung aller Eingriffe.

Code: Implementierung der Override-API

Hier ist ein FastAPI-Endpunkt, ueber den menschliche Operatoren KI-Entscheidungen ueberschreiben koennen:

from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
from typing import Optional, List
from datetime import datetime
import json

class OverrideRequest(BaseModel):
    system_id: str
    request_id: str
    operator_id: str
    override_reason: str
    reason_code: str  # e.g., "incorrect_risk_score", "missing_context", "appeal"
    original_ai_decision: str
    human_decision: str
    additional_notes: Optional[str] = None

class OverrideResponse(BaseModel):
    override_id: str
    timestamp: str
    status: str

app = FastAPI(title="AI System Human Override API")

AUDIT_LOG_PATH = "/var/log/ai-act-audit/"

def log_override(override_data: dict):
    """Log human override to audit trail"""

    override_id = f"override-{datetime.utcnow().timestamp()}"

    log_entry = {
        "override_id": override_id,
        "timestamp": datetime.utcnow().isoformat(),
        "system_id": override_data["system_id"],
        "original_request_id": override_data["request_id"],
        "operator_id": override_data["operator_id"],
        "override_reason": override_data["override_reason"],
        "reason_code": override_data["reason_code"],
        "original_ai_decision": override_data["original_ai_decision"],
        "human_decision": override_data["human_decision"],
        "additional_notes": override_data.get("additional_notes"),
        "metadata": {
            "compliance": "Article 14 Human Oversight",
            "audit_ready": True
        }
    }

    # Save to audit log
    filename = f"{AUDIT_LOG_PATH}{override_id}.json"
    with open(filename, 'w') as f:
        json.dump(log_entry, f, indent=2)

    return override_id

@app.post("/override", response_model=OverrideResponse)
async def create_override(
    request: OverrideRequest,
    background_tasks: BackgroundTasks
):
    """Create human override record for AI system decision"""

    # Validate request
    if request.reason_code not in [
        "incorrect_risk_score",
        "missing_context",
        "appeal",
        "regulatory_requirement",
        "other"
    ]:
        raise HTTPException(status_code=400, detail="Invalid reason code")

    # Log override in background
    override_id = log_override(request.dict())

    # TODO: In production, trigger downstream actions:
    # - Update system behavior if pattern detected
    # - Notify compliance team if override rate exceeds threshold
    # - Retrain model if systematic bias identified

    return OverrideResponse(
        override_id=override_id,
        timestamp=datetime.utcnow().isoformat(),
        status="recorded"
    )

@app.get("/overrides/{system_id}")
async def get_system_overrides(
    system_id: str,
    limit: int = 100,
    reason_code: Optional[str] = None
):
    """Retrieve override history for a specific system"""

    overrides = []

    for filename in os.listdir(AUDIT_LOG_PATH):
        if not filename.startswith("override-") or not filename.endswith('.json'):
            continue

        with open(os.path.join(AUDIT_LOG_PATH, filename), 'r') as f:
            override = json.load(f)

        if override['system_id'] != system_id:
            continue

        if reason_code and override['reason_code'] != reason_code:
            continue

        overrides.append(override)

        if len(overrides) >= limit:
            break

    # Sort by timestamp descending
    overrides.sort(key=lambda x: x['timestamp'], reverse=True)

    return {
        "system_id": system_id,
        "total_overrides": len(overrides),
        "overrides": overrides[:limit]
    }

@app.get("/overrides/{system_id}/analytics")
async def get_override_analytics(system_id: str, days: int = 30):
    """Calculate override analytics for monitoring"""

    end_date = datetime.utcnow()
    start_date = end_date - timedelta(days=days)

    overrides = await get_system_overrides(system_id, limit=10000)
    filtered_overrides = [
        o for o in overrides['overrides']
        if datetime.fromisoformat(o['timestamp']) >= start_date
    ]

    # Calculate metrics
    total_overrides = len(filtered_overrides)

    reason_counts = {}
    for override in filtered_overrides:
        reason = override['reason_code']
        reason_counts[reason] = reason_counts.get(reason, 0) + 1

    top_operators = {}
    for override in filtered_overrides:
        operator = override['operator_id']
        top_operators[operator] = top_operators.get(operator, 0) + 1

    return {
        "system_id": system_id,
        "period_days": days,
        "total_overrides": total_overrides,
        "override_rate_per_day": round(total_overrides / days, 2),
        "top_reasons": sorted(reason_counts.items(), key=lambda x: x[1], reverse=True),
        "top_operators": sorted(top_operators.items(), key=lambda x: x[1], reverse=True),
        "compliance_note": "High override rates may indicate need for model retraining"
    }

Diese API bietet eine vollstaendige Schnittstelle fuer menschliche Eingriffe mit Audit-Protokollierung, Verlaufsabfragen und Analysefunktionen zur Ueberwachung der Konformitaet mit Artikel 14.

Generator fuer Compliance-Dokumentation

Technische Dokumentation ist fuer die Konformitaetsbewertung erforderlich (Artikel 11). Die Dokumentation muss eine Systembeschreibung, den Verwendungszweck, die Risikobewertung, Testergebnisse und Compliance-Massnahmen umfassen.

Erforderliche Dokumentation fuer die Konformitaetsbewertung

Die technische Dokumentation muss Folgendes enthalten:

  • Allgemeine Beschreibung des KI-Systems
  • Detaillierte Erklaerung der Systemarchitektur
  • Ergebnisse der Risikobewertung
  • Test- und Validierungsergebnisse
  • Dokumentation der Daten-Governance
  • Massnahmen zur menschlichen Aufsicht
  • Transparenz- und Erklaerbarkeitsmassnahmen
  • Massnahmen zu Robustheit, Genauigkeit und Cybersicherheit

Code: CLI-Tool zur Dokumentationsgenerierung

Hier ist ein Kommandozeilen-Tool, das Compliance-Dokumentation aus Audit-Logs und System-Metadaten generiert:

#!/usr/bin/env python3
"""
EU AI Act Compliance Documentation Generator
Generates technical documentation for conformity assessment
"""

import argparse
import json
from datetime import datetime
from pathlib import Path
from typing import Dict, Any

class DocumentationGenerator:
    """Generate EU AI Act compliance documentation from system metadata"""

    def __init__(self, system_metadata_path: str, audit_log_path: str):
        self.system_metadata = self._load_json(system_metadata_path)
        self.audit_log_path = audit_log_path

    def _load_json(self, path: str) -> Dict[str, Any]:
        with open(path, 'r') as f:
            return json.load(f)

    def _generate_section_1_general_description(self) -> str:
        """Article 11.1: General description of the AI system"""

        metadata = self.system_metadata

        return f"""
# 1. Allgemeine Beschreibung des KI-Systems

## 1.1 Systemuebersicht

**System-ID:** {metadata['system_id']}
**Systemname:** {metadata['system_name']}
**Version:** {metadata['version']}
**Anbieter:** {metadata['provider']}
**Datum der Dokumentation:** {datetime.utcnow().isoformat()}

## 1.2 Bestimmungsgemaesser Zweck

{metadata['intended_purpose']}

## 1.3 Vorgesehene Nutzer

{metadata['intended_users']}

## 1.4 Risikoklassifizierung

**Risikostufe:** {metadata['risk_tier']}
**Anhang-III-Kategorie:** {metadata.get('annex_iii_category', 'N/A')}

## 1.5 Einsatzumgebung

**Hosting-Standort:** {metadata['hosting_location']}
**EU-gehostet:** {metadata['eu_hosted']}
**Datenresidenz:** {metadata['data_residency']}
"""

    def _generate_section_2_architecture(self) -> str:
        """Article 11.2: System architecture description"""

        metadata = self.system_metadata

        return f"""
# 2. Systemarchitektur

## 2.1 Modellinformationen

**Modelltyp:** {metadata['model']['type']}
**Modellname:** {metadata['model']['name']}
**Modellversion:** {metadata['model']['version']}
**Anbieter-API:** {metadata['model']['api_endpoint']}

## 2.2 Datenfluss

{self._describe_data_flow()}

## 2.3 Integrationspunkte

{self._describe_integrations()}
"""

    def _describe_data_flow(self) -> str:
        """Describe system data flow"""

        return """
1. Nutzer sendet Anfrage an die Anwendungsoberflaeche
2. Anfrage durchlaeuft die Pruefung auf menschlichen Eingriff (Artikel 14)
3. Anfrage wird an die JuiceFactory API weitergeleitet
4. Audit-Logging-Middleware erfasst Ein-/Ausgabe (Artikel 10)
5. Modell generiert Antwort
6. Antwort wird protokolliert und an den Nutzer zurueckgegeben
7. Menschlicher Operator kann Entscheidung bei Bedarf ueberschreiben
8. Alle Eingriffe werden im Audit-Trail protokolliert
"""

    def _describe_integrations(self) -> str:
        """Describe system integrations"""

        metadata = self.system_metadata

        integrations = []
        for integration in metadata.get('integrations', []):
            integrations.append(f"- {integration['name']}: {integration['purpose']}")

        return "\n".join(integrations) if integrations else "Keine externen Integrationen"

    def _generate_section_3_risk_assessment(self) -> str:
        """Article 11.3: Risk assessment results"""

        metadata = self.system_metadata

        return f"""
# 3. Risikobewertung

## 3.1 Identifizierte Risiken

{self._list_risks()}

## 3.2 Risikominderungsmassnahmen

{self._list_mitigations()}

## 3.3 Restrisiko

**Akzeptabel:** {metadata['risk_assessment']['residual_risk_acceptable']}
**Begruendung:** {metadata['risk_assessment']['residual_risk_justification']}
"""

    def _list_risks(self) -> str:
        """List identified risks"""

        risks = self.system_metadata.get('risk_assessment', {}).get('identified_risks', [])
        return "\n".join([f"- **{r['category']}:** {r['description']} (Schweregrad: {r['severity']})" for r in risks])

    def _list_mitigations(self) -> str:
        """List risk mitigation measures"""

        mitigations = self.system_metadata.get('risk_assessment', {}).get('mitigations', [])
        return "\n".join([f"- **{m['risk_category']}:** {m['measure']}" for m in mitigations])

    def _generate_section_4_testing(self) -> str:
        """Article 11.4: Testing and validation results"""

        metadata = self.system_metadata

        return f"""
# 4. Test und Validierung

## 4.1 Testmethodik

{metadata['testing']['methodology']}

## 4.2 Testergebnisse

**Genauigkeit (Accuracy):** {metadata['testing']['accuracy']}%
**Praezision (Precision):** {metadata['testing']['precision']}%
**Trefferquote (Recall):** {metadata['testing']['recall']}%
**F1-Score:** {metadata['testing']['f1_score']}%

## 4.3 Bias-Tests

{self._describe_bias_testing()}

## 4.4 Robustheitstests

{self._describe_robustness_testing()}
"""

    def _describe_bias_testing(self) -> str:
        """Describe bias testing results"""

        bias_tests = self.system_metadata.get('testing', {}).get('bias_tests', [])
        if not bias_tests:
            return "Keine Ergebnisse aus Bias-Tests vorhanden."

        return "\n".join([f"- **{t['attribute']}:** {t['result']}" for t in bias_tests])

    def _describe_robustness_testing(self) -> str:
        """Describe robustness testing results"""

        robustness = self.system_metadata.get('testing', {}).get('robustness', {})
        return f"""
- **Adversariale Angriffe:** {robustness.get('adversarial_attacks', 'Nicht getestet')}
- **Datendrift:** {robustness.get('data_drift', 'Nicht getestet')}
- **Grenzfaelle:** {robustness.get('edge_cases', 'Nicht getestet')}
"""

    def _generate_section_5_compliance_measures(self) -> str:
        """Article 11.5: Compliance with AI Act requirements"""

        metadata = self.system_metadata

        return f"""
# 5. Compliance-Massnahmen gemaess KI-Verordnung

## 5.1 Artikel 9: Risikomanagementsystem

**Implementiert:** {metadata['compliance']['article_9_risk_management']['implemented']}
**Beschreibung:** {metadata['compliance']['article_9_risk_management']['description']}

## 5.2 Artikel 10: Daten und Daten-Governance

**Implementiert:** {metadata['compliance']['article_10_data_governance']['implemented']}
**Beschreibung:** {metadata['compliance']['article_10_data_governance']['description']}

## 5.3 Artikel 11: Technische Dokumentation

**Implementiert:** JA (Dieses Dokument)

## 5.4 Artikel 12: Aufzeichnungspflichten (Automatische Protokollierung)

**Implementiert:** {metadata['compliance']['article_12_record_keeping']['implemented']}
**Beschreibung:** {metadata['compliance']['article_12_record_keeping']['description']}

## 5.5 Artikel 14: Menschliche Aufsicht

**Implementiert:** {metadata['compliance']['article_14_human_oversight']['implemented']}
**Beschreibung:** {metadata['compliance']['article_14_human_oversight']['description']}

## 5.6 Artikel 15: Genauigkeit, Robustheit, Cybersicherheit

**Implementiert:** {metadata['compliance']['article_15_accuracy_robustness']['implemented']}
**Beschreibung:** {metadata['compliance']['article_15_accuracy_robustness']['description']}

## 5.7 Artikel 13: Transparenz

**Implementiert:** {metadata['compliance']['article_13_transparency']['implemented']}
**Beschreibung:** {metadata['compliance']['article_13_transparency']['description']}
"""

    def generate_full_documentation(self) -> str:
        """Generate complete technical documentation"""

        sections = [
            "# Technische Dokumentation gemaess EU AI Act",
            f"**Erstellt:** {datetime.utcnow().isoformat()}",
            "",
            self._generate_section_1_general_description(),
            "",
            self._generate_section_2_architecture(),
            "",
            self._generate_section_3_risk_assessment(),
            "",
            self._generate_section_4_testing(),
            "",
            self._generate_section_5_compliance_measures(),
            "",
            "---",
            "",
            "Diese Dokumentation wird automatisch aus System-Metadaten und Audit-Logs generiert. ",
            "Reichen Sie dieses Dokument zusammen mit dem vollstaendigen Audit-Log-Archiv ",
            "bei Ihrer benannten Stelle (Notified Body) zur Konformitaetsbewertung ein."
        ]

        return "\n".join(sections)

def main():
    parser = argparse.ArgumentParser(description='Generate EU AI Act compliance documentation')
    parser.add_argument('--metadata', required=True, help='Path to system metadata JSON file')
    parser.add_argument('--audit-log-path', default='/var/log/ai-act-audit/', help='Path to audit log directory')
    parser.add_argument('--output', default='eu-ai-act-compliance-doc.md', help='Output filename')

    args = parser.parse_args()

    generator = DocumentationGenerator(args.metadata, args.audit_log_path)
    documentation = generator.generate_full_documentation()

    with open(args.output, 'w') as f:
        f.write(documentation)

    print(f"Documentation generated: {args.output}")

if __name__ == '__main__':
    main()

Speichern Sie dieses Skript als generate-compliance-doc.py, machen Sie es ausfuehrbar (chmod +x) und fuehren Sie es mit Ihrer System-Metadaten-JSON-Datei aus.

Ueberwachung und Reporting-Dashboard

Echtzeit-Ueberwachung ist essenziell fuer die Konformitaet mit Artikel 12 und um nachzuweisen, dass Ihr KI-System bestimmungsgemaess arbeitet.

Wichtige zu ueberwachende Metriken

Fuer die Konformitaet mit dem EU AI Act sollten Sie folgende Metriken ueberwachen:

  • Anfrage-/Antwortraten und Fehlerquoten
  • Override-Raten und Begruendungscodes
  • Modell-Leistungskennzahlen (Genauigkeit, Latenz)
  • Vollstaendigkeit der Audit-Logs
  • Indikatoren fuer Datendrift

Code: Prometheus-Metriken und Grafana-Dashboard

Hier ist ein Prometheus-Metrik-Collector und eine Grafana-Dashboard-Konfiguration:

from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time

# Prometheus metrics for EU AI Act compliance
REQUESTS_TOTAL = Counter(
    'ai_act_requests_total',
    'Total number of AI system requests',
    ['system_id', 'status', 'model']
)

REQUEST_LATENCY = Histogram(
    'ai_act_request_latency_seconds',
    'Request latency in seconds',
    ['system_id', 'model'],
    buckets=[0.01, 0.05, 0.1, 0.5, 1.0, 5.0, 10.0]
)

HUMAN_OVERRIDES_TOTAL = Counter(
    'ai_act_human_overrides_total',
    'Total number of human overrides',
    ['system_id', 'reason_code', 'operator_id']
)

COMPLIANCE_STATUS = Gauge(
    'ai_act_compliance_status',
    'Compliance status (1 = compliant, 0 = non-compliant)',
    ['system_id', 'requirement']
)

class PrometheusMetrics:
    """Collect and expose EU AI Act compliance metrics"""

    @staticmethod
    def record_request(system_id: str, model: str, latency_ms: float, status: str):
        """Record request metric"""

        REQUESTS_TOTAL.labels(
            system_id=system_id,
            status=status,
            model=model
        ).inc()

        REQUEST_LATENCY.labels(
            system_id=system_id,
            model=model
        ).observe(latency_ms / 1000.0)  # Convert to seconds

    @staticmethod
    def record_override(system_id: str, reason_code: str, operator_id: str):
        """Record human override metric"""

        HUMAN_OVERRIDES_TOTAL.labels(
            system_id=system_id,
            reason_code=reason_code,
            operator_id=operator_id
        ).inc()

    @staticmethod
    def set_compliance_status(system_id: str, requirement: str, compliant: bool):
        """Set compliance status for specific requirement"""

        COMPLIANCE_STATUS.labels(
            system_id=system_id,
            requirement=requirement
        ).set(1 if compliant else 0)

    @staticmethod
    def start_metrics_server(port: int = 8000):
        """Start Prometheus metrics HTTP server"""

        start_http_server(port)
        print(f"Prometheus metrics server started on port {port}")
        print(f"Access metrics at http://localhost:{port}/metrics")

# Example usage
if __name__ == "__main__":
    PrometheusMetrics.start_metrics_server(port=9090)

    # Simulate some requests
    system_id = "credit-scoring-system-v1"
    model = "juicefactory/qwen3-vl"

    for i in range(10):
        latency = 50 + (i * 10)  # Simulated latency
        status = "success" if i < 9 else "error"

        PrometheusMetrics.record_request(
            system_id=system_id,
            model=model,
            latency_ms=latency,
            status=status
        )

        time.sleep(0.1)

    # Simulate override
    PrometheusMetrics.record_override(
        system_id=system_id,
        reason_code="incorrect_risk_score",
        operator_id="operator-001"
    )

    # Set compliance status
    PrometheusMetrics.set_compliance_status(
        system_id=system_id,
        requirement="article_10_logging",
        compliant=True
    )

    # Keep server running
    while True:
        time.sleep(1)

Grafana-Dashboard-JSON (in Grafana importieren):

{
  "dashboard": {
    "title": "EU AI Act Compliance Dashboard",
    "panels": [
      {
        "title": "Request Rate",
        "type": "graph",
        "targets": [
          {
            "expr": "rate(ai_act_requests_total[5m])",
            "legendFormat": "{{system_id}} - {{status}}"
          }
        ]
      },
      {
        "title": "Request Latency (P95)",
        "type": "graph",
        "targets": [
          {
            "expr": "histogram_quantile(0.95, rate(ai_act_request_latency_seconds_bucket[5m]))",
            "legendFormat": "{{system_id}}"
          }
        ]
      },
      {
        "title": "Human Override Rate",
        "type": "graph",
        "targets": [
          {
            "expr": "rate(ai_act_human_overrides_total[1h])",
            "legendFormat": "{{reason_code}}"
          }
        ]
      },
      {
        "title": "Compliance Status",
        "type": "stat",
        "targets": [
          {
            "expr": "ai_act_compliance_status",
            "legendFormat": "{{requirement}}"
          }
        ]
      }
    ]
  }
}

Dieses Monitoring-Setup bietet Echtzeit-Einblick in den Compliance-Status Ihres KI-Systems — unverzichtbar fuer die Anforderungen von Artikel 12 und fuer den Nachweis der Konformitaet gegenueber Aufsichtsbehoerden.

FAQ

Muss ich die Verordnung einhalten, wenn mein Unternehmen nicht in der EU ansaessig ist?

Ja. Der EU AI Act hat eine extraterritoriale Wirkung und gilt fuer jedes KI-System, das von EU-Buergerinnen und -Buergern genutzt wird — unabhaengig vom Firmensitz. Wenn Ihre KI-Produkte oder -Dienste von Personen in der Europaeischen Union genutzt werden, oder wenn KI-Ergebnisse in der EU verwendet werden, unterliegen Sie diesem Gesetz.

Was passiert, wenn ich die Frist im August 2026 versaeume?

Verstoesse bei Hochrisiko-KI-Systemen koennen Bussgelder von bis zu 15 Millionen Euro oder 3 % des weltweiten Jahresumsatzes nach sich ziehen. Organisationen, die heute beginnen, haben gerade noch genug Zeit fuer die Konformitaetsbewertung, die 6 bis 12 Monate in Anspruch nimmt. Ueber finanzielle Strafen hinaus koennen nationale Aufsichtsbehoerden die Aussetzung des KI-Systembetriebs anordnen und den verpflichtenden Rueckruf nicht konformer Produkte vom EU-Markt verlangen.

Wie unterstuetzt JuiceFactory die Konformitaet mit dem EU AI Act?

JuiceFactory bietet standardmaessig zustandslose KI-Inferenz ohne Datenspeicherung, vollstaendig gehostet in der EU (Schweden). Das vereinfacht die Einhaltung der DSGVO und des AI Act fuer Hochrisiko-Systeme, die strikte Datensouveraenitaet und Audit-Trails erfordern. Die OpenAI-kompatible API ermoeglicht eine nahtlose Integration mit bestehender Compliance-Infrastruktur, waehrend die gesamte Inferenz auf EU-Infrastruktur stattfindet — mit voller Unterstuetzung der Audit-Logging-Anforderungen gemaess Artikel 10.

Was ist der Unterschied zwischen Hochrisiko- und KI-Systemen mit begrenztem Risiko?

Hochrisiko-Systeme (Artikel 6 und Anhang III) werden in kritischen Bereichen wie Personalauswahl, Kreditwuerdigkeitsbewertung, medizinischer Triage, Bildung, kritischer Infrastruktur und Strafverfolgung eingesetzt. Sie erfordern die vollstaendige Einhaltung von Risikomanagementsystemen, Daten-Governance, technischer Dokumentation, automatischer Protokollierung und menschlicher Aufsicht (Artikel 9-15). KI-Systeme mit begrenztem Risiko (Artikel 50) wie Chatbots und Content-Generatoren unterliegen nur Transparenzpflichten — sie muessen die KI-Interaktion gegenueber Nutzern offenlegen und KI-generierte Inhalte kennzeichnen.

Benoetige ich eine benannte Stelle (Notified Body) fuer die Konformitaetsbewertung?

Ja, fuer Hochrisiko-KI-Systeme muessen Sie eine benannte Stelle fuer die Konformitaetsbewertung beauftragen. Benannte Stellen sind unabhaengige Organisationen, die von EU-Mitgliedstaaten benannt werden, um KI-Systeme anhand der Anforderungen des AI Act zu zertifizieren. Dieser Prozess dauert in der Regel 6 bis 12 Monate und sollte umgehend eingeleitet werden, um die Frist im August 2026 einzuhalten. Einige Hochrisiko-KI-Systeme (Anhang III, Teil A) koennen sich fuer eine Selbstbewertung qualifizieren, wenn sie bestimmte Kriterien erfuellen, aber die meisten Enterprise-KI-Systeme werden eine Zertifizierung durch eine benannte Stelle benoetigen.


Fazit

Die Entwicklung EU-AI-Act-konformer KI-Systeme erfordert die systematische Implementierung von Risikoklassifizierung, Audit-Logging, Daten-Governance, menschlicher Aufsicht und Compliance-Dokumentation. Die Frist am 2. August 2026 rueckt schnell naeher, und die Konformitaetsbewertung dauert 6 bis 12 Monate — Organisationen, die jetzt starten, haben gerade noch ausreichend Zeit.

Zentrale Erkenntnisse:

  1. Die Risikoklassifizierung ist das Fundament der Konformitaet. Klassifizieren Sie Ihre Systeme fruehzeitig und gestalten Sie die Architektur entsprechend.
  2. Audit-Logging und Daten-Governance sind fuer Hochrisiko-Systeme nicht verhandelbar. Implementieren Sie die Kontrollen gemaess Artikel 10 von Tag eins an.
  3. Schnittstellen zur menschlichen Aufsicht muessen von Anfang an in die Systemarchitektur integriert werden — nicht nachtraeglich hinzugefuegt.
  4. Die Compliance-Dokumentation muss vor Ablauf der Frist erstellt und fuer die Pruefung durch die benannte Stelle bereit sein.

JuiceFactory's EU-gehostete, standardmaessig zustandslose Inferenz-Infrastruktur vereinfacht die Konformitaet fuer Hochrisiko-Systeme. Mit standardmaessig keiner Datenspeicherung und voller API-Kompatibilitaet mit OpenAI-SDKs erhalten Sie DSGVO-konforme KI-Inferenz — ohne die Komplexitaet eigener Infrastrukturverwaltung.

Starten Sie mit JuiceFactory's EU-gehosteter, DSGVO-konformer KI-Inferenz

Preisvergleich fuer Enterprise-KI-Infrastruktur

Related Guides

Ship GDPR-Compliant AI Today

Zero-retention inference in Stockholm. DPA included. Same OpenAI SDK, two lines change.