Python을 사용한 인간 감독 메커니즘 구현 예제입니다.
"""
인간 감독(Human Oversight) 시스템 구현
EU AI Act 제14조 준수
"""
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Dict, List, Optional, Callable, Any
import logging
import time
class OversightMode(Enum):
"""인간 감독 모드 (EU AI Act 기준)"""
HUMAN_IN_THE_LOOP = "hitl" # 매 결정마다 인간 승인 필요
HUMAN_ON_THE_LOOP = "hotl" # 실시간 모니터링 + 개입 가능
HUMAN_IN_COMMAND = "hic" # 전체 운영 사이클 감독
class DecisionStatus(Enum):
"""AI 결정 상태"""
PENDING_REVIEW = "pending" # 인간 검토 대기
APPROVED = "approved" # 승인됨
REJECTED = "rejected" # 거부됨
OVERRIDDEN = "overridden" # 무효화됨
AUTO_EXECUTED = "auto" # 자동 실행됨
HALTED = "halted" # 중단됨
@dataclass
class AIDecision:
"""AI 결정 데이터 구조"""
decision_id: str
model_name: str
input_data: Dict
prediction: Any
confidence: float
reasoning: str
timestamp: datetime = field(default_factory=datetime.now)
status: DecisionStatus = DecisionStatus.PENDING_REVIEW
reviewer_id: Optional[str] = None
review_notes: Optional[str] = None
review_timestamp: Optional[datetime] = None
class HumanOversightSystem:
"""
EU AI Act 제14조 준수 인간 감독 시스템
고위험 AI 시스템에 필요한 모든 감독 기능 제공:
- 결정 검토 및 승인/거부
- 시스템 중단 및 무효화
- 감독자 알림 및 경고
- 자동화 편향 방지
"""
# 신뢰도 임계값 (이 이하면 반드시 인간 검토 필요)
CONFIDENCE_THRESHOLD = 0.85
# 자동화 편향 경고 메시지
AUTOMATION_BIAS_WARNING = """
⚠️ 자동화 편향 주의
AI의 추천을 맹목적으로 수용하지 마세요.
입력 데이터와 맥락을 독립적으로 검토하세요.
"""
def __init__(
self,
mode: OversightMode,
high_risk_domain: str,
alert_handlers: List[Callable] = None
):
self.mode = mode
self.high_risk_domain = high_risk_domain
self.alert_handlers = alert_handlers or []
self.decisions: Dict[str, AIDecision] = {}
self.is_system_active = True
self.halt_reason: Optional[str] = None
self.logger = logging.getLogger("human_oversight")
# 감독 통계
self.stats = {
"total_decisions": 0,
"human_reviewed": 0,
"auto_executed": 0,
"overridden": 0,
"halted": 0
}
def register_decision(
self,
decision: AIDecision
) -> str:
"""
AI 결정 등록 및 감독 프로세스 시작
Returns:
decision_id: 결정 추적용 ID
"""
if not self.is_system_active:
raise SystemError(f"System halted: {self.halt_reason}")
self.decisions[decision.decision_id] = decision
self.stats["total_decisions"] += 1
# 모드별 처리
if self.mode == OversightMode.HUMAN_IN_THE_LOOP:
# HITL: 모든 결정에 인간 검토 필요
decision.status = DecisionStatus.PENDING_REVIEW
self._notify_reviewers(decision)
elif self.mode == OversightMode.HUMAN_ON_THE_LOOP:
# HOTL: 저신뢰도 결정만 검토, 나머지는 자동 실행
if decision.confidence < self.CONFIDENCE_THRESHOLD:
decision.status = DecisionStatus.PENDING_REVIEW
self._notify_reviewers(decision)
else:
decision.status = DecisionStatus.AUTO_EXECUTED
self.stats["auto_executed"] += 1
self._log_auto_execution(decision)
elif self.mode == OversightMode.HUMAN_IN_COMMAND:
# HIC: 배치 검토를 위해 대기
decision.status = DecisionStatus.AUTO_EXECUTED
self.stats["auto_executed"] += 1
self.logger.info(
f"Decision registered: {decision.decision_id}, "
f"mode={self.mode.value}, status={decision.status.value}"
)
return decision.decision_id
def review_decision(
self,
decision_id: str,
reviewer_id: str,
approved: bool,
notes: str = ""
) -> AIDecision:
"""
인간 검토자의 결정 승인/거부
Args:
decision_id: 결정 ID
reviewer_id: 검토자 ID
approved: 승인 여부
notes: 검토 메모
"""
decision = self.decisions.get(decision_id)
if not decision:
raise ValueError(f"Decision not found: {decision_id}")
# 자동화 편향 경고 표시
self.logger.warning(self.AUTOMATION_BIAS_WARNING)
decision.reviewer_id = reviewer_id
decision.review_notes = notes
decision.review_timestamp = datetime.now()
decision.status = (
DecisionStatus.APPROVED if approved
else DecisionStatus.REJECTED
)
self.stats["human_reviewed"] += 1
self.logger.info(
f"Decision reviewed: {decision_id}, "
f"reviewer={reviewer_id}, approved={approved}"
)
return decision
def override_decision(
self,
decision_id: str,
supervisor_id: str,
new_decision: Any,
reason: str
) -> AIDecision:
"""
AI 결정 무효화 및 인간 결정으로 대체
(EU AI Act 제14조 제4항 c호)
"""
decision = self.decisions.get(decision_id)
if not decision:
raise ValueError(f"Decision not found: {decision_id}")
# 원래 결정 기록 유지
original_prediction = decision.prediction
decision.status = DecisionStatus.OVERRIDDEN
decision.prediction = new_decision
decision.reviewer_id = supervisor_id
decision.review_notes = f"OVERRIDE: {reason}. Original: {original_prediction}"
decision.review_timestamp = datetime.now()
self.stats["overridden"] += 1
self.logger.warning(
f"Decision overridden: {decision_id}, "
f"supervisor={supervisor_id}, reason={reason}"
)
return decision
def halt_system(
self,
supervisor_id: str,
reason: str,
emergency: bool = False
) -> bool:
"""
AI 시스템 즉시 중단
(EU AI Act 제14조 제4항 c호 - "stop" 버튼 기능)
Args:
supervisor_id: 중단 권한자 ID
reason: 중단 사유
emergency: 긴급 중단 여부
"""
self.is_system_active = False
self.halt_reason = reason
# 대기 중인 모든 결정 중단
for decision in self.decisions.values():
if decision.status == DecisionStatus.PENDING_REVIEW:
decision.status = DecisionStatus.HALTED
self.stats["halted"] += 1
self.logger.critical(
f"SYSTEM HALTED by {supervisor_id}: {reason}"
)
# 긴급 알림 발송
if emergency:
self._send_emergency_alert(supervisor_id, reason)
return True
def resume_system(
self,
supervisor_id: str,
confirmation_code: str
) -> bool:
"""시스템 재개 (이중 확인 필요)"""
# 보안: 확인 코드 검증 필요
if not self._verify_confirmation(supervisor_id, confirmation_code):
raise PermissionError("Invalid confirmation")
self.is_system_active = True
self.halt_reason = None
self.logger.info(
f"System resumed by {supervisor_id}"
)
return True
def get_oversight_dashboard(self) -> Dict:
"""
감독자 대시보드 데이터
(EU AI Act 제14조 제4항 a호 - 감독 인터페이스)
"""
pending = [
d for d in self.decisions.values()
if d.status == DecisionStatus.PENDING_REVIEW
]
return {
"system_status": "active" if self.is_system_active else "halted",
"halt_reason": self.halt_reason,
"mode": self.mode.value,
"domain": self.high_risk_domain,
"statistics": self.stats,
"pending_reviews": len(pending),
"pending_decisions": [
{
"id": d.decision_id,
"model": d.model_name,
"confidence": d.confidence,
"reasoning": d.reasoning,
"timestamp": d.timestamp.isoformat()
}
for d in pending
],
"automation_bias_warning": self.AUTOMATION_BIAS_WARNING,
"oversight_capabilities": {
"can_review": True,
"can_override": True,
"can_halt": True,
"can_resume": True
}
}
def _notify_reviewers(self, decision: AIDecision):
"""검토자에게 알림 발송"""
for handler in self.alert_handlers:
try:
handler(
event="REVIEW_REQUIRED",
decision=decision,
priority="high" if decision.confidence < 0.5 else "normal"
)
except Exception as e:
self.logger.error(f"Alert handler failed: {e}")
def _log_auto_execution(self, decision: AIDecision):
"""자동 실행 결정 감사 로그"""
self.logger.info(
f"Auto-executed decision: {decision.decision_id}, "
f"confidence={decision.confidence:.2%}"
)
def _send_emergency_alert(self, supervisor_id: str, reason: str):
"""긴급 알림 발송"""
for handler in self.alert_handlers:
handler(
event="EMERGENCY_HALT",
supervisor=supervisor_id,
reason=reason,
priority="critical"
)
def _verify_confirmation(
self,
supervisor_id: str,
code: str
) -> bool:
"""확인 코드 검증 (실제 구현 필요)"""
# TODO: 실제 인증 로직 구현
return len(code) == 6 and code.isdigit()
# 사용 예시
if __name__ == "__main__":
import uuid
logging.basicConfig(level=logging.INFO)
def alert_handler(**kwargs):
print(f"ALERT: {kwargs}")
# 인간 감독 시스템 초기화 (HITL 모드)
oversight = HumanOversightSystem(
mode=OversightMode.HUMAN_IN_THE_LOOP,
high_risk_domain="credit_scoring",
alert_handlers=[alert_handler]
)
# AI 결정 등록
decision = AIDecision(
decision_id=str(uuid.uuid4())[:8],
model_name="credit_score_model_v2",
input_data={"income": 50000, "debt": 10000},
prediction="APPROVE",
confidence=0.78,
reasoning="Low debt-to-income ratio suggests low default risk"
)
oversight.register_decision(decision)
# 대시보드 확인
dashboard = oversight.get_oversight_dashboard()
print(f"Pending reviews: {dashboard['pending_reviews']}")
# 인간 검토
reviewed = oversight.review_decision(
decision_id=decision.decision_id,
reviewer_id="analyst_001",
approved=True,
notes="Verified income documentation"
)
print(f"Decision status: {reviewed.status.value}")
print(f"Stats: {oversight.stats}")