249 lines
11 KiB
Python
249 lines
11 KiB
Python
"""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)
|