"""Lifecycle-Pipeline: ein Ablauf für Basismodell, Adapter, Quantisierung, Embedding-Modell, Engine.""" from __future__ import annotations from pathlib import Path from typing import Any, Literal, Protocol from pydantic import BaseModel from .cases import pass_rate from .registry import AdapterSpec, Registry from .store import Store from .util import load_yaml, now Kind = Literal["base_model", "adapter", "quantization", "embedding_model", "engine"] TERMINAL = {"promoted", "rejected", "rolled_back", "failed"} # erlaubte Übergänge durch Job-Reports REPORT_FLOW = { "build": ("building", "evaluating"), "eval": ("evaluating", "shadow"), "shadow": ("shadow", "canary"), "canary": ("canary", "awaiting_promotion"), } class LifecycleConfig(BaseModel): tolerance: float = 0.02 min_pass_rate: float = 0.85 # wenn keine Baseline bekannt ist min_agreement: float = 0.85 min_n: int = 50 canary_share: float = 0.1 @classmethod def from_file(cls, path: str | Path) -> LifecycleConfig: return cls(**load_yaml(path)) class JobRunner(Protocol): def submit(self, job: str, candidate: dict[str, Any]) -> str: ... class LoggingJobRunner: """Dry-Run: merkt sich Aufträge; im Betrieb ersetzt durch Webhook an n8n/Runner.""" def __init__(self) -> None: self.jobs: list[tuple[str, dict[str, Any]]] = [] def submit(self, job: str, candidate: dict[str, Any]) -> str: self.jobs.append((job, candidate)) return f"[dry-run] Job {job} für Kandidat {candidate['id']}" class WebhookJobRunner: def __init__(self, url: str, auth: str = "", client: Any | None = None): import httpx self.url = url self.headers = {"Authorization": auth} if auth else {} self.client = client or httpx.Client(timeout=15.0) def submit(self, job: str, candidate: dict[str, Any]) -> str: resp = self.client.post(self.url, json={"job": job, "candidate": candidate}, headers=self.headers) resp.raise_for_status() return f"Job {job} übergeben" class LifecycleError(Exception): pass class Lifecycle: def __init__(self, store: Store, registry: Registry, jobs: JobRunner, config: LifecycleConfig | None = None): self.store = store self.registry = registry self.jobs = jobs self.config = config or LifecycleConfig() # --- Anlegen / Starten ------------------------------------------------------------------ def create(self, kind: Kind, slot: str, target: dict[str, Any], baseline: dict[str, Any] | None = None) -> str: if not self.registry.has_slot(slot): raise LifecycleError(f"Unbekannter Slot {slot}") if kind == "adapter" and not target.get("cluster"): raise LifecycleError("Adapter-Kandidat braucht target.cluster") if kind == "base_model" and not target.get("base_model"): raise LifecycleError("Basismodell-Kandidat braucht target.base_model") if baseline is None: baseline = {"pass_rate": self._baseline(kind, slot, target)} return self.store.add_candidate(kind=kind, slot=slot, target=target, baseline=baseline, state="proposed", metrics={}, history=[], snapshot=None) def _baseline(self, kind: str, slot: str, target: dict[str, Any]) -> float | None: rows = [r for r in self.store.list_cases(stage=1, split="eval") if r["slot"] == slot] if kind == "adapter": rows = [r for r in rows if r["cluster"] == target["cluster"]] return pass_rate(rows) def get(self, cid: str) -> dict[str, Any]: cand = self.store.get_candidate(cid) if cand is None: raise LifecycleError(f"Kandidat {cid} unbekannt") return cand def _move(self, cand: dict[str, Any], to: str, note: str = "", **extra: Any) -> dict[str, Any]: history = list(cand.get("history") or []) history.append({"ts": now(), "from": cand["state"], "to": to, "note": note}) self.store.update_candidate(cand["id"], state=to, history=history, **extra) return self.get(cand["id"]) def start(self, cid: str) -> dict[str, Any]: cand = self.get(cid) if cand["state"] != "proposed": raise LifecycleError(f"Kandidat ist {cand['state']}, nicht proposed") cand = self._move(cand, "building", "freigegeben") self.jobs.submit("build", cand) return cand # --- Job-Reports ------------------------------------------------------------------------ def report(self, cid: str, step: str, payload: dict[str, Any]) -> dict[str, Any]: cand = self.get(cid) if payload.get("error"): self._cleanup_canary(cand) return self._move(cand, "failed", f"{step}: {payload['error']}") if step not in REPORT_FLOW: raise LifecycleError(f"Unbekannter Schritt {step}") expected, nxt = REPORT_FLOW[step] if cand["state"] != expected: raise LifecycleError(f"Report {step} passt nicht zum Zustand {cand['state']}") metrics = {**(cand.get("metrics") or {}), step: payload} ok, why = self._gate(cand, step, payload) if not ok: self._cleanup_canary(cand) return self._move(cand, "rejected", why, metrics=metrics) target = dict(cand["target"]) if step == "build" and payload.get("model"): target["model"] = payload["model"] # Name, unter dem der Kandidat bereitsteht cand = self._move(cand, nxt, why, metrics=metrics, target=target) if nxt == "canary": self._start_canary(cand) elif nxt in {"evaluating", "shadow"}: self.jobs.submit({"evaluating": "eval", "shadow": "shadow"}[nxt], cand) elif nxt == "awaiting_promotion": self._cleanup_canary(cand) return cand def _gate(self, cand: dict[str, Any], step: str, p: dict[str, Any]) -> tuple[bool, str]: c = self.config if step == "build": return bool(p.get("ok", True)), "Build ok" n = int(p.get("n", 0)) if step == "shadow": agreement = float(p.get("agreement", 0)) ok = agreement >= c.min_agreement and n >= c.min_n return ok, f"Shadow-Übereinstimmung {agreement:.2f} (n={n})" rate = float(p.get("pass_rate", 0)) base = (cand.get("baseline") or {}).get("pass_rate") needed = c.min_pass_rate if base is None else base - c.tolerance min_n = c.min_n if step == "canary" else 1 ok = rate >= needed and n >= min_n return ok, f"{step}: Pass-Rate {rate:.2f} (benötigt {needed:.2f}, n={n})" # --- Canary ----------------------------------------------------------------------------- def _start_canary(self, cand: dict[str, Any]) -> None: model = cand["target"].get("model") if not model: raise LifecycleError("Kandidat ohne bereitgestelltes Modell kann nicht in den Canary") cluster = cand["target"].get("cluster") if cand["kind"] == "adapter" else None self.registry.set_canary(cand["slot"], model, self.config.canary_share, cluster) def _cleanup_canary(self, cand: dict[str, Any]) -> None: slot = self.registry.slot(cand["slot"]) if slot.canary_model and slot.canary_model == cand["target"].get("model"): self.registry.set_canary(cand["slot"], None) def canary_metrics(self, cid: str) -> dict[str, Any]: """Pass-Rate der Fälle, die über das Canary-Modell liefen.""" cand = self.get(cid) model = cand["target"].get("model") since = next((h["ts"] for h in cand.get("history") or [] if h["to"] == "canary"), None) column = "adapter" if cand["kind"] == "adapter" else "model" rows = [r for r in self.store.list_cases(stage=1, since=since) if r[column] == model] return {"pass_rate": pass_rate(rows), "n": len([r for r in rows if r["passed"] is not None])} # --- Promote / Rollback ----------------------------------------------------------------- def promote(self, cid: str) -> dict[str, Any]: cand = self.get(cid) if cand["state"] != "awaiting_promotion": raise LifecycleError(f"Kandidat ist {cand['state']}, nicht awaiting_promotion") slot = self.registry.slot(cand["slot"]) snapshot: dict[str, Any] = { "base_model": slot.base_model_override, "adapters": {a.name: a.status for a in self.registry.adapters()}, } followups: list[str] = [] kind, target = cand["kind"], cand["target"] if kind == "base_model": old_base = slot.base_model self.registry.set_base_model(slot.name, target["base_model"]) for a in self.registry.adapters(): if a.base_model == old_base and a.status == "active": self.registry.set_adapter(a.model_copy(update={"status": "retired"})) followups.append(self.create("adapter", slot.name, {"cluster": a.cluster, "base_model": target["base_model"]})) elif kind == "adapter": name = target.get("model") or target.get("adapter") if not name: raise LifecycleError("Adapter-Kandidat ohne Adapter-Namen") base = target.get("base_model") or slot.base_model current = self.registry.active_adapter(base, target["cluster"]) version = (current.version + 1) if current else 1 if current: self.registry.set_adapter(current.model_copy(update={"status": "retired"})) self.registry.set_adapter(AdapterSpec(name=name, base_model=base, cluster=target["cluster"], version=version, status="active")) self.jobs.submit("promote", cand) return self._move(cand, "promoted", "freigegeben", snapshot={**snapshot, "followups": followups}) def rollback(self, cid: str) -> dict[str, Any]: cand = self.get(cid) if cand["state"] != "promoted": raise LifecycleError(f"Kandidat ist {cand['state']}, nicht promoted") snap = cand.get("snapshot") or {} if cand["kind"] == "base_model": self.registry.set_base_model(cand["slot"], snap.get("base_model")) statuses = snap.get("adapters") or {} for a in self.registry.adapters(): old = statuses.get(a.name) if old is None and a.status == "active": self.registry.set_adapter(a.model_copy(update={"status": "retired"})) elif old is not None and old != a.status: self.registry.set_adapter(a.model_copy(update={"status": old})) for follow in snap.get("followups") or []: f = self.store.get_candidate(follow) if f and f["state"] not in TERMINAL: self._move(f, "rejected", "Rollback des Basismodells") self.jobs.submit("rollback", cand) return self._move(cand, "rolled_back", "Rollback") def latest_promoted(self, slot: str) -> dict[str, Any] | None: rows = [c for c in self.store.list_candidates(state="promoted") if c["slot"] == slot] return rows[-1] if rows else None def reject(self, cid: str, note: str = "abgelehnt") -> dict[str, Any]: cand = self.get(cid) if cand["state"] in TERMINAL: raise LifecycleError(f"Kandidat ist bereits {cand['state']}") self._cleanup_canary(cand) return self._move(cand, "rejected", note)