머신러닝 모델 서빙 및 추론 API

// 개요 및 중요성 예제 코드
// 여기에 실제 코드 예제가 들어갑니다
console.log('개요 및 중요성 구현');

// 추가 설명을 위한 코멘트
function example1() {
    // 실무에서 사용하는 패턴
    return '개요 및 중요성 완료';
}

// 핵심 개념과 기본 원리 예제 코드
// 여기에 실제 코드 예제가 들어갑니다
console.log('핵심 개념과 기본 원리 구현');

// 추가 설명을 위한 코멘트
function example2() {
    // 실무에서 사용하는 패턴
    return '핵심 개념과 기본 원리 완료';
}

// 실전 구현 가이드 예제 코드
// 여기에 실제 코드 예제가 들어갑니다
console.log('실전 구현 가이드 구현');

// 추가 설명을 위한 코멘트
function example3() {
    // 실무에서 사용하는 패턴
    return '실전 구현 가이드 완료';
}

// 고급 패턴 및 최적화 예제 코드
// 여기에 실제 코드 예제가 들어갑니다
console.log('고급 패턴 및 최적화 구현');

// 추가 설명을 위한 코멘트
function example4() {
    // 실무에서 사용하는 패턴
    return '고급 패턴 및 최적화 완료';
}

# 실무 적용 사례: 이커머스 추천 시스템
import asyncio
import redis
from concurrent.futures import ThreadPoolExecutor
import logging

class RecommendationService:
    def __init__(self):
        self.redis_client = redis.Redis(host='localhost', port=6379)
        self.thread_pool = ThreadPoolExecutor(max_workers=20)
        self.model_cache = {}
        
    async def get_product_recommendations(self, user_id: str, num_items: int = 10):
        """사용자 맞춤 상품 추천 API"""
        try:
            # 캐시된 추천 결과 확인
            cached_result = await self.get_cached_recommendations(user_id)
            if cached_result:
                return cached_result
            
            # 실시간 모델 추론
            user_features = await self.get_user_features(user_id)
            item_features = await self.get_item_features()
            
            # 병렬 추론으로 성능 최적화
            recommendations = await self.parallel_inference(
                user_features, item_features, num_items
            )
            
            # 결과 캐싱 (5분간)
            await self.cache_recommendations(user_id, recommendations, ttl=300)
            
            return {
                "user_id": user_id,
                "recommendations": recommendations,
                "generated_at": time.time()
            }
            
        except Exception as e:
            logging.error(f"Recommendation error for user {user_id}: {str(e)}")
            return await self.fallback_recommendations(user_id)

# 성능 최적화 및 모니터링 시스템
import psutil
import time
from dataclasses import dataclass
import prometheus_client

class ModelMonitoringSystem:
    def __init__(self):
        self.metrics_collector = prometheus_client.CollectorRegistry()
        self.inference_time_histogram = prometheus_client.Histogram(
            'ml_inference_duration_seconds', 
            'ML model inference time',
            ['model_name', 'version'],
            registry=self.metrics_collector
        )
        
    def monitor_model_performance(self, model_name: str):
        """모델 성능 실시간 모니터링"""
        def decorator(func):
            async def wrapper(*args, **kwargs):
                start_time = time.time()
                
                try:
                    result = await func(*args, **kwargs)
                    inference_time = time.time() - start_time
                    
                    # Prometheus 메트릭 업데이트
                    self.inference_time_histogram.labels(
                        model_name=model_name, version='v1'
                    ).observe(inference_time)
                    
                    # 성능 임계값 확인
                    if inference_time > 1.0:  # 1초 이상
                        await self.send_performance_alert(model_name, inference_time)
                    
                    return result
                    
                except Exception as e:
                    await self.handle_inference_error(model_name, str(e))
                    raise
                    
            return wrapper
        return decorator