core: Policy, Pflege-Agent (HITL), Healer, Lifecycle-Pipeline, Graduierung + Konfiguration
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -0,0 +1,246 @@
|
||||
"""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, client: Any | None = None):
|
||||
import httpx
|
||||
|
||||
self.url = url
|
||||
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})
|
||||
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)
|
||||
rows = [r for r in self.store.list_cases(stage=1, since=since) if r["model"] == 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)
|
||||
Reference in New Issue
Block a user