MLOps: 실험에서 프로덕션까지
모든 데이터 과학자는 이 순간을 경험했습니다. 모델은 Jupyter 노트북에서 완벽하게 작동합니다. 측정항목이 훌륭해서 데모 중에 팀에서 박수를 보냈습니다. 그러면 다음과 같은 치명적인 질문이 나옵니다. "이것을 언제 생산에 투입합니까?". 그리고 침묵이 시작됩니다. 업계 추정에 따르면, 기계 학습 프로젝트의 최대 85%는 프로덕션 환경에 도달하지 않습니다. 모델이 작동하지 않아서가 아니라 인프라, 프로세스 및 일하게 만드는 규율 안정적이고 지속적으로.
MLOps (Machine Learning Operations)은 이러한 격차를 메우기 위해 정확하게 만들어졌습니다. 이는 단일 기술이 아니라, 이를 가능하게 하는 일련의 관행, 도구 및 문화입니다. 그들은 고립된 실험을 강력한 프로덕션 ML 시스템으로 전환합니다. 이 기사에서 우리는 볼 것입니다 MLOps가 무엇인지, 왜 필수가 되었는지, 어떻게 구체적으로 적용해야 하는지, 제한된 예산으로도.
무엇을 배울 것인가
- 대부분의 ML 프로젝트가 프로덕션에 적용되지 않는 이유와 MLOps가 문제를 해결하는 방법
- DevOps와 MLOps의 주요 차이점
- Google 모델에 따른 MLOps 성숙도의 3가지 수준
- 프로덕션에서 ML 모델의 전체 수명 주기
- MLflow를 사용하여 실험 추적을 수행하는 방법
- FastAPI 및 Docker를 사용하여 모델을 제공하는 방법
- 연간 5,000 EUR 미만으로 시작할 수 있는 오픈 소스 스택
MLOps란 무엇이며 왜 필요한가요?
MLOps 및 원칙 적용 데브옵스 머신러닝 수명주기에 DevOps가 기존 소프트웨어의 개발과 운영을 통합한 것처럼 MLOps도 통합합니다. ML 시스템을 위한 데이터 과학, 엔지니어링 및 운영. 목표는 전자를 자동화하는 것입니다. 데이터 준비부터 교육, 검증까지 모든 단계를 재현 가능하게 만듭니다. 배포, 모니터링부터 재교육까지.
DevOps와 MLOps: 주요 차이점
소프트웨어 업계에서 온 사람들은 동일한 DevOps 방식을 적용하는 것만으로도 충분하다고 생각할 수 있습니다. ML 모델에. 실제로 MLOps를 그 자체로 하나의 분야로 만드는 근본적인 차이점이 있습니다.
| 나는 기다린다 | 데브옵스 | MLOps |
|---|---|---|
| 인공물 | 소스 코드 | 코드 + 데이터 + 모델 |
| 버전 관리 | 코드에 대한 Git | 데이터 및 모델용 Git + DVC |
| 테스트 | 단위 테스트, 통합 테스트 | 데이터 검증, 모델 검증, A/B 테스트 |
| CI/CD | 코드 빌드, 테스트, 배포 | 모델 학습, 검증, 배포 |
| 모니터링 | 지연 시간, 오류, 가동 시간 | 데이터 드리프트, 개념 드리프트, 모델 성능 |
| 하락 | 명시적인 버그 | 시간이 지남에 따라 자동 성능 저하 |
| 재현성 | 동일한 코드 = 동일한 출력 | 동일한 코드 + 동일한 데이터 + 동일한 시드 = 동일한 출력 |
가장 결정적인 차이점은 조용한 저하. 소프트웨어 서비스 기존 방식은 작동하거나 작동하지 않습니다. 버그가 있으면 오류가 발생합니다. ML 모델은 다음을 수행할 수 있습니다. 기술적 오류 없이 계속해서 예측을 반환하지만 점진적인 정확성을 가지고 있습니다. 훈련에 비해 입력 데이터가 변경되었기 때문에 더 나쁩니다. 모니터링하지 않고 구체적으로, 사용자가 불평을 시작할 때까지 아무도 이를 알아차리지 못합니다.
"데스 밸리" 문제 ML
Gartner는 생성 AI 프로젝트의 30%가 단계 이후에 폐기될 것으로 추정합니다. 데이터 품질 저하로 인해 2025년 말까지 개념 증명 완료 부적절한 위험, 비용 증가 또는 불분명한 비즈니스 가치. MLOps 주소 각 문제를 체계적으로 해결합니다.
MLOps 시장: 수치 및 동향
MLOps 시장은 놀라운 속도로 성장하고 있습니다. 업계 분석에 따르면, 2025년 MLOps 시장의 글로벌 가치는 20억~30억 달러로 추산됩니다. 2035년까지 그 규모가 250억~560억 달러에 이를 것으로 예상됩니다. 소스에 따라 29%에서 42% 사이의 CAGR(연간 복합 성장률)을 보입니다.
이 수치는 구체적인 현실을 반영합니다. 기업은 막대한 투자를 하고 있습니다. ML 모델을 프로덕션에 도입합니다. 시장 추정에 따르면 70% 이상이 북미의 대기업은 프로덕션 환경에서 AI 워크로드를 실행하고 있으며 55% 이상이 통합 자동 모델 모니터링 시스템. 그러나 거의 3분의 2에 달하는 조직은 여전히 파일럿 단계에 머물러 있어 AI를 확장할 수 없습니다. 기업 수준에서.
MLOps 성숙도의 3가지 수준
Google은 표준이 된 3계층 MLOps 성숙도 모델을 정의했습니다. 사실상 해당 부문. 각 수준은 자동화 수준의 증가를 나타내며 ML 수명주기의 안정성.
수준 0: 수동 프로세스
레벨 0에서는 각 단계가 수동입니다. 데이터 과학자는 노트북과 기차에서 일합니다. 모델을 로컬에서 파일로 내보낸 후 엔지니어링 팀에 전달합니다. API로 래핑됩니다. 자동화도, 모니터링도, 자동 재교육도 없습니다.
| 특성 | 레벨 0 |
|---|---|
| 훈련 | 설명서, 수첩에 |
| 배포 | 수동, .pkl 또는 .h5 파일 전달 |
| 모니터링 | 없음 또는 수동 |
| 재교육 | 명시적인 요청이 있는 경우에만 |
| 재현성 | 가난하거나 결석 |
이 수준은 ML을 사용 사례에 적용하기 시작하는 조직에서 일반적입니다. 이는 모델이 거의 업데이트되지 않고 데이터가 거의 변경되지 않는 경우 충분할 수 있습니다. 하지만 확장되지는 않습니다.
레벨 1: ML 파이프라인 자동화
레벨 1에서는 교육이 다음을 통해 자동화됩니다. ML 파이프라인. 아니요 예 더 이상 단일 모델이 아니라 이를 생성하는 전체 파이프라인을 배포합니다. 이를 통해 지속적인 훈련: 새 데이터가 도착하면 파이프라인이 다시 학습됩니다. 자동으로 모델이 됩니다.
| 특성 | 레벨 1 |
|---|---|
| 훈련 | 파이프라인을 통해 자동 |
| 배포 | 자동화된 파이프라인 |
| 모니터링 | 모델 성능 + 트리거 재훈련 |
| 재교육 | 새로운 데이터 또는 성능 저하 시 자동 |
| 재현성 | 좋음(파이프라인 버전이 지정됨) |
데이터가 자주 변경되지만 ML 접근 방식이 남아 있는 경우 수준 1이면 충분합니다. 안정. 파이프라인은 동일하지만 새로운 데이터를 사용하여 주기적으로 다시 실행됩니다.
레벨 2: 기계 학습을 위한 CI/CD
레벨 2에서는 완전한 시스템이 추가됩니다. ML 특정 CI/CD. 데이터뿐만 아니라 파이프라인 코드, 기능, 하이퍼파라미터도 변경됩니다. 모델의 아키텍처. 각 변경 사항은 자동 테스트, 검증 및 통제된 배포.
| 특성 | 레벨 2 |
|---|---|
| 훈련 | 파이프라인 자체의 자동 + CI/CD |
| 배포 | 블루/그린, 카나리아, A/B 테스트 |
| 모니터링 | 포괄적: 데이터 드리프트, 개념 드리프트, 성능, 대기 시간 |
| 재교육 | 검증 및 롤백을 통한 자동 |
| 재현성 | 완료(코드 + 데이터 + 환경 버전 지정) |
성숙한 조직을 위한 레벨 2 및 목표 달성. 투자가 필요하다 인프라와 문화 측면에서 중요하지만 수십, 수백 개를 관리할 수 있는 유일한 방법입니다. 지속 가능한 생산 모델.
MLOps 수명 주기
프로덕션에서 ML 모델의 수명 주기는 반복적인 프로세스입니다. 여섯 가지 주요 단계. 전통적인 소프트웨어 개발과 달리 이 주기는 끝이 없습니다. 생산 중인 모델에는 지속적인 유지 관리가 필요합니다.
+----------+ +---------+ +----------+
| DATA |---->| TRAIN |---->| EVALUATE |
| Collect | | Feature | | Validate |
| Clean | | Train | | Compare |
| Version | | Tune | | Approve |
+----------+ +---------+ +----------+
^ |
| v
+----------+ +---------+ +----------+
| RETRAIN |<----| MONITOR |<----| DEPLOY |
| Trigger | | Drift | | Stage |
| Schedule | | Metrics | | Canary |
| Auto | | Alert | | Release |
+----------+ +---------+ +----------+
1. 날짜: 수집, 정리 및 버전 관리
모든 것은 데이터에서 시작됩니다. 이 단계에서는 원시 데이터가 수집되고 정리됩니다. (결측값, 이상값, 중복 처리) 유용한 기능으로 변환되어 그들은 스스로 버전을 만듭니다. 데이터 버전 관리는 기본입니다. 모델을 재현하는 것입니다. 정확히 알아야 해 어떤 데이터 훈련용으로 사용되었습니다. 다음과 같은 도구 DVC (데이터 버전 관리) 버전 관리 허용 Git과 유사한 대규모 데이터 세트.
2. 훈련: 특성 엔지니어링 및 훈련
데이터가 준비되면 기능을 구축하고 알고리즘을 선택하고 훈련합니다. 모델. 각 실험(하이퍼파라미터, 기능, 아키텍처의 조합) 관련 매개변수 및 지표로 추적됩니다. 다음과 같은 도구 MLflow 이 프로세스를 체계적이고 재현 가능하게 만듭니다.
3. 평가: 검증 및 비교
훈련된 모델은 사전 정의된 지표(정확도, F1 점수, RMSE, AUC)를 현재 생산 중인 버전과 비교했습니다. 새로운 모델이라면 최소 기준을 초과하지 않거나 이전 기준보다 향상되지 않으면 승진되지 않습니다.
4. 배포: 스테이징, Canary 및 릴리스
승인된 모델은 점진적인 환경을 거칩니다: 테스트를 위한 준비 통합, 제한된 실제 트래픽으로 검증을 위한 카나리아, 최종적으로 프로덕션 완료. 다음과 같은 전략 블루/그린 배포 e 카나리아 릴리스 위험을 최소화하십시오.
5. 모니터링: 드리프트, 지표 및 경고
생산 과정에서 모델은 지속적으로 모니터링됩니다. 측정항목이 추적됩니다. 기술(지연 시간, 처리량, 오류) 및 ML 지표(실제 데이터의 정확성, 예측 분포, 데이터 드리프트). 측정항목이 다음과 같을 때 경고가 트리거됩니다. 그들은 임계값 아래로 떨어집니다.
6. 재교육: 트리거 및 자동화
모니터링을 통해 성능 저하가 감지되면 재훈련이 활성화됩니다. 이것은 예약(예: 매주), 트리거 기반(예: 90% 미만의 정확도) 또는 수동. 새 모델은 평가 및 배포 단계를 다시 거칩니다.
오픈 소스 MLOps 스택
MLOps의 장점 중 하나는 이를 포괄하는 성숙한 오픈 소스 생태계가 있다는 것입니다. 수명주기의 각 단계. 값비싼 엔터프라이즈 플랫폼을 구입할 필요가 없습니다 시작하려면 올바른 도구를 사용하여 완전한 MLOps 파이프라인을 구축할 수 있습니다.
| 단계 | 기구 | 기능 |
|---|---|---|
| 데이터 버전 관리 | DVC | Git과 통합된 데이터 세트 및 모델 버전 관리 |
| 실험 추적 | MLflow | 각 실험에 대한 매개변수, 측정항목, 아티팩트 로깅 |
| 모델 레지스트리 | MLflow 모델 레지스트리 | 모델 버전 관리 및 승격(스테이징/프로덕션) |
| 파이프라인 오케스트레이션 | 지사 / 공기 흐름 | 워크플로 조정, 예약, 재시도 |
| 모델 제공 | FastAPI + 도커 | 예측 제공을 위한 REST API, 컨테이너화 |
| 컨테이너화 | 도커 + K8 | 재현 가능한 환경, 수평적 확장성 |
| 모니터링 | 프로메테우스 + 그라파나 | 측정항목, 대시보드, 알림 |
| 데이터 검증 | 큰 기대 | 자동 데이터 품질 테스트 |
노트북에서 파이프라인으로: 실제 사례
모든 ML 팀이 직면하는 가장 일반적인 단계인 코드 변환을 살펴보겠습니다. 모듈식이며 재현 가능한 파이프라인으로 Jupyter 노트북에 작성되었습니다. 한 번 해보자 분류의 실제 예를 살펴보고 이를 재구성하는 방법을 살펴보겠습니다.
이전: 모놀리식 노트북
다음은 모든 항목이 분리되지 않고 단일 파일에 있는 일반적인 노트북입니다. 로깅이나 버전 관리 없이 책임을 집니다.
# Cella 1: Tutto nello stesso notebook
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, f1_score
import pickle
# Caricamento dati
df = pd.read_csv("data/customers.csv")
# Feature engineering inline
df["age_group"] = pd.cut(df["age"], bins=[0, 25, 45, 65, 100],
labels=["young", "adult", "senior", "elderly"])
df["total_spend"] = df["orders"] * df["avg_order_value"]
# Split
X = df[["age", "total_spend", "visits", "days_since_last"]]
y = df["churned"]
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)
# Training - parametri hardcoded
model = RandomForestClassifier(n_estimators=100, max_depth=10)
model.fit(X_train, y_train)
# Valutazione - print a schermo
y_pred = model.predict(X_test)
print(f"Accuracy: {accuracy_score(y_test, y_pred)}")
print(f"F1: {f1_score(y_test, y_pred)}")
# Salvataggio - pickle senza versioning
with open("model.pkl", "wb") as f:
pickle.dump(model, f)
print("Modello salvato!")
모놀리식 노트북 문제
- 재생할 수 없음: 시드 없음, 데이터 버전 관리 없음
- 추적되지 않음: 매개변수 및 측정항목은 노트북 출력에만 존재합니다.
- 테스트할 수 없음: 테스트할 격리된 기능이 없습니다.
- 배포할 수 없음: 피클은 API가 아닙니다
- 유지 관리할 수 없음: 기능을 수정하려면 모든 것을 다시 실행해야 합니다.
이후: 모듈식 파이프라인
우리는 코드를 각각 특정 책임이 있는 별도의 모듈로 재구성합니다. 모든 기능을 테스트할 수 있고, 모든 매개변수를 구성할 수 있으며, 모든 지표를 추적할 수 있습니다.
"""Modulo per la preparazione e trasformazione dei dati."""
import pandas as pd
from pathlib import Path
from typing import Tuple
def load_data(path: str) -> pd.DataFrame:
"""Carica il dataset dal path specificato."""
filepath = Path(path)
if not filepath.exists():
raise FileNotFoundError(f"Dataset non trovato: {path}")
return pd.read_csv(filepath)
def create_features(df: pd.DataFrame) -> pd.DataFrame:
"""Crea le feature derivate per il modello."""
result = df.copy()
result["age_group"] = pd.cut(
result["age"],
bins=[0, 25, 45, 65, 100],
labels=["young", "adult", "senior", "elderly"]
)
result["total_spend"] = result["orders"] * result["avg_order_value"]
return result
def split_data(
df: pd.DataFrame,
target_col: str = "churned",
feature_cols: list = None,
test_size: float = 0.2,
random_state: int = 42
) -> Tuple[pd.DataFrame, pd.DataFrame, pd.Series, pd.Series]:
"""Split dei dati in train/test con seed fisso per riproducibilità."""
from sklearn.model_selection import train_test_split
if feature_cols is None:
feature_cols = ["age", "total_spend", "visits", "days_since_last"]
X = df[feature_cols]
y = df[target_col]
return train_test_split(X, y, test_size=test_size, random_state=random_state)
"""Modulo per il training e la valutazione del modello."""
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, f1_score, precision_score, recall_score
from typing import Dict, Any
import pandas as pd
def train_model(
X_train: pd.DataFrame,
y_train: pd.Series,
n_estimators: int = 100,
max_depth: int = 10,
random_state: int = 42
) -> RandomForestClassifier:
"""Addestra un RandomForestClassifier con parametri configurabili."""
model = RandomForestClassifier(
n_estimators=n_estimators,
max_depth=max_depth,
random_state=random_state
)
model.fit(X_train, y_train)
return model
def evaluate_model(
model: RandomForestClassifier,
X_test: pd.DataFrame,
y_test: pd.Series
) -> Dict[str, float]:
"""Valuta il modello e restituisce un dizionario di metriche."""
y_pred = model.predict(X_test)
return {
"accuracy": accuracy_score(y_test, y_pred),
"f1_score": f1_score(y_test, y_pred),
"precision": precision_score(y_test, y_pred),
"recall": recall_score(y_test, y_pred),
}
"""Pipeline principale che orchestra tutte le fasi."""
from src.data.preprocessing import load_data, create_features, split_data
from src.models.trainer import train_model, evaluate_model
import yaml
from pathlib import Path
def run_pipeline(config_path: str = "config.yaml") -> None:
"""Esegue l'intera pipeline ML con configurazione esterna."""
# 1. Carica configurazione
with open(config_path) as f:
config = yaml.safe_load(f)
# 2. Data preparation
print("[1/4] Caricamento dati...")
df = load_data(config["data"]["path"])
df = create_features(df)
# 3. Split
print("[2/4] Split train/test...")
X_train, X_test, y_train, y_test = split_data(
df,
test_size=config["data"]["test_size"],
random_state=config["data"]["random_state"]
)
# 4. Training
print("[3/4] Training modello...")
model = train_model(
X_train, y_train,
n_estimators=config["model"]["n_estimators"],
max_depth=config["model"]["max_depth"],
random_state=config["model"]["random_state"]
)
# 5. Evaluation
print("[4/4] Valutazione...")
metrics = evaluate_model(model, X_test, y_test)
for name, value in metrics.items():
print(f" {name}: {value:.4f}")
if __name__ == "__main__":
run_pipeline()
# config.yaml - Tutti i parametri in un unico file
data:
path: "data/customers.csv"
test_size: 0.2
random_state: 42
feature_cols:
- age
- total_spend
- visits
- days_since_last
model:
algorithm: "random_forest"
n_estimators: 100
max_depth: 10
random_state: 42
evaluation:
metrics:
- accuracy
- f1_score
- precision
- recall
min_accuracy: 0.85
모듈식 파이프라인의 장점
- 재생할 수 있는: 고정 시드, 아웃소싱 구성, 버전 관리된 데이터
- 테스트 가능: 각 기능은 격리되어 있으며 전용 단위 테스트를 가질 수 있습니다.
- 유지 관리 가능: 기능을 수정해도 훈련에는 영향을 미치지 않으며 그 반대의 경우도 마찬가지입니다.
- 구성 가능: 코드를 건드리지 않고 하이퍼파라미터 변경
- 자동화 가능: 파이프라인은 CI/CD로 실행될 수 있습니다.
MLflow를 사용한 실험 추적
하이퍼파라미터를 변경한 후 어떤 조합이 있었는지 기억하지 못하는 경우가 몇 번이나 있습니까? 최선의 결과가 나왔나요? 그만큼'실험적 추적 이 문제를 해결합니다 각 실험의 매개변수, 지표 및 아티팩트를 자동으로 기록합니다.
MLflow 실험 추적을 위한 가장 인기 있는 오픈 소스 도구입니다. 실험을 보고 비교할 수 있는 웹 UI가 있는 서버, Python API를 제공합니다. 모델의 수명주기를 관리하기 위한 로깅 및 모델 레지스트리.
설정 및 첫 번째 실험
# Installazione
pip install mlflow
# Avvio del tracking server locale
mlflow server --host 127.0.0.1 --port 5000
"""Pipeline ML con experiment tracking via MLflow."""
import mlflow
import mlflow.sklearn
from src.data.preprocessing import load_data, create_features, split_data
from src.models.trainer import train_model, evaluate_model
def run_tracked_pipeline(config: dict) -> None:
"""Esegue la pipeline tracciando tutto con MLflow."""
# Imposta il tracking URI (server locale o remoto)
mlflow.set_tracking_uri("http://127.0.0.1:5000")
mlflow.set_experiment("churn-prediction")
with mlflow.start_run(run_name="rf-baseline") as run:
# Log dei parametri
mlflow.log_param("algorithm", "RandomForest")
mlflow.log_param("n_estimators", config["model"]["n_estimators"])
mlflow.log_param("max_depth", config["model"]["max_depth"])
mlflow.log_param("test_size", config["data"]["test_size"])
mlflow.log_param("random_state", config["data"]["random_state"])
# Data preparation
df = load_data(config["data"]["path"])
df = create_features(df)
X_train, X_test, y_train, y_test = split_data(
df,
test_size=config["data"]["test_size"],
random_state=config["data"]["random_state"]
)
# Log dimensioni dataset
mlflow.log_param("train_samples", len(X_train))
mlflow.log_param("test_samples", len(X_test))
mlflow.log_param("n_features", X_train.shape[1])
# Training
model = train_model(
X_train, y_train,
n_estimators=config["model"]["n_estimators"],
max_depth=config["model"]["max_depth"]
)
# Evaluation
metrics = evaluate_model(model, X_test, y_test)
# Log delle metriche
for name, value in metrics.items():
mlflow.log_metric(name, value)
# Log del modello come artefatto
mlflow.sklearn.log_model(
model,
artifact_path="model",
registered_model_name="churn-classifier"
)
# Log della configurazione come artefatto
mlflow.log_artifact("config.yaml")
print(f"Run ID: {run.info.run_id}")
print(f"Metriche: {metrics}")
여러번의 실험을 거친 후, http://127.0.0.1:5000 브라우저에서.
MLflow UI는 모든 실험이 포함된 표를 표시하므로 비교할 수 있습니다.
측정항목을 보고, 실적별로 정렬하고 측정항목과 측정항목 그래프를 확인하세요.
모델 레지스트리: 모델 버전 관리
코드의 버전이 Git으로 관리되는 것처럼 ML 모델도 버전이 관리되어야 합니다. 와 모델 레지스트리. MLflow Model Registry는 중앙 집중식 시스템을 제공합니다. 모델의 라이프사이클을 3단계로 관리합니다.
| 경기장 | 설명 | 누가 그것을 사용합니까? |
|---|---|---|
| 없음/스테이징 | 테스트 및 검증 중인 모델 | 데이터 과학자, QA |
| 생산 | 승인된 모델, 실제 트래픽 제공 | API 제공, 최종 사용자 |
| 보관됨 | 모델이 폐기되고 감사를 위해 보관됨 | 규정 준수, 롤백 |
"""Gestione del ciclo di vita del modello con MLflow Model Registry."""
from mlflow.tracking import MlflowClient
client = MlflowClient("http://127.0.0.1:5000")
# Recupera l'ultima versione del modello in staging
latest_versions = client.get_latest_versions(
name="churn-classifier",
stages=["Staging"]
)
if latest_versions:
version = latest_versions[0].version
print(f"Modello in staging: v{version}")
# Promuovi a Production dopo validazione
client.transition_model_version_stage(
name="churn-classifier",
version=version,
stage="Production",
archive_existing_versions=True # Archivia versione precedente
)
print(f"Modello v{version} promosso a Production")
# Carica il modello in produzione per inference
import mlflow.pyfunc
model = mlflow.pyfunc.load_model("models:/churn-classifier/Production")
prediction = model.predict(new_data)
배포: FastAPI + Docker
프로덕션 중인 ML 모델은 일반적으로 다음과 같이 노출됩니다. REST API. FastAPI Python을 위한 이상적인 선택: 그리고 빠릅니다(ASGI 기반). 자동 문서(OpenAPI/Swagger)를 생성하고 검증이 뛰어납니다. Pydantic을 통한 데이터. 컨테이너화 도커, 우리는 얻는다 어디에나 배치할 수 있는 아티팩트.
"""API REST per servire predizioni del modello ML."""
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
import mlflow.pyfunc
import pandas as pd
import logging
from typing import List
# Configurazione logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = FastAPI(
title="Churn Prediction API",
description="API per predizioni di churn basata su ML",
version="1.0.0"
)
class PredictionRequest(BaseModel):
"""Schema della richiesta di predizione."""
age: int = Field(..., ge=0, le=120, description="Eta del cliente")
total_spend: float = Field(..., ge=0, description="Spesa totale")
visits: int = Field(..., ge=0, description="Numero visite")
days_since_last: int = Field(..., ge=0, description="Giorni dall'ultima visita")
class PredictionResponse(BaseModel):
"""Schema della risposta di predizione."""
prediction: int
probability: float
model_version: str
# Caricamento modello all'avvio
MODEL_NAME = "churn-classifier"
MODEL_STAGE = "Production"
model = None
model_version = "unknown"
@app.on_event("startup")
async def load_model():
"""Carica il modello MLflow all'avvio del server."""
global model, model_version
try:
model_uri = f"models:/{MODEL_NAME}/{MODEL_STAGE}"
model = mlflow.pyfunc.load_model(model_uri)
model_version = model.metadata.run_id[:8]
logger.info(f"Modello caricato: {MODEL_NAME} ({model_version})")
except Exception as e:
logger.error(f"Errore caricamento modello: {e}")
raise
@app.get("/health")
async def health_check():
"""Endpoint di health check."""
return {"status": "healthy", "model_loaded": model is not None}
@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
"""Genera una predizione di churn per un cliente."""
if model is None:
raise HTTPException(status_code=503, detail="Modello non caricato")
try:
input_data = pd.DataFrame([request.model_dump()])
prediction = model.predict(input_data)
probability = float(prediction[0]) if hasattr(prediction[0], '__float__') else 0.0
return PredictionResponse(
prediction=int(prediction[0]),
probability=probability,
model_version=model_version
)
except Exception as e:
logger.error(f"Errore predizione: {e}")
raise HTTPException(status_code=500, detail="Errore nella predizione")
@app.post("/predict/batch", response_model=List[PredictionResponse])
async def predict_batch(requests: List[PredictionRequest]):
"""Genera predizioni batch per più clienti."""
if model is None:
raise HTTPException(status_code=503, detail="Modello non caricato")
input_data = pd.DataFrame([r.model_dump() for r in requests])
predictions = model.predict(input_data)
return [
PredictionResponse(
prediction=int(p),
probability=float(p),
model_version=model_version
)
for p in predictions
]
# Dockerfile per il serving del modello ML
FROM python:3.11-slim
WORKDIR /app
# Dipendenze di sistema
RUN apt-get update && apt-get install -y --no-install-recommends \
gcc \
&& rm -rf /var/lib/apt/lists/*
# Dipendenze Python
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Codice applicazione
COPY src/serving/ ./serving/
COPY config.yaml .
# Porta del servizio
EXPOSE 8000
# Healthcheck
HEALTHCHECK --interval=30s --timeout=5s --retries=3 \
CMD curl -f http://localhost:8000/health || exit 1
# Avvio con uvicorn
CMD ["uvicorn", "serving.app:app", "--host", "0.0.0.0", "--port", "8000"]
# Build dell'immagine
docker build -t churn-api:v1.0.0 .
# Avvio del container
docker run -d \
--name churn-api \
-p 8000:8000 \
-e MLFLOW_TRACKING_URI=http://mlflow-server:5000 \
churn-api:v1.0.0
# Test dell'API
curl -X POST http://localhost:8000/predict \
-H "Content-Type: application/json" \
-d '{"age": 35, "total_spend": 1250.50, "visits": 12, "days_since_last": 45}'
생산 중 모니터링
배포는 작업의 끝이 아니라 새로운 중요한 단계의 시작입니다. 모니터링. 생산 중인 모델은 시간이 지남에 따라 품질이 저하됩니다. 변경 사항과 그에 따른 데이터. 모니터링은 세 가지 주요 영역을 다루어야 합니다.
추적할 측정항목
| 범주 | 측정항목 | 기구 |
|---|---|---|
| 하부 구조 | 지연 시간(p50, p95, p99), 처리량, HTTP 오류, CPU/RAM | 프로메테우스 + 그라파나 |
| 모델 | 정확도, F1 점수, 예측 분포, 신뢰도 | MLflow + 사용자 지정 측정항목 |
| 데이터 | 데이터 드리프트, 특징 드리프트, 결측값, 입력 분포 | 분명히 AI / 큰 기대 |
데이터 드리프트와 개념 드리프트
두 가지 유형의 모델 저하를 구별하는 것이 중요합니다.
- 날짜 드리프트: 입력 데이터의 분포는 다음과 같이 변경됩니다. 트레이닝 세트. 예: 25~45세 고객을 대상으로 훈련된 모델이 수신을 시작합니다. 60세 이상 고객을 위한 요청.
- 컨셉 드리프트: 입력과 출력의 관계가 변경됩니다. 예: 이후 팬데믹 상황에서 고객 이탈 패턴은 완전히 다르지만, 입력 시에는 동일한 분포를 갖습니다.
"""Rilevamento data drift con test statistici."""
import numpy as np
from scipy import stats
from typing import Dict, Tuple
def detect_drift(
reference_data: np.ndarray,
production_data: np.ndarray,
feature_names: list,
threshold: float = 0.05
) -> Dict[str, Dict]:
"""
Rileva data drift confrontando distribuzioni con il test KS.
Args:
reference_data: dati di training (riferimento)
production_data: dati di produzione (attuali)
feature_names: nomi delle feature
threshold: soglia p-value per il drift (default 0.05)
Returns:
Report di drift per ogni feature
"""
drift_report = {}
for i, feature in enumerate(feature_names):
ref_values = reference_data[:, i]
prod_values = production_data[:, i]
# Test Kolmogorov-Smirnov
ks_stat, p_value = stats.ks_2samp(ref_values, prod_values)
drift_detected = p_value < threshold
drift_report[feature] = {
"ks_statistic": round(ks_stat, 4),
"p_value": round(p_value, 4),
"drift_detected": drift_detected,
"ref_mean": round(float(np.mean(ref_values)), 4),
"prod_mean": round(float(np.mean(prod_values)), 4),
}
if drift_detected:
print(f"DRIFT RILEVATO su '{feature}': "
f"KS={ks_stat:.4f}, p={p_value:.4f}")
return drift_report
재교육을 재활성화해야 하는 시기
모든 드리프트에 즉각적인 재교육이 필요한 것은 아닙니다. 명확한 임계값 정의: 데이터 드리프트 중요한 기능에 대한 정확도가 5% 이상 떨어지거나 예측 분포가 상당히 불균형합니다. 이로 인해 발생할 수 있는 과도한 재교육을 피하세요. 불안정하다.
연간 5,000 EUR 미만의 예산으로 시작하는 방법
MLOps가 반드시 수십만 유로의 비용이 드는 엔터프라이즈 플랫폼을 의미하는 것은 아닙니다. 이탈리아 SME 또는 소규모 팀의 경우 MLOps 인프라 구축이 가능합니다. 오픈 소스 도구와 최소 비용으로 효과적입니다.
중소기업을 위해 제안된 스택
| 요소 | 해결책 | 연간 비용 |
|---|---|---|
| 암호 | GitHub 무료 / GitLab CE | 0유로 |
| 데이터 버전 관리 | DVC + Google Cloud Storage(5GB 무료) | 0~50유로 |
| 실험 추적 | 저렴한 VM의 MLflow | 200~500유로 |
| 훈련 | Google Colab Pro/스팟 VM | 120~600유로 |
| 피복재 | VM의 FastAPI(2 vCPU, 4GB RAM) | 300~800유로 |
| 모니터링 | Prometheus + Grafana(자체 호스팅) | 0 EUR(동일한 VM에서) |
| CI/CD | GitHub Actions(월 2,000분 무료) | 0유로 |
| 컨테이너 레지스트리 | GitHub 컨테이너 레지스트리 | 0유로 |
예상 총액: 620 - 1,950 EUR/년, EUR 5,000 기준점보다 훨씬 낮습니다. 이 스택은 중간 정도의 트래픽 볼륨으로 프로덕션에서 최대 5~10개의 모델을 지원합니다. (하루에 수천 개의 예측).
비용 절감을 위한 팁
- 스팟/선점형 VM: 긴급하지 않은 교육의 경우 최대 70% 절감
- 자동 확장: 요청이 없으면 0으로 조정
- 모델 압축: 더 작은 모델 = 더 적은 제공 리소스
- 일괄 추론: 실시간 예측이 필요하지 않은 경우 야간 배치를 사용하세요.
- 다중 테넌트: 모든 프로젝트를 위한 단일 MLflow/Grafana 인프라
MLOps 프로젝트의 구조
즉시 사용할 수 있는 것으로 끝내기 위한 폴더 구조는 다음과 같습니다. MLOps 프로젝트에 권장됩니다. 이 조직은 분리를 따릅니다. 책임과 자동화, 테스트 및 협업의 용이성.
churn-prediction/
data/
raw/ # Dati grezzi (versionati con DVC)
processed/ # Dati trasformati
data.dvc # File di tracking DVC
src/
data/
preprocessing.py # Pulizia e feature engineering
validation.py # Validazione qualità dati
models/
trainer.py # Logica di training
evaluator.py # Valutazione e metriche
serving/
app.py # FastAPI application
schemas.py # Pydantic schemas
monitoring/
drift_detector.py # Rilevamento drift
metrics.py # Metriche custom
pipeline.py # Orchestrazione pipeline
tests/
test_preprocessing.py
test_trainer.py
test_api.py
config.yaml # Configurazione pipeline
Dockerfile # Container per serving
docker-compose.yaml # Stack locale completo
requirements.txt # Dipendenze Python
.dvc/ # Configurazione DVC
.github/
workflows/
train.yaml # CI/CD per training
deploy.yaml # CI/CD per deployment
mlflow/ # Artefatti MLflow (locale)
README.md
결론 및 다음 단계
MLOps는 대규모 기술 회사에만 국한된 사치품이 아닙니다. 그리고 꼭 필요한 안정적이고 지속 가능한 방식으로 ML 모델을 프로덕션에 도입하려는 사람. 이 기사에서 우리는 문제를 이해하는 것부터 기본 사항을 다뤘습니다. (ML 프로젝트가 실패하는 이유) 구체적인 솔루션(모듈식 파이프라인, 실험) 추적, 모델 레지스트리, 컨테이너화된 제공 및 모니터링).
핵심은 점진적으로 시작하는 것입니다. 모델의 레벨 2에 도달할 필요가 없습니다. 첫날부터 Google의 성숙도를 확인하세요. 모범 사례를 활용하여 레벨 0부터 시작하세요.
- 곧: 노트북 코드를 모듈로 분리하세요. config.yaml을 사용하세요.
- 1주차: 실험을 추적하려면 MLflow를 추가하세요.
- 2주차: FastAPI + Docker를 사용하여 모델을 컨테이너화합니다.
- 1개월차: GitHub Actions를 사용하여 CI/CD 파이프라인을 구현합니다.
- 2개월차: Prometheus 및 기본 알림을 사용하여 모니터링을 추가합니다.
- 3개월: 데이터 버전 관리를 위해 DVC를 구현합니다.
시리즈의 다음 기사에서는 각 구성 요소인 데이터 관리에 대해 더 자세히 살펴보겠습니다. DVC를 통한 ML 전용 CI/CD 파이프라인 생성, 고급 모니터링 Kubernetes에서 AI 및 확장 가능한 배포가 가능합니다. 각 항목은 실용적이며, 작업 코드 및 단계별 지침.
시리즈 로드맵
- 제2조: DVC - ML용 데이터 버전 관리
- 제3조: MLflow 심층 분석 - 고급 실험 추적
- 제4조: GitHub Actions를 사용한 기계 학습용 CI/CD
- 제5조: 프로덕션의 Feature Store 및 Feature Engineering
- 제6조: Kubernetes를 사용한 확장 가능한 모델 제공
- 제7조: 고급 모니터링: 데이터 드리프트와 AI
- 제8조: 거버넌스, 규정 준수 및 책임 있는 ML