AI 개발 전략: Human-in-the-Loop와 Crawl-Walk-Run

# ❌ 나쁜 예: AI 단독 판단으로 중요한 결정
def process_loan_application(application):
    """대출 신청 자동 처리"""
    risk_score = ai_model.predict(application)

    if risk_score > 0.7:
        approve_loan(application)  # 자동 승인
        send_money(application.account)
    else:
        reject_loan(application)  # 자동 거부

    return risk_score

# ✅ 좋은 예: 사람의 검토 포함
def process_loan_application(application):
    """대출 신청을 AI로 분석하고 사람이 최종 결정"""
    risk_score = ai_model.predict(application)
    risk_factors = ai_model.explain_prediction(application)

    # AI 추천을 담당자에게 제시
    recommendation = {
        'score': risk_score,
        'factors': risk_factors,
        'ai_suggestion': 'approve' if risk_score > 0.7 else 'review_needed',
        'confidence': ai_model.get_confidence(application)
    }

    # 담당자가 검토 후 결정
    queue_for_human_review(application, recommendation)

    return recommendation

# ✅ 피드백 루프가 있는 시스템
class EmailClassifier:
    """스팸 메일 분류기 with HITL"""

    def classify_email(self, email):
        """이메일을 분류하고 불확실한 경우 사람에게 요청"""
        prediction = self.model.predict(email.content)
        confidence = self.model.predict_proba(email.content)

        # 확신이 낮으면 사람이 검토
        if confidence < 0.8:
            human_label = self.request_human_review(email, prediction)

            # 사람의 판단을 학습 데이터로 저장
            self.training_queue.add({
                'email': email,
                'ai_prediction': prediction,
                'human_label': human_label,
                'timestamp': datetime.now()
            })

            return human_label

        return prediction

    def retrain_model(self):
        """주기적으로 사람의 피드백으로 모델 재학습"""
        new_data = self.training_queue.get_all()

        # 특히 AI가 틀렸던 케이스에 집중
        hard_examples = [d for d in new_data
                        if d['ai_prediction'] != d['human_label']]

        self.model.fit(new_data + hard_examples * 2)  # 어려운 예제 가중치 증가

# 콘텐츠 자동 게시 시스템
def publish_content(article):
    """AI가 콘텐츠를 생성하고 편집자가 승인"""

    # AI가 초안 작성
    draft = ai_writer.generate_article(article.topic)
    seo_optimized = ai_optimizer.optimize_seo(draft)

    # 편집자 승인 대기
    approval = editor_review_queue.add(seo_optimized)

    if approval.status == 'approved':
        publish_to_website(seo_optimized)
    elif approval.status == 'needs_revision':
        # 편집자의 피드백을 반영
        revised = apply_editor_feedback(seo_optimized, approval.feedback)
        publish_to_website(revised)

# 고객 지원 챗봇
class CustomerSupportBot:
    """자동 응답하지만 사람이 모니터링"""

    def handle_conversation(self, customer_message):
        """AI가 응답하되, 위험 신호 감지시 에스컬레이션"""

        response = self.ai_model.generate_response(customer_message)
        sentiment = self.sentiment_analyzer.analyze(customer_message)

        # 위험 신호 감지
        if self.needs_human_intervention(customer_message, sentiment):
            self.escalate_to_human(customer_message, response)
            return "상담원 연결 중입니다..."

        # 자동 응답
        self.send_response(response)

        # 모니터링 대시보드에 기록
        self.log_interaction(customer_message, response, sentiment)

        return response

    def needs_human_intervention(self, message, sentiment):
        """에스컬레이션이 필요한지 판단"""
        return (
            sentiment['anger'] > 0.7 or  # 화난 고객
            'refund' in message.lower() or  # 환불 요청
            'legal' in message.lower() or  # 법적 문제
            self.conversation_length > 10  # 대화가 길어짐
        )

# 상품 추천 시스템
class ProductRecommender:
    """자동 추천하고 주기적으로 성과 분석"""

    def recommend_products(self, user_id):
        """실시간 추천"""
        recommendations = self.model.predict(user_id)
        self.log_recommendations(user_id, recommendations)
        return recommendations

    def weekly_performance_review(self):
        """주간 성과 리포트 생성"""
        metrics = {
            'click_through_rate': self.calculate_ctr(),
            'conversion_rate': self.calculate_conversion(),
            'revenue_impact': self.calculate_revenue(),
            'unusual_patterns': self.detect_anomalies()
        }

        # 이상 패턴 발견시 알림
        if metrics['unusual_patterns']:
            self.alert_ml_team(metrics)

        # 주간 리포트를 팀에 공유
        self.send_report(metrics)

        return metrics

# ✅ 좋은 예: 위험도에 따른 차등 검토
class ContentModerator:
    """콘텐츠 자동 검토 시스템"""

    def moderate_content(self, post):
        """위험도에 따라 다른 처리"""
        risk_score = self.ai_model.predict_risk(post)

        if risk_score > 0.9:
            # 높은 위험: 즉시 차단하고 사람이 검토
            self.block_content(post)
            self.queue_for_review(post, priority='high')
            return 'blocked_pending_review'

        elif risk_score > 0.5:
            # 중간 위험: 사람이 먼저 검토
            decision = self.human_review(post)
            return decision

        else:
            # 낮은 위험: 자동 승인
            self.publish_content(post)
            # 하지만 샘플링해서 주기적 검토
            if random.random() < 0.05:  # 5% 샘플링
                self.queue_for_audit(post)
            return 'published'

# ✅ 좋은 예: AI가 사람의 검토를 돕기
class ReviewAssistant:
    """사람의 검토를 도와주는 AI"""

    def prepare_review_interface(self, application):
        """검토자에게 필요한 정보를 정리해서 제공"""

        # AI 분석 결과
        analysis = {
            'risk_score': self.calculate_risk(application),
            'key_concerns': self.identify_concerns(application),
            'similar_cases': self.find_similar_cases(application),
            'recommendation': self.get_recommendation(application),
            'confidence': self.get_confidence(application)
        }

        # 시각화된 인터페이스 생성
        interface = {
            'application_summary': self.create_summary(application),
            'ai_analysis': analysis,
            'relevant_rules': self.fetch_relevant_rules(application),
            'decision_buttons': ['Approve', 'Reject', 'Need More Info'],
            'feedback_field': 'Why did you make this decision?'
        }

        return interface

# Crawl 단계 예시: 이메일 자동 분류 시작
class EmailClassifierCrawl:
    """초기 버전: 간단한 규칙과 키워드 기반"""

    def __init__(self):
        # 매우 간단한 규칙
        self.urgent_keywords = ['urgent', 'asap', 'emergency']
        self.spam_keywords = ['winner', 'prize', 'click here']

    def classify(self, email):
        """규칙 기반 분류 (ML 없음)"""
        subject = email.subject.lower()

        # 간단한 if-else 규칙
        if any(keyword in subject for keyword in self.spam_keywords):
            return 'spam'
        elif any(keyword in subject for keyword in self.urgent_keywords):
            return 'urgent'
        else:
            return 'normal'

    def test_with_team(self):
        """한 팀에서만 테스트"""
        test_team = 'engineering_team'
        results = []

        for email in self.get_team_emails(test_team):
            classification = self.classify(email)

            # 모든 분류를 사람이 검토
            human_feedback = self.ask_user_feedback(email, classification)
            results.append({
                'email': email,
                'ai_class': classification,
                'human_class': human_feedback,
                'correct': classification == human_feedback
            })

        # 정확도 측정
        accuracy = sum(r['correct'] for r in results) / len(results)
        print(f"Crawl 단계 정확도: {accuracy:.2%}")

        return results

# Walk 단계 예시: ML 모델 도입과 확장
class EmailClassifierWalk:
    """개선된 버전: ML 모델 + 자동화"""

    def __init__(self):
        # 실제 ML 모델 사용
        self.model = self.load_trained_model()
        self.confidence_threshold = 0.8

        # 모니터링 시스템
        self.metrics = MetricsCollector()
        self.alerting = AlertSystem()

    def classify(self, email):
        """ML 기반 분류 with confidence"""
        features = self.extract_features(email)
        prediction = self.model.predict(features)
        confidence = self.model.predict_proba(features).max()

        # 메트릭 기록
        self.metrics.record('prediction', prediction)
        self.metrics.record('confidence', confidence)

        # 낮은 확신은 사람에게
        if confidence < self.confidence_threshold:
            self.metrics.record('human_review_needed', 1)
            return self.request_human_review(email, prediction)

        return prediction

    def deploy_to_multiple_teams(self):
        """여러 팀으로 확장"""
        teams = ['engineering', 'sales', 'support']

        for team in teams:
            # 팀별 성능 추적
            team_metrics = {}

            for email in self.get_team_emails(team):
                classification = self.classify(email)
                team_metrics[classification] = team_metrics.get(classification, 0) + 1

            # 팀별 리포트
            self.generate_team_report(team, team_metrics)

            # 이상 패턴 감지
            if self.detect_anomaly(team_metrics):
                self.alerting.send_alert(
                    f"Unusual pattern detected for {team}",
                    team_metrics
                )

    def continuous_improvement(self):
        """주간 모델 업데이트"""
        # 지난 주 피드백 수집
        feedback_data = self.collect_weekly_feedback()

        # 틀린 예측들로 재학습
        if len(feedback_data) > 100:
            self.retrain_model(feedback_data)

            # A/B 테스트
            self.run_ab_test(
                control=self.current_model,
                treatment=self.new_model
            )

# Run 단계 예시: 완전 자동화 시스템
class EmailClassifierRun:
    """프로덕션 그레이드: 고급 기능 + 자동화"""

    def __init__(self):
        # 앙상블 모델
        self.models = {
            'transformer': self.load_transformer_model(),
            'gradient_boost': self.load_gb_model(),
            'neural_net': self.load_nn_model()
        }

        # 고급 기능
        self.active_learning = ActiveLearningSystem()
        self.auto_scaling = AutoScalingManager()
        self.incident_manager = IncidentManager()

    def classify(self, email):
        """앙상블 예측"""
        predictions = {}

        # 여러 모델의 예측
        for name, model in self.models.items():
            pred = model.predict(email)
            conf = model.predict_proba(email).max()
            predictions[name] = {'prediction': pred, 'confidence': conf}

        # 앙상블 결정 (투표 또는 가중 평균)
        final_prediction = self.ensemble_decision(predictions)

        # 자동 스케일링
        self.auto_scaling.adjust_resources()

        return final_prediction

    def autonomous_operation(self):
        """자율 운영 시스템"""

        # 실시간 모니터링
        health = self.check_system_health()

        if health['status'] == 'degraded':
            # 자동 복구 시도
            self.auto_heal()

        # 능동 학습: AI가 스스로 학습할 데이터 선택
        uncertain_cases = self.active_learning.identify_uncertain_cases()
        for case in uncertain_cases:
            self.request_annotation(case)

        # 자동 모델 업데이트
        if self.should_update_model():
            self.deploy_new_model_version()

        # 비즈니스 메트릭 최적화
        self.optimize_for_business_goals()

    def optimize_for_business_goals(self):
        """비즈니스 목표에 맞게 자동 최적화"""
        current_metrics = {
            'accuracy': 0.94,
            'latency': 150,  # ms
            'cost_per_classification': 0.001,  # $
            'user_satisfaction': 4.5
        }

        # 목표에 따라 파라미터 자동 조정
        if current_metrics['latency'] > 200:
            self.optimize_for_speed()
        elif current_metrics['accuracy'] < 0.90:
            self.optimize_for_accuracy()

class StageTransitionChecker:
    """단계 전환 판단 도구"""

    def should_move_to_walk(self, crawl_metrics):
        """Crawl → Walk 전환 판단"""
        criteria = {
            'basic_functionality_works': crawl_metrics['success_rate'] > 0.7,
            'user_feedback_positive': crawl_metrics['user_satisfaction'] > 3.5,
            'clear_value': crawl_metrics['time_saved'] > 0 or crawl_metrics['cost_saved'] > 0,
            'team_ready': crawl_metrics['team_buy_in'] == True,
            'learned_lessons': len(crawl_metrics['lessons_learned']) > 5
        }

        return all(criteria.values())

    def should_move_to_run(self, walk_metrics):
        """Walk → Run 전환 판단"""
        criteria = {
            'stable_performance': walk_metrics['uptime'] > 0.99,
            'good_accuracy': walk_metrics['accuracy'] > 0.85,
            'user_trust': walk_metrics['adoption_rate'] > 0.6,
            'automation_ready': walk_metrics['automation_rate'] > 0.5,
            'monitoring_in_place': walk_metrics['has_monitoring'] == True,
            'incident_response': walk_metrics['mean_time_to_recover'] < 3600  # 1시간
        }

        return all(criteria.values())

class AISystemWithStages:
    """단계별 HITL 전략을 통합한 AI 시스템"""

    def __init__(self, current_stage='crawl'):
        self.stage = current_stage
        self.model = self.load_model()
        self.metrics = MetricsTracker()

    def process(self, input_data):
        """단계에 따라 다른 처리 전략"""

        if self.stage == 'crawl':
            return self._crawl_process(input_data)
        elif self.stage == 'walk':
            return self._walk_process(input_data)
        elif self.stage == 'run':
            return self._run_process(input_data)

    def _crawl_process(self, input_data):
        """Crawl: 모든 것을 사람이 검토"""

        # AI 예측
        prediction = self.model.predict(input_data)
        explanation = self.model.explain(input_data)

        # 사람에게 제시
        decision = self.request_human_decision(
            input_data=input_data,
            ai_suggestion=prediction,
            explanation=explanation,
            message="AI가 학습 중입니다. 판단해주세요."
        )

        # 피드백 저장
        self.save_feedback(input_data, prediction, decision)

        # 충분한 데이터 수집시 Walk로 전환 고려
        if self.metrics.get_sample_count() > 1000:
            self.consider_stage_transition()

        return decision

    def _walk_process(self, input_data):
        """Walk: 확신 높은 것만 자동화"""

        prediction = self.model.predict(input_data)
        confidence = self.model.get_confidence(input_data)

        # 확신도에 따른 처리
        if confidence > 0.9:
            # 높은 확신: 자동 실행
            result = self.auto_execute(prediction)

            # 샘플링 모니터링
            if random.random() < 0.1:  # 10% 샘플
                self.log_for_review(input_data, prediction, result)

        elif confidence > 0.6:
            # 중간 확신: 사람이 빠르게 검토
            result = self.quick_human_review(input_data, prediction)

        else:
            # 낮은 확신: 상세 검토
            result = self.detailed_human_review(input_data, prediction)

        # 성능 추적
        self.metrics.track_performance(confidence, result)

        # Run 단계로 전환 고려
        if self.metrics.get_automation_rate() > 0.8:
            self.consider_stage_transition()

        return result

    def _run_process(self, input_data):
        """Run: 대부분 자동, 예외만 개입"""

        try:
            # 자동 처리
            prediction = self.model.predict(input_data)
            result = self.auto_execute(prediction)

            # 이상 패턴 감지
            if self.anomaly_detector.is_anomaly(input_data, prediction):
                self.alert_human_for_review(input_data, prediction, result)

            # 주기적 품질 체크
            if self.metrics.get_count() % 1000 == 0:
                self.schedule_quality_audit()

            return result

        except Exception as e:
            # 예외 발생시 사람 개입
            self.incident_manager.create_incident(input_data, e)
            return self.emergency_human_fallback(input_data)

    def consider_stage_transition(self):
        """다음 단계로 전환 고려"""
        current_performance = self.metrics.get_overall_performance()

        if self.stage == 'crawl':
            if self.should_move_to_walk(current_performance):
                self.transition_to_walk()

        elif self.stage == 'walk':
            if self.should_move_to_run(current_performance):
                self.transition_to_run()

    def transition_to_walk(self):
        """Walk 단계로 전환"""
        print("📈 Crawl → Walk 전환 시작")

        # 모델 재학습 (수집한 피드백으로)
        feedback_data = self.get_all_feedback()
        self.model.retrain(feedback_data)

        # 모니터링 시스템 설정
        self.setup_monitoring()

        # 알림 시스템 설정
        self.setup_alerting()

        self.stage = 'walk'
        print("✅ Walk 단계로 전환 완료")

    def transition_to_run(self):
        """Run 단계로 전환"""
        print("🚀 Walk → Run 전환 시작")

        # 고급 모델로 업그레이드
        self.upgrade_to_production_model()

        # 자동 스케일링 설정
        self.setup_autoscaling()

        # 능동 학습 시스템 활성화
        self.enable_active_learning()

        # 사고 대응 시스템 설정
        self.setup_incident_response()

        self.stage = 'run'
        print("✅ Run 단계로 전환 완료")

# ❌ 나쁜 예: 충분한 검증 없이 자동화
def deploy_ai_system_wrong():
    """2주 만에 완전 자동화 시도"""

    # 소규모 테스트만 진행
    test_on_small_dataset()  # 100개 샘플

    # 바로 전체 배포
    deploy_to_all_users()  # 위험!
    enable_full_automation()  # 위험!

    # 문제 발생시:
    # - 대규모 실패
    # - 사용자 신뢰 손실
    # - 비즈니스 영향

# ✅ 좋은 예: 단계적 확장
def deploy_ai_system_right():
    """점진적 배포"""

    # Crawl: 소규모 테스트 (1-2개월)
    pilot_test_with_team(size=50)
    collect_feedback()
    improve_model()

    # Walk: 제한적 배포 (2-3개월)
    deploy_to_subset(percentage=10)
    monitor_closely()
    adjust_parameters()

    # Run: 전면 배포 (검증 후)
    if ready_for_production():
        gradual_rollout(percentage=[20, 50, 100])
        enable_full_features()

# ❌ 나쁜 예: 피드백 루프 없음
class AIWithoutLearning:
    def predict(self, data):
        prediction = self.model.predict(data)
        # 사람이 수정해도 기록하지 않음
        return prediction

    # 결과: 계속 같은 실수 반복

# ✅ 좋은 예: 피드백 루프 구축
class AIWithLearning:
    def predict(self, data):
        prediction = self.model.predict(data)
        return prediction

    def record_feedback(self, data, ai_prediction, human_correction):
        """사람의 수정 사항을 기록"""
        self.feedback_db.insert({
            'data': data,
            'ai_prediction': ai_prediction,
            'correct_answer': human_correction,
            'timestamp': datetime.now()
        })

    def retrain_monthly(self):
        """매월 피드백으로 재학습"""
        feedback = self.feedback_db.get_last_month()

        # 특히 틀린 케이스에 집중
        mistakes = [f for f in feedback
                   if f['ai_prediction'] != f['correct_answer']]

        # 재학습
        self.model.fit(feedback + mistakes * 3)  # 실수에 가중치

# ❌ 나쁜 예: 모니터링 없음
def run_without_monitoring():
    while True:
        user_input = get_user_request()
        result = ai_model.predict(user_input)
        send_result(result)
        # 성능 추적 없음!

# ✅ 좋은 예: 종합 모니터링
class MonitoredAISystem:
    def __init__(self):
        self.metrics = MetricsCollector()
        self.alerting = AlertSystem()

    def process(self, user_input):
        start_time = time.time()

        # 예측
        result = self.ai_model.predict(user_input)

        # 메트릭 기록
        self.metrics.record({
            'latency': time.time() - start_time,
            'prediction': result,
            'confidence': self.ai_model.get_confidence(user_input),
            'timestamp': datetime.now()
        })

        # 이상 감지
        if self.detect_anomaly(result):
            self.alerting.send_alert(
                "Unusual prediction detected",
                {'input': user_input, 'output': result}
            )

        return result

    def hourly_health_check(self):
        """매시간 시스템 건강 체크"""
        metrics = self.metrics.get_last_hour()

        checks = {
            'latency_ok': metrics['avg_latency'] < 500,  # 500ms 이하
            'accuracy_ok': metrics['accuracy'] > 0.85,
            'volume_ok': metrics['request_count'] > 100,
            'error_rate_ok': metrics['error_rate'] < 0.01
        }

        if not all(checks.values()):
            self.alerting.send_alert("Health check failed", checks)

# ❌ 위험한 예: 편향 체크 없음
class BiasedHiringAI:
    def evaluate_candidate(self, resume):
        # 과거 채용 데이터로 학습
        # 문제: 과거 데이터에 성별/인종 편향이 있으면?
        score = self.model.predict(resume)
        return score

# ✅ 좋은 예: 공정성 체크
class FairHiringAI:
    def evaluate_candidate(self, resume):
        score = self.model.predict(resume)

        # 편향 체크
        fairness_metrics = self.check_fairness(resume, score)

        return {
            'score': score,
            'fairness_check': fairness_metrics,
            'needs_review': fairness_metrics['has_concern']
        }

    def check_fairness(self, resume, score):
        """공정성 체크"""

        # 민감 속성 제거 후 재평가
        anonymized_resume = self.remove_sensitive_info(resume)
        anonymous_score = self.model.predict(anonymized_resume)

        # 점수 차이 확인
        score_difference = abs(score - anonymous_score)

        return {
            'original_score': score,
            'anonymous_score': anonymous_score,
            'difference': score_difference,
            'has_concern': score_difference > 0.1  # 10% 이상 차이
        }

    def audit_for_bias(self):
        """주기적 편향 감사"""
        recent_evaluations = self.get_recent_evaluations(days=30)

        # 그룹별 통계
        stats_by_group = self.calculate_acceptance_rates(recent_evaluations)

        # 불공정한 차이가 있는지 확인
        if self.has_disparate_impact(stats_by_group):
            self.alert_ml_team("Potential bias detected", stats_by_group)
            self.pause_auto_decisions()  # 자동 결정 중단

# ❌ 나쁜 예: 사람을 단순 검토자로만 활용
class BadHITL:
    def process(self, items):
        for item in items:
            prediction = self.ai_model.predict(item)

            # 사람에게 단순히 "맞다/틀리다"만 물어봄
            human_says_correct = self.ask_human("Is this correct?", prediction)

            if human_says_correct:
                self.execute(prediction)

        # 문제:
        # - 지루함
        # - 피로
        # - 주의력 감소
        # - 품질 저하

# ✅ 좋은 예: 사람의 전문성을 활용
class GoodHITL:
    def process(self, items):
        # AI가 먼저 분류
        easy_cases = []
        hard_cases = []

        for item in items:
            prediction = self.ai_model.predict(item)
            confidence = self.ai_model.get_confidence(item)

            if confidence > 0.9:
                # 쉬운 케이스: 자동 처리
                easy_cases.append((item, prediction))
            else:
                # 어려운 케이스: 사람의 전문성 필요
                hard_cases.append((item, prediction))

        # 자동 처리
        for item, prediction in easy_cases:
            self.execute(prediction)

        # 사람은 복잡한 케이스에만 집중
        for item, ai_prediction in hard_cases:
            # 풍부한 컨텍스트 제공
            context = self.prepare_review_context(item, ai_prediction)

            # 사람의 전문적 판단 요청
            expert_decision = self.request_expert_review(context)

            # 왜 그렇게 판단했는지도 기록
            self.record_reasoning(item, ai_prediction, expert_decision)

        # 결과:
        # - 사람은 흥미로운 케이스에 집중
        # - 높은 주의력 유지
        # - AI도 사람의 추론을 학습

# HITL 도구
- Prodigy: 어노테이션 도구
- Label Studio: 오픈소스 레이블링
- Amazon SageMaker Ground Truth: 관리형 레이블링

# 모니터링 도구
- MLflow: ML 실험 추적
- Weights & Biases: 모델 성능 추적
- Evidently AI: ML 모델 모니터링

# 공정성 체크
- Fairlearn: 공정성 평가 및 완화
- AI Fairness 360 (IBM): 편향 감지 및 완화