AI 감사

# AI 감사 추적 시스템 구현 예제
import logging
import hashlib
from datetime import datetime
from dataclasses import dataclass
from typing import Dict, Any, List

@dataclass
class AuditRecord:
    """AI 의사결정 감사 기록"""
    timestamp: datetime
    model_id: str
    model_version: str
    input_hash: str
    output_hash: str
    decision: Any
    confidence: float
    processing_time_ms: float
    user_id: str

class AIAuditLogger:
    """EU AI Act 준수를 위한 감사 로거"""

    def __init__(self, system_name: str, risk_level: str = "high"):
        self.system_name = system_name
        self.risk_level = risk_level
        self.logger = logging.getLogger(f"ai_audit_{system_name}")
        self._setup_secure_logging()

    def _setup_secure_logging(self):
        """변조 방지 로깅 설정"""
        handler = logging.FileHandler(
            f"audit_logs/{self.system_name}_{datetime.now().strftime('%Y%m')}.log",
            encoding='utf-8'
        )
        formatter = logging.Formatter(
            '%(asctime)s|%(levelname)s|%(message)s|%(checksum)s'
        )
        handler.setFormatter(formatter)
        self.logger.addHandler(handler)

    def log_decision(self, model_id: str, model_version: str,
                     inputs: Dict, outputs: Dict,
                     confidence: float, user_id: str) -> AuditRecord:
        """AI 의사결정 기록 - 감사 추적성 확보"""
        record = AuditRecord(
            timestamp=datetime.utcnow(),
            model_id=model_id,
            model_version=model_version,
            input_hash=self._hash_data(inputs),
            output_hash=self._hash_data(outputs),
            decision=outputs.get('decision'),
            confidence=confidence,
            processing_time_ms=outputs.get('processing_time', 0),
            user_id=user_id
        )

        # 무결성 체크섬 포함 로깅
        log_entry = self._serialize_record(record)
        self.logger.info(log_entry, extra={'checksum': self._compute_checksum(log_entry)})

        return record

    def _hash_data(self, data: Dict) -> str:
        """입출력 데이터 해시화 (개인정보 보호)"""
        return hashlib.sha256(str(data).encode()).hexdigest()[:16]

    def _compute_checksum(self, data: str) -> str:
        """로그 무결성 검증용 체크섬"""
        return hashlib.sha256(data.encode()).hexdigest()[:8]

    def generate_audit_report(self, start_date: datetime,
                              end_date: datetime) -> Dict[str, Any]:
        """감사 보고서 생성 - 규제 당국 제출용"""
        return {
            "system_name": self.system_name,
            "risk_level": self.risk_level,
            "report_period": {"start": start_date, "end": end_date},
            "total_decisions": self._count_decisions(start_date, end_date),
            "accuracy_metrics": self._calculate_accuracy(),
            "fairness_metrics": self._calculate_fairness(),
            "incident_summary": self._get_incidents(start_date, end_date),
            "compliance_status": "COMPLIANT",
            "generated_at": datetime.utcnow().isoformat()
        }

# 사용 예시
audit_logger = AIAuditLogger("credit_scoring_ai", risk_level="high")

# 신용 평가 결정 로깅
audit_logger.log_decision(
    model_id="credit_model_v2",
    model_version="2.1.0",
    inputs={"applicant_id": "HASH_12345", "income": "bracket_3"},
    outputs={"decision": "APPROVED", "credit_limit": 50000},
    confidence=0.92,
    user_id="analyst_001"
)