MLOps: utrullning, versioner, rollback

En modell i drift är inte ett artefaktfilnamn utan en version med en livscykel.

Utrullningsstrategier:

Shadow först, sedan canary är den vanligaste kombinationen: shadow visar att den nya modellen inte kraschar och hur dess svar skiljer sig, canary visar hur användarna faktiskt reagerar.

Det viktigaste kravet: återgången ska ta sekunder, inte en omdeploy. Är rollback ett aliasbyte vågar man rulla ut oftare — och det är i sig den största säkerhetsförbättringen.

Vad som ska versioneras tillsammans. En «modellversion» som bara är viktfilen räcker inte:

Att bara byta viktfil är ett klassiskt fel: en ny modell med gammal förbehandling ger tyst sämre resultat, eftersom inget går sönder.

Träning/serverings-skevhet är samma fenomen i en annan form: förbehandlingen i träningen och i produktionen implementeras på två olika ställen och glider isär. Det enda robusta motmedlet är att dela kod — samma funktion används på båda ställena, testad mot samma fixtur.

Vad som övervakas i drift:

Utdatafördelningen är den bästa tidiga varningen. Kvalitetsmått kräver ofta facit som dröjer dagar eller veckor; att andelen positiva prediktioner plötsligt går från 4 % till 11 % märks samma dag.

Automatisk rollback bör utlösas av tydliga tröskelbrott:

Modellregister. Varje version har ett stadium: staging → produktion → arkiverad. Övergångarna loggas med vem som beslutade och varför. Det är samma spårbarhetskrav som AI-förordningen ställer på högrisksystem, så det är arbete som ändå måste göras.

import json, time
from dataclasses import dataclass, asdict, field
from pathlib import Path

@dataclass
class Modellversion:
    version: str
    vikter_sha: str
    forbehandling_sha: str
    config: dict
    datahash: str
    git_commit: str
    beroenden: dict
    stadium: str = "staging"          # staging | produktion | arkiverad
    skapad: float = field(default_factory=time.time)

class Register:
    def __init__(self, fil="register.jsonl"):
        self.fil = Path(fil)

    def registrera(self, mv: Modellversion, av: str, anledning: str):
        with self.fil.open("a", encoding="utf-8") as f:
            f.write(json.dumps({**asdict(mv), "av": av, "anledning": anledning},
                               ensure_ascii=False) + "\n")

    def befordra(self, version: str, till: str, av: str, anledning: str):
        self.registrera(Modellversion(version=version, vikter_sha="", forbehandling_sha="",
                                      config={}, datahash="", git_commit="", beroenden={},
                                      stadium=till), av, anledning)

# Shadow: kör den nya modellen parallellt, använd inte svaren
async def shadow(begaran, produktion, kandidat, logg):
    svar = await produktion(begaran)
    try:
        kandidat_svar = await kandidat(begaran)
        logg.append({"lika": svar == kandidat_svar,
                     "prod": svar, "kandidat": kandidat_svar})
    except Exception as e:
        logg.append({"fel": str(type(e).__name__)})
    return svar                         # användaren får ALLTID produktionssvaret

# Canary med automatisk rollback
class Canary:
    def __init__(self, baslinje, steg=(0.01, 0.05, 0.25, 1.0)):
        self.baslinje = baslinje        # {"felfrekvens": .., "p95_ms": .., "positiv_andel": ..}
        self.steg = steg
        self.index = 0

    def andel(self):
        return self.steg[self.index]

    def utvardera(self, aktuell: dict) -> tuple[str, str]:
        if aktuell["felfrekvens"] > 2 * self.baslinje["felfrekvens"]:
            return "rollback", "felfrekvensen har fördubblats"
        if aktuell["p95_ms"] > 2 * self.baslinje["p95_ms"]:
            return "rollback", "p95-latensen har fördubblats"
        avvikelse = abs(aktuell["positiv_andel"] - self.baslinje["positiv_andel"])
        if avvikelse > 0.5 * self.baslinje["positiv_andel"]:
            return "rollback", f"utdatafördelningen avviker {avvikelse:.3f}"
        if self.index + 1 < len(self.steg):
            self.index += 1
            return "fortsatt", f"höjer till {self.steg[self.index]:.0%}"
        return "klar", "full utrullning"

# Rollback ska vara ett aliasbyte, inte en omdeploy
def rulla_tillbaka(register, till_version, av, anledning):
    register.befordra(till_version, "produktion", av, f"ROLLBACK: {anledning}")
    byt_alias("produktion", till_version)        # sekunder, inte minuter

# Dela förbehandlingskoden mellan träning och servering
def forbehandla(ra: dict) -> dict:
    """ENDA implementationen. Importeras av både träningspipelinen och API:t."""
    return {"alder": float(ra["alder"]),
            "stad": (ra.get("stad") or "").strip().lower() or "<okand>"}

def test_forbehandling_fixtur():
    """Samma fixtur körs i både tränings- och serverings-CI."""
    assert forbehandla({"alder": "15", "stad": " Malmö "}) == {"alder": 15.0, "stad": "malmö"}
    assert forbehandla({"alder": 16, "stad": None}) == {"alder": 16.0, "stad": "<okand>"}