AI Engineering

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

# Apache Airflow를 사용한 데이터 수집 파이프라인 예시
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta

# 기본 인자 설정
default_args = {
    'owner': 'ai_engineer',
    'depends_on_past': False,
    'start_date': datetime(2023, 1, 1),
    'email_on_failure': True,
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}

# DAG 정의
dag = DAG(
    'data_collection_pipeline',
    default_args=default_args,
    description='데이터 수집 및 전처리 파이프라인',
    schedule_interval=timedelta(days=1),
)

# 데이터 수집 함수
def collect_data(**context):
    # 데이터 소스에서 데이터 수집 로직
    print("데이터 수집 중…")
    # 수집된 데이터 반환

# 데이터 전처리 함수
def preprocess_data(**context):
    # 데이터 전처리 로직
    print("데이터 전처리 중…")
    # 전처리된 데이터 반환

# 데이터 검증 함수
def validate_data(**context):
    # 데이터 품질 검증 로직
    print("데이터 검증 중…")
    # 검증 결과 반환

# 태스크 생성 및 연결
collect_task = PythonOperator(
    task_id='collect_data',
    python_callable=collect_data,
    provide_context=True,
    dag=dag,
)

preprocess_task = PythonOperator(
    task_id='preprocess_data',
    python_callable=preprocess_data,
    provide_context=True,
    dag=dag,
)

validate_task = PythonOperator(
    task_id='validate_data',
    python_callable=validate_data,
    provide_context=True,
    dag=dag,
)

# 태스크 의존성 설정
collect_task >> preprocess_task >> validate_task

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99

# PyTorch와 MLflow를 사용한 모델 훈련 및 실험 추적 예시
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader
import mlflow
import mlflow.pytorch

# 모델 정의
class SimpleNN(nn.Module):
    def __init__(self, input_size, hidden_size, output_size):
        super(SimpleNN, self).__init__()
        self.layer1 = nn.Linear(input_size, hidden_size)
        self.relu = nn.ReLU()
        self.layer2 = nn.Linear(hidden_size, output_size)
    
    def forward(self, x):
        x = self.layer1(x)
        x = self.relu(x)
        x = self.layer2(x)
        return x

# 훈련 함수
def train_model(model, train_loader, val_loader, criterion, optimizer, epochs):
    # MLflow 실험 시작
    mlflow.start_run()
    
    # 하이퍼파라미터 로깅
    mlflow.log_param("hidden_size", model.layer1.out_features)
    mlflow.log_param("learning_rate", optimizer.param_groups[0]['lr'])
    mlflow.log_param("epochs", epochs)
    
    for epoch in range(epochs):
        # 훈련 모드
        model.train()
        train_loss = 0
        for inputs, targets in train_loader:
            # 순전파
            outputs = model(inputs)
            loss = criterion(outputs, targets)
            
            # 역전파 및 최적화
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
            
            train_loss += loss.item()
        
        # 평가 모드
        model.eval()
        val_loss = 0
        correct = 0
        total = 0
        with torch.no_grad():
            for inputs, targets in val_loader:
                outputs = model(inputs)
                loss = criterion(outputs, targets)
                val_loss += loss.item()
                
                _, predicted = outputs.max(1)
                total += targets.size(0)
                correct += predicted.eq(targets).sum().item()
        
        # 메트릭 계산 및 로깅
        train_loss /= len(train_loader)
        val_loss /= len(val_loader)
        accuracy = 100. * correct / total
        
        mlflow.log_metric("train_loss", train_loss, step=epoch)
        mlflow.log_metric("val_loss", val_loss, step=epoch)
        mlflow.log_metric("accuracy", accuracy, step=epoch)
        
        print(f"Epoch {epoch+1}/{epochs}: "
              f"Train Loss: {train_loss:f}, "
              f"Val Loss: {val_loss:f}, "
              f"Accuracy: {accuracy:f}%")
    
    # 모델 저장
    mlflow.pytorch.log_model(model, "model")
    mlflow.end_run()
    
    return model

# 모델 훈련 실행
input_size = 784  # MNIST 이미지 크기
hidden_size = 128
output_size = 10  # 0-9 숫자 분류
model = SimpleNN(input_size, hidden_size, output_size)

# 손실 함수 및 옵티마이저 설정
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)

# 데이터 로더 설정 (실제 코드에서는 실제 데이터 사용)
# train_loader = DataLoader(…)
# val_loader = DataLoader(…)

# 모델 훈련
# trained_model = train_model(model, train_loader, val_loader, criterion, optimizer, epochs=10)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

# Flask와 Docker를 사용한 모델 배포 예시

# app.py
from flask import Flask, request, jsonify
import torch
import torch.nn as nn
import numpy as np

app = Flask(__name__)

# 모델 정의 (실제 배포에서는 저장된 모델을 로드)
class SimpleNN(nn.Module):
    def __init__(self, input_size, hidden_size, output_size):
        super(SimpleNN, self).__init__()
        self.layer1 = nn.Linear(input_size, hidden_size)
        self.relu = nn.ReLU()
        self.layer2 = nn.Linear(hidden_size, output_size)
    
    def forward(self, x):
        x = self.layer1(x)
        x = self.relu(x)
        x = self.layer2(x)
        return x

# 모델 로드
def load_model():
    model = SimpleNN(784, 128, 10)
    model.load_state_dict(torch.load('model.pth'))
    model.eval()
    return model

model = load_model()

@app.route('/predict', methods=['POST'])
def predict():
    try:
        # 요청에서 데이터 추출
        data = request.json
        input_data = np.array(data['input'])
        
        # 입력 데이터 전처리
        input_tensor = torch.FloatTensor(input_data)
        
        # 예측 수행
        with torch.no_grad():
            output = model(input_tensor)
            probabilities = torch.nn.functional.softmax(output, dim=0)
            predicted_class = torch.argmax(probabilities).item()
        
        # 결과 반환
        return jsonify({
            'predicted_class': predicted_class,
            'probabilities': probabilities.tolist()
        })
    
    except Exception as e:
        return jsonify({'error': str(e)}), 400

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

# Dockerfile
FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 5000

CMD ["python", "app.py"]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77

# GitHub Actions를 사용한 CI/CD 파이프라인 예시
# .github/workflows/model-pipeline.yml

name: Model Training and Deployment Pipeline

on:
  push:
    branches: [ main ]
  pull_request:
    branches: [ main ]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v2
    - name: Set up Python
      uses: actions/setup-python@v2
      with:
        python-version: '3.9'
    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install -r requirements.txt
    - name: Run tests
      run: |
        pytest tests/

  train:
    needs: test
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v2
    - name: Set up Python
      uses: actions/setup-python@v2
      with:
        python-version: '3.9'
    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install -r requirements.txt
    - name: Train model
      run: |
        python scripts/train.py
    - name: Evaluate model
      run: |
        python scripts/evaluate.py
    - name: Upload model artifact
      uses: actions/upload-artifact@v2
      with:
        name: model-artifact
        path: models/model.pth

  deploy:
    needs: train
    if: github.event_name == 'push' && github.ref == 'refs/heads/main'
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v2
    - name: Download model artifact
      uses: actions/download-artifact@v2
      with:
        name: model-artifact
        path: models/
    - name: Set up Docker Buildx
      uses: docker/setup-buildx-action@v1
    - name: Login to DockerHub
      uses: docker/login-action@v1
      with:
        username: ${{ secrets.DOCKERHUB_USERNAME }}
        password: ${{ secrets.DOCKERHUB_TOKEN }}
    - name: Build and push
      uses: docker/build-push-action@v2
      with:
        context: .
        push: true
        tags: user/model-api:latest

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80

# Prometheus와 Grafana를 사용한 모델 모니터링 예시
from flask import Flask, request, jsonify
import torch
import numpy as np
from prometheus_client import Counter, Histogram, Gauge, Summary, start_http_server
import time

app = Flask(__name__)

# Prometheus 메트릭 정의
PREDICTION_COUNT = Counter('model_predictions_total', 'Total number of predictions')
PREDICTION_LATENCY = Histogram('model_prediction_latency_seconds', 'Prediction latency in seconds')
PREDICTION_ERRORS = Counter('model_prediction_errors_total', 'Total number of prediction errors')
MODEL_ACCURACY = Gauge('model_accuracy', 'Current model accuracy')
INPUT_DRIFT = Gauge('input_drift_score', 'Input data drift score')

# 모델 로드 및 설정
# (이전 예시 참조)

@app.route('/predict', methods=['POST'])
def predict():
    start_time = time.time()
    
    try:
        # 요청에서 데이터 추출
        data = request.json
        input_data = np.array(data['input'])
        
        # 입력 데이터 드리프트 감지
        drift_score = check_data_drift(input_data)
        INPUT_DRIFT.set(drift_score)
        
        # 예측 수행
        with torch.no_grad():
            input_tensor = torch.FloatTensor(input_data)
            output = model(input_tensor)
            probabilities = torch.nn.functional.softmax(output, dim=0)
            predicted_class = torch.argmax(probabilities).item()
        
        # 메트릭 업데이트
        PREDICTION_COUNT.inc()
        
        # 결과 반환
        response = {
            'predicted_class': predicted_class,
            'probabilities': probabilities.tolist()
        }
        
        return jsonify(response)
    
    except Exception as e:
        PREDICTION_ERRORS.inc()
        return jsonify({'error': str(e)}), 400
    
    finally:
        # 예측 지연 시간 측정
        latency = time.time() - start_time
        PREDICTION_LATENCY.observe(latency)

def check_data_drift(input_data):
    # 입력 데이터 드리프트 감지 로직
    # 기준 데이터와 현재 데이터의 통계적 특성 비교
    # 간단한 예시로, 평균 차이를 기반으로 점수 계산
    reference_mean = 0.5  # 기준 데이터의 평균 (예시 값)
    current_mean = np.mean(input_data)
    
    drift_score = abs(current_mean - reference_mean)
    return drift_score

# 모델 정확도 업데이트 함수 (주기적으로 호출)
def update_model_accuracy():
    # 테스트 데이터로 모델 정확도 평가
    accuracy = evaluate_model_on_test_data()
    MODEL_ACCURACY.set(accuracy)

# 메트릭 서버 시작
start_http_server(8000)

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152

# 데이터 파이프라인 예시 코드 (Airflow 사용)
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta
import pandas as pd
import numpy as np
from sklearn.preprocessing import StandardScaler
import logging

default_args = {
    'owner': 'ai_team',
    'depends_on_past': False,
    'start_date': datetime(2023, 1, 1),
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'customer_churn_data_pipeline',
    default_args=default_args,
    description='Customer churn data processing pipeline',
    schedule_interval='@daily',
)

def extract_customer_data(**kwargs):
    """고객 데이터 추출 함수"""
    try:
        # 데이터베이스에서 고객 데이터 추출
        # 실제 구현에서는 SQL 쿼리 또는 API 호출 사용
        logging.info("Extracting customer data...")
        
        # 예시 데이터 (실제로는 DB에서 가져옴)
        customers = pd.DataFrame({
            'customer_id': range(1000, 1100),
            'subscription_length': np.random.randint(1, 60, 100),
            'monthly_charges': np.random.uniform(20, 200, 100),
            'total_charges': np.random.uniform(100, 5000, 100),
            'contract_type': np.random.choice(['Monthly', 'One year', 'Two year'], 100),
            'payment_method': np.random.choice(['Credit card', 'Bank transfer', 'Electronic check'], 100),
            'tech_support_calls': np.random.randint(0, 15, 100),
            'last_active_days': np.random.randint(0, 30, 100),
        })
        
        # 결과 저장
        customers.to_csv('/tmp/raw_customer_data.csv', index=False)
        return '/tmp/raw_customer_data.csv'
    
    except Exception as e:
        logging.error(f"Error extracting customer data: {e}")
        raise

def validate_data(**kwargs):
    """데이터 검증 함수"""
    try:
        input_path = kwargs['ti'].xcom_pull(task_ids='extract_customer_data')
        logging.info(f"Validating data from {input_path}")
        
        # 데이터 로드
        df = pd.read_csv(input_path)
        
        # 데이터 검증 로직
        # 1. 필수 열 존재 확인
        required_columns = ['customer_id', 'subscription_length', 'monthly_charges']
        missing_columns = [col for col in required_columns if col not in df.columns]
        if missing_columns:
            raise ValueError(f"Missing required columns: {missing_columns}")
        
        # 2. 중복 고객 ID 확인
        if df['customer_id'].duplicated().any():
            raise ValueError("Duplicate customer IDs found")
        
        # 3. 데이터 범위 확인
        if (df['monthly_charges'] < 0).any():
            raise ValueError("Negative monthly charges found")
        
        # 4. 결측치 확인
        missing_values = df.isnull().sum()
        if missing_values.sum() > 0:
            logging.warning(f"Missing values found: {missing_values[missing_values > 0]}")
        
        # 검증 통과 - 데이터 저장
        df.to_csv('/tmp/validated_customer_data.csv', index=False)
        return '/tmp/validated_customer_data.csv'
    
    except Exception as e:
        logging.error(f"Data validation error: {e}")
        raise

def transform_features(**kwargs):
    """특성 엔지니어링 함수"""
    try:
        input_path = kwargs['ti'].xcom_pull(task_ids='validate_data')
        logging.info(f"Transforming features from {input_path}")
        
        # 데이터 로드
        df = pd.read_csv(input_path)
        
        # 특성 엔지니어링
        # 1. 범주형 변수 원-핫 인코딩
        df = pd.get_dummies(df, columns=['contract_type', 'payment_method'], drop_first=True)
        
        # 2. 새로운 특성 생성
        df['avg_monthly_charge'] = df['total_charges'] / df['subscription_length']
        df['inactive_ratio'] = df['last_active_days'] / 30  # 활동 비율 계산
        
        # 3. 수치형 특성 정규화
        scaler = StandardScaler()
        numeric_features = ['subscription_length', 'monthly_charges', 'total_charges', 
                           'tech_support_calls', 'avg_monthly_charge']
        df[numeric_features] = scaler.fit_transform(df[numeric_features])
        
        # 4. 특성 선택 (중요한 특성만 유지)
        selected_columns = ['customer_id', 'subscription_length', 'monthly_charges', 
                           'tech_support_calls', 'last_active_days', 'inactive_ratio', 
                           'avg_monthly_charge'] + [col for col in df.columns if 'contract_type' in col 
                                                  or 'payment_method' in col]
        df = df[selected_columns]
        
        # 변환된 데이터 저장
        df.to_csv('/tmp/features_customer_data.csv', index=False)
        return '/tmp/features_customer_data.csv'
    
    except Exception as e:
        logging.error(f"Feature transformation error: {e}")
        raise

# DAG 태스크 정의
extract_task = PythonOperator(
    task_id='extract_customer_data',
    python_callable=extract_customer_data,
    provide_context=True,
    dag=dag,
)

validate_task = PythonOperator(
    task_id='validate_data',
    python_callable=validate_data,
    provide_context=True,
    dag=dag,
)

transform_task = PythonOperator(
    task_id='transform_features',
    python_callable=transform_features,
    provide_context=True,
    dag=dag,
)

# 태스크 의존성 설정
extract_task >> validate_task >> transform_task

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167

# 모델 훈련 및 평가 예시 코드 (MLflow 사용)
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score, roc_auc_score
from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
import mlflow
import mlflow.sklearn
import logging

# 로깅 설정
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

def load_train_data(path='/tmp/features_customer_data.csv', target_path='/tmp/customer_churn_labels.csv'):
    """훈련 데이터 및 레이블 로드"""
    try:
        # 특성 데이터 로드
        features = pd.read_csv(path)
        
        # 타겟 데이터 로드 (이탈 여부)
        # 실제 구현에서는 이 데이터가 별도 소스에서 올 수 있음
        targets = pd.read_csv(target_path)
        
        # 고객 ID로 데이터 병합
        data = features.merge(targets, on='customer_id')
        
        # ID 열 분리
        X = data.drop(['customer_id', 'churn'], axis=1)
        y = data['churn']
        
        return X, y
    
    except Exception as e:
        logger.error(f"Error loading training data: {e}")
        raise

def train_evaluate_model(X, y, model_type='random_forest', experiment_name='churn_prediction'):
    """모델 훈련 및 평가"""
    try:
        # MLflow 실험 설정
        mlflow.set_experiment(experiment_name)
        
        # 데이터 분할
        X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
        
        # MLflow 실행 시작
        with mlflow.start_run() as run:
            # 모델 선택
            if model_type == 'random_forest':
                model = RandomForestClassifier(n_estimators=100, random_state=42)
                params = {
                    'n_estimators': 100,
                    'max_depth': None,
                    'min_samples_split': 2,
                    'min_samples_leaf': 1
                }
            elif model_type == 'gradient_boosting':
                model = GradientBoostingClassifier(n_estimators=100, random_state=42)
                params = {
                    'n_estimators': 100,
                    'learning_rate': 0.1,
                    'max_depth': 3
                }
            elif model_type == 'logistic_regression':
                model = LogisticRegression(max_iter=1000, random_state=42)
                params = {
                    'C': 1.0,
                    'penalty': 'l2',
                    'solver': 'lbfgs'
                }
            else:
                raise ValueError(f"Unsupported model type: {model_type}")
            
            # 파라미터 로깅
            mlflow.log_params(params)
            
            # 모델 훈련
            model.fit(X_train, y_train)
            
            # 예측
            y_pred = model.predict(X_test)
            y_prob = model.predict_proba(X_test)[:, 1]
            
            # 평가 메트릭 계산
            accuracy = accuracy_score(y_test, y_pred)
            precision = precision_score(y_test, y_pred)
            recall = recall_score(y_test, y_pred)
            f1 = f1_score(y_test, y_pred)
            roc_auc = roc_auc_score(y_test, y_prob)
            
            # 메트릭 로깅
            metrics = {
                'accuracy': accuracy,
                'precision': precision,
                'recall': recall,
                'f1_score': f1,
                'roc_auc': roc_auc
            }
            mlflow.log_metrics(metrics)
            
            # 모델 저장
            mlflow.sklearn.log_model(model, "model")
            
            # 결과 출력
            logger.info(f"Model training completed with metrics: {metrics}")
            
            # 교차 검증 수행
            cv_scores = cross_val_score(model, X, y, cv=5, scoring='roc_auc')
            mlflow.log_metric("cv_roc_auc_mean", cv_scores.mean())
            mlflow.log_metric("cv_roc_auc_std", cv_scores.std())
            
            logger.info(f"Cross-validation ROC AUC: {cv_scores.mean():.4f} ± {cv_scores.std():.4f}")
            
            return {
                'model': model,
                'metrics': metrics,
                'run_id': run.info.run_id
            }
    
    except Exception as e:
        logger.error(f"Error in model training and evaluation: {e}")
        raise

def main():
    """메인 실행 함수"""
    try:
        # 데이터 로드
        logger.info("Loading training data...")
        X, y = load_train_data()
        
        # 여러 모델 학습 및 평가
        models = ['random_forest', 'gradient_boosting', 'logistic_regression']
        results = {}
        
        for model_type in models:
            logger.info(f"Training and evaluating {model_type} model...")
            results[model_type] = train_evaluate_model(X, y, model_type)
        
        # 최고 성능 모델 선택
        best_model_type = max(results, key=lambda k: results[k]['metrics']['roc_auc'])
        best_model_results = results[best_model_type]
        
        logger.info(f"Best model: {best_model_type} with ROC AUC: {best_model_results['metrics']['roc_auc']:.4f}")
        
        # 최고 성능 모델 등록 (모델 레지스트리 사용)
        mlflow.set_tracking_uri("http://localhost:5000")  # MLflow 서버 주소
        model_name = "customer_churn_predictor"
        model_version = mlflow.register_model(
            f"runs:/{best_model_results['run_id']}/model",
            model_name
        )
        
        logger.info(f"Registered model: {model_name} version {model_version.version}")
        
        return best_model_results
    
    except Exception as e:
        logger.error(f"Error in main function: {e}")
        raise

if __name__ == "__main__":
    main()

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249

# FastAPI를 사용한 모델 서빙 예시 코드
from fastapi import FastAPI, HTTPException, Depends, BackgroundTasks
from pydantic import BaseModel
from typing import List, Dict, Optional
import pandas as pd
import numpy as np
import pickle
import json
import logging
from datetime import datetime
import mlflow.pyfunc
import redis
import os
from prometheus_client import Counter, Histogram, Gauge
import time

# 로깅 설정
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# Redis 연결 설정
redis_client = redis.Redis(
    host=os.getenv("REDIS_HOST", "localhost"),
    port=int(os.getenv("REDIS_PORT", 6379)),
    db=0
)

# Prometheus 메트릭 정의
PREDICTION_COUNT = Counter('churn_predictions_total', 'Total number of churn predictions', ['result'])
PREDICTION_LATENCY = Histogram('churn_prediction_latency_seconds', 'Churn prediction latency in seconds')
MODEL_ACCURACY = Gauge('churn_model_accuracy', 'Current churn model accuracy')

# 기본 정확도 설정 (모니터링 시스템에서 주기적으로 업데이트됨)
MODEL_ACCURACY.set(0.85)

# FastAPI 앱 초기화
app = FastAPI(title="Customer Churn Prediction API")

# 요청 및 응답 모델
class CustomerFeatures(BaseModel):
    customer_id: str
    subscription_length: int
    monthly_charges: float
    total_charges: float
    contract_type: str
    payment_method: str
    tech_support_calls: int
    last_active_days: int

class PredictionResponse(BaseModel):
    customer_id: str
    churn_probability: float
    churn_risk_level: str
    model_version: str
    prediction_id: str
    prediction_time: str

# 모델 로드 함수
def load_model():
    """MLflow에서 최신 모델 로드"""
    try:
        # 모델 레지스트리에서 'Production' 단계의 모델 로드
        model = mlflow.pyfunc.load_model("models:/customer_churn_predictor/Production")
        logger.info("Successfully loaded churn prediction model")
        return model
    except Exception as e:
        logger.error(f"Error loading model: {e}")
        # 백업 방법: 로컬에 저장된 모델 파일 로드
        with open('models/churn_model_backup.pkl', 'rb') as f:
            model = pickle.load(f)
        logger.info("Loaded backup model")
        return model

# 의존성 주입을 위한 모델 제공자
def get_model():
    """싱글톤 패턴으로 모델 제공"""
    if not hasattr(get_model, "model"):
        get_model.model = load_model()
    return get_model.model

# 전처리 함수
def preprocess_features(customer: CustomerFeatures) -> pd.DataFrame:
    """요청 데이터 전처리"""
    # 데이터 프레임 변환
    data = pd.DataFrame([customer.dict()])
    
    # 1. 범주형 변수 처리
    contract_types = ['Monthly', 'One year', 'Two year']
    payment_methods = ['Credit card', 'Bank transfer', 'Electronic check']
    
    # 원-핫 인코딩
    for contract in contract_types[1:]:  # 첫 번째 카테고리는 기준으로 사용
        data[f'contract_type_{contract}'] = (data['contract_type'] == contract).astype(int)
    
    for method in payment_methods[1:]:
        data[f'payment_method_{method}'] = (data['payment_method'] == method).astype(int)
    
    # 2. 새로운 특성 생성
    data['avg_monthly_charge'] = data['total_charges'] / data['subscription_length']
    data['inactive_ratio'] = data['last_active_days'] / 30
    
    # 3. 원본 범주형 변수 제거
    data = data.drop(['contract_type', 'payment_method'], axis=1)
    
    # 고객 ID 별도 저장
    customer_id = data['customer_id']
    data = data.drop(['customer_id'], axis=1)
    
    return data, customer_id

# 예측 결과를 Redis에 캐싱하는 함수
def cache_prediction(customer_id: str, prediction: dict):
    """예측 결과를 Redis에 캐싱"""
    try:
        # 12시간 유효기간으로 캐싱
        redis_client.setex(
            f"churn_prediction:{customer_id}",
            43200,  # 12시간 (초 단위)
            json.dumps(prediction)
        )
    except Exception as e:
        logger.warning(f"Failed to cache prediction: {e}")

# 캐시된 예측 결과 조회 함수
def get_cached_prediction(customer_id: str) -> Optional[dict]:
    """Redis에서 캐시된 예측 결과 조회"""
    try:
        cached = redis_client.get(f"churn_prediction:{customer_id}")
        if cached:
            return json.loads(cached)
        return None
    except Exception as e:
        logger.warning(f"Failed to retrieve cached prediction: {e}")
        return None

# 마케팅 시스템에 알림 전송 함수 (백그라운드 작업)
def send_to_marketing_system(customer_id: str, churn_probability: float, risk_level: str):
    """고위험 고객 정보를 마케팅 시스템에 전송"""
    try:
        # 실제 구현에서는 외부 시스템 API 호출
        logger.info(f"Sending alert to marketing system for customer {customer_id}")
        # requests.post("https://marketing-api.example.com/churn-alerts",
        #     json={
        #         "customer_id": customer_id,
        #         "churn_probability": churn_probability,
        #         "risk_level": risk_level,
        #         "timestamp": datetime.now().isoformat()
        #     },
        #     headers={"Authorization": "Bearer " + API_KEY}
        # )
    except Exception as e:
        logger.error(f"Failed to send alert to marketing system: {e}")

@app.get("/")
def read_root():
    return {"message": "Customer Churn Prediction API"}

@app.get("/health")
def health_check():
    """헬스 체크 엔드포인트"""
    try:
        # 모델이 로드되었는지 확인
        model = get_model()
        # Redis 연결 확인
        redis_client.ping()
        return {"status": "healthy", "model_loaded": True, "redis_connected": True}
    except Exception as e:
        logger.error(f"Health check failed: {e}")
        return {"status": "unhealthy", "error": str(e)}, 503

@app.post("/predict", response_model=PredictionResponse)
async def predict_churn(
    customer: CustomerFeatures,
    background_tasks: BackgroundTasks,
    model=Depends(get_model)
):
    """고객 이탈 예측 엔드포인트"""
    prediction_start_time = time.time()
    
    try:
        # 1. 캐시된 결과 확인
        cached_prediction = get_cached_prediction(customer.customer_id)
        if cached_prediction:
            logger.info(f"Returning cached prediction for customer {customer.customer_id}")
            # 캐시된 결과를 반환하되, 예측 시간은 현재 시간으로 업데이트
            cached_prediction["prediction_time"] = datetime.now().isoformat()
            return PredictionResponse(**cached_prediction)
        
        # 2. 특성 전처리
        features, customer_id = preprocess_features(customer)
        
        # 3. 예측 수행
        churn_probability = model.predict(features)[0]
        
        # 4. 결과 후처리
        risk_level = "high" if churn_probability > 0.7 else "medium" if churn_probability > 0.3 else "low"
        
        # 5. 응답 생성
        prediction_id = f"pred_{int(time.time())}_{customer.customer_id}"
        response = {
            "customer_id": customer.customer_id,
            "churn_probability": float(churn_probability),
            "churn_risk_level": risk_level,
            "model_version": getattr(model, 'version', 'unknown'),
            "prediction_id": prediction_id,
            "prediction_time": datetime.now().isoformat()
        }
        
        # 6. 결과 캐싱
        background_tasks.add_task(cache_prediction, customer.customer_id, response)
        
        # 7. 고위험 고객 마케팅 알림
        if risk_level == "high":
            background_tasks.add_task(
                send_to_marketing_system,
                customer.customer_id,
                churn_probability,
                risk_level
            )
        
        # 8. 메트릭 업데이트
        prediction_label = "churn" if churn_probability > 0.5 else "no_churn"
        PREDICTION_COUNT.labels(result=prediction_label).inc()
        
        # 9. 응답 반환
        return PredictionResponse(**response)
    
    except Exception as e:
        logger.error(f"Prediction error: {e}")
        raise HTTPException(status_code=500, detail=str(e))
    
    finally:
        # 지연 시간 측정 및 기록
        prediction_time = time.time() - prediction_start_time
        PREDICTION_LATENCY.observe(prediction_time)

# 애플리케이션 시작 시 필요한 초기화
@app.on_event("startup")
async def startup_event():
    """애플리케이션 시작 시 실행되는 작업"""
    logger.info("Starting Churn Prediction API")
    # 모델 미리 로드
    get_model()

# 애플리케이션 종료 시 정리 작업
@app.on_event("shutdown")
async def shutdown_event():
    """애플리케이션 종료 시 실행되는 작업"""
    logger.info("Shutting down Churn Prediction API")