"""Pflege-Agent: Service-Routine, Vorschläge, Freigaben per Prompt.""" from __future__ import annotations import re from collections.abc import Callable from datetime import datetime from pathlib import Path from typing import Any, Protocol from pydantic import BaseModel, Field from ..cases import pass_rate from ..embeddings import SkillIndex from ..lifecycle import Lifecycle from ..llm import ChatClient, LLMError from ..policy import Policy from ..registry import Registry from ..skills import SkillLibrary from ..store import Store from ..util import load_yaml, now from .actions import ActionError, ActionExecutor CONTEXT_SIGNALS = ("context_exceeded", "compact_prompt", "larger_slot", "output_truncated", "context_overflow") class MaintenanceConfig(BaseModel): demote_below: float = 0.80 min_cases: int = 20 window_cases: int = 50 promote_min_rate: float = 0.95 merge_similarity: float = 0.92 degraded_hours: float = 1.0 context_signal_limit: int = 10 adapter_min_cases: int = 300 adapter_target_rate: float = 0.90 lookback_days: float = 7.0 ontology_max_findings: int = 15 @classmethod def from_file(cls, path: str | Path) -> MaintenanceConfig: return cls(**load_yaml(path)) class Proposal(BaseModel): seq: int kind: str params: dict[str, Any] = Field(default_factory=dict) summary: str = "" reason: str = "" risk: str = "medium" state: str = "open" source: str = "" created_at: float | None = None decided_at: float | None = None decided_by: str | None = None result: dict[str, Any] | None = None @property def id(self) -> str: return f"P-{self.seq}" def line(self) -> str: return f"{self.id} [{self.risk}] {self.summary}" + (f" – {self.reason}" if self.reason else "") class RoutineReport(BaseModel): findings: list[str] = Field(default_factory=list) proposals: list[Proposal] = Field(default_factory=list) auto_executed: list[str] = Field(default_factory=list) text: str = "" class CommandReply(BaseModel): text: str proposals: list[Proposal] = Field(default_factory=list) class IntentMapper(Protocol): def map(self, text: str) -> str | None: ... HELP = """Befehle: - status · hilfe · routine · vorschläge - genehmige P-3 [P-5 …] · genehmige alle mit risiko niedrig · lehne P-3 ab: - wechsle auf · trainiere adapter für · rollback - stufe zurück · markiere als ok|degraded|unavailable - ordne P-7 zu (Vokabel-Vorschlag einem vorhandenen Konzept zuordnen)""" class LLMIntentMapper: """Bildet Freitext auf genau einen bekannten Befehl ab – führt nie selbst etwas aus.""" def __init__(self, chat: ChatClient, model: str = "tier1-general"): self.chat = chat self.model = model def map(self, text: str) -> str | None: system = ("Übersetze die Nachricht in GENAU einen Befehl aus dieser Liste oder antworte NONE.\n" + HELP + "\nAntworte nur mit dem Befehl.") try: res = self.chat.chat(self.model, [{"role": "system", "content": system}, {"role": "user", "content": text}], max_tokens=40, temperature=0) except LLMError: return None out = res.content.strip().strip("`").splitlines()[0] if res.content.strip() else "" return None if not out or out.upper() == "NONE" else out _ID = r"p-?\d+" COMMANDS: list[tuple[str, re.Pattern[str]]] = [ (name, re.compile(pattern, re.IGNORECASE)) for name, pattern in [ ("status", r"^(status|überblick|overview)$"), ("help", r"^(hilfe|help|\?)$"), ("routine", r"^((starte\s+(die\s+)?)?service-?routine|routine)$"), ("list", r"^((zeige\s+)?(offene\s+)?vorschläge|proposals)$"), ("approve_low", r"^(genehmige|approve)\s+alle\s+(mit\s+)?risiko\s+(niedrig|low)$"), ("approve", rf"^(genehmige|approve)\s+(?P{_ID}([\s,]+(und\s+)?{_ID})*)$"), ("reject", rf"^(lehne\s+(?P{_ID})\s+ab|reject\s+(?P{_ID}))(\s*[:,]?\s*(weil\s+)?(?P.+))?$"), ("switch_base", r"^wechsle\s+(?P[\w.-]+)\s+auf\s+(?P[\w./:@-]+)$"), ("train_adapter", r"^trainiere\s+(einen\s+)?adapter\s+für\s+(den\s+cluster\s+)?(?P[\w.-]+)$"), ("rollback", r"^rollback\s+(?P[\w.-]+)$"), ("demote", r"^(stufe\s+(?P[\w.-]+)\s+zurück|degradiere\s+(?P[\w.-]+))$"), ("mark", r"^markiere\s+(?P[\w.-]+)\s+(als\s+)?(?Pok|degraded|unavailable)$"), ("map_concept", rf"^ordne\s+(?P{_ID})\s+(dem\s+konzept\s+)?(?P[\w.-]+)\s+zu$"), ("confirm", r"^(ja|yes|j|ok)$"), ("cancel", r"^(nein|no|n|abbrechen)$"), ] ] def parse_command(text: str) -> tuple[str, dict[str, str]] | None: cleaned = text.strip().rstrip(".!") for name, pattern in COMMANDS: m = pattern.match(cleaned) if m: return name, {k: v for k, v in m.groupdict().items() if v is not None} return None def _seq(token: str) -> int: return int(re.sub(r"\D", "", token)) class MaintenanceAgent: def __init__(self, *, store: Store, registry: Registry, library: SkillLibrary, index: SkillIndex, policy: Policy, actions: ActionExecutor, lifecycle: Lifecycle, config: MaintenanceConfig | None = None, intent_mapper: IntentMapper | None = None, ontology_check: Callable[[], list[Any]] | None = None, clock: Callable[[], float] = now, wallclock: Callable[[], datetime] = datetime.now): self.store = store self.registry = registry self.library = library self.index = index self.policy = policy self.actions = actions self.lifecycle = lifecycle self.config = config or MaintenanceConfig() self.intent_mapper = intent_mapper self.ontology_check = ontology_check self.clock = clock self.wallclock = wallclock self._pending: dict[str, str] = {} # Sitzung → zu bestätigender Befehl # --- Vorschläge ------------------------------------------------------------------------- def _load(self, seq: int) -> Proposal: row = self.store.get_proposal(seq) if row is None: raise KeyError(f"P-{seq} unbekannt") return Proposal(**row) def proposals(self, state: str | None = "open") -> list[Proposal]: return [Proposal(**r) for r in self.store.list_proposals(state=state)] def propose(self, kind: str, params: dict[str, Any], summary: str, *, reason: str = "", source: str = "routine", dedup_key: str | None = None, force_approval: bool = False) -> Proposal: key = dedup_key or f"{kind}:{sorted(params.items())}" existing = self.store.open_proposal_with_key(key) if existing: return Proposal(**existing) mode = self.policy.decide(kind) if kind == "mark_slot" and params.get("status") == "ok": force_approval = True # Wieder-Freischalten ist die unsichere Richtung seq = self.store.add_proposal(kind=kind, params=params, summary=summary, reason=reason, risk=self.policy.risk(kind), source=source, dedup_key=key) if mode == "deny": self.store.update_proposal(seq, state="rejected", decided_by="policy", decided_at=self.clock(), result={"message": "durch Richtlinie verboten"}) elif mode == "auto" and not force_approval: return self.approve(seq, by="policy") return self._load(seq) def approve(self, seq: int, by: str = "human") -> Proposal: p = self._load(seq) if p.state != "open": raise ValueError(f"{p.id} ist {p.state}") self.store.update_proposal(seq, state="approved", decided_by=by, decided_at=self.clock()) if not self.policy.in_window(p.kind, self.wallclock()): self.store.update_proposal(seq, result={"deferred": True, "message": "wartet auf Wartungsfenster"}) return self._load(seq) return self._execute(seq) def _execute(self, seq: int) -> Proposal: p = self._load(seq) params = {**p.params, "decided_by": p.decided_by} if p.kind == "ontology_change" else p.params try: result = self.actions.execute(p.kind, params) self.store.update_proposal(seq, state="executed", result=result) except ActionError as exc: self.store.update_proposal(seq, state="failed", result={"ok": False, "message": str(exc)}) return self._load(seq) def reject(self, seq: int, by: str = "human", reason: str = "") -> Proposal: p = self._load(seq) if p.state != "open": raise ValueError(f"{p.id} ist {p.state}") self.store.update_proposal(seq, state="rejected", decided_by=by, decided_at=self.clock(), result={"message": reason or "abgelehnt"}) if p.kind == "promote_candidate": self.lifecycle.reject(p.params["candidate"], f"abgelehnt: {reason}") if p.kind == "ontology_change": params = {k: v for k, v in p.params.items() if k != "key"} self.store.set_ontology_decision(p.params["key"], "reject", params, by) return self._load(seq) def run_deferred(self) -> list[Proposal]: done = [] for p in self.proposals(state="approved"): if (p.result or {}).get("deferred") and self.policy.in_window(p.kind, self.wallclock()): done.append(self._execute(p.seq)) return done # --- Service-Routine -------------------------------------------------------------------- def run_routine(self) -> RoutineReport: report = RoutineReport() c = self.config since = self.clock() - c.lookback_days * 86400 def add(p: Proposal) -> None: if p.state == "executed" and p.decided_by == "policy": report.auto_executed.append(f"{p.id}: {p.summary} → {(p.result or {}).get('message', '')}") else: report.proposals.append(p) # 1. Drift / 2. Graduierungskandidaten for skill in self.library.routable(): if skill.meta.status == "active" and skill.meta.stage >= 1: rows = self.store.list_cases(skill=skill.name, min_stage=1, limit=c.window_cases) rate = pass_rate(rows) n = len([r for r in rows if r["passed"] is not None]) if rate is not None and n >= c.min_cases and rate < c.demote_below: report.findings.append(f"Drift: {skill.name} Pass-Rate {rate:.0%} (n={n})") add(self.propose("demote_skill", {"skill": skill.name, "reason": f"Pass-Rate {rate:.0%}"}, f"Skill {skill.name} auf Stufe 0 zurückstufen", reason=f"Pass-Rate {rate:.0%} < {c.demote_below:.0%} (n={n})")) else: rows = self.store.list_cases(skill=skill.name, stage=0, limit=c.window_cases) rate = pass_rate(rows) n = len([r for r in rows if r["passed"] is not None]) if rate is not None and n >= c.min_cases and rate >= c.promote_min_rate: report.findings.append(f"Graduierungskandidat: {skill.name} ({rate:.0%} auf Stufe 0, n={n})") add(self.propose("request_graduation", {"skill": skill.name}, f"Graduierungslauf für {skill.name} anstoßen", reason=f"stabil auf Stufe 0 ({rate:.0%}, n={n})")) # 3. ähnliche Skills for a, b, sim in self.index.pairwise_similar(c.merge_similarity): if self.library.has(a) and self.library.has(b) and self.library.get(a).routable \ and self.library.get(b).routable: report.findings.append(f"Ähnliche Skills: {a} ~ {b} ({sim:.2f})") add(self.propose("merge_skills", {"a": a, "b": b}, f"Skills {a} und {b} zusammenführen", reason=f"Ähnlichkeit {sim:.2f}")) # 4. degradierte Slots for slot in self.registry.slots(): state = self.store.get_slot_state(slot.name) or {} age_h = (self.clock() - (state.get("updated_at") or self.clock())) / 3600 if not slot.available and age_h >= c.degraded_hours: report.findings.append(f"Slot {slot.name} seit {age_h:.1f} h {slot.status} ({slot.reason or '-'})") add(self.propose("mark_slot", {"slot": slot.name, "status": "ok", "reason": "nach Prüfung"}, f"Slot {slot.name} wieder auf ok setzen", reason=f"{slot.status} seit {age_h:.1f} h – vorher Ursache prüfen")) # 5. Kontext-Signale totals: dict[str, int] = {} for kind in CONTEXT_SIGNALS: for subject, count in self.store.event_subjects(kind, since=since).items(): totals[subject] = totals.get(subject, 0) + count for subject, count in sorted(totals.items()): if self.registry.has_slot(subject) and count >= c.context_signal_limit: report.findings.append(f"Kontext-Signale auf {subject}: {count} in {c.lookback_days:g} Tagen") add(self.propose("recalibrate_context", {"slot": subject}, f"Effektiven Kontext von {subject} neu kalibrieren", reason=f"{count} Signale")) # 6. Adapter-Bedarf open_clusters = {c_["target"].get("cluster") for c_ in self.store.list_candidates() if c_["kind"] == "adapter" and c_["state"] not in {"promoted", "rejected", "failed", "rolled_back"}} clusters: dict[str, str] = {} for skill in self.library.routable(): clusters.setdefault(skill.cluster, skill.meta.slot) for cluster, slot in sorted(clusters.items()): if cluster in open_clusters or not self.registry.has_slot(slot): continue if not self.registry.slot(slot).spec.supports_lora: continue train = self.store.list_cases(cluster=cluster, split="train", passed=True) if len(train) < c.adapter_min_cases: continue weak = self.store.list_cases(cluster=cluster, min_stage=1, limit=c.window_cases) rate = pass_rate(weak) if rate is None or rate < c.adapter_target_rate: report.findings.append(f"Adapter-Bedarf: {cluster} ({len(train)} Trainingsfälle)") add(self.propose("train_adapter", {"cluster": cluster, "slot": slot}, f"LoRA-Adapter für {cluster} trainieren", reason=f"{len(train)} verifizierte Fälle, Stufe-1-Rate " f"{'–' if rate is None else f'{rate:.0%}'}")) # 7. Lifecycle: Canary auswerten, Freigaben vorschlagen for cand in self.store.list_candidates(state="canary"): m = self.lifecycle.canary_metrics(cand["id"]) if m["n"] >= self.lifecycle.config.min_n and m["pass_rate"] is not None: after = self.lifecycle.report(cand["id"], "canary", m) report.auto_executed.append(f"Canary {cand['id'][:8]} ausgewertet → {after['state']}") for cand in self.store.list_candidates(state="awaiting_promotion"): add(self.propose("promote_candidate", {"candidate": cand["id"]}, f"{cand['kind']} für {cand['slot']} übernehmen ({cand['target']})", reason="alle Prüfstufen bestanden", source="lifecycle")) # 8. Ontologie ↔ Datenhaltung (SPEC-ontology §3) if self.ontology_check is not None: try: ontology_findings = self.ontology_check() except Exception as exc: # noqa: BLE001 – externe Quelle (n8n-API) darf die Routine nicht abbrechen report.findings.append(f"Ontologie-Abgleich nicht möglich: {type(exc).__name__}: {exc}") ontology_findings = [] violations = [f for f in ontology_findings if not f.change] for f in violations[:c.ontology_max_findings]: report.findings.append(f"Ontologie [{f.severity}]: {f.summary}") if len(violations) > c.ontology_max_findings: report.findings.append(f"Ontologie: {len(violations) - c.ontology_max_findings} weitere Befunde " "(knaster ontology check)") for f in ontology_findings: if f.change: add(self.propose("ontology_change", {"key": f.key, **f.change}, f.summary, reason=f.detail, dedup_key=f"ontology:{f.key}")) for p in self.run_deferred(): report.auto_executed.append(f"{p.id} im Wartungsfenster ausgeführt: {p.summary}") open_all = self.proposals("open") report.text = self._render_report(report, open_all) return report @staticmethod def _render_report(report: RoutineReport, open_all: list[Proposal]) -> str: lines = ["Service-Routine"] lines.append("\nBefunde:" if report.findings else "\nKeine Auffälligkeiten.") lines += [f"- {f}" for f in report.findings] if report.auto_executed: lines.append("\nAutomatisch erledigt:") lines += [f"- {a}" for a in report.auto_executed] if open_all: lines.append("\nOffene Vorschläge (genehmige P-x / lehne P-x ab):") lines += [f"- {p.line()}" for p in open_all] return "\n".join(lines) # --- Befehle ---------------------------------------------------------------------------- def handle_command(self, text: str, user: str = "human", session: str = "default") -> CommandReply: parsed = parse_command(text) if parsed is None: if self.intent_mapper is not None: mapped = self.intent_mapper.map(text) if mapped and parse_command(mapped) and parse_command(mapped)[0] not in {"confirm", "cancel"}: self._pending[session] = mapped return CommandReply(text=f"Meinst du: `{mapped}`? Antworte mit „ja“.") return CommandReply(text="Das habe ich nicht verstanden.\n" + HELP) name, args = parsed if name == "confirm": pending = self._pending.pop(session, None) if not pending: return CommandReply(text="Es gibt nichts zu bestätigen.") return self.handle_command(pending, user, session) if name == "cancel": self._pending.pop(session, None) return CommandReply(text="Abgebrochen.") try: return getattr(self, f"_cmd_{name}")(args, user) except (KeyError, ValueError) as exc: return CommandReply(text=f"Fehler: {exc}") def _cmd_help(self, args: dict[str, str], user: str) -> CommandReply: return CommandReply(text=HELP) def _cmd_status(self, args: dict[str, str], user: str) -> CommandReply: lines = ["Slots:"] lines += [f"- {s.name}: {s.status} (Basis {s.base_model}, Kontext {s.effective_context})" for s in self.registry.slots()] by_status: dict[str, int] = {} for s in self.library.all(): by_status[s.meta.status] = by_status.get(s.meta.status, 0) + 1 lines.append("Skills: " + ", ".join(f"{k} {v}" for k, v in sorted(by_status.items()))) running = [c for c in self.store.list_candidates() if c["state"] not in {"promoted", "rejected", "failed", "rolled_back"}] lines.append(f"Kandidaten in Arbeit: {len(running)}") lines.append(f"Offene Vorschläge: {len(self.proposals('open'))}") return CommandReply(text="\n".join(lines)) def _cmd_routine(self, args: dict[str, str], user: str) -> CommandReply: report = self.run_routine() return CommandReply(text=report.text, proposals=report.proposals) def _cmd_list(self, args: dict[str, str], user: str) -> CommandReply: props = self.proposals("open") if not props: return CommandReply(text="Keine offenen Vorschläge.") return CommandReply(text="\n".join(p.line() for p in props), proposals=props) def _decided_text(self, p: Proposal) -> str: msg = (p.result or {}).get("message", "") return f"{p.id} {p.state}" + (f": {msg}" if msg else "") def _cmd_approve(self, args: dict[str, str], user: str) -> CommandReply: ids = re.findall(_ID, args["ids"], re.IGNORECASE) done = [self.approve(_seq(i), by=user) for i in ids] return CommandReply(text="\n".join(self._decided_text(p) for p in done), proposals=done) def _cmd_approve_low(self, args: dict[str, str], user: str) -> CommandReply: done = [self.approve(p.seq, by=user) for p in self.proposals("open") if p.risk == "low"] if not done: return CommandReply(text="Keine offenen Vorschläge mit niedrigem Risiko.") return CommandReply(text="\n".join(self._decided_text(p) for p in done), proposals=done) def _cmd_reject(self, args: dict[str, str], user: str) -> CommandReply: p = self.reject(_seq(args.get("id") or args["id2"]), by=user, reason=args.get("reason", "")) return CommandReply(text=self._decided_text(p), proposals=[p]) def _new(self, kind: str, params: dict[str, Any], summary: str, user: str) -> CommandReply: p = self.propose(kind, params, summary, reason=f"Auftrag von {user}", source="command", force_approval=True) return CommandReply(text=f"Vorschlag {p.line()}\nBestätige mit „genehmige {p.id}“.", proposals=[p]) def _require_slot(self, slot: str) -> None: if not self.registry.has_slot(slot): raise KeyError(f"Slot {slot} unbekannt") def _cmd_switch_base(self, args: dict[str, str], user: str) -> CommandReply: self._require_slot(args["slot"]) return self._new("create_candidate", {"kind": "base_model", "slot": args["slot"], "target": {"base_model": args["base"]}}, f"Basismodell von {args['slot']} auf {args['base']} wechseln (Pipeline, inkl. " "Neutraining der Adapter)", user) def _cmd_train_adapter(self, args: dict[str, str], user: str) -> CommandReply: slot = next((s.meta.slot for s in self.library.all() if s.cluster == args["cluster"]), "tier1-general") return self._new("train_adapter", {"cluster": args["cluster"], "slot": slot}, f"LoRA-Adapter für {args['cluster']} trainieren", user) def _cmd_rollback(self, args: dict[str, str], user: str) -> CommandReply: self._require_slot(args["slot"]) return self._new("rollback_candidate", {"slot": args["slot"]}, f"Letzte Übernahme auf {args['slot']} zurückrollen", user) def _cmd_demote(self, args: dict[str, str], user: str) -> CommandReply: skill = args.get("skill") or args["skill2"] if not self.library.has(skill): raise KeyError(f"Skill {skill} unbekannt") return self._new("demote_skill", {"skill": skill, "reason": f"Auftrag von {user}"}, f"Skill {skill} auf Stufe 0 zurückstufen", user) def _cmd_mark(self, args: dict[str, str], user: str) -> CommandReply: self._require_slot(args["slot"]) return self._new("mark_slot", {"slot": args["slot"], "status": args["status"].lower(), "reason": f"Auftrag von {user}", "set_by": user}, f"Slot {args['slot']} als {args['status'].lower()} markieren", user) def _cmd_map_concept(self, args: dict[str, str], user: str) -> CommandReply: p = self._load(_seq(args["id"])) if p.kind != "ontology_change" or p.params.get("type") != "vocab_value": raise ValueError(f"{p.id} ist kein Vokabel-Vorschlag") if p.state != "open": raise ValueError(f"{p.id} ist {p.state}") self.store.update_proposal(p.seq, params={**p.params, "decision": "map", "concept": args["concept"]}) done = self.approve(p.seq, by=user) return CommandReply(text=self._decided_text(done), proposals=[done])