12 - Deployment ML in produzione: serving e monitoring

Un modello è finito quando qualcun altro può chiamarlo, e continua a funzionare mentre il mondo cambia. Il notebook che ti ha dato 97% di accuratezza non risponde a nessuna delle due domande: non dice come il modello arriva a chi ne ha bisogno, né come ti accorgi che ha smesso di funzionare. Il deployment è il lavoro che porta il modello fuori dal notebook; il serving è il modo in cui risponde alle richieste (online, una alla volta, o a blocchi); il monitoring è l'insieme delle misure che ti dicono, dopo il rilascio, se risponde ancora bene.

Questo articolo chiude la serie ML Fundamentals e usa un solo modello piccolo per eseguire ciascuno di questi passi: salvare un artefatto, esporlo con un servizio che valida gli ingressi, misurare la latenza, produrre di proposito un errore di preparazione dei dati, osservare due tipi di cambiamento dei dati in produzione, e calcolare quanti esempi servono per fidarsi di un modello nuovo.

Tutto è eseguito in un solo processo, senza server in ascolto e senza porte aperte: il servizio è provato con il client di test di FastAPI, che chiama l'applicazione direttamente. Non c'è un cloud né un cluster, e non ne servono per capire dove si rompe un modello in produzione.

Il modello è la regressione logistica del terzo articolo sulle cifre scritte a mano, e le metriche sono quelle dell'articolo precedente. La parte infrastrutturale (container, registro dei modelli, pipeline di rilascio) ha un posto a parte nella serie Machine Learning Engineer: Dockerfile per un servizio ML, model registry e dove gira il modello.

Dal notebook all'artefatto

Quello che porti in produzione è un file, l'artefatto, più le informazioni per usarlo senza sorprese. Per un modello scikit-learn il formato più comune è joblib, un formato basato su pickle. La documentazione di scikit-learn elenca i rischi: il file richiede lo stesso ambiente di quello di addestramento (non si garantisce che un modello salvato con una versione di scikit-learn si carichi con un'altra) e caricare un pickle può eseguire codice arbitrario.

Come alternative la guida cita ONNX, che permette di servire senza Python e che definisce l'opzione più sicura, ma non copre tutti i modelli, e skops, che permette di validare in parte il contenuto prima di caricarlo.

Il listato salva una pipeline (lo StandardScaler e il modello insieme, per una ragione che vedrai fra poco), scrive un file di metadati con le versioni delle librerie e l'impronta SHA-256 dell'artefatto, ricarica il modello dopo aver verificato l'impronta e controlla che le predizioni coincidano. L'ultima parte mostra che cosa vuol dire «caricare è eseguire»: un oggetto innocuo, con un metodo __reduce__ che chiama print, stampa una riga mentre viene deserializzato.

# Un modello da portare in produzione: artefatto, metadati, e che cosa significa "caricare" un file pickle
import hashlib, json, tempfile
from pathlib import Path
import joblib
import numpy as np
import scipy, sklearn
from sklearn.datasets import load_digits
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler

X, y = load_digits(return_X_y=True)
Xtr, Xte, ytr, yte = train_test_split(X / 16.0, y, test_size=0.3, random_state=0, stratify=y)
modello = make_pipeline(StandardScaler(), LogisticRegression(max_iter=1000)).fit(Xtr, ytr)   # scaler E modello insieme

with tempfile.TemporaryDirectory() as d:
    percorso = Path(d) / "digits-v1.joblib"
    joblib.dump(modello, percorso)
    meta = {"nome": "digits", "versione": "1", "sklearn": sklearn.__version__, "numpy": np.__version__,
            "scipy": scipy.__version__, "n_feature": 64, "sha256": hashlib.sha256(percorso.read_bytes()).hexdigest()}
    (Path(d) / "digits-v1.json").write_text(json.dumps(meta))
    # ricarico come farebbe il servizio, dopo aver controllato che l'impronta coincida
    atteso = json.loads((Path(d) / "digits-v1.json").read_text())["sha256"]
    assert hashlib.sha256(percorso.read_bytes()).hexdigest() == atteso
    caricato = joblib.load(percorso)
    print("dimensione dell'artefatto:", percorso.stat().st_size, "byte")
print("metadati salvati:", {k: v for k, v in meta.items() if k != "sha256"}, "+ sha256 (", len(meta["sha256"]), "caratteri )")
print("predizioni identiche prima e dopo il salvataggio:", bool((caricato.predict(Xte) == modello.predict(Xte)).all()))
print(f"accuratezza sul test: {caricato.score(Xte, yte):.4f}")

# Caricare un pickle e' ESEGUIRE codice: questo oggetto, innocuo, stampa una riga mentre lo si deserializza
import pickle
class Innocuo:
    def __reduce__(self):
        return (print, ("  >>> codice eseguito durante pickle.loads",))
blob = pickle.dumps(Innocuo())
print("sto per chiamare pickle.loads su un file che non ho scritto io:")
pickle.loads(blob)

# Output:
# dimensione dell'artefatto: 8137 byte
# metadati salvati: {'nome': 'digits', 'versione': '1', 'sklearn': '1.9.0', 'numpy': '2.4.6', 'scipy': '1.17.1', 'n_feature': 64} + sha256 ( 64 caratteri )
# predizioni identiche prima e dopo il salvataggio: True
# accuratezza sul test: 0.9722
# sto per chiamare pickle.loads su un file che non ho scritto io:
#   >>> codice eseguito durante pickle.loads

L'artefatto pesa 8.137 byte ed è stato ricaricato con le stesse predizioni (accuratezza 0,9722 sul test). I metadati dicono con quali versioni è stato creato: se in produzione ne trovi altre, scikit-learn avvisa con un InconsistentVersionWarning che puoi trasformare in errore. L'impronta SHA-256 dimostra che il file non è cambiato dopo che l'hai salvato, ma non che sia sicuro: dimostra solo che è lo stesso.

L'ultima riga stampata, «codice eseguito durante pickle.loads», è il motivo per cui un file pickle di cui non conosci l'origine non si carica mai, come avverte la documentazione di Python: l'artefatto che il tuo servizio carica deve venire solo da un luogo che controlli.

Un servizio che valida l'ingresso

Un servizio di predizione ha tre compiti: caricare il modello una volta sola, controllare che l'ingresso sia ammissibile e rispondere. Il listato lo scrive con FastAPI. Il modello viene caricato in un gestore del ciclo di vita (lifespan), all'avvio, e non a ogni richiesta, che sarebbe costoso.

L'ingresso è descritto da un modello Pydantic: una lista di esattamente 64 numeri, ciascuno fra 0 e 1, con un controllo esplicito sull'intervallo. Il client TestClient di FastAPI, usato come gestore di contesto con with, esegue il ciclo di vita e chiama l'applicazione nel processo, senza aprire una porta (la versione installata avvisa che l'uso di httpx da parte del TestClient è deprecato: non cambia i risultati).

# Un servizio di predizione in FastAPI, provato dentro il processo con TestClient (nessun server, nessuna porta)
import tempfile
from contextlib import asynccontextmanager
from pathlib import Path
import joblib
from fastapi import FastAPI
from fastapi.testclient import TestClient
from pydantic import BaseModel, Field, field_validator
from sklearn.datasets import load_digits
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler

X, y = load_digits(return_X_y=True)
Xtr, Xte, ytr, yte = train_test_split(X / 16.0, y, test_size=0.3, random_state=0, stratify=y)
tmp = tempfile.TemporaryDirectory()
ARTEFATTO = Path(tmp.name) / "digits-v1.joblib"
joblib.dump(make_pipeline(StandardScaler(), LogisticRegression(max_iter=1000)).fit(Xtr, ytr), ARTEFATTO)

stato = {}

@asynccontextmanager
async def ciclo_di_vita(app):
    stato["modello"] = joblib.load(ARTEFATTO)       # caricato UNA volta all'avvio, non a ogni richiesta
    yield
    stato.clear()

app = FastAPI(title="digits", lifespan=ciclo_di_vita)

class Richiesta(BaseModel):
    pixel: list[float] = Field(min_length=64, max_length=64)

    @field_validator("pixel")
    @classmethod
    def nel_range(cls, v):
        if any(not 0.0 <= p <= 1.0 for p in v):
            raise ValueError("i pixel devono stare in [0, 1]")
        return v

@app.get("/health")
def health():
    return {"stato": "ok", "modello": "digits-v1"}

@app.post("/predict")
def predict(r: Richiesta):
    p = stato["modello"].predict_proba([r.pixel])[0]
    return {"classe": int(p.argmax()), "probabilita": round(float(p.max()), 4), "modello": "digits-v1"}

with TestClient(app) as client:                      # "with" esegue il ciclo di vita: il modello viene caricato
    print("GET /health ->", client.get("/health").status_code, client.get("/health").json())
    r = client.post("/predict", json={"pixel": Xte[0].tolist()})
    print("POST /predict valida ->", r.status_code, r.json(), " (etichetta vera:", int(yte[0]), ")")
    r = client.post("/predict", json={"pixel": [0.5] * 63})
    print("63 valori invece di 64 ->", r.status_code, r.json()["detail"][0]["type"])
    r = client.post("/predict", json={"pixel": [2.0] + [0.0] * 63})
    print("un pixel fuori da [0, 1] ->", r.status_code, r.json()["detail"][0]["msg"])
    r = client.post("/predict", json={"pixel": ["a"] * 64})
    print("testo al posto dei numeri ->", r.status_code, r.json()["detail"][0]["type"])
    giuste = sum(client.post("/predict", json={"pixel": x.tolist()}).json()["classe"] == t for x, t in zip(Xte[:100], yte[:100]))
    print("prime 100 immagini del test passate dal servizio: classi giuste =", giuste)

# Output:
# GET /health -> 200 {'stato': 'ok', 'modello': 'digits-v1'}
# POST /predict valida -> 200 {'classe': 1, 'probabilita': 0.9525, 'modello': 'digits-v1'}  (etichetta vera: 1 )
# 63 valori invece di 64 -> 422 too_short
# un pixel fuori da [0, 1] -> 422 Value error, i pixel devono stare in [0, 1]
# testo al posto dei numeri -> 422 float_parsing
# prime 100 immagini del test passate dal servizio: classi giuste = 98

La richiesta valida riceve 200 e la classe 1 con probabilità 0,9525, che è l'etichetta vera. Le tre richieste non valide non arrivano al modello: 63 valori al posto di 64 danno 422 con errore too_short, un pixel a 2,0 dà 422 con il messaggio del tuo controllo, e il testo al posto dei numeri dà 422 con float_parsing.

Senza questo livello, una lista di 63 valori arriverebbe alla pipeline e solleverebbe un'eccezione interna, un errore 500 che nasconde il problema vero. Passando 100 immagini del test dal servizio, 98 sono classificate correttamente. Questo è un servizio minimale: manca l'autenticazione, manca il limite di frequenza e manca la gestione dei registri. Per esporlo davvero serve un server ASGI come uvicorn, che qui non ho avviato per non aprire porte.

Latenza: una richiesta alla volta contro un blocco

La latenza è il tempo fra la richiesta e la risposta; per un servizio online di solito si guarda il percentile alto (il 99° percentile, cioè il tempo sotto cui stanno 99 richieste su 100) più che la media. In un servizio che riceve richieste da blocchi (batch), conta invece quanto lavoro si fa per secondo.

Il listato misura lo stesso modello in tre modi: 500 chiamate da una riga, una chiamata da 500 righe e blocchi di dimensioni diverse. I tempi sono misure e cambiano da una esecuzione all'altra, perché dipendono dalla macchina e dal carico: li ho misurati in cinque esecuzioni su un AMD Ryzen 5 5600H con 2 thread e riporto gli intervalli.

# Latenza: una predizione alla volta contro blocchi (TEMPI: variano da una esecuzione all'altra)
import time, tempfile
from pathlib import Path
import joblib
import numpy as np
from sklearn.datasets import load_digits
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler

X, y = load_digits(return_X_y=True)
Xtr, Xte, ytr, yte = train_test_split(X / 16.0, y, test_size=0.3, random_state=0, stratify=y)
m = make_pipeline(StandardScaler(), LogisticRegression(max_iter=1000)).fit(Xtr, ytr)
righe = Xte[:500]
m.predict_proba(righe[:5])                                           # riscaldamento

t0 = time.perf_counter()
singole = [m.predict_proba(r.reshape(1, -1))[0] for r in righe]      # 500 chiamate da una riga
t_singole = time.perf_counter() - t0
t0 = time.perf_counter()
blocco = m.predict_proba(righe)                                      # una chiamata da 500 righe
t_blocco = time.perf_counter() - t0
print("stesse probabilita':", np.allclose(np.array(singole), blocco))
print(f"500 chiamate singole: {t_singole * 1000:7.1f} ms in tutto, {t_singole / 500 * 1000:.3f} ms a richiesta")
print(f"1 chiamata da 500:    {t_blocco * 1000:7.1f} ms in tutto, {t_blocco / 500 * 1000:.3f} ms a riga")
print(f"rapporto fra i costi per riga: {t_singole / t_blocco:.0f}x")

# Dimensione del blocco: tempo per riga
for n in (1, 10, 100, 1000):
    reps = max(1, 2000 // n)
    t0 = time.perf_counter()
    for _ in range(reps):
        m.predict_proba(Xte[:n])
    print(f"blocco da {n:4d} righe: {(time.perf_counter() - t0) / reps / n * 1000:.4f} ms a riga")

# Output:
# stesse probabilita': True
# 500 chiamate singole:   127.5 ms in tutto, 0.255 ms a richiesta
# 1 chiamata da 500:        0.5 ms in tutto, 0.001 ms a riga
# rapporto fra i costi per riga: 262x
# blocco da    1 righe: 0.2351 ms a riga
# blocco da   10 righe: 0.0204 ms a riga
# blocco da  100 righe: 0.0024 ms a riga
# blocco da 1000 righe: 0.0004 ms a riga

Le probabilità sono le stesse nei due modi. Cinquecento chiamate singole hanno impiegato 100-135 ms in tutto, cioè circa 0,20-0,27 ms a richiesta, mentre una chiamata da 500 righe ne ha impiegato 0,4-0,5: il costo per riga è 240-300 volte più basso. Non è il calcolo a essere lento, è il costo fisso di ogni chiamata (validazione, conversione degli array, chiamate Python), che per un blocco si paga una volta sola.

Il costo per riga scende da 0,20-0,22 ms con una riga a 0,0004-0,0005 ms con mille. In un servizio online raccogliere più richieste in un blocco (micro-batching) può aumentare molto il numero di richieste al secondo, ma aggiunge l'attesa necessaria a riempire il blocco: è un compromesso fra latenza di ciascuna richiesta e throughput complessivo. Questa misura riguarda un modello lineare su un laptop; per un altro modello i numeri cambiano e vanno rimisurati.

Il bug silenzioso: training-serving skew

Il training-serving skew è la differenza fra come i dati sono preparati nell'addestramento e come lo sono in produzione. La guida «Rules of Machine Learning» di Google lo definisce come una differenza fra le prestazioni durante l'addestramento e durante il serving, che può nascere da una differenza nel modo in cui i dati sono trattati nelle due pipeline, da un cambiamento dei dati fra i due momenti o da un ciclo di retroazione fra modello e algoritmo, e consiglia di misurarlo esplicitamente.

Un errore classico di preparazione è reimplementare nel servizio la standardizzazione: dimenticarla, oppure calcolarla sul blocco di richieste invece che con le statistiche del training. Il listato lo prova sul dataset del tumore al seno (30 feature con scale molto diverse), con tre varianti e un modello che si aspetta ingressi standardizzati.

# Training-serving skew: lo stesso modello, tre modi di preparare gli ingressi in produzione
import numpy as np
from sklearn.datasets import load_breast_cancer
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler

X, y = load_breast_cancer(return_X_y=True)
Xtr, Xte, ytr, yte = train_test_split(X, y, test_size=0.3, random_state=0, stratify=y)

scaler = StandardScaler().fit(Xtr)
lr = LogisticRegression(max_iter=2000).fit(scaler.transform(Xtr), ytr)        # il modello vede dati standardizzati
pipeline = make_pipeline(scaler, lr)                                          # lo scaler viaggia con il modello
print(f"classe positiva nel test: {yte.mean():.3f}   campioni di test: {len(yte)}")

print(f"A) pipeline salvata intera:                      accuratezza {pipeline.score(Xte, yte):.3f}")
print(f"B) il servizio dimentica lo scaling:             accuratezza {lr.score(Xte, yte):.3f}")

def con_statistiche_del_blocco(batch):
    """Errore tipico: standardizzare le richieste con media e deviazione del BLOCCO ricevuto, non del training."""
    return (batch - batch.mean(axis=0)) / (batch.std(axis=0) + 1e-9)

for n in (1, 5, 50):
    giuste = 0
    for i in range(0, len(Xte) - len(Xte) % n, n):
        giuste += (lr.predict(con_statistiche_del_blocco(Xte[i:i + n])) == yte[i:i + n]).sum()
    tot = len(Xte) - len(Xte) % n
    print(f"C) statistiche del blocco, richieste a blocchi da {n:2d}: accuratezza {giuste / tot:.3f}")

# Output:
# classe positiva nel test: 0.626   campioni di test: 171
# A) pipeline salvata intera:                      accuratezza 0.959
# B) il servizio dimentica lo scaling:             accuratezza 0.374
# C) statistiche del blocco, richieste a blocchi da  1: accuratezza 0.626
# C) statistiche del blocco, richieste a blocchi da  5: accuratezza 0.847
# C) statistiche del blocco, richieste a blocchi da 50: accuratezza 0.940

Con la pipeline salvata intera l'accuratezza è 0,959. Se il servizio dimentica lo scaling scende a 0,374, meno del caso (il 62,6% dei campioni è di classe positiva: predire sempre «positivo» farebbe 0,626, sempre «negativo» 0,374): il modello vede numeri sulla scala sbagliata. L'errore C è subdolo: standardizzare con le statistiche del blocco funziona più o meno se il blocco è grande (0,940 con 50 richieste), peggio con 5 (0,847) e fallisce con una richiesta sola, perché la deviazione è zero e tutti i valori diventano zero, quindi il modello risponde sempre la stessa classe, con accuratezza 0,626.

Nessuna eccezione, nessun errore 500: il servizio risponde bene, con numeri sbagliati. La difesa è la regola di questa sezione: salva la preparazione dei dati con il modello, nella stessa pipeline, e verifica con un test che il servizio dia le stesse predizioni dell'ambiente di addestramento sugli stessi esempi.

Monitorare: ingressi, uscite ed esiti

Dopo il rilascio i dati cambiano. Sculley e colleghi, in «Hidden Technical Debt in Machine Learning Systems», sostengono che i test non bastano in un mondo che cambia e che serve un monitoraggio in tempo reale del comportamento del sistema; fra i punti di partenza indicano il confronto fra la distribuzione delle etichette predette e quella delle etichette osservate (prediction bias).

Nel linguaggio comune i cambiamenti sono di due tipi: il covariate shift, in cui cambia la distribuzione degli ingressi, e il concept drift, in cui cambia il legame fra ingressi e risultato. Per confrontare gli ingressi si usa il test di Kolmogorov-Smirnov a due campioni (in SciPy, ks_2samp) o il PSI (Population Stability Index), $\sum_i (a_i - e_i)\ln(a_i/e_i)$ sulle quote $e_i$ del riferimento e $a_i$ dell'attuale in intervalli a quantili: il PSI è una misura di uso comune e non uno standard di una norma.

Il listato addestra una regressione logistica su dati sintetici con sei feature, e poi la «mette in produzione» su tre lotti da 5.000 righe: uno senza cambiamenti, uno in cui la feature 0 trasla di una deviazione standard (covariate shift), e uno con gli stessi ingressi ma con il 30% degli esiti ribaltati (concept drift).

Per ogni lotto stampa la statistica KS e il suo p-value, il PSI della feature 0, l'accuratezza e le quote di classe 1 prevista e osservata. In produzione l'accuratezza richiede le etichette, che arrivano tardi o mai: qui le conosciamo perché è una simulazione.

# Monitoraggio: che cosa vedono gli ingressi e che cosa vede l'accuratezza (con le etichette, che in produzione arrivano tardi)
import numpy as np
from scipy.stats import ks_2samp
from sklearn.datasets import make_classification
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split

X, y = make_classification(n_samples=30000, n_features=6, n_informative=4, n_redundant=0, class_sep=1.2,
                           flip_y=0.02, random_state=0)
Xtr, Xrest, ytr, yrest = train_test_split(X, y, test_size=0.8, random_state=0)
modello = LogisticRegression(max_iter=1000).fit(Xtr, ytr)
rif = Xtr[:, 0]                                                     # distribuzione di riferimento (training), feature 0

def psi(riferimento, attuale, bins=10):
    """Population Stability Index: somma di (a - e) * ln(a / e) su intervalli a quantili del riferimento."""
    tagli = np.quantile(riferimento, np.linspace(0, 1, bins + 1)[1:-1])
    e = np.bincount(np.searchsorted(tagli, riferimento), minlength=bins) / len(riferimento)
    a = np.bincount(np.searchsorted(tagli, attuale), minlength=bins) / len(attuale)
    e, a = np.clip(e, 1e-6, None), np.clip(a, 1e-6, None)
    return float(((a - e) * np.log(a / e)).sum())

lotti = {}
lotti["nessun cambiamento"] = (Xrest[:5000], yrest[:5000])
Xs = Xrest[5000:10000].copy(); Xs[:, 0] += 1.0                      # la feature 0 trasla di +1 deviazione standard
lotti["covariate shift (feature 0 + 1)"] = (Xs, yrest[5000:10000])
Xc, yc = Xrest[10000:15000], yrest[10000:15000].copy()              # stessi ingressi, ma nel 30% dei casi l'esito si ribalta
rng = np.random.default_rng(0); ribalta = rng.random(len(yc)) < 0.30
yc[ribalta] = 1 - yc[ribalta]
lotti["concept drift (30% di esiti ribaltati)"] = (Xc, yc)

print(f"{'lotto di produzione':42s} {'KS feat.0':>9s} {'p-value':>9s} {'PSI feat.0':>10s} {'accuratezza':>12s} {'classe 1 prevista':>18s} {'classe 1 osservata':>19s}")
for nome, (Xl, yl) in lotti.items():
    ks = ks_2samp(rif, Xl[:, 0])
    print(f"{nome:42s} {ks.statistic:9.3f} {ks.pvalue:9.2e} {psi(rif, Xl[:, 0]):10.3f} {modello.score(Xl, yl):12.3f}"
          f" {modello.predict(Xl).mean():18.3f} {yl.mean():19.3f}")
print(f"accuratezza sul riferimento (test del training, 5000 righe): {modello.score(Xrest[:5000], yrest[:5000]):.3f}")

# Output:
# lotto di produzione                        KS feat.0   p-value PSI feat.0  accuratezza  classe 1 prevista  classe 1 osservata
# nessun cambiamento                             0.015  5.98e-01      0.002        0.747              0.453               0.489
# covariate shift (feature 0 + 1)                0.213 2.63e-108      0.389        0.731              0.592               0.499
# concept drift (30% di esiti ribaltati)         0.014  6.12e-01      0.003        0.603              0.461               0.504
# accuratezza sul riferimento (test del training, 5000 righe): 0.747

Il lotto senza cambiamenti dà KS 0,015 (p-value 0,60), PSI 0,002 e accuratezza 0,747. Il covariate shift è evidentissimo negli ingressi: KS 0,213 con p-value di 1e-108 e PSI 0,389, ma l'accuratezza scende solo a 0,731, perché il modello lineare regge bene la traslazione. Il concept drift è invisibile negli ingressi (KS 0,014, p-value 0,61, PSI 0,003, identici al lotto stabile) ma fa crollare l'accuratezza a 0,603.

Due lezioni opposte, dallo stesso esperimento. Primo: un allarme sugli ingressi non è un allarme sulla qualità del modello, e con migliaia di righe il p-value diventa minuscolo anche per spostamenti irrilevanti, quindi guarda l'ampiezza (KS, PSI) oltre al p-value. Secondo: nessun controllo sugli ingressi vede il concept drift, che si scopre solo con le etichette; la quota di classe 1 prevista contro l'osservata dà 0,461 contro 0,504, un divario di 4,3 punti, quasi uguale a quello del lotto stabile (3,6), e non è un segnale utile qui. Servono entrambi i livelli, e un modo per ottenere etichette, anche parziali.

Rilasciare un modello nuovo: quanti esempi servono

Un modello nuovo, che sul test offline è un punto percentuale migliore, va confrontato con quello in produzione su traffico reale: prima in shadow (riceve le stesse richieste ma le sue risposte non vengono usate), poi con una quota di traffico. Ma un punto di accuratezza su 90 è una differenza piccola, e un esperimento con troppo pochi esempi non la vede.

Per un confronto fra due proporzioni, con livello di significatività 5% e potenza 80%, il numero di richieste per braccio è $n = \left(z_{\alpha/2}\sqrt{2\overline{p}\overline{q}} + z_{\beta}\sqrt{p_1 q_1 + p_2 q_2}\right)^2 / (p_2 - p_1)^2$. Il listato calcola $n$ per 0,90 contro 0,91 e controlla la formula con una simulazione: 2.000 esperimenti per ciascuna dimensione, e la frazione in cui un test z a due proporzioni riconosce la differenza (la potenza). Controlla anche i falsi allarmi con due modelli uguali.

# Quanti esempi servono per accorgersi che il modello B e' meglio di A di un punto di accuratezza?
import numpy as np
from scipy.stats import norm

p1, p2, alfa, potenza = 0.90, 0.91, 0.05, 0.80
za, zb = norm.ppf(1 - alfa / 2), norm.ppf(potenza)
pm = (p1 + p2) / 2
n = (za * np.sqrt(2 * pm * (1 - pm)) + zb * np.sqrt(p1 * (1 - p1) + p2 * (1 - p2))) ** 2 / (p2 - p1) ** 2
n = int(np.ceil(n))
print(f"formula: servono circa {n} richieste PER BRACCIO ({2 * n} in tutto) per vedere {p1:.2f} -> {p2:.2f} (alpha {alfa}, potenza {potenza})")

def z_a_due_proporzioni(a, b, n):
    """a e b sono i successi dei due bracci (n richieste ciascuno)."""
    pp = (a + b) / (2 * n)
    return (b / n - a / n) / np.sqrt(2 * pp * (1 - pp) / n)

def frazione_significativa(n, p_a, p_b, prove=2000, seme=0):
    rng = np.random.default_rng(seme)
    z = z_a_due_proporzioni(rng.binomial(n, p_a, prove), rng.binomial(n, p_b, prove), n)
    return float((np.abs(z) > za).mean())

for k in (n // 4, n // 2, n, 2 * n):
    print(f"n per braccio = {k:6d}: l'esperimento riconosce la differenza nel {frazione_significativa(k, p1, p2):.1%} dei casi")
print(f"se i due modelli fossero uguali (0.90 contro 0.90), falsi allarmi con n = {n}: {frazione_significativa(n, 0.90, 0.90, seme=1):.1%}")

# Output:
# formula: servono circa 13495 richieste PER BRACCIO (26990 in tutto) per vedere 0.90 -> 0.91 (alpha 0.05, potenza 0.8)
# n per braccio =   3373: l'esperimento riconosce la differenza nel 28.3% dei casi
# n per braccio =   6747: l'esperimento riconosce la differenza nel 50.2% dei casi
# n per braccio =  13495: l'esperimento riconosce la differenza nel 80.4% dei casi
# n per braccio =  26990: l'esperimento riconosce la differenza nel 98.1% dei casi
# se i due modelli fossero uguali (0.90 contro 0.90), falsi allarmi con n = 13495: 5.1%

Servono circa 13.495 richieste per braccio, 26.990 in tutto. La simulazione conferma la formula: con questo $n$ l'esperimento riconosce la differenza nell'80,4% dei casi, con un quarto dell'$n$ solo nel 28,3%, con metà nel 50,2%. Se i due modelli fossero identici, il test dà un falso allarme nel 5,1% dei casi, come il livello di significatività dichiarato. Se il servizio riceve, per ipotesi, 5.000 richieste al giorno e ne mandi il 10% al modello nuovo, in un mese hai circa 15.000 richieste sul braccio nuovo, lo stesso ordine di grandezza del numero per braccio calcolato; con una ripartizione 10/90 la formula va rifatta, ma l'ordine di grandezza resta, e serve avere le etichette.

Questo è il motivo per cui gli aumenti di un punto di accuratezza sono difficili da dimostrare in produzione, e per cui conviene decidere in anticipo quale differenza ti interessa e con quale traffico puoi rilevarla.

Esercizio svolto

Su un'unica feature il riferimento (training) ha quote 0,5, 0,3 e 0,2 in tre intervalli; in produzione le quote sono 0,4, 0,3 e 0,3. Calcola a mano il PSI. Poi dì se il valore, da solo, ti basta per scegliere se riaddestrare il modello.

# Esercizio: PSI a mano su tre intervalli
import numpy as np
atteso = np.array([0.5, 0.3, 0.2])
attuale = np.array([0.4, 0.3, 0.3])
termini = (attuale - atteso) * np.log(attuale / atteso)
print("termini per intervallo:", np.round(termini, 5).tolist())
print("PSI =", round(float(termini.sum()), 5))

# Output:
# termini per intervallo: [0.02231, 0.0, 0.04055]
# PSI = 0.06286

Che cosa sai fare adesso, e che cosa hai percorso

Con questo articolo sai fare cinque cose, con una misura eseguita per ciascuna:

• salvare un modello con i suoi metadati e la sua impronta, e sapere perché non si carica un pickle di origine ignota;
• esporlo con un servizio che valida l'ingresso e provarlo senza aprire una porta;
• misurare la latenza e capire quando raccogliere richieste in blocchi;
• riconoscere il training-serving skew e prevenirlo salvando la preparazione dei dati con il modello;
• monitorare ingressi ed esiti, distinguendo covariate shift e concept drift, e calcolare quanti esempi servono per un confronto fra modelli.

Le cinque capacità hanno in comune che nessuna dà un risultato da sola: l'artefatto senza metadati si rompe al primo cambio di versione, il servizio senza validazione nasconde gli errori, il monitoraggio senza etichette non vede il concept drift. La distanza fra il notebook e la produzione è fatta di queste piccole verifiche, ciascuna delle quali puoi eseguire in un minuto, come hai fatto qui.

La serie ML Fundamentals finisce qui. Le dodici tappe, nell'ordine in cui si costruiscono l'una sull'altra:

1. Introduzione al Machine Learning: tassonomia dei problemi e flusso di lavoro;
2. Regressione lineare from scratch: il primo modello, scritto in Python;
3. Regressione logistica: dalla retta alla classificazione binaria;
4. Bias-variance tradeoff: perché un modello più flessibile non è sempre migliore;
5. Cross-validation: stimare le prestazioni e scegliere il modello;
6. Alberi decisionali e Random Forest: modelli non lineari a regole;
7. Support Vector Machines: margine massimo e kernel;
8. K-Means e clustering gerarchico: raggruppare senza etichette;
9. PCA: ridurre le dimensioni;
10. Reti neurali feedforward: dal perceptron alla backpropagation;
11. Metriche e valutazione: giudicare un modello con la domanda giusta;
12. questo articolo: portare il modello in produzione e continuare a misurarlo.

L'ordine non è casuale: le prime cinque tappe insegnano a costruire e a giudicare un modello semplice (regressione, classificazione, errore, validazione), le tappe dalla sesta alla decima allargano la famiglia dei modelli (alberi, margini, gruppi, assi principali, reti), e le ultime due chiudono il ciclo con la valutazione e la produzione. Se vuoi proseguire fuori dalla serie, la serie Machine Learning Engineer ripercorre molti degli stessi temi con più dettagli sul lavoro di ingegneria, comprese le tappe di tracciamento degli esperimenti e di test dei modelli.