🚀 Deploy: Uplift 모델 학습 파이프라인 및 예측 API 추가

.github/workflows/cd-production.yaml

@@ -7,6 +7,7 @@ on:
     paths:
       - 'services/model-api/**'
       - 'services/alert-bridge/**'
+      - 'k8s/**'
 
 jobs:
   deploy:
@@ -30,6 +31,9 @@ jobs:
       - name: model-api 배포 확인
         run: kubectl rollout status deployment/model-api -n model-api
 
+      - name: alert-bridge kubectl apply
+        run: kubectl apply -f k8s/alert-bridge/deployment.yaml
+
       - name: alert-bridge 파드 재시작
         run: kubectl rollout restart deployment/alert-bridge -n monitoring
 

.github/workflows/deploy-churn.yaml

@@ -0,0 +1,22 @@
+name: Deploy (churn)
+
+on:
+  repository_dispatch:
+    types:
+      - deploy-churn
+
+jobs:
+  deploy:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: kubectl 설정
+        run: |
+          mkdir -p $HOME/.kube
+          echo "${{ secrets.KUBECONFIG_PRODUCTION }}" | base64 -d > $HOME/.kube/config
+
+      - name: model-api 파드 재시작
+        run: kubectl rollout restart deployment/model-api -n model-api
+
+      - name: model-api 배포 확인
+        run: kubectl rollout status deployment/model-api -n model-api

.github/workflows/deploy-uplift.yaml

@@ -0,0 +1,22 @@
+name: Deploy (uplift)
+
+on:
+  repository_dispatch:
+    types:
+      - deploy-uplift
+
+jobs:
+  deploy:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: kubectl 설정
+        run: |
+          mkdir -p $HOME/.kube
+          echo "${{ secrets.KUBECONFIG_PRODUCTION }}" | base64 -d > $HOME/.kube/config
+
+      - name: model-api 파드 재시작
+        run: kubectl rollout restart deployment/model-api -n model-api
+
+      - name: model-api 배포 확인
+        run: kubectl rollout status deployment/model-api -n model-api

k8s/alert-bridge/deployment.yaml

@@ -19,12 +19,33 @@ spec:
           ports:
             - containerPort: 8080
           env:
-            - name: GITHUB_TOKEN
+            - name: SMTP_HOST
               valueFrom:
                 secretKeyRef:
-                  name: github-credentials
-                  key: token
-            - name: GITHUB_REPO_OWNER
-              value: simGPT
-            - name: GITHUB_REPO_NAME
-              value: sw-mlops
+                  name: smtp-credentials
+                  key: host
+            - name: SMTP_PORT
+              valueFrom:
+                secretKeyRef:
+                  name: smtp-credentials
+                  key: port
+            - name: SMTP_USER
+              valueFrom:
+                secretKeyRef:
+                  name: smtp-credentials
+                  key: user
+            - name: SMTP_PASSWORD
+              valueFrom:
+                secretKeyRef:
+                  name: smtp-credentials
+                  key: password
+            - name: ALERT_MAIL_FROM
+              valueFrom:
+                secretKeyRef:
+                  name: smtp-credentials
+                  key: mail-from
+            - name: ALERT_MAIL_TO
+              valueFrom:
+                secretKeyRef:
+                  name: smtp-credentials
+                  key: mail-to

services/alert-bridge/main.py

@@ -1,36 +1,57 @@
 import os
-import httpx
+import smtplib
+import ssl
+from datetime import datetime, timezone
+from email.mime.text import MIMEText
 from fastapi import FastAPI, Request
 
 app = FastAPI(title="Alert Bridge")
 
-GITHUB_TOKEN = os.getenv("GITHUB_TOKEN")
-GITHUB_REPO_OWNER = os.getenv("GITHUB_REPO_OWNER")
-GITHUB_REPO_NAME = os.getenv("GITHUB_REPO_NAME")
+SMTP_HOST = os.getenv("SMTP_HOST")
+SMTP_PORT = int(os.getenv("SMTP_PORT", "465"))
+SMTP_USER = os.getenv("SMTP_USER")
+SMTP_PASSWORD = os.getenv("SMTP_PASSWORD")
+ALERT_MAIL_FROM = os.getenv("ALERT_MAIL_FROM")
+ALERT_MAIL_TO = os.getenv("ALERT_MAIL_TO")
 
-# Alertmanager에서 알림을 받아서 GitHub Actions로 전달하는 역할을 하는 api 엔드포인트(Alertmanager가 요청을 보냄)
+# 알림 내용을 이메일로 작성하는 함수
+def build_email_body(alerts: list) -> str:
+    lines = [f"[sw-mlops 알림] {len(alerts)}개의 이상 감지 - {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S UTC')}\n"]
+    for a in alerts:
+        labels = a.get("labels", {})
+        annotations = a.get("annotations", {})
+        lines.append(f"• Alert   : {labels.get('alertname', '-')}")
+        lines.append(f"  Severity: {labels.get('severity', '-')}")
+        lines.append(f"  Summary : {annotations.get('summary', '-')}")
+        lines.append(f"  Detail  : {annotations.get('description', '-')}\n")
+    return "\n".join(lines)
+
+# 이메일 전송하는 함수
+def send_email(subject: str, body: str):
+    msg = MIMEText(body, "plain", "utf-8")
+    msg["Subject"] = subject
+    msg["From"] = ALERT_MAIL_FROM
+    msg["To"] = ALERT_MAIL_TO
+
+    with smtplib.SMTP(SMTP_HOST, SMTP_PORT) as server:
+        server.starttls(context=ssl.create_default_context())
+        server.login(SMTP_USER, SMTP_PASSWORD)
+        server.sendmail(ALERT_MAIL_FROM, ALERT_MAIL_TO, msg.as_string())
+
+
+# 알림 받아서 이메일 전송하는 api
 @app.post("/alert")
 async def receive_alert(request: Request):
     payload = await request.json()
 
-    # alertmanager에서 받은 알림 중에서 firing 상태인 알림만 필터링
     firing_alerts = [a for a in payload.get("alerts", []) if a["status"] == "firing"]
     if not firing_alerts:
-        return {"message": "no firing alerts"}
+        return {"message": "발생중인 알림이 없습니다."}
 
     alert_names = [a["labels"].get("alertname", "") for a in firing_alerts]
+    subject = f"[sw-mlops 알림] {', '.join(alert_names)}"
+    body = build_email_body(firing_alerts) # 알림 내용을 이메일 body로 작성
+
+    send_email(subject, body) # 이메일 전송
 
-    async with httpx.AsyncClient() as client:
-        response = await client.post(
-            f"https://api.github.com/repos/{GITHUB_REPO_OWNER}/{GITHUB_REPO_NAME}/dispatches",
-            headers={
-                "Authorization": f"Bearer {GITHUB_TOKEN}",
-                "Accept": "application/vnd.github+json",
-            },
-            json={
-                "event_type": "model-retrain",
-                "client_payload": {"alerts": alert_names},
-            },
-        )
-
-    return {"status": response.status_code, "alerts": alert_names}
+    return {"message": "이메일이 전송되었습니다.", "alerts": alert_names}

services/alert-bridge/requirements.txt

@@ -1,3 +1,2 @@
 fastapi
 uvicorn
-httpx

services/model-api/app/main.py

@@ -61,3 +61,27 @@ def churn_predict(request: RequestSchema):
             metadata=None,
             error=str(e),
         )
+
+
+# Uplift 예측 API
+@app.post("/uplift_predict", response_model=ResponseSchema)
+def uplift_predict(request: RequestSchema):
+    try:
+        service = get_service("uplift")
+        output = service.predict(request.data)
+
+        return ResponseSchema(
+            success=True,
+            model="uplift",
+            result=output["result"],
+            metadata=output["metadata"],
+            error=None,
+        )
+    except Exception as e:
+        return ResponseSchema(
+            success=False,
+            model="uplift",
+            result=None,
+            metadata=None,
+            error=str(e),
+        )

services/model-api/app/models/loader.py

@@ -2,6 +2,7 @@
 import mlflow.pytorch
 import mlflow.sklearn
 
+from app.models.uplift_model import TLearner  # noqa: F401 MLflow 역직렬화 시 필요
 
 _model_cache: dict = {}
 
@@ -11,7 +12,7 @@ def load_model(model_name: str):
     if cache_key in _model_cache:
         return _model_cache[cache_key]
 
-    mlflow_uri = os.getenv("MLFLOW_TRACKING_URI", "http://mlflow:5000") # 환경변수에서 mlflow tracking uri 가져오기, 없으면 기본값으로 http://mlflow:5000 사용
+    mlflow_uri = os.getenv("MLFLOW_TRACKING_URI", "http://mlflow:5000") #  deployment에서 mlflow tracking uri 가져오기, 없으면 기본값으로 http://mlflow:5000 사용
     mlflow.set_tracking_uri(mlflow_uri)
 
     model_uri = f"models:/{model_name}/latest"
@@ -27,7 +28,23 @@ def load_churn_model(model_name: str):
     if cache_key in _model_cache:
         return _model_cache[cache_key]
 
-    mlflow_uri = os.getenv("MLFLOW_TRACKING_URI", "http://mlflow:5000") # 환경변수에서 mlflow tracking uri 가져오기, 없으면 기본값으로 http://mlflow:5000 사용
+    mlflow_uri = os.getenv("MLFLOW_TRACKING_URI", "http://mlflow:5000") # deployment에서 mlflow tracking uri 가져오기, 없으면 기본값으로 http://mlflow:5000 사용
+    mlflow.set_tracking_uri(mlflow_uri)
+
+    model_uri = f"models:/{model_name}/latest"
+    model = mlflow.sklearn.load_model(model_uri)
+
+    _model_cache[cache_key] = model
+    return model
+
+
+# mlflow에서 uplift 모델 로드하는 함수
+def load_uplift_model(model_name: str):
+    cache_key = f"{model_name}_uplift"
+    if cache_key in _model_cache:
+        return _model_cache[cache_key]
+
+    mlflow_uri = os.getenv("MLFLOW_TRACKING_URI", "http://mlflow:5000")
     mlflow.set_tracking_uri(mlflow_uri)
 
     model_uri = f"models:/{model_name}/latest"

services/model-api/app/models/uplift_model.py

@@ -0,0 +1,31 @@
+import numpy as np
+from sklearn.base import BaseEstimator
+from sklearn.linear_model import LogisticRegression
+from sklearn.pipeline import Pipeline
+from sklearn.preprocessing import StandardScaler
+
+# 각 그룹을 로지스틱 회귀모델로 학습 후 예측값의 차이로 uplift 점수 계산하는 T-learner 모델
+class TLearner(BaseEstimator):
+    def __init__(self, C=1.0):
+        self.C = C
+        # treatment(사용자에게 프로모션을 제공한 그룹)과 control(프로모션을 제공하지 않은 그룹) 각각에 대해 모델을 학습
+        self.treatment_model = Pipeline([
+            ('scaler', StandardScaler()),
+            ('clf', LogisticRegression(C=C, random_state=42, max_iter=1000)),
+        ])
+        self.control_model = Pipeline([
+            ('scaler', StandardScaler()),
+            ('clf', LogisticRegression(C=C, random_state=42, max_iter=1000)),
+        ])
+    # fit 메서드에서는 treatment 그룹과 control 그룹을 나누어 각각의 모델을 학습
+    def fit(self, X, y, treatment):
+        treatment = np.array(treatment)
+        self.treatment_model.fit(X[treatment == 1], y[treatment == 1])
+        self.control_model.fit(X[treatment == 0], y[treatment == 0])
+        return self
+    # predict 메서드에서는 두 모델의 예측값의 차이로 uplift 점수를 계산하여 반환
+    # (마케팅을 하면 구매할 확률) - (마케팅을 안 하면 구매할 확률) = uplift 점수
+    def predict(self, X):
+        p_treatment = self.treatment_model.predict_proba(X)[:, 1]
+        p_control = self.control_model.predict_proba(X)[:, 1]
+        return p_treatment - p_control

services/model-api/app/services/service_factory.py

@@ -1,9 +1,10 @@
-from app.services import mnist_service, churn_service
+from app.services import mnist_service, churn_service, uplift_service
 
 
 _services = {
     "mnist": mnist_service,
     "churn": churn_service,
+    "uplift": uplift_service,
 }
 
 # 모델의 종류에 맞게 service 파일 불러오는 라우터

services/model-api/app/services/uplift_service.py

@@ -0,0 +1,43 @@
+import time
+from datetime import datetime, timezone
+
+from app.models.loader import load_uplift_model
+
+MODEL_NAME = "uplift"
+
+FEATURES = [
+    'account_age_months',
+    'avg_order_value',
+    'total_orders',
+    'days_since_last_purchase',
+    'discount_usage_rate',
+    'return_rate',
+    'browsing_frequency_per_week',
+    'cart_abandonment_rate',
+]
+
+
+def predict(data: dict) -> dict:
+    missing = [f for f in FEATURES if f not in data]
+    if missing:
+        raise ValueError(f"누락된 피처: {missing}")
+
+    model = load_uplift_model(MODEL_NAME) # 모델 로드
+
+    x = [[data[f] for f in FEATURES]]
+
+    start = time.time()
+    uplift_score = float(model.predict(x)[0])
+    duration = time.time() - start
+
+    return {
+        "result": {
+            "uplift_score": uplift_score,
+            "is_persuadable": uplift_score > 0, # 마케팅 액션이 효과가 있는 고객인지 여부
+        },
+        "metadata": {
+            "model": MODEL_NAME,
+            "inference_time_ms": round(duration * 1000, 3),
+            "timestamp": datetime.now(timezone.utc).isoformat(),
+        },
+    }