MLOps: deney takibi, model kaydı, servisleme ve izleme

Model eğitmek işin yarısı bile değil

Pedal İzmir veri ekibi Haziran verisiyle saatlik bisiklet talebini tahmin eden bir model eğitti; doğrulama MAE'si iyi, herkes memnun. Temmuz'da İzmir'i sıcak hava dalgası vuruyor. Model hiç görmediği sıcaklıklarla karşılaşıyor ve tahminleri sistematik olarak yukarı kayıyor; istasyonlara gereğinden çok bisiklet taşınıyor. Kimse fark etmiyor çünkü kimse izlemiyor. Üstelik "hangi model canlıda, hangi veriyle, hangi parametrelerle eğitildi?" sorusunun cevabı bir not defterinin 47. hücresinde.

MLOps, makine öğrenmesi modellerini tekrarlanabilir biçimde eğitmek, sürümlemek, yayına almak ve canlıda izlemek için kullanılan uygulamalar bütünüdür. Bu derste yaşam döngüsünün her adımını tarayıcıda çalışan küçük araçlarla (JSON tabanlı deney kaydı, model kayıt defteri, PSI/KS kayma testleri, pydantic şema doğrulama) kuracak, gerçek araçların (MLflow, DVC, FastAPI) kodunu da göreceksin. Ön işleme, pipeline ve model değerlendirme dersinde modeli joblib ile kaydetmeyi, Docker ve dağıtım: konteyner, compose, bulut dersinde kapsayıcıyı gördün; burada bunları birbirine bağlıyoruz.

Yaşam döngüsü ve olgunluk seviyeleri

Bir ML sistemi şu döngüden geçer: problem tanımı → veri toplama ve sürümleme → özellik üretimi → deney ve eğitim → değerlendirme ve kalite kapısı → kayıt ve yayın → servisleme → izleme → (tetiklenirse) yeniden eğitim. Ekiplerin bu döngüyü ne kadar otomatik yürüttüğü olgunluk seviyesi olarak anılır:

Çoğu ekip için hedef seviye 1'dir: her eğitim tekrarlanabilir, her canlı modelin kökeni bellidir.

Deney takibi: parametre, metrik, artefakt

Her eğitim koşusu (run) üç şey üretir: parametreler (alpha=1.0, özellik listesi, veri sürümü), metrikler (doğrulama MAE'si) ve artefaktlar (model dosyası, grafik, özellik önemi tablosu). Bunları elle bir tabloya yazmak yerine bir izleme aracı kullanırsın. Endüstri standardı MLflow'dur:

import mlflow
from sklearn.linear_model import Ridge

mlflow.set_experiment("pedal-talep")
with mlflow.start_run(run_name="ridge-a1"):
    mlflow.log_params({"alpha": 1.0, "veri_surumu": "a3f9c2"})
    model = Ridge(alpha=1.0).fit(X_tr, y_tr)
    mlflow.log_metric("mae_dogrulama", 15.27)
    mlflow.sklearn.log_model(model, "model", input_example=X_tr.head(2))

import json, time

class DeneyKaydi:
    def __init__(self, ad):
        self.ad, self.kosular = ad, []

    def baslat(self, params):
        kosu = {"id": f"kosu-{len(self.kosular) + 1}", "params": dict(params), "metrikler": {}}
        self.kosular.append(kosu)
        return kosu["id"]

    def metrik(self, kosu_id, ad, deger):
        next(k for k in self.kosular if k["id"] == kosu_id)["metrikler"][ad] = deger

    def en_iyi(self, metrik, buyuk_iyi=False):
        adaylar = [k for k in self.kosular if metrik in k["metrikler"]]
        return (max if buyuk_iyi else min)(adaylar, key=lambda k: k["metrikler"][metrik])

kayit = DeneyKaydi("pedal-talep")
for alpha, mae in [(0.1, 15.9), (1.0, 15.3), (10.0, 16.4)]:
    kid = kayit.baslat({"alpha": alpha})
    kayit.metrik(kid, "mae", mae)
print(kayit.en_iyi("mae"))
print(json.dumps(kayit.kosular[0], ensure_ascii=False))

Model kayıt defteri, aşamalar ve imza

Deney takibi "ne denedik?" sorusunu, model kayıt defteri (model registry) "canlıda ne var?" sorusunu cevaplar. Her kayıtlı model bir ad ve artan sürüm numarası taşır; sürümlere takma ad (alias) verilir: canlıdaki sürüm @champion, deneme kuyruğundaki @challenger. Servis modeli models:/talep@champion adresinden yükler; yeni sürüme geçmek ya da geri dönmek (rollback) takma adı başka sürüme taşımaktan ibarettir, kod değişmez. Eski anlatımlarda None → Staging → Production → Archived aşamalarını (stages) görürsün: MLflow bunları 2.9'dan beri kullanımdan kaldırıyor, yeni projede alias kullan.

Model imzası (signature) modelin beklediği girdi sütunlarını, türlerini ve çıktısını kaydeder. Servis tarafı gelen isteği bu imzayla karşılaştırır; sütun sırası ya da türü kayarsa model sessizce yanlış tahmin üretmek yerine açık bir hata verir. Bu, eğitim-servis çarpıklığına (training-serving skew) karşı ilk savunma hattıdır.

Veri ve model sürümleme

"Bu model hangi veriyle eğitildi?" sorusunun tek güvenilir cevabı verinin içerik özetidir (hash). DVC (Data Version Control) büyük veri dosyalarını Git dışında (S3, disk) tutar ve Git'e yalnız dosyanın MD5 özetini içeren küçük bir .dvc dosyası koyar. Kod commit'i ile veri özeti birlikte bir modeli tam olarak yeniden üretmeye yeter.

import hashlib
import veri

def parmak_izi(yol, n=12):
    h = hashlib.sha256()
    with open(yol, "rb") as f:
        for parca in iter(lambda: f.read(65536), b""):
            h.update(parca)
    return h.hexdigest()[:n]

for tablo in ["bk_weather", "bk_stations"]:
    print(tablo, parmak_izi(veri.yol("pedal_izmir", tablo)))

dvc init
dvc add data/bk_trips.csv          # data/bk_trips.csv.dvc oluşur (md5 + boyut)
git add data/bk_trips.csv.dvc .gitignore
git commit -m "veri: Haziran-Ağustos yolculukları"
dvc push                           # dosyayı uzak depoya (S3, GCS, disk) gönderir

Servisleme: toplu mu, çevrim içi mi?

from fastapi import FastAPI
from pydantic import BaseModel, Field
import joblib, pandas as pd

app = FastAPI()
model = joblib.load("model_v3.joblib")            # süreç başında bir kez yükle
OZELLIKLER = ["saat", "haftasonu", "temp_c", "rain_mm", "wind_kmh"]

class TalepIstegi(BaseModel):
    saat: int = Field(ge=0, le=23)
    haftasonu: bool
    temp_c: float = Field(ge=-15, le=50)
    rain_mm: float = Field(ge=0)
    wind_kmh: float = Field(ge=0)

@app.post("/predict")
def predict(istek: TalepIstegi):
    satir = pd.DataFrame([istek.model_dump()])[OZELLIKLER]   # sıra imzadan
    return {"talep": float(model.predict(satir)[0]), "model_surumu": "v3"}

from pydantic import BaseModel, Field, ValidationError

class TalepIstegi(BaseModel):
    saat: int = Field(ge=0, le=23)
    temp_c: float = Field(ge=-15, le=50)

print(TalepIstegi(saat=8, temp_c=31.5))
try:
    TalepIstegi(saat=26, temp_c="sıcak")
except ValidationError as e:
    print(e.error_count(), "hata:", [h["loc"][0] for h in e.errors()])

İzleme: veri kayması, tahmin dağılımı, gecikmeli etiket

Canlı model üç katmanda izlenir:

1. Girdi (veri) kayması (data drift): özelliklerin dağılımı değişti mi? Ölçüler: PSI (Population Stability Index) ve KS testi (Kolmogorov-Smirnov).
2. Tahmin dağılımı: modelin çıktısı kaydı mı? Etiket gelmeden bile görülebilir; en erken alarmdır.
3. Performans: gerçek etiketler geldiğinde (bisiklette bir saat sonra, kredide aylar sonra: gecikmeli etiket) MAE, AUC gibi metrikleri yeniden hesapla.

Bir de kavram kayması (concept drift) var: girdiler aynı kalsa bile girdi ile hedef arasındaki ilişki değişir (ör. yeni bir metro hattı açılınca aynı saatte talep düşer). Bunu yalnız performans izlemesi yakalar.

PSI, referans dağılımı kovalara (genellikle referansın 10 yüzdelik dilimi) böler ve her kovada referans oranı r ile güncel oran g için Σ (g − r) · ln(g / r) toplar. Yaygın yorum: < 0.1 kararlı, 0.1–0.25 orta kayma (incele), > 0.25 büyük kayma (aksiyon). Boş kovada ln(0) patlamasın diye oranlar küçük bir değerle (1e-4) alttan kırpılır.

import numpy as np
import pandas as pd
from scipy.stats import ks_2samp

hava = veri.oku("pedal_izmir", "bk_weather", parse_dates=["obs_hour"])
haziran = hava[hava["obs_hour"].dt.month == 6]["temp_c"]
temmuz = hava[hava["obs_hour"].dt.month == 7]["temp_c"]
agustos = hava[hava["obs_hour"].dt.month == 8]["temp_c"]

def psi(ref, guncel, kova=10):
    sinir = np.unique(np.quantile(ref, np.linspace(0, 1, kova + 1)))
    sinir[0], sinir[-1] = -np.inf, np.inf
    r = np.histogram(ref, sinir)[0] / len(ref)
    g = np.histogram(guncel, sinir)[0] / len(guncel)
    r, g = np.clip(r, 1e-4, None), np.clip(g, 1e-4, None)
    return float(np.sum((g - r) * np.log(g / r)))

for ad, ay in [("Temmuz", temmuz), ("Ağustos", agustos)]:
    ks = ks_2samp(haziran, ay)
    print(f"{ad}: PSI={psi(haziran, ay):.3f}  KS={ks.statistic:.3f}  p={ks.pvalue:.4f}")

Temmuz'da sıcaklık PSI'ı 1.35: büyük bir kayma. Ağustos'ta hava normale döndüğü için PSI 0.02. KS istatistiği (0.24) daha ılımlı görünüyor; çünkü PSI, Haziran'ın serin gecelerine ait kovaların Temmuz'da boş kalmasını ağır cezalandırır (kırpılan 1e-4 oranının logu çok büyüktür). İki ölçüyü birlikte okumak bu yüzden faydalıdır. KS testi de aynı şeyi söyler; ama dikkat: binlerce satırda KS p-değeri en küçük farkta bile 0'a iner. Bu yüzden üretimde p-değerinden çok etki büyüklüğüne (KS istatistiği, PSI) ve iş eşiğine bakılır.

Yeniden eğitim ve güvenli yayın

Yeniden eğitim tetikleyicileri: takvim (her pazartesi), veri hacmi (son eğitimden beri 50 bin yeni etiket), kayma (PSI > 0.25 üç gün üst üste) ya da performans (MAE referansın %20 üstünde iki gün). Tek günlük sıçramaya tepki vermemek için "ardışık gün" kuralı kullanılır.

Güvenli yayın kalıpları: shadow (gölge) modda yeni model gerçek trafiği alır, tahminini kaydeder ama kullanıcıya eski modelin cevabı döner; risksiz karşılaştırma yapılır. Canary (kanarya) modda kullanıcıların küçük bir yüzdesi (%5) yeni modele yönlendirilir; metrikler iyiyse oran artırılır. Kullanıcı kimliğinin hash'iyle yönlendirmek, aynı kullanıcının hep aynı modeli görmesini sağlar. A/B testi ise iş metriğindeki farkı istatistiksel olarak ölçer (İstatistik ve hipotez testi dersi).

ML için CI: davranış testleri ve kalite kapıları

Klasik birim testleri kodu sınar; model için ek olarak davranış testleri yazılır:

• Yön testi (directional): yağış artınca talep tahmini düşmeli.
• Değişmezlik testi (invariance): kullanıcının adı ya da cinsiyeti değişince kredi skoru değişmemeli.
• Minimum işlev testi: bilinen basit örneklerde beklenen sonuç (sabah 8'de hafta içi talep gece 3'ten yüksek).

Kalite kapısı (quality gate) CI hattında yeni modeli canlıdakiyle karşılaştırır ve kurallar sağlanmazsa yayını durdurur: "aday MAE'si mevcut MAE'den en fazla %2 kötü olabilir", "p95 gecikme 50 ms altında", "model dosyası 100 MB altında".

from sklearn.linear_model import Ridge

hava["talep_ornek"] = 20 + 0.9 * hava["temp_c"] - 13 * hava["rain_mm"]
model = Ridge().fit(hava[["temp_c", "rain_mm"]], hava["talep_ornek"])
X = hava[["temp_c", "rain_mm"]].head(200)
yagisli = X.assign(rain_mm=X["rain_mm"] + 2)
oran = (model.predict(yagisli) < model.predict(X)).mean()
print("Yağış artınca talep düşen satır oranı:", oran)
assert oran == 1.0, "yön testi başarısız"