Testowanie A/B modeli uczenia maszynowego: metodologia, metryki i wdrożenie
Wytrenowałeś dwie wersje swojego modelu rekomendacji. Nowy model oparty na transformatora, wykazuje o 3% wyższe AUC w zestawie wstrzymanym. Wydaje się, że jest wyraźna poprawa ale czy ta różnica naprawdę przekłada się na pozytywny wpływ na rzeczywistych użytkowników? Model może działać lepiej w niektórych kohortach demograficznych, a gorzej w innych. Może to zmniejszyć współczynnik klikalności przy jednoczesnym zwiększeniu długoterminowej satysfakcji. Może mieć opóźnienie wyższa, co niweluje korzyści związane z dokładnością. Mówią, że dane offline nie kłamią tylko część historii.
L'Testy A/B dla modeli ML oraz metodologia, która pozwala na reakcję te pytania rygorystycznie, porównując wersje modelu dotyczącego rzeczywistego ruchu z prawdziwych użytkowników, mierzących naprawdę ważne wskaźniki biznesowe. Według badań 2025 autorstwa Aimpoint Digital Labs, organizacji, które przyjmują ustrukturyzowane strategie testów A/B dla modeli ML zmniejszają się o 40% ryzyko regresji w produkcji w porównaniu z wdrożeniami bezpośrednimi opartymi wyłącznie na wskaźnikach offline. Rynek MLOps, który jest tego wart 4,38 miliarda dolarów w 2026 r. i oczekuje się, że do 2035 r. osiągnie 89,18 miliarda dolarów, a koszt A/B testowanie jest jednym z podstawowych elementów tego wzrostu.
W tym przewodniku zbudujemy kompletny system testów A/B dla modeli ML: od teorii statystyki do routera FastAPI, od wdrożenia Canary po tryb cienia, z testów częstości do podejścia bayesowskiego z Thompson Sampling, aż do monitorowania metryk w trakcie testy z Prometheusem i Grafaną.
Czego się nauczysz
- Różnica między testami ML A/B a klasycznymi testami A/B sieci
- Projektowanie eksperymentów: wielkość próby, moc statystyczna, metryki sukcesu
- Podział ruchu za pomocą routerów FastAPI i stopniowe wdrażanie Canary
- Tryb cienia: testuj bez wpływu na użytkowników
- Wieloręcy bandyci i Thompson Sampling jako alternatywa dla klasycznych testów A/B
- Analiza statystyczna: wartość p, przedziały ufności, wielkość efektu
- Bayesowskie testy A/B umożliwiające szybsze podejmowanie decyzji
- Monitoring podczas testów z Prometheusem i Grafaną
- Najlepsze praktyki i anty-wzorce, których należy unikać
Testowanie A/B uczenia maszynowego a testowanie A/B sieci Web: różnice krytyczne
Testy A/B narodziły się w analityce internetowej w celu porównywania odmian stron docelowych, przycisków i kopia. Podstawowe ramy statystyczne są takie same, ale testy A/B dla modeli ML już tak dodatkowe komplikacje, które znacznie różnią się w praktyce.
W testach sieciowych porównywane są dyskretne doświadczenia wizualne: wariant A i wariant B są wyraźnie oddzielone. W modelach ML przewidywania są ciągłe, rozproszone i często skorelowane w czasie. Model rekomendacji, który obsługuje tego samego użytkownika w różnych sesjach nie daje niezależnych przewidywań: istnieje korelacja czasowa co narusza założenia niezależności klasycznych testów statystycznych.
Kluczowe różnice ML a internetowe testy A/B
- Metryka: w sieci zoptymalizowany jest CTR, czyli współczynnik konwersji; w ML tak jednocześnie optymalizuj wskaźniki offline (AUC, RMSE) i wskaźniki biznesowe (przychody, wskaźnik rezygnacji, NPS), często sprzeczne.
- Opóźnienie odpowiedzi: w sieci wynik jest natychmiastowy (kliknij); w ML może zająć kilka dni lub tygodni (rezygnacja po 30 dniach, przychody po kwartale).
- Rozkład efektów: model może średnio działać lepiej ale gorzej w określonych kohortach (ageizm, uprzedzenia geograficzne), co wymaga analizy segmentowej.
- Efekty systemowe: w systemach sprzężenia zwrotnego (zalecenia, dynamiczna wycena), model B wpływa na dane, które następnie szkolą model C.
- Ryzyka operacyjne: błąd w wariancie internetowym powoduje zły UX; błąd w modelu ML wykrywania oszustw może spowodować znaczne straty finansowe.
Projekt eksperymentu: przed kodem
Źle zaprojektowany test A/B jest gorszy niż brak testu A/B: daje fałszywe poczucie rygor naukowy przy jednoczesnym formułowaniu błędnych wniosków. Projekt eksperymentu musi poprzedzać jakiekolwiek wdrożenie techniczne.
Zdefiniuj metryki sukcesu
Każdy eksperyment musi mieć Podstawowe metryki jedyny, który decyduje zwycięzcy plus zero do dwóch Metryki poręczy ten model B nie może być gorszy niż A. Podstawowy miernik musi być bezpośredni związek przyczynowo-skutkowy z celem biznesowym.
Przykładowe wskaźniki dla różnych scenariuszy:
- Model rezygnacji: podstawowy = wskaźnik retencji przez 30 dni; poręcz = opóźnienie P95, koszt kampanii
- Model rekomendacji: podstawowy = przychód na sesję; poręcz = CTR, zalecenia dotyczące różnorodności
- Model oszustwa: podstawowy = wskaźnik niewykrytych oszustw; poręcz = współczynnik wyników fałszywie dodatnich, opóźnienie
- Model cenowy: podstawowa = marża brutto; poręcz = współczynnik konwersji, NPS
Obliczanie wielkości próbki
Niezbędna wielkość próbki zależy od trzech czynników:wielkość efektu minimalne który chcesz wykryć (minimalny wykrywalny efekt, MDE), poziom znaczenia alfa (zwykle 0,05) i la moc statystyczna 1-beta (zwykle 0,80).
# sample_size_calculator.py
# Calcolo del sample size per A/B test ML
import numpy as np
from scipy import stats
from scipy.stats import norm
import math
def calculate_sample_size(
baseline_rate: float,
minimum_detectable_effect: float,
alpha: float = 0.05,
power: float = 0.80,
two_tailed: bool = True
) -> int:
"""
Calcola il sample size per un A/B test su proporzioni (es. conversion rate).
Args:
baseline_rate: Tasso attuale del modello A (es. 0.15 per 15% churn)
minimum_detectable_effect: Variazione minima relativa da rilevare (es. 0.05 per +5%)
alpha: Livello di significativita (type I error rate)
power: Potenza statistica (1 - type II error rate)
two_tailed: True per test bidirezionale (default raccomandato)
Returns:
Sample size per ciascuna delle due varianti
"""
p1 = baseline_rate
p2 = baseline_rate * (1 + minimum_detectable_effect)
# Calcolo basato su formula di Cohen
z_alpha = norm.ppf(1 - alpha / (2 if two_tailed else 1))
z_beta = norm.ppf(power)
p_avg = (p1 + p2) / 2
q_avg = 1 - p_avg
numerator = (z_alpha * math.sqrt(2 * p_avg * q_avg) + z_beta * math.sqrt(p1 * (1-p1) + p2 * (1-p2))) ** 2
denominator = (p2 - p1) ** 2
n = math.ceil(numerator / denominator)
return n
def calculate_duration_days(
sample_size_per_variant: int,
daily_requests: int,
traffic_split: float = 0.5
) -> float:
"""Stima la durata del test in giorni."""
requests_per_variant_per_day = daily_requests * traffic_split
return sample_size_per_variant / requests_per_variant_per_day
# --- Esempio pratico: modello churn ---
baseline_churn_rate = 0.18 # 18% churn attuale (modello A)
mde = 0.10 # vogliamo rilevare un miglioramento del 10% relativo
# (da 18% a 16.2%)
n_per_variant = calculate_sample_size(
baseline_rate=baseline_churn_rate,
minimum_detectable_effect=-mde, # negativo = riduzione del churn
alpha=0.05,
power=0.80
)
daily_traffic = 5000 # richieste al giorno
test_duration = calculate_duration_days(n_per_variant, daily_traffic, 0.5)
print(f"Sample size per variante: {n_per_variant:,} campioni")
print(f"Durata stimata del test: {test_duration:.1f} giorni")
print(f"Traffico totale necessario: {n_per_variant * 2:,} richieste")
# Output tipico:
# Sample size per variante: 8,744 campioni
# Durata stimata del test: 3.5 giorni
# Traffico totale necessario: 17,488 richiestePułapka podglądająca: nie patrz zbyt wcześnie na wyniki
Problem „podglądania” (lub opcjonalnego zatrzymywania) to jeden z najczęstszych błędów w testach A/B: patrzymy na wyniki pośrednie i zatrzymujemy test, gdy tylko zostanie osiągnięty poziom istotności statystyki. Jeśli spojrzysz na dane, drastycznie zwiększa się odsetek fałszywych alarmów Każdego dnia prawdopodobieństwo znalezienia znaczącego wyniku przez przypadek wzrasta do 30%, nawet jeśli oba warianty są identyczne. Zawsze używaj rozmiaru próbki z góry ustalone i sprawdzić wyniki dopiero na koniec testu, lub przyjąć metody testów sekwencyjnych, takich jak sekwencyjne testy współczynnika prawdopodobieństwa (SPRT).
Podział ruchu za pomocą FastAPI
Router do testów A/B jest centralnym elementem infrastruktury. Należy rozpowszechniać ruch w sposób deterministyczny (ten sam użytkownik musi zawsze korzystać z tego samego wariant przez cały czas trwania testu), zapisz, do jakiego wariantu zostałeś przydzielony każdego użytkownika i każdą prognozę, i bądź niezwykle szybki, aby nie powodować opóźnień na ścieżkę krytyczną.
# ab_router.py
# Router A/B testing per modelli ML con FastAPI
from fastapi import FastAPI, Request, Header
from pydantic import BaseModel
import hashlib
import json
import time
import logging
from typing import Optional, Literal
from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST
from starlette.responses import Response
logger = logging.getLogger(__name__)
app = FastAPI(title="ML A/B Testing Router")
# --- Prometheus Metrics ---
AB_REQUESTS = Counter(
"ab_test_requests_total",
"Numero totale di richieste per variante",
labelnames=["experiment_id", "variant", "model_version"]
)
AB_LATENCY = Histogram(
"ab_test_latency_seconds",
"Latenza inference per variante",
labelnames=["experiment_id", "variant"],
buckets=[0.01, 0.05, 0.1, 0.25, 0.5, 1.0]
)
AB_PREDICTIONS = Counter(
"ab_test_predictions_total",
"Distribuzione delle predizioni per variante",
labelnames=["experiment_id", "variant", "prediction_bucket"]
)
# --- Configurazione esperimento ---
ACTIVE_EXPERIMENT = {
"experiment_id": "churn_model_v2_vs_v3",
"model_a": {
"name": "churn-model-v2",
"endpoint": "http://model-a-service:8080/predict",
"traffic_weight": 0.5
},
"model_b": {
"name": "churn-model-v3",
"endpoint": "http://model-b-service:8080/predict",
"traffic_weight": 0.5
},
"start_time": "2025-03-01T00:00:00Z",
"end_time": "2025-03-15T00:00:00Z"
}
class PredictionRequest(BaseModel):
user_id: str
features: dict
def assign_variant(user_id: str, experiment_id: str, traffic_split: float = 0.5) -> str:
"""
Assegna deterministicamente un utente a una variante.
Lo stesso user_id + experiment_id producono sempre lo stesso risultato.
"""
hash_input = f"{user_id}:{experiment_id}"
hash_value = int(hashlib.md5(hash_input.encode()).hexdigest(), 16)
normalized = (hash_value % 10000) / 10000.0
if normalized < traffic_split:
return "A"
else:
return "B"
async def call_model(endpoint: str, features: dict) -> dict:
"""Chiama il servizio del modello."""
import httpx
async with httpx.AsyncClient(timeout=2.0) as client:
response = await client.post(endpoint, json=features)
response.raise_for_status()
return response.json()
@app.post("/predict")
async def predict(request: PredictionRequest):
"""
Endpoint principale: smista le richieste alle varianti A/B.
"""
exp = ACTIVE_EXPERIMENT
exp_id = exp["experiment_id"]
# Assegna variante in modo deterministico
variant = assign_variant(
user_id=request.user_id,
experiment_id=exp_id,
traffic_split=exp["model_a"]["traffic_weight"]
)
# Seleziona il modello corretto
model_config = exp["model_a"] if variant == "A" else exp["model_b"]
# Registra richiesta
AB_REQUESTS.labels(
experiment_id=exp_id,
variant=variant,
model_version=model_config["name"]
).inc()
# Chiama il modello con misura della latenza
start_time = time.time()
try:
result = await call_model(model_config["endpoint"], request.features)
except Exception as e:
logger.error(f"Errore chiamata modello {variant}: {e}")
raise
latency = time.time() - start_time
AB_LATENCY.labels(experiment_id=exp_id, variant=variant).observe(latency)
# Bucketing della predizione per distribuzione
score = result.get("churn_probability", 0)
bucket = "high" if score > 0.7 else ("medium" if score > 0.3 else "low")
AB_PREDICTIONS.labels(
experiment_id=exp_id, variant=variant, prediction_bucket=bucket
).inc()
return {
"prediction": result,
"variant": variant,
"model_version": model_config["name"],
"experiment_id": exp_id,
"latency_ms": round(latency * 1000, 2)
}
@app.get("/metrics")
async def metrics():
return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)
@app.get("/experiment/status")
async def experiment_status():
"""Ritorna lo stato corrente dell'esperimento."""
return {
"experiment": ACTIVE_EXPERIMENT["experiment_id"],
"active": True
}Wdrożenie w systemie Canary: wydanie progresywne
Il rozmieszczenie kanarków oraz strategia stopniowego wydawania, gdzie nowy model („kanarek”) początkowo otrzymuje tylko niewielki procent ruchu produkcyjnego, zazwyczaj 1-5%. Jeśli wskaźniki pozostaną stabilne, procent jest stopniowo zwiększany: 5% → 10% → 25% → 50% → 100%. W przypadku anomalii wycofanie jest natychmiastowe i przywraca cały ruch do stabilnego modelu.
W przeciwieństwie do klasycznego testu A/B 50/50, kanarek jest zorientowany redukcja ryzyka więcej niż statystyczne wykrywanie różnic. Celem nie jest demonstracja że nowy model jest lepszy pod względem istotności statystycznej, ale sprawdź, czy tak nie jest powodować zauważalne problemy techniczne lub regresje przed skalowaniem wdrożenia.
# canary_deployment.py
# Implementazione canary deployment con rollback automatico
import asyncio
import time
import logging
from dataclasses import dataclass, field
from typing import Optional
from prometheus_client import Gauge
logger = logging.getLogger(__name__)
# Gauge per monitorare la percentuale di traffico canary
CANARY_TRAFFIC_WEIGHT = Gauge(
"canary_traffic_weight_percent",
"Percentuale di traffico al modello canary",
labelnames=["experiment_id"]
)
ERROR_RATE_GAUGE = Gauge(
"canary_error_rate",
"Tasso di errore del modello canary",
labelnames=["experiment_id"]
)
@dataclass
class CanaryConfig:
experiment_id: str
stable_model_endpoint: str
canary_model_endpoint: str
initial_canary_weight: float = 0.05 # Inizia al 5%
max_canary_weight: float = 1.0 # Target finale: 100%
step_size: float = 0.10 # Incremento per ogni step
step_interval_minutes: int = 30 # Ogni 30 minuti aumenta
max_error_rate: float = 0.02 # Rollback se errori > 2%
max_latency_p99_ms: float = 500.0 # Rollback se P99 > 500ms
current_weight: float = field(init=False)
def __post_init__(self):
self.current_weight = self.initial_canary_weight
class CanaryController:
"""
Controlla progressivamente il traffico al modello canary.
Esegue rollback automatico se le metriche superano le soglie.
"""
def __init__(self, config: CanaryConfig):
self.config = config
self.error_count = 0
self.total_count = 0
self.latencies = []
self.is_rolled_back = False
self.is_promoted = False
def should_route_to_canary(self, user_id: str) -> bool:
"""Determina se questa richiesta va al canary."""
if self.is_rolled_back:
return False
hash_val = int(hashlib.md5(
f"{user_id}:{self.config.experiment_id}".encode()
).hexdigest(), 16)
normalized = (hash_val % 10000) / 10000.0
return normalized < self.config.current_weight
def record_outcome(self, is_canary: bool, success: bool, latency_ms: float):
"""Registra l'esito di una chiamata al canary."""
if not is_canary:
return
self.total_count += 1
if not success:
self.error_count += 1
self.latencies.append(latency_ms)
# Aggiorna metriche Prometheus
error_rate = self.error_count / max(self.total_count, 1)
ERROR_RATE_GAUGE.labels(
experiment_id=self.config.experiment_id
).set(error_rate)
# Controlla soglie per rollback automatico
if error_rate > self.config.max_error_rate and self.total_count > 100:
logger.critical(
f"Error rate {error_rate:.2%} exceeded threshold "
f"{self.config.max_error_rate:.2%}. Initiating rollback."
)
self.rollback()
if len(self.latencies) >= 100:
p99 = sorted(self.latencies)[-1] # semplificato
if p99 > self.config.max_latency_p99_ms:
logger.critical(f"P99 latency {p99:.0f}ms exceeded threshold. Rollback.")
self.rollback()
def advance_canary(self):
"""Incrementa il peso del canary se le metriche sono OK."""
if self.is_rolled_back or self.is_promoted:
return
new_weight = min(
self.config.current_weight + self.config.step_size,
self.config.max_canary_weight
)
self.config.current_weight = new_weight
CANARY_TRAFFIC_WEIGHT.labels(
experiment_id=self.config.experiment_id
).set(new_weight * 100)
logger.info(
f"Canary weight increased to {new_weight:.0%} "
f"for experiment {self.config.experiment_id}"
)
if new_weight >= self.config.max_canary_weight:
self.is_promoted = True
logger.info("Canary fully promoted to production!")
def rollback(self):
"""Esegue rollback immediato al modello stabile."""
self.config.current_weight = 0.0
self.is_rolled_back = True
CANARY_TRAFFIC_WEIGHT.labels(
experiment_id=self.config.experiment_id
).set(0)
logger.warning(f"ROLLBACK executed for {self.config.experiment_id}")Tryb cienia: testowanie bez wpływu na użytkowników
Lo tryb cienia (lub wdrażanie w tle) jest najbardziej konserwatywną techniką a jednocześnie najpotężniejszy w sprawdzaniu nowego modelu przed udostępnieniem go użytkownikom. Ruch produkcyjny jest zduplikowany: Model A obsługuje rzeczywiste i własne żądania prognozy są zwracane użytkownikom, podczas gdy model B otrzymuje te same żądania równolegle, ale przychodzą jego przewidywania odrzucone lub tylko zalogowane.
Takie podejście pozwala porównać oba modele na rzeczywistym ruchu bez żadnego ruchu ryzyko dla użytkowników lub firmy. Idealnie nadaje się do sprawdzenia, czy nowy model: nie posiada błędów krytycznych, spełnia wymogi latencji pod rzeczywistym obciążeniem, nie generuje prognoz anomalnych lub wykraczających poza dystrybucję i zachowuje się zgodnie z oczekiwaniami we wszystkich segmentach użytkowników.
# shadow_mode.py
# Implementazione shadow deployment con logging asincrono
import asyncio
import httpx
import logging
import json
from datetime import datetime
from typing import Any
logger = logging.getLogger(__name__)
class ShadowModeRouter:
"""
Router che invia le richieste sia al modello produzione che al modello shadow.
Il modello produzione risponde agli utenti; il shadow solo logga.
"""
def __init__(
self,
production_endpoint: str,
shadow_endpoint: str,
shadow_log_file: str = "shadow_predictions.jsonl"
):
self.production_endpoint = production_endpoint
self.shadow_endpoint = shadow_endpoint
self.shadow_log_file = shadow_log_file
async def predict(self, request_data: dict, request_id: str) -> dict:
"""
Invia la richiesta al modello produzione e in parallelo al shadow.
Restituisce solo la risposta del modello produzione.
"""
# Esegui produzione e shadow in parallelo
prod_task = asyncio.create_task(
self._call_model(self.production_endpoint, request_data, "production")
)
shadow_task = asyncio.create_task(
self._call_model(self.shadow_endpoint, request_data, "shadow")
)
# Aspetta la risposta produzione (non bloccante per shadow)
prod_result = await prod_task
# Logga la risposta shadow in background senza bloccare
asyncio.create_task(
self._log_shadow_result(shadow_task, request_id, request_data, prod_result)
)
return prod_result
async def _call_model(
self, endpoint: str, data: dict, label: str
) -> dict:
"""Chiama un endpoint modello con gestione degli errori."""
start = asyncio.get_event_loop().time()
try:
async with httpx.AsyncClient(timeout=2.0) as client:
response = await client.post(endpoint, json=data)
response.raise_for_status()
result = response.json()
result["_latency_ms"] = (asyncio.get_event_loop().time() - start) * 1000
result["_model"] = label
return result
except Exception as e:
logger.error(f"Error calling {label} model: {e}")
return {"error": str(e), "_model": label, "_latency_ms": -1}
async def _log_shadow_result(
self,
shadow_task: asyncio.Task,
request_id: str,
input_data: dict,
production_result: dict
):
"""Logga la risposta shadow per analisi offline."""
try:
shadow_result = await shadow_task
except Exception as e:
shadow_result = {"error": str(e)}
log_entry = {
"timestamp": datetime.utcnow().isoformat(),
"request_id": request_id,
"input_features": input_data,
"production_prediction": production_result.get("prediction"),
"production_latency_ms": production_result.get("_latency_ms"),
"shadow_prediction": shadow_result.get("prediction"),
"shadow_latency_ms": shadow_result.get("_latency_ms"),
"shadow_error": shadow_result.get("error"),
"predictions_agree": (
production_result.get("prediction") == shadow_result.get("prediction")
)
}
# Scrivi su file JSONL per analisi batch
with open(self.shadow_log_file, "a") as f:
f.write(json.dumps(log_entry) + "\n")
# --- Analisi dei risultati shadow ---
def analyze_shadow_results(log_file: str):
"""Analizza i log shadow per validare il nuovo modello."""
import pandas as pd
records = []
with open(log_file) as f:
for line in f:
records.append(json.loads(line))
df = pd.DataFrame(records)
total = len(df)
agreement_rate = df["predictions_agree"].mean()
shadow_errors = df["shadow_error"].notna().sum()
print(f"Totale richieste analizzate: {total:,}")
print(f"Tasso di accordo produzione/shadow: {agreement_rate:.1%}")
print(f"Errori modello shadow: {shadow_errors} ({shadow_errors/total:.1%})")
print(f"Latenza media produzione: {df['production_latency_ms'].mean():.1f}ms")
print(f"Latenza media shadow: {df['shadow_latency_ms'].mean():.1f}ms")
# Identifica i casi di disaccordo per analisi manuale
disagreements = df[~df["predictions_agree"]]
print(f"\nCasi di disaccordo: {len(disagreements)}")
return dfWieloręcy bandyci: poza klasycznymi testami A/B
Głównym ograniczeniem klasycznych testów A/B jest koszt eksploracji: na czas trwania testu ułamek użytkowników potencjalnie otrzyma model gorzej. Jeżeli model B jest wyraźnie lepszy, to „marnujemy” konwersje użytkowników przypisanych do A w tygodniach testowych.
I Wieloręcy bandyci (MAB) rozwiązują problem eksploracja-eksploatacja: zamiast utrzymywać stały podział przez cały czas trwania testu, algorytm dostosować się dynamicznie ruchu do działającego modelu lepiej, maksymalizując całkowitą liczbę konwersji podczas samego testu. Poszukiwanie z 2025 roku przez Aimpoint Digital Labs pokazuje, że podejścia bandyckie, takie jak Thompson Sampling może zmniejszyć skumulowany żal 20-35% w porównaniu do klasycznych testów A/B w scenariuszach z silnymi skutkami.
# thompson_sampling_bandit.py
# Multi-Armed Bandit con Thompson Sampling per selezione modello ML
import numpy as np
from dataclasses import dataclass, field
from typing import List, Tuple
import json
import logging
logger = logging.getLogger(__name__)
@dataclass
class ModelArm:
"""Rappresenta un modello come braccio del bandit."""
name: str
endpoint: str
alpha: float = 1.0 # Successi (Beta distribution prior)
beta: float = 1.0 # Fallimenti (Beta distribution prior)
@property
def estimated_success_rate(self) -> float:
"""Stima puntuale del tasso di successo (media della distribuzione Beta)."""
return self.alpha / (self.alpha + self.beta)
@property
def total_observations(self) -> int:
return int(self.alpha + self.beta - 2) # sottrai i prior
def sample(self) -> float:
"""Campiona dalla distribuzione Beta posteriore (Thompson Sampling)."""
return np.random.beta(self.alpha, self.beta)
def update(self, reward: float):
"""
Aggiorna la distribuzione con il nuovo outcome.
reward = 1.0 per successo (churn evitato, conversione, etc.)
reward = 0.0 per fallimento
"""
if reward >= 0.5: # successo
self.alpha += 1
else: # fallimento
self.beta += 1
class ThompsonSamplingBandit:
"""
Multi-Armed Bandit con Thompson Sampling.
Ottimale per selezione adattiva di modelli ML.
"""
def __init__(self, models: List[ModelArm]):
self.models = models
self.selection_history = []
def select_model(self) -> Tuple[int, ModelArm]:
"""
Seleziona il modello campionando dalle distribuzioni Beta.
Il modello con il sample più alto viene selezionato.
"""
samples = [arm.sample() for arm in self.models]
best_idx = int(np.argmax(samples))
self.selection_history.append(best_idx)
return best_idx, self.models[best_idx]
def update(self, arm_idx: int, reward: float):
"""Aggiorna la distribuzione del braccio selezionato."""
self.models[arm_idx].update(reward)
def get_traffic_allocation(self) -> dict:
"""
Stima la distribuzione del traffico corrente
basata sulla storia delle selezioni recenti.
"""
if not self.selection_history:
return {arm.name: 1/len(self.models) for arm in self.models}
recent = self.selection_history[-1000:] # ultime 1000 selezioni
total = len(recent)
allocation = {}
for i, arm in enumerate(self.models):
allocation[arm.name] = recent.count(i) / total
return allocation
def get_status(self) -> dict:
"""Ritorna lo stato corrente del bandit."""
return {
"models": [
{
"name": arm.name,
"estimated_rate": round(arm.estimated_success_rate, 4),
"alpha": arm.alpha,
"beta": arm.beta,
"observations": arm.total_observations
}
for arm in self.models
],
"traffic_allocation": self.get_traffic_allocation(),
"total_selections": len(self.selection_history)
}
def check_convergence(self, min_observations: int = 500) -> Optional[str]:
"""
Verifica se il bandit e convergito verso un vincitore chiaro.
Restituisce il nome del modello vincitore o None se ancora incerto.
"""
for arm in self.models:
if arm.total_observations < min_observations:
return None # Non abbastanza dati
# Controlla se un modello domina chiaramente
rates = [(arm.name, arm.estimated_success_rate) for arm in self.models]
rates.sort(key=lambda x: x[1], reverse=True)
best_name, best_rate = rates[0]
second_name, second_rate = rates[1]
# Margine di 3% di distanza per dichiarare un vincitore
if best_rate - second_rate > 0.03:
logger.info(f"Bandit converged: {best_name} wins ({best_rate:.2%} vs {second_rate:.2%})")
return best_name
return None
# --- Esempio di utilizzo ---
models = [
ModelArm(name="churn-model-v2", endpoint="http://model-v2:8080/predict"),
ModelArm(name="churn-model-v3", endpoint="http://model-v3:8080/predict"),
]
bandit = ThompsonSamplingBandit(models)
# Simulazione di 1000 interazioni
np.random.seed(42)
true_rates = {"churn-model-v2": 0.72, "churn-model-v3": 0.78} # v3 e migliore
for i in range(1000):
arm_idx, selected_model = bandit.select_model()
# Simula outcome (in produzione viene dal feedback reale)
reward = float(np.random.random() < true_rates[selected_model.name])
bandit.update(arm_idx, reward)
if (i + 1) % 200 == 0:
status = bandit.get_status()
print(f"\nStep {i+1}:")
for m in status["models"]:
print(f" {m['name']}: rate={m['estimated_rate']:.3f}, obs={m['observations']}")
winner = bandit.check_convergence(min_observations=100)
if winner:
print(f" => WINNER: {winner}")Analiza statystyczna: wartość p, przedziały ufności i wielkość efektu
Na koniec okresu testowego analiza statystyczna musi odpowiedzieć na trzy różne pytania: Czy zaobserwowana różnica jest istotna statystycznie? Jak duży jest efekt? Czy efekt ma praktyczne znaczenie dla biznesu?
# statistical_analysis.py
# Analisi statistica dei risultati di un A/B test ML
import numpy as np
import pandas as pd
from scipy import stats
from scipy.stats import norm, t
import math
from typing import Tuple, Optional
def analyze_ab_test_results(
conversions_a: int,
total_a: int,
conversions_b: int,
total_b: int,
alpha: float = 0.05
) -> dict:
"""
Analisi statistica completa di un A/B test su proporzioni.
Returns:
Dizionario con tutti i risultati statistici
"""
p_a = conversions_a / total_a
p_b = conversions_b / total_b
# --- Test z per differenza di proporzioni ---
# Pooled proportion sotto H0 (le due proporzioni sono uguali)
p_pool = (conversions_a + conversions_b) / (total_a + total_b)
se_pool = math.sqrt(p_pool * (1 - p_pool) * (1/total_a + 1/total_b))
z_statistic = (p_b - p_a) / se_pool
p_value = 2 * (1 - norm.cdf(abs(z_statistic))) # two-tailed
# --- Confidence Interval per la differenza ---
se_diff = math.sqrt(p_a * (1-p_a)/total_a + p_b * (1-p_b)/total_b)
z_critical = norm.ppf(1 - alpha/2)
diff = p_b - p_a
ci_lower = diff - z_critical * se_diff
ci_upper = diff + z_critical * se_diff
# --- Effect size (Cohen's h per proporzioni) ---
phi_a = 2 * math.asin(math.sqrt(p_a))
phi_b = 2 * math.asin(math.sqrt(p_b))
cohens_h = phi_b - phi_a
effect_magnitude = (
"negligible" if abs(cohens_h) < 0.2
else "small" if abs(cohens_h) < 0.5
else "medium" if abs(cohens_h) < 0.8
else "large"
)
# --- Relative lift ---
relative_lift = (p_b - p_a) / p_a if p_a > 0 else 0
# --- Potenza statistica osservata ---
z_beta = (abs(z_statistic) - z_critical)
observed_power = norm.cdf(z_beta)
is_significant = p_value < alpha
return {
"variant_a": {
"conversions": conversions_a,
"total": total_a,
"rate": round(p_a, 4),
"rate_pct": f"{p_a:.2%}"
},
"variant_b": {
"conversions": conversions_b,
"total": total_b,
"rate": round(p_b, 4),
"rate_pct": f"{p_b:.2%}"
},
"difference": {
"absolute": round(diff, 4),
"relative_lift": round(relative_lift, 4),
"relative_lift_pct": f"{relative_lift:.2%}",
"confidence_interval_95": (round(ci_lower, 4), round(ci_upper, 4))
},
"statistics": {
"z_statistic": round(z_statistic, 4),
"p_value": round(p_value, 6),
"is_significant": is_significant,
"alpha": alpha,
"cohens_h": round(cohens_h, 4),
"effect_magnitude": effect_magnitude,
"observed_power": round(observed_power, 4)
},
"conclusion": (
f"Modello B e statisticamente migliore (p={p_value:.4f}, lift={relative_lift:.2%})"
if is_significant and diff > 0
else f"Nessuna differenza significativa rilevata (p={p_value:.4f})"
)
}
# --- Esempio pratico ---
results = analyze_ab_test_results(
conversions_a=1380, # Modello A: 1380 churn evitati
total_a=8500, # su 8500 utenti a rischio
conversions_b=1545, # Modello B: 1545 churn evitati
total_b=8200 # su 8200 utenti
)
print("=== RISULTATI A/B TEST ===")
print(f"Modello A: {results['variant_a']['rate_pct']} retention rate")
print(f"Modello B: {results['variant_b']['rate_pct']} retention rate")
print(f"Lift relativo: {results['difference']['relative_lift_pct']}")
print(f"CI 95%: {results['difference']['confidence_interval_95']}")
print(f"p-value: {results['statistics']['p_value']}")
print(f"Significativo: {results['statistics']['is_significant']}")
print(f"Effect size: {results['statistics']['effect_magnitude']} (h={results['statistics']['cohens_h']})")
print(f"\nConclusione: {results['conclusion']}")Bayesowskie testy A/B
Częstotliwe podejście oparte na wartości p ma znane ograniczenia: wartość p nie jest prawdopodobieństwem że model B jest lepszy (i prawdopodobieństwo zaobserwowania takich ekstremalnych danych gdyby H0 było prawdą). Podejście Bayesowski odpowiada bezpośrednio na interesujące nas pytanie: jakie jest prawdopodobieństwo, że model B jest lepszy od A, i o ile?
Podejście Bayesa pozwala również zatrzymać test, gdy osiągnie on wartość a wystarczająco duże prawdopodobieństwo (np. 95%), że model jest najlepszy, bez typowego dla częstości problemu zaglądania.
# bayesian_ab_test.py
# A/B Testing Bayesiano per modelli ML
import numpy as np
from scipy import stats
from scipy.stats import beta as beta_dist
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
def bayesian_ab_test(
successes_a: int, trials_a: int,
successes_b: int, trials_b: int,
prior_alpha: float = 1.0,
prior_beta: float = 1.0,
n_samples: int = 100_000,
credible_interval: float = 0.95
) -> dict:
"""
A/B test bayesiano usando distribuzione Beta come prior/posterior.
Modella il tasso di successo come Beta(alpha, beta).
Returns:
Risultati con probabilità che B > A e credible intervals
"""
# Aggiorna i prior con i dati osservati (prior Beta + dati binomiali = posterior Beta)
alpha_a = prior_alpha + successes_a
beta_a = prior_beta + (trials_a - successes_a)
alpha_b = prior_alpha + successes_b
beta_b = prior_beta + (trials_b - successes_b)
# Campiona dalle distribuzioni posterior
samples_a = np.random.beta(alpha_a, beta_a, n_samples)
samples_b = np.random.beta(alpha_b, beta_b, n_samples)
# Probabilità che B sia migliore di A
prob_b_better = np.mean(samples_b > samples_a)
# Distribuzione del lift relativo
lift_samples = (samples_b - samples_a) / samples_a
lift_mean = np.mean(lift_samples)
lift_std = np.std(lift_samples)
# Credible interval per il lift
ci_lower = float(np.percentile(lift_samples, (1 - credible_interval) / 2 * 100))
ci_upper = float(np.percentile(lift_samples, (1 - (1 - credible_interval) / 2) * 100))
# Probabilità di un lift minimo (es. almeno +2%)
prob_lift_2pct = np.mean(lift_samples > 0.02)
# Expected loss: quanto perdiamo se scegliamo il modello sbagliato
expected_loss_a = np.mean(np.maximum(samples_b - samples_a, 0)) # perdita se scegliamo A
expected_loss_b = np.mean(np.maximum(samples_a - samples_b, 0)) # perdita se scegliamo B
return {
"posterior_a": {"alpha": alpha_a, "beta": beta_a, "mean": alpha_a/(alpha_a+beta_a)},
"posterior_b": {"alpha": alpha_b, "beta": beta_b, "mean": alpha_b/(alpha_b+beta_b)},
"prob_b_better_than_a": round(float(prob_b_better), 4),
"lift": {
"mean": round(float(lift_mean), 4),
"std": round(float(lift_std), 4),
f"credible_interval_{int(credible_interval*100)}pct": (
round(ci_lower, 4), round(ci_upper, 4)
),
"prob_lift_above_2pct": round(float(prob_lift_2pct), 4)
},
"expected_loss": {
"choose_a": round(float(expected_loss_a), 6),
"choose_b": round(float(expected_loss_b), 6),
"recommended_choice": "B" if expected_loss_b < expected_loss_a else "A"
},
"decision": (
"Scegli B" if prob_b_better > 0.95
else "Scegli A" if prob_b_better < 0.05
else f"Incerto (P(B>A) = {prob_b_better:.1%}) - continua a raccogliere dati"
)
}
# --- Esempio ---
result = bayesian_ab_test(
successes_a=1380, trials_a=8500,
successes_b=1545, trials_b=8200,
credible_interval=0.95
)
print("=== A/B TEST BAYESIANO ===")
print(f"P(B > A) = {result['prob_b_better_than_a']:.1%}")
print(f"Lift medio: {result['lift']['mean']:.2%}")
print(f"Credible interval 95%: {result['lift']['credible_interval_95pct']}")
print(f"P(lift > 2%): {result['lift']['prob_lift_above_2pct']:.1%}")
print(f"Expected loss se scegli A: {result['expected_loss']['choose_a']:.6f}")
print(f"Expected loss se scegli B: {result['expected_loss']['choose_b']:.6f}")
print(f"Decisione: {result['decision']}")Monitorowanie podczas testów z Prometheusem i Grafaną
Aktywny test A/B w produkcji musi być stale monitorowany. To nie wystarczy poczekaj do końca testu, aby przeanalizować wyniki: należy zagwarantować, że oba warianty działają poprawnie na poziomie technicznym (opóźnienie, poziom błędów, dostępność) oraz że wskaźniki biznesowe są zgodne z początkowymi oczekiwaniami.
# ab_test_monitoring.yml
# Dashboard Grafana per A/B test ML - configurazione panel
# Esempio di PromQL queries per i panel Grafana:
# 1. Distribuzione del traffico tra varianti (dovrebbe essere ~50/50)
# sum by (variant) (rate(ab_test_requests_total[5m]))
# 2. Latenza P95 per variante
# histogram_quantile(0.95, sum by (variant, le) (rate(ab_test_latency_seconds_bucket[5m])))
# 3. Error rate per variante
# sum by (variant) (rate(ab_test_errors_total[5m])) /
# sum by (variant) (rate(ab_test_requests_total[5m]))
# 4. Distribuzione delle predizioni per variante (prediction drift indicator)
# sum by (variant, prediction_bucket) (rate(ab_test_predictions_total[1h]))
---
# prometheus_ab_alerts.yml
groups:
- name: ab_test_alerts
rules:
# Alert se il traffico non e bilanciato (sbilanciamento > 10%)
- alert: ABTestTrafficImbalance
expr: |
abs(
sum(rate(ab_test_requests_total{variant="A"}[10m]))
/
sum(rate(ab_test_requests_total[10m]))
- 0.5
) > 0.10
for: 5m
labels:
severity: warning
annotations:
summary: "A/B test traffic imbalance detected"
description: "Traffic split deviates more than 10% from 50/50"
# Alert se error rate variante B supera il doppio di A
- alert: ABTestVariantBHighErrors
expr: |
(
sum(rate(ab_test_errors_total{variant="B"}[5m]))
/
sum(rate(ab_test_requests_total{variant="B"}[5m]))
) > 2 * (
sum(rate(ab_test_errors_total{variant="A"}[5m]))
/
sum(rate(ab_test_requests_total{variant="A"}[5m]))
)
for: 10m
labels:
severity: critical
annotations:
summary: "Variant B has significantly higher error rate than A"
description: "Consider rolling back variant B"
# Alert se latenza P95 di B supera 200ms più di A
- alert: ABTestVariantBHighLatency
expr: |
(
histogram_quantile(0.95, sum by (le) (
rate(ab_test_latency_seconds_bucket{variant="B"}[5m])
))
-
histogram_quantile(0.95, sum by (le) (
rate(ab_test_latency_seconds_bucket{variant="A"}[5m])
))
) > 0.2
for: 5m
labels:
severity: warning
annotations:
summary: "Variant B P95 latency is 200ms+ higher than A"Budżet <5 tys. EUR/rok dla MŚP: pełne testy stosu A/B
Kompletny system testów A/B dla modeli ML nie wymaga budżetu przedsiębiorstwa. Dzięki stosowi open source i małemu VPS możesz mieć wszystko, czego potrzebujesz:
- Router FastAPI + statystyki Pythona: Oprogramowanie typu open source, bezpłatne
- Prometeusz + Grafana: Oprogramowanie typu open source, bezpłatne
- VPS dla hostingu (Hetzner/OVH): 20-40 EUR/miesiąc (240-480 EUR/rok)
- Usługa flag funkcji (samodzielny hosting Unleash): Oprogramowanie typu open source, bezpłatne
- MLflow dla rejestru modeli: Oprogramowanie typu open source, bezpłatne
- Całkowita szacunkowa infrastruktura: 300-600 EUR/rok
Najlepsze praktyki i anty-wzorce
Lista kontrolna przed eksperymentem
- Przed rozpoczęciem oblicz wielkość próbki: nigdy nie rób testów „dopóki Nie widzę nic ciekawego.” Wielkość próbki jest stała i nie podlega negocjacjom.
- Zdefiniuj JEDNĄ podstawową metrykę: optymalizuj pod kątem dwóch wskaźników jednocześnie czyni decyzję niejednoznaczną. Istnieją metryki poręczy zapobiegać regresjom, a nie wybierać zwycięzców.
- Sprawdź ważność zadania: przed rozpoczęciem prawdziwego testu, wykonaj test A/A (ten sam model w obu wariantach), aby sprawdzić, czy tak nie jest w zadaniu występują błędy, które powodują sztuczne różnice.
- Udokumentuj swoje hipotezy: zauważ, dlaczego oczekujemy modelu B jest lepszy i o ile. Pozwala to uniknąć stronniczości „Widziałem odpowiedź i teraz”. Wymyślę wyjaśnienie.”
- Sprawdź nowych użytkowników osobno: nowi użytkownicy nie mają w poprzedniej historii nie mieli żadnego z modeli i zachowywali się inaczej. Analizuj ich wyniki oddzielnie.
Anty-wzorce, których należy bezwzględnie unikać
- Ciągłe podglądanie: sprawdzaj wyniki codziennie, np zatrzymanie testu przy pierwszej istotności statystycznej zwiększa liczbę wyników fałszywie dodatnich do 30%. Jeśli potrzebujesz wcześniejszego zatrzymania, użyj testów sekwencyjnych (SPRT).
- HARKing (hipotezowanie po poznaniu wyników): analizować danych, aby znaleźć istotne różnice, a następnie opowiedz historię jakby zostało to postawione a priori. W teście 20 segmentów jeden będzie się wyróżniał istotne tylko przypadkowo przy alfa = 0,05.
- Ignoruj wariancję metryki: niektóre wskaźniki, takie jak przychodów na użytkownika mają bardzo duże kolejki. Może to zrobić pojedynczy użytkownik wieloryba efekt, który nie istnieje, wydaje się znaczący. Użyj bootstrapu lub nietestowania parametryczny dla metryk z rozkładem innym niż Gaussa.
- Testy są za krótkie: efekty tygodniowe (użytkownicy korzystający z usługa tylko w poniedziałki) i efekt nowości (pozytywnie reagują użytkownicy do nowości na 1-2 dni, a następnie powrót do wartości wyjściowych) wymagają przetestowania co najmniej 2 tygodnie na odszkodowanie.
- Pętle sprzężenia zwrotnego w systemach ML: w układach z pętlami sprzężenia zwrotnego (zalecenia zmieniające zachowania użytkowników), przewidywania dwa warianty nie są niezależne. Modeluj tę korelację jawnie.
Kiedy stosować które podejście
Przewodnik po wyborze strategii
- Tryb cienia: stosować, gdy model jest zupełnie nowy, jeszcze niesprawdzone lub gdy ryzyko wystąpienia błędu jest zbyt wysokie. I zawsze pierwszy krok przed jakimikolwiek testami z prawdziwymi użytkownikami.
- Wdrożenie na Wyspach Kanaryjskich: użyj, gdy chcesz zmniejszyć ryzyko operacyjne nowego wdrożenia. Doskonały do modeli krytycznych (oszustwa, ustalanie cen), gdzie regresja miałaby natychmiastowe skutki finansowe.
- Klasyczny test A/B (50/50): użyj, jeśli chcesz zmierzyć efekt biznes o maksymalnej mocy statystycznej i niskim ryzyku operacyjnym. Wymaga wystarczającej wielkości próbki i szybkiej pętli sprzężenia zwrotnego.
- Wieloręki bandyta: stosować, gdy informacja zwrotna jest szybka (w ciągu godzin/dni), koszt eksploracji jest wysoki i wolisz maksymalizować konwersje podczas testu. Nie jest idealny do małych efektów z wolnym sprzężeniem zwrotnym.
- Bayesowski A/B: korzystaj z elastycznych reguł zatrzymywania kiedy tylko chcesz, interpretować prawdopodobieństwa bezpośrednio lub mieć wcześniejsze informacje z eksperymentów precedensy. Idealny dla zespołów, dla których wartości p są mylące.
Wnioski i dalsze kroki
Testy A/B dla modeli ML to znacznie więcej niż zwykły podział ruchu. Wymaga rygorystycznego projektowania statystycznego przed każdą implementacją, wyboru właściwa strategia oparta na kontekście (cień, kanarek, 50/50, bandyta), ciągłe monitorowanie w trakcie testu i poprawna analiza statystyczna na koniec.
Różnica między zespołem, który poprawnie przeprowadza testy A/B, a zespołem, który robi to źle nie chodzi o złożoność kodu, ale o dyscyplinę procesu: definiowanie najpierw hipotezy, nie patrz na dane w trakcie, dopiero potem wszystko poprawnie przeanalizuj. Dzięki stosowi open source opisanemu w tym przewodniku (FastAPI, Prometheus, Grafana, scipy, numpy), możesz wdrożyć system klasy produkcyjnej przy minimalnym budżecie.
Naturalnym kolejnym krokiem jest zintegrowanie testów A/B z zarządzaniem ML: każda decyzja o wprowadzeniu modelu do produkcji musi być udokumentowana, podlegających audytowi i zgodnych ze standardami etycznymi i regulacyjnymi. Zobaczymy to w artykule następny w temacie ML Governance.
Seria MLOps trwa
- Poprzedni artykuł: Skalowanie ML na Kubernetesie - koordynuj wdrożenia za pomocą KubeFlow i Seldon Core
- Następny artykuł: Zarządzanie ML: zgodność, audyt, etyka - Ustawa o sztucznej inteligencji UE, wyjaśnialność i uczciwość
- Powiązany: Wykrywanie dryfu modelu i automatyczne ponowne uczenie - wykrywać i reagować na degradację modelu
- Powiązany: Obsługa modeli: FastAPI + Uvicorn w produkcji - budować skalowalne interfejsy API wnioskowania
- Powiązane serie: Zaawansowane głębokie uczenie się - Testy A/B dla złożonych modeli neuronowych