모니터링 및 감사

공정성은 일회성 성취가 아닙니다—지속적인 모니터링과 정기적인 감사가 필요합니다. 이 장에서는 弘益人間 원칙에 따라 운영 중인 공정한 AI를 유지하기 위한 시스템과 관행을 다룹니다.

지속적인 모니터링이 중요한 이유

데이터 분포가 변화하고, 사용자 모집단이 진화하며, 사회적 규범이 변화함에 따라 AI 시스템은 시간이 지남에 따라 드리프트할 수 있습니다. 지속적인 모니터링 없이는 배포 시 공정했던 시스템이 시간이 지남에 따라 편향될 수 있습니다.

모니터링 인프라 설정

1. 실시간 공정성 지표

전통적인 성능 지표와 함께 공정성 지표를 지속적으로 추적합니다.

# 공정성 모니터링 대시보드
import numpy as np
from datetime import datetime, timedelta

class FairnessMonitor:
    """
    실시간 공정성 모니터링 시스템
    """
    def __init__(self, protected_attributes, fairness_thresholds):
        self.protected_attributes = protected_attributes
        self.fairness_thresholds = fairness_thresholds
        self.metrics_history = []
        self.alerts = []

    def log_prediction(self, features, prediction, true_label=None, timestamp=None):
        """
        모니터링을 위한 개별 예측 로그
        """
        if timestamp is None:
            timestamp = datetime.now()

        record = {
            'timestamp': timestamp,
            'features': features,
            'prediction': prediction,
            'true_label': true_label,
            'protected_attrs': {
                attr: features.get(attr) for attr in self.protected_attributes
            }
        }

        self.metrics_history.append(record)

    def compute_realtime_metrics(self, window_hours=24):
        """
        최근 예측에 대한 공정성 지표 계산
        """
        cutoff_time = datetime.now() - timedelta(hours=window_hours)
        recent_records = [r for r in self.metrics_history
                         if r['timestamp'] > cutoff_time]

        if not recent_records:
            return {}

        # 데이터 추출
        predictions = np.array([r['prediction'] for r in recent_records])

        metrics = {}
        for attr in self.protected_attributes:
            groups = {}
            for record in recent_records:
                group = record['protected_attrs'][attr]
                if group not in groups:
                    groups[group] = []
                groups[group].append(record['prediction'])

            # 인구통계학적 균등성 계산
            positive_rates = {}
            for group, preds in groups.items():
                positive_rates[group] = np.mean(preds)

            if len(positive_rates) > 1:
                max_rate = max(positive_rates.values())
                min_rate = min(positive_rates.values())
                disparity = max_rate - min_rate

                metrics[attr] = {
                    'positive_rates': positive_rates,
                    'disparity': disparity,
                    'status': '정상' if disparity < self.fairness_thresholds.get(attr, 0.1) else '경고'
                }

                if disparity >= self.fairness_thresholds.get(attr, 0.1):
                    self.raise_alert(attr, disparity, positive_rates)

        return metrics

    def raise_alert(self, attribute, disparity, rates):
        """
        공정성 경고 발생
        """
        alert = {
            'timestamp': datetime.now(),
            'attribute': attribute,
            'disparity': disparity,
            'rates': rates,
            'severity': '높음' if disparity > 0.2 else '중간'
        }
        self.alerts.append(alert)
        print(f"⚠️ 공정성 경고: {attribute} 불균형 = {disparity:.3f}")

    def generate_report(self):
        """
        모니터링 보고서 생성
        """
        print("공정성 모니터링 보고서")
        print("=" * 70)
        print(f"보고 시간: {datetime.now()}")
        print(f"전체 로그된 예측: {len(self.metrics_history)}")
        print(f"활성 경고: {len([a for a in self.alerts if (datetime.now() - a['timestamp']).seconds < 86400])}")

        current_metrics = self.compute_realtime_metrics()

        print("\n현재 공정성 상태:")
        print("-" * 70)
        for attr, data in current_metrics.items():
            print(f"\n{attr}:")
            print(f"  불균형: {data['disparity']:.3f} [{data['status']}]")
            for group, rate in data['positive_rates'].items():
                print(f"    {group}: {rate:.3f}")

        print("\n弘益人間 - 모든 사용자를 위한 공정성 유지")
        print("=" * 70)

# 사용
monitor = FairnessMonitor(
    protected_attributes=['gender', 'race'],
    fairness_thresholds={'gender': 0.1, 'race': 0.1}
)

2. 드리프트 탐지

데이터 분포 또는 모델 동작이 크게 변경되는 시기를 탐지합니다.

# 분포 드리프트 탐지
from scipy import stats

class DriftDetector:
    """
    피처 및 예측의 분포 드리프트 탐지
    """
    def __init__(self, reference_data, significance_level=0.05):
        self.reference_data = reference_data
        self.significance_level = significance_level

    def detect_feature_drift(self, current_data, feature_name):
        """
        Kolmogorov-Smirnov 테스트를 사용하여 피처 드리프트 탐지
        """
        reference_values = self.reference_data[feature_name]
        current_values = current_data[feature_name]

        # KS 테스트
        statistic, p_value = stats.ks_2samp(reference_values, current_values)

        drift_detected = p_value < self.significance_level

        result = {
            'feature': feature_name,
            'statistic': statistic,
            'p_value': p_value,
            'drift_detected': drift_detected
        }

        if drift_detected:
            print(f"⚠️ {feature_name}에서 드리프트 탐지됨")
            print(f"   KS 통계: {statistic:.4f}, p-값: {p_value:.4f}")

        return result

    def detect_prediction_drift(self, current_predictions):
        """
        예측 분포의 드리프트 탐지
        """
        reference_predictions = self.reference_data['predictions']

        # 분포 비교
        ref_mean = reference_predictions.mean()
        curr_mean = current_predictions.mean()

        ref_std = reference_predictions.std()
        curr_std = current_predictions.std()

        # 평균 차이에 대한 Z-테스트
        z_score = (curr_mean - ref_mean) / (ref_std / np.sqrt(len(current_predictions)))
        p_value = 2 * (1 - stats.norm.cdf(abs(z_score)))

        drift_detected = p_value < self.significance_level

        print(f"예측 드리프트 분석:")
        print(f"  참조 평균: {ref_mean:.3f}")
        print(f"  현재 평균: {curr_mean:.3f}")
        print(f"  Z-점수: {z_score:.3f}")
        print(f"  드리프트 탐지됨: {drift_detected}")

        return drift_detected

    def comprehensive_drift_check(self, current_data, protected_attrs):
        """
        모든 피처와 그룹에 대한 드리프트 확인
        """
        print("\n포괄적인 드리프트 확인")
        print("=" * 70)

        drift_report = {'features': {}, 'groups': {}}

        # 각 피처 확인
        for feature in current_data.columns:
            if feature in protected_attrs:
                continue
            result = self.detect_feature_drift(current_data, feature)
            drift_report['features'][feature] = result

        # 그룹별 드리프트 확인
        for attr in protected_attrs:
            for group in current_data[attr].unique():
                mask = current_data[attr] == group
                group_predictions = current_data[mask]['predictions']

                ref_mask = self.reference_data[attr] == group
                ref_group_predictions = self.reference_data[ref_mask]['predictions']

                # 그룹 분포 비교
                statistic, p_value = stats.ks_2samp(ref_group_predictions, group_predictions)

                drift_report['groups'][f"{attr}_{group}"] = {
                    'statistic': statistic,
                    'p_value': p_value,
                    'drift_detected': p_value < self.significance_level
                }

        return drift_report

자동화된 감사 시스템

예약된 공정성 감사

정기적으로 포괄적인 공정성 감사를 실행합니다.

# 자동화된 감사 스케줄러
import schedule
import time
from datetime import datetime

class AutomatedAuditor:
    """
    자동화된 공정성 감사 시스템
    """
    def __init__(self, model, data_source, protected_attrs):
        self.model = model
        self.data_source = data_source
        self.protected_attrs = protected_attrs
        self.audit_history = []

    def run_audit(self):
        """
        포괄적인 공정성 감사 실행
        """
        print(f"\n{'='*70}")
        print(f"자동화된 공정성 감사 - {datetime.now()}")
        print(f"{'='*70}")

        # 최근 데이터 가져오기
        data = self.data_source.get_recent_data()
        X = data[self.model.feature_names]
        y_true = data['label']

        # 예측 생성
        y_pred = self.model.predict(X)

        audit_results = {
            'timestamp': datetime.now(),
            'sample_size': len(data),
            'overall_accuracy': np.mean(y_true == y_pred),
            'group_metrics': {},
            'alerts': []
        }

        # 각 보호 속성에 대한 공정성 평가
        for attr in self.protected_attrs:
            groups = data[attr].unique()

            group_metrics = {}
            for group in groups:
                mask = data[attr] == group

                accuracy = np.mean(y_true[mask] == y_pred[mask])
                positive_rate = np.mean(y_pred[mask])

                group_metrics[group] = {
                    'accuracy': accuracy,
                    'positive_rate': positive_rate,
                    'sample_size': mask.sum()
                }

            # 불균형 확인
            accuracies = [m['accuracy'] for m in group_metrics.values()]
            acc_disparity = max(accuracies) - min(accuracies)

            rates = [m['positive_rate'] for m in group_metrics.values()]
            rate_disparity = max(rates) - min(rates)

            audit_results['group_metrics'][attr] = {
                'metrics': group_metrics,
                'accuracy_disparity': acc_disparity,
                'rate_disparity': rate_disparity
            }

            # 경고 생성
            if acc_disparity > 0.1:
                audit_results['alerts'].append({
                    'type': 'accuracy_disparity',
                    'attribute': attr,
                    'severity': '높음',
                    'value': acc_disparity
                })

            if rate_disparity > 0.1:
                audit_results['alerts'].append({
                    'type': 'demographic_parity',
                    'attribute': attr,
                    'severity': '중간',
                    'value': rate_disparity
                })

        # 감사 결과 저장
        self.audit_history.append(audit_results)

        # 요약 출력
        self.print_audit_summary(audit_results)

        # 필요시 경고 전송
        if audit_results['alerts']:
            self.send_alerts(audit_results['alerts'])

        return audit_results

    def print_audit_summary(self, results):
        """
        감사 결과 요약 출력
        """
        print(f"\n감사 요약:")
        print(f"  샘플 크기: {results['sample_size']}")
        print(f"  전체 정확도: {results['overall_accuracy']:.3f}")
        print(f"  생성된 경고: {len(results['alerts'])}")

        for attr, data in results['group_metrics'].items():
            print(f"\n{attr}:")
            print(f"  정확도 불균형: {data['accuracy_disparity']:.3f}")
            print(f"  비율 불균형: {data['rate_disparity']:.3f}")

        print("\n弘益人間 - 정기적인 감사는 지속적인 공정성을 보장합니다")

    def send_alerts(self, alerts):
        """
        이해관계자에게 경고 전송
        """
        for alert in alerts:
            print(f"\n🚨 경고: {alert['type']} - {alert['severity']}")
            print(f"   속성: {alert['attribute']}")
            print(f"   값: {alert['value']:.3f}")

    def schedule_audits(self, frequency='daily'):
        """
        정기 감사 예약
        """
        if frequency == 'daily':
            schedule.every().day.at("02:00").do(self.run_audit)
        elif frequency == 'weekly':
            schedule.every().monday.at("02:00").do(self.run_audit)
        elif frequency == 'hourly':
            schedule.every().hour.do(self.run_audit)

        print(f"자동화된 감사 예약됨: {frequency}")

        # 스케줄러 실행
        while True:
            schedule.run_pending()
            time.sleep(60)

# 사용
# auditor = AutomatedAuditor(model, data_source, ['gender', 'race'])
# auditor.schedule_audits('daily')

성능 저하 탐지

# 성능 저하 모니터
class PerformanceDegradationDetector:
    """
    시간이 지남에 따른 점진적인 성능 저하 탐지
    """
    def __init__(self, baseline_metrics, degradation_threshold=0.05):
        self.baseline_metrics = baseline_metrics
        self.degradation_threshold = degradation_threshold
        self.metric_history = []

    def check_degradation(self, current_metrics):
        """
        기준선 대비 현재 지표 비교
        """
        degradations = {}

        for metric_name, baseline_value in self.baseline_metrics.items():
            current_value = current_metrics.get(metric_name)

            if current_value is None:
                continue

            degradation = baseline_value - current_value
            degradation_pct = (degradation / baseline_value) * 100

            degradations[metric_name] = {
                'baseline': baseline_value,
                'current': current_value,
                'degradation': degradation,
                'degradation_pct': degradation_pct,
                'threshold_exceeded': degradation > self.degradation_threshold
            }

        # 지표 로그
        self.metric_history.append({
            'timestamp': datetime.now(),
            'metrics': current_metrics,
            'degradations': degradations
        })

        # 유의미한 저하 보고
        significant_degradations = {
            k: v for k, v in degradations.items()
            if v['threshold_exceeded']
        }

        if significant_degradations:
            print("⚠️ 성능 저하 탐지됨")
            for metric, data in significant_degradations.items():
                print(f"  {metric}: {data['baseline']:.3f} → {data['current']:.3f} "
                      f"({data['degradation_pct']:+.1f}%)")

        return degradations

    def analyze_trends(self, window_size=30):
        """
        시간이 지남에 따른 성능 추세 분석
        """
        if len(self.metric_history) < window_size:
            print("추세 분석을 위한 이력 부족")
            return

        recent_history = self.metric_history[-window_size:]

        print("\n성능 추세 분석")
        print("=" * 70)

        for metric_name in self.baseline_metrics.keys():
            values = [h['metrics'].get(metric_name, 0) for h in recent_history]

            # 추세 탐지를 위한 선형 회귀
            x = np.arange(len(values))
            slope, intercept = np.polyfit(x, values, 1)

            trend = "개선" if slope > 0 else "저하"
            trend_strength = abs(slope) * window_size

            print(f"\n{metric_name}:")
            print(f"  추세: {trend}")
            print(f"  기울기: {slope:.6f}")
            print(f"  {window_size}주기 동안 예상 변화: {trend_strength:.3f}")

            if slope < -0.001:
                print(f"  ⚠️ 저하 추세 탐지됨")

        return True

이해관계자 보고

# 이해관계자 보고서 생성기
class StakeholderReportGenerator:
    """
    이해관계자를 위한 포괄적인 공정성 보고서 생성
    """
    def __init__(self, monitor, auditor):
        self.monitor = monitor
        self.auditor = auditor

    def generate_executive_summary(self):
        """
        공정성 상태에 대한 요약 보고서 생성
        """
        print("\n" + "=" * 70)
        print("요약 보고서 - AI 공정성 상태")
        print("=" * 70)
        print(f"보고 날짜: {datetime.now().strftime('%Y-%m-%d')}")

        # 최신 감사 가져오기
        if self.auditor.audit_history:
            latest_audit = self.auditor.audit_history[-1]

            print(f"\n전체 상태: ", end="")
            if not latest_audit['alerts']:
                print("✓ 준수")
            else:
                high_alerts = [a for a in latest_audit['alerts'] if a['severity'] == '높음']
                if high_alerts:
                    print("✗ 중대 문제")
                else:
                    print("⚠ 주의 필요")

            print(f"\n주요 지표:")
            print(f"  전체 예측: {latest_audit['sample_size']:,}")
            print(f"  전체 정확도: {latest_audit['overall_accuracy']:.1%}")
            print(f"  활성 경고: {len(latest_audit['alerts'])}")

            print(f"\n보호 속성별 공정성 평가:")
            for attr, data in latest_audit['group_metrics'].items():
                print(f"\n{attr}:")
                print(f"  최대 정확도 불균형: {data['accuracy_disparity']:.1%}")
                print(f"  최대 비율 불균형: {data['rate_disparity']:.1%}")

                status = "✓" if (data['accuracy_disparity'] < 0.1 and
                               data['rate_disparity'] < 0.1) else "⚠"
                print(f"  상태: {status}")

        print("\n" + "=" * 70)
        print("弘益人間 - 모든 사용자에게 공정하게 서비스하겠다는 약속")
        print("=" * 70)

    def generate_detailed_report(self, output_file):
        """
        시각화가 포함된 상세 보고서 생성
        """
        # 프로덕션에서는 차트, 표 등을 생성함
        with open(output_file, 'w') as f:
            f.write("상세 공정성 감사 보고서\n")
            f.write("=" * 70 + "\n\n")

            # 모든 감사 데이터 포함
            for audit in self.auditor.audit_history[-10:]:  # 마지막 10개 감사
                f.write(f"\n감사 날짜: {audit['timestamp']}\n")
                f.write(f"정확도: {audit['overall_accuracy']:.3f}\n")
                f.write(f"경고: {len(audit['alerts'])}\n")
                f.write("-" * 70 + "\n")

        print(f"상세 보고서 저장됨: {output_file}")

# 사용
# report_gen = StakeholderReportGenerator(monitor, auditor)
# report_gen.generate_executive_summary()

지속적인 모니터링 모범 사례