MLOps: od eksperymentu do produkcji
Każdy analityk danych przeżył ten moment: model sprawdza się doskonale w notatniku Jupyter, wskaźniki są doskonałe, zespół bije brawa podczas demonstracji. Potem pojawia się fatalne pytanie: „Kiedy wprowadzimy to do produkcji?”. I zaczyna się cisza. Według szacunków branży, aż 85% projektów uczenia maszynowego nigdy nie trafia do środowiska produkcyjnego. Nie dlatego, że modele nie działają, ale dlatego, że infrastruktura, proces i dyscyplinę, aby zmusić ich do pracy niezawodnie i stale.
MLOps (Machine Learning Operations) został stworzony właśnie po to, aby wypełnić tę lukę. Nie jest to pojedyncza technologia, ale zbiór praktyk, narzędzi i kultury przekształcają izolowane eksperymenty w solidne produkcyjne systemy ML. W tym artykule zobaczymy co oznacza MLOps, dlaczego stał się niezbędny i jak zacząć go konkretnie stosować, nawet przy ograniczonym budżecie.
Czego się nauczysz
- dlaczego większość projektów ML nie trafia do produkcji i jak MLOps rozwiązuje problem
- Kluczowe różnice pomiędzy DevOps i MLOps
- 3 poziomy dojrzałości MLOps według modelu Google
- Pełny cykl życia modelu ML w produkcji
- Jak wykonać eksperymentalne śledzenie za pomocą MLflow
- Jak obsłużyć model za pomocą FastAPI i Dockera
- Zestaw open source na początek za mniej niż 5000 EUR rocznie
Co to jest MLOps i dlaczego jest potrzebny
MLOps i stosowanie zasad DevOps do cyklu życia uczenia maszynowego. Tak jak DevOps zjednoczył rozwój i operacje tradycyjnego oprogramowania, tak MLOps jednoczy analityka danych, inżynieria i operacje dla systemów ML. Celem jest automatyzacja m.in zapewnić powtarzalność wszystkich faz: od przygotowania danych po szkolenie, od walidacji po wdrożenie, od monitorowania po przekwalifikowanie.
DevOps vs MLOps: kluczowe różnice
Osoby wywodzące się ze świata oprogramowania mogą pomyśleć, że wystarczy stosowanie tych samych praktyk DevOps do modeli ML. W rzeczywistości istnieją zasadnicze różnice, które czynią MLOps dyscypliną samą w sobie.
| Czekam | DevOps | MLOps |
|---|---|---|
| Artefakt | Kod źródłowy | Kod + Dane + Model |
| Wersjonowanie | Git dla kodu | Git + DVC dla danych i modeli |
| Testowanie | Testy jednostkowe, testy integracyjne | Walidacja danych, walidacja modelu, testy A/B |
| CI/CD | Kompiluj, testuj, wdrażaj kod | Trenuj, sprawdzaj, wdrażaj model |
| Monitorowanie | Opóźnienia, błędy, czas pracy | Dryf danych, dryf koncepcji, wydajność modelu |
| Degradacja | Wyraźne błędy | Cicha degradacja z upływem czasu |
| Powtarzalność | Ten sam kod = to samo wyjście | Ten sam kod + te same dane + to samo ziarno = to samo wyjście |
Najbardziej krytyczną różnicą jest cicha degradacja. Usługa oprogramowania tradycyjne działa lub nie działa: jeśli wystąpi błąd, zgłasza błąd. Model ML może w dalszym ciągu zwracaj prognozy bez błędów technicznych, ale z coraz większą dokładnością gorzej, ponieważ dane wejściowe uległy zmianie w porównaniu do treningu. Bez monitorowania specyficzne, nikt tego nie zauważa, dopóki użytkownicy nie zaczną narzekać.
Problem „Doliny Śmierci” ML
Gartner szacuje, że 30% projektów generatywnej AI zostanie porzuconych po tej fazie weryfikacji koncepcji do końca 2025 r. ze względu na słabą jakość danych, kontrole niewystarczające ryzyko, rosnące koszty lub niejasna wartość biznesowa. Adresy MLOps każdy z tych problemów systematycznie.
Rynek MLOps: Liczby i trendy
Rynek MLOps rośnie w imponującym tempie. Według analiz branżowych, globalną wartość rynku MLOps oszacowano na 2–3 miliardy dolarów w 2025 roku, z przewidywaniami, że do 2035 r. osiągnie ona wartość od 25 do 56 miliardów dolarów, z CAGR (złożona roczna stopa wzrostu) od 29% do 42% w zależności od źródeł.
Liczby te odzwierciedlają konkretną rzeczywistość: firmy inwestują masowo w celu wprowadzenia modeli ML do produkcji. Według szacunków rynkowych, ponad 70 proc Duże przedsiębiorstwa w Ameryce Północnej wykorzystują w produkcji obciążenia oparte na sztucznej inteligencji, a ponad 55% z nich tak robi zintegrowane systemy automatycznego monitorowania modeli. Jednak prawie dwie trzecie organizacje nadal utknęły w fazie pilotażowej i nie są w stanie skalować sztucznej inteligencji na poziomie przedsiębiorstwa.
3 poziomy dojrzałości MLOps
Google zdefiniowało 3-poziomowy model dojrzałości MLOps, który stał się standardem de facto sektora. Każdy poziom reprezentuje rosnący stopień automatyzacji i niezawodność w cyklu życia uczenia maszynowego.
Poziom 0: Proces ręczny
Na poziomie 0 każdy etap jest ręczny. Analityk danych pracuje w swoim notatniku, trenuje model lokalnie, eksportuje go jako plik i dostarcza zespołowi inżynieryjnemu, który zawija się w API. Nie ma automatyzacji, nie ma monitorowania, nie ma automatycznego przekwalifikowania.
| Charakterystyczny | Poziom 0 |
|---|---|
| Szkolenie | Podręcznik, w notatniku |
| Wdrożyć | Instrukcja, dostawa pliku .pkl lub .h5 |
| Monitorowanie | Brak lub ręczne |
| Przekwalifikowanie | Tylko na wyraźne żądanie |
| Powtarzalność | Słaby lub nieobecny |
Ten poziom jest powszechny w organizacjach rozpoczynających stosowanie ML w swoich przypadkach użycia. Może to być wystarczające, gdy modele są rzadko aktualizowane, a dane zmieniają się niewiele, ale to nie ma skali.
Poziom 1: Automatyzacja rurociągów ML
Na poziomie 1 szkolenie jest zautomatyzowane poprzez: Rurociągi ML. Nie tak nie wdraża już pojedynczego modelu, ale cały potok, który go tworzy. Pozwala to ciągłe szkolenie: Po nadejściu nowych danych potok zostaje ponownie przeszkolony model automatycznie.
| Charakterystyczny | Poziom 1 |
|---|---|
| Szkolenie | Automatycznie poprzez rurociąg |
| Wdrożyć | Zautomatyzowany rurociąg |
| Monitorowanie | Wydajność modelu + ponowne szkolenie wyzwalacza |
| Przekwalifikowanie | Automatycznie w przypadku nowych danych lub degradacji |
| Powtarzalność | Dobra (wersja potokowa) |
Poziom 1 jest wystarczający, gdy dane zmieniają się często, ale podejście ML pozostaje stabilny. Potok jest taki sam, ale jest okresowo uruchamiany ze świeżymi danymi.
Poziom 2: CI/CD dla uczenia maszynowego
Na poziomie 2 dodawany jest kompletny system CI/CD specyficzne dla ML. Zmieniają się nie tylko dane, ale także kod potoku, funkcje, hiperparametry, architekturę modelu. Każda zmiana przechodzi automatyczne testy, walidację i kontrolowane rozmieszczenie.
| Charakterystyczny | Poziom 2 |
|---|---|
| Szkolenie | Automatyczny + CI/CD na samym rurociągu |
| Wdrożyć | Niebieski/zielony, kanarek, testy A/B |
| Monitorowanie | Kompleksowe: dryf danych, dryf koncepcji, wydajność, opóźnienia |
| Przekwalifikowanie | Automatycznie z walidacją i wycofywaniem |
| Powtarzalność | Kompletny (wersja kodu + danych + środowiska) |
Osiągnięcie poziomu 2 i celu dla dojrzałych organizacji. Wymaga inwestycji znaczących w infrastrukturze i kulturze, ale to jedyny sposób na zarządzanie dziesiątkami czy setkami modeli zrównoważonej produkcji.
Cykl życia MLOps
Cykl życia modelu uczenia maszynowego w środowisku produkcyjnym jest procesem iteracyjnym, przez który przechodzi sześć głównych faz. W przeciwieństwie do tradycyjnego tworzenia oprogramowania, ten cykl tego nie robi nigdy się nie kończy: model w produkcji wymaga ciągłej konserwacji.
+----------+ +---------+ +----------+
| 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. Data: Gromadzenie, czyszczenie i wersjonowanie
Wszystko zaczyna się od danych. Na tym etapie zbierane i oczyszczane są surowe dane (obsługa brakujących wartości, wartości odstających, duplikatów), są one przekształcane w przydatne funkcje i weryfikują siebie. Wersjonowanie danych ma fundamentalne znaczenie: umożliwia odtworzenie modelu musisz wiedzieć dokładnie jakie dane służyły do treningu. Narzędzia takie jak DVC (Kontrola wersji danych) umożliwia wersjonowanie duże zbiory danych podobne do Git.
2. Pociąg: Inżynieria funkcji i szkolenie
Mając gotowe dane, tworzysz funkcje, wybierasz algorytm i trenujesz model. Każdy eksperyment (kombinacja hiperparametrów, cech, architektury) jest śledzone za pomocą odpowiednich parametrów i metryk. Narzędzia takie jak MLflow aby proces ten był systematyczny i powtarzalny.
3. Ocena: walidacja i porównanie
Wyszkolony model jest sprawdzany pod kątem predefiniowanych metryk (dokładność, wynik F1, RMSE, AUC) i porównano z wersją aktualnie produkowaną. Jeśli nowy model nie przekracza progów minimalnych lub nie poprawia się od poprzedniego, nie otrzymuje awansu.
4. Wdróż: Staging, Canary i Release
Zatwierdzony model przechodzi przez środowiska progresywne: etap testowy integracja, canary do walidacji przy ograniczonym rzeczywistym ruchu i wreszcie produkcja kompletny. Strategie takie jak wdrożenie w kolorze niebieskim/zielonym e uwolnienie kanarków zminimalizować ryzyko.
5. Monitoruj: dryf, metryki i alerty
W produkcji model jest monitorowany w sposób ciągły. Metryki są śledzone techniki (opóźnienie, przepustowość, błędy) i metryki ML (dokładność na rzeczywistych danych, dystrybucja prognoz, dryf danych). Alerty są wyzwalane, gdy metryki spadają poniżej progów.
6. Ponowne szkolenie: wyzwalanie i automatyzacja
Gdy monitorowanie wykryje degradację, aktywowane jest ponowne szkolenie. To może być zaplanowane (np. cotygodniowe), wyzwalane (np. dokładność poniżej 90%) lub ręczne. Nowy model ponownie przechodzi przez fazę oceny i wdrożenia.
Stos MLOps typu open source
Jedną z zalet MLOps jest to, że istnieje dojrzały ekosystem open source, który go obejmuje każdym etapie cyklu życia. Nie ma potrzeby kupowania drogich platform dla przedsiębiorstw na początek: dzięki odpowiednim narzędziom możesz zbudować kompletny potok MLOps.
| Faza | Instrument | Funkcjonować |
|---|---|---|
| Wersjonowanie danych | DVC | Wersjonowanie zbiorów danych i modeli zintegrowane z Git |
| Śledzenie eksperymentu | MLflow | Rejestrowanie parametrów, metryk i artefaktów dla każdego eksperymentu |
| Rejestr modeli | Rejestr modelu MLflow | Wersjonowanie i promocja modelu (inscenizacja/produkcja) |
| Orkiestracja rurociągów | Prefekt / Przepływ powietrza | Orkiestracja przepływu pracy, planowanie, ponowna próba |
| Porcja modelowa | FastAPI + Docker | Interfejs API REST do obsługi prognoz, skonteneryzowany |
| Konteneryzacja | Doker + K8 | Powtarzalne środowiska, skalowalność pozioma |
| Monitorowanie | Prometeusz + Grafana | Metryki, dashboardy, alerty |
| Walidacja danych | Wielkie oczekiwania | Automatyczne testy jakości danych |
Od notatnika do rurociągu: przykład praktyczny
Przyjrzyjmy się najczęstszemu etapowi, przed którym staje każdy zespół ML: przekształcaniu kodu napisane w notatniku Jupyter w sposób modułowy i powtarzalny. Weźmy A prawdziwy przykład klasyfikacji i zobaczmy, jak ją zrestrukturyzować.
Przed: Notatnik monolityczny
Oto typowy notatnik, w którym wszystko znajduje się w jednym pliku, bez rozdzielania odpowiedzialność, bez logowania, bez wersjonowania.
# 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!")
Problemy z notebookiem monolitycznym
- Nie można grać: bez nasion, bez wersjonowania danych
- Nie śledzone: parametry i metryki znajdują się wyłącznie w wynikach notebooka
- Nie testowalne: brak izolowanych funkcji do przetestowania
- Nie można wdrożyć: marynata nie jest interfejsem API
- Nie do utrzymania: modyfikowanie funkcji wymaga ponownego uruchomienia wszystkiego
Po: Rurociąg modułowy
Restrukturyzujemy kod w osobne moduły, z których każdy ma określoną odpowiedzialność. Każda funkcja jest testowalna, każdy parametr konfigurowalny, każda metryka jest śledzona.
"""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
Zalety rurociągu modułowego
- Odtwarzalny: naprawione ziarno, konfiguracja zlecona na zewnątrz, dane wersjonowane
- Testowalne: każda funkcja jest izolowana i może mieć dedykowane testy jednostkowe
- Utrzymywane: modyfikowanie funkcji nie ma wpływu na szkolenie i odwrotnie
- Konfigurowalne: zmieniaj hiperparametry bez dotykania kodu
- Automatyczne: potok może być obsługiwany przez CI/CD
Śledzenie eksperymentów za pomocą MLflow
Ile razy zmieniałeś hiperparametr, a potem nie pamiętałeś, jaką miał kombinację? biorąc pod uwagę najlepszy wynik? THE'śledzenie eksperymentalne rozwiązuje ten problem automatycznie rejestruje parametry, metryki i artefakty każdego eksperymentu.
MLflow i najpopularniejsze narzędzie typu open source do eksperymentalnego śledzenia. Oferuje serwer z interfejsem internetowym do przeglądania i porównywania eksperymentów, API Pythona do rejestrowania danych oraz Rejestr Modeli do zarządzania cyklem życia modeli.
Konfiguracja i pierwszy eksperyment
# 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}")
Po przeprowadzeniu kilku eksperymentów otwórz http://127.0.0.1:5000 w przeglądarce.
Interfejs użytkownika MLflow wyświetla tabelę ze wszystkimi eksperymentami, umożliwiając porównanie
metryki, sortuj według wydajności i przeglądaj wykresy metryk i metryk.
Rejestr modeli: modele wersjonowania
Podobnie jak kod jest wersjonowany w Git, modele ML muszą być wersjonowane z Rejestr modeli. Rejestr modelu MLflow oferuje scentralizowany system zarządzać cyklem życia modeli w trzech etapach.
| Stadion | Opis | Kto tego używa |
|---|---|---|
| Brak / Inscenizacja | Model w fazie testów i walidacji | Analityk danych, kontrola jakości |
| Produkcja | Zatwierdzony model, obsługuje rzeczywisty ruch | Obsługa interfejsu API, użytkowników końcowych |
| Zarchiwizowane | Model wycofany ze służby, zachowany do kontroli | Zgodność, wycofanie |
"""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)
Wdrożenie: FastAPI + Docker
Model ML w produkcji jest zwykle eksponowany jako API RESTOWE. FastAPI i idealny wybór dla Pythona: i szybki (w oparciu o ASGI), generuje automatyczną dokumentację (OpenAPI/Swagger) i ma doskonałą walidację danych poprzez Pydantic. Kontenerowanie z Doker, otrzymujemy artefakt, który można rozmieścić w dowolnym miejscu.
"""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}'
Monitorowanie w produkcji
Wdrożenie nie oznacza końca pracy, ale początek nowej, krytycznej fazy: monitorowanie. Model w produkcji ulega degradacji z biegiem czasu, ponieważ ulega zniszczeniu zmiany, a wraz z nimi dane. Monitoring musi obejmować trzy główne obszary.
Metryki do śledzenia
| Kategoria | Metryka | Instrument |
|---|---|---|
| Infrastruktura | Opóźnienie (p50, p95, p99), przepustowość, błędy HTTP, procesor/RAM | Prometeusz + Grafana |
| Model | Dokładność, wynik F1, rozkład przewidywań, pewność | MLflow + niestandardowe metryki |
| Dane | Dryf danych, dryf cech, brakujące wartości, rozkład danych wejściowych | Ewidentnie AI / Wielkie oczekiwania |
Dryf danych a dryf koncepcji
Istotne jest rozróżnienie dwóch typów degradacji modelu:
- Przesunięcie daty: rozkład zmian danych wejściowych w zakresie zestaw treningowy. Przykład: modelka szkolona na klientach w wieku 25-45 lat zaczyna przyjmować zapytania dla klientów 60+.
- Dryf koncepcji: związek między zmianami wejściowymi i wyjściowymi. Przykład: po pandemia, wzorce odpływu klientów są zupełnie inne, ale cechy na wejściu mają ten sam rozkład.
"""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
Kiedy ponownie aktywować przekwalifikowanie
Nie każdy drift wymaga natychmiastowego przekwalifikowania. Zdefiniuj jasne progi: dryf danych w przypadku cech krytycznych, spadek dokładności większy niż 5% lub rozkład przewidywań znacząco niezrównoważony. Unikaj nadmiernego przekwalifikowania, jakie może to wprowadzić niestabilny.
Jak zacząć z budżetem mniejszym niż 5000 EUR rocznie
MLOps nie musi oznaczać platform korporacyjnych kosztujących setki tysięcy euro. Dla włoskiego MŚP lub małego zespołu istnieje możliwość zbudowania infrastruktury MLOps skuteczne dzięki narzędziom open source i minimalnym kosztom.
Stos proponowany dla MŚP
| Część | Rozwiązanie | Koszt roczny |
|---|---|---|
| Kod | GitHub za darmo / GitLab CE | 0 EUR |
| Wersjonowanie danych | DVC + Google Cloud Storage (5 GB bezpłatnie) | 0 - 50 EUR |
| Śledzenie eksperymentu | MLflow na taniej maszynie wirtualnej | 200 - 500 EUR |
| Szkolenie | Google Colab Pro / maszyna wirtualna typu spot | 120 - 600 EUR |
| Porcja | FastAPI na maszynie wirtualnej (2 vCPU, 4 GB RAM) | 300 - 800 EUR |
| Monitorowanie | Prometeusz + Grafana (własny gospodarz) | 0 EUR (na tej samej maszynie wirtualnej) |
| CI/CD | Akcje GitHub (2000 min/miesiąc bezpłatnie) | 0 EUR |
| Rejestr kontenerów | Rejestr kontenerów GitHub | 0 EUR |
Szacunkowa suma: 620 - 1950 EUR/rok, znacznie poniżej progu 5 000 EUR. Ten stos obsługuje do 5-10 modeli w produkcji o umiarkowanym natężeniu ruchu (tysiące prognoz dziennie).
Wskazówki dotyczące obniżania kosztów
- Maszyna wirtualna typu spot/możliwa do wywłaszczenia: do 70% oszczędności w przypadku niepilnych szkoleń
- Automatyczne skalowanie: skaluj do zera, gdy nie ma żądań
- Kompresja modelu: mniejsze modele = mniej zasobów do obsługi
- Wnioskowanie wsadowe: jeśli nie potrzebujesz prognoz w czasie rzeczywistym, użyj partii nocnych
- Wielu najemców: jedna infrastruktura MLflow/Grafana dla wszystkich projektów
Struktura projektu MLOps
Na koniec coś, co można natychmiast wykorzystać, oto struktura folderów rekomendowany do projektu MLOps. Organizacja ta następuje po oddzieleniu odpowiedzialność i łatwość automatyzacji, testowania i współpracy.
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
Wnioski i dalsze kroki
MLOps nie jest luksusem zarezerwowanym dla dużych firm technologicznych. I konieczność każdego, kto chce wprowadzić do produkcji modele ML w sposób niezawodny i zrównoważony. W tym artykule omówiliśmy podstawy: od zrozumienia problemu (dlaczego projekty ML zawodzą) do konkretnego rozwiązania (rurociągi modułowe, eksperyment śledzenie, rejestracja modeli, obsługa kontenerowa i monitorowanie).
Najważniejsze jest, aby zacząć stopniowo. Nie ma potrzeby osiągania poziomu 2 modelu dojrzałości Google od pierwszego dnia. Zacznij od poziomu 0, stosując dobre praktyki:
- Zaraz: Podziel kod notesu na moduły. Użyj pliku config.yaml.
- Tydzień 1: Dodaj MLflow, aby śledzić eksperymenty.
- Tydzień 2: Konteneryzuj model za pomocą FastAPI + Docker.
- Miesiąc 1: Zaimplementuj potok CI/CD za pomocą GitHub Actions.
- Miesiąc 2: Dodaj monitorowanie za pomocą Prometheusa i podstawowe alerty.
- Miesiąc 3: Zaimplementuj DVC do wersjonowania danych.
W kolejnych artykułach z tej serii zagłębimy się w każdy komponent: zarządzanie danymi z DVC, tworzenie potoków CI/CD specyficznych dla ML, zaawansowane monitorowanie za pomocą Ewidentnie sztuczna inteligencja i skalowalne wdrożenie na Kubernetesie. Każdy przedmiot będzie praktyczny, m.in Działający kod i instrukcje krok po kroku.
Plan działania serii
- Artykuł 2: DVC — wersjonowanie danych dla ML
- Artykuł 3: Głębokie nurkowanie MLflow — zaawansowane śledzenie eksperymentów
- Artykuł 4: CI/CD dla uczenia maszynowego z akcjami GitHub
- Artykuł 5: Sklep z funkcjami i inżynieria funkcji w produkcji
- Artykuł 6: Skalowalne udostępnianie modelu w Kubernetes
- Artykuł 7: Zaawansowane monitorowanie: dryf danych i ewidentna sztuczna inteligencja
- Artykuł 8: Zarządzanie, zgodność i odpowiedzialne ML