177 lines
6.9 KiB
Python
177 lines
6.9 KiB
Python
"""pipeline.yaml laden + statisch validieren (R8): Normal-Kanten müssen ein DAG
|
|
sein (graphlib), Rück-/Reparatur-Kanten nur von Gates und stufen-abwärts,
|
|
Worker registriert, Skill-Dateien vorhanden, genau ein Gate je Stufe (außer
|
|
letzter). Liefert auch das vorberechnete Layout für die Graphensicht."""
|
|
import graphlib
|
|
from dataclasses import dataclass, field
|
|
|
|
import yaml
|
|
|
|
from . import config, skills
|
|
|
|
|
|
class GraphFehler(Exception):
|
|
pass
|
|
|
|
|
|
@dataclass
|
|
class Knoten:
|
|
id: str
|
|
typ: str # system | llm | gate
|
|
worker: str
|
|
stufe: str
|
|
titel: str = ""
|
|
beschreibung: str = ""
|
|
skills: list = field(default_factory=list)
|
|
extra_skills: dict = field(default_factory=dict) # name → Skill-Liste
|
|
klasse: str = "lang" # lang | mittel | kurz → Bündel-Größe je Call
|
|
timeout: list | None = None # [basis_s, pro_item_s]
|
|
runden_max: int = 3 # nur Gates: Runden-Cap
|
|
max_parallel: int = 4
|
|
max_versuche: int = config.MAX_VERSUCHE
|
|
barriere: bool = False
|
|
|
|
@property
|
|
def buendel(self) -> int:
|
|
return config.BUENDEL[self.klasse]
|
|
|
|
|
|
@dataclass
|
|
class Graph:
|
|
stufen: list[str]
|
|
knoten: dict[str, Knoten]
|
|
kanten: list[dict] # {von, nach, art}
|
|
|
|
def stufen_index(self, stufe: str) -> int:
|
|
return self.stufen.index(stufe)
|
|
|
|
def gate_der_stufe(self, stufe: str) -> Knoten | None:
|
|
return next((k for k in self.knoten.values()
|
|
if k.typ == "gate" and k.stufe == stufe), None)
|
|
|
|
def normal_nachfolger(self, knoten_id: str) -> set[str]:
|
|
return {k["nach"] for k in self.kanten
|
|
if k["von"] == knoten_id and k["art"] == "normal"}
|
|
|
|
def erlaubte_ziele(self, knoten_id: str) -> set[str]:
|
|
"""Alle Knoten, für die dieser Knoten Tasks erzeugen darf."""
|
|
return {k["nach"] for k in self.kanten if k["von"] == knoten_id}
|
|
|
|
def layout(self) -> dict:
|
|
"""Fluss-Layout (R7): Zeile = längster Pfad über Normal-Kanten,
|
|
Spalte = Spur innerhalb der Zeile. Reine Reparatur-Ziele (ohne
|
|
Normal-Kanten) rücken neben ihr Gate."""
|
|
vor: dict[str, list[str]] = {k: [] for k in self.knoten}
|
|
for ka in self.kanten:
|
|
if ka["art"] == "normal":
|
|
vor[ka["nach"]].append(ka["von"])
|
|
ebene: dict[str, int] = {}
|
|
for kid in graphlib.TopologicalSorter(
|
|
{k: set(v) for k, v in vor.items()}).static_order():
|
|
ebene[kid] = max((ebene[v] + 1 for v in vor[kid]), default=0)
|
|
an_normal = {ka[e] for ka in self.kanten if ka["art"] == "normal"
|
|
for e in ("von", "nach")}
|
|
for ka in self.kanten:
|
|
if ka["art"] != "normal" and ka["nach"] not in an_normal:
|
|
ebene[ka["nach"]] = ebene[ka["von"]]
|
|
|
|
pos = {}
|
|
spuren: dict[int, int] = {}
|
|
for kid in sorted(ebene, key=lambda i: ebene[i]):
|
|
k = self.knoten[kid]
|
|
spur = spuren.get(ebene[kid], 0)
|
|
spuren[ebene[kid]] = spur + 1
|
|
pos[kid] = {"zeile": ebene[kid], "spalte": spur, "typ": k.typ,
|
|
"barriere": k.barriere, "stufe": k.stufe,
|
|
"titel": k.titel or kid,
|
|
"beschreibung": k.beschreibung,
|
|
"skills": [{"name": s, "art": skills.lade(s)["art"]}
|
|
for s in k.skills]}
|
|
return {"stufen": self.stufen, "knoten": pos, "kanten": self.kanten}
|
|
|
|
|
|
_graph: Graph | None = None
|
|
|
|
|
|
def laden(worker_namen: set[str] | None = None, pfad=None) -> Graph:
|
|
global _graph
|
|
daten = yaml.safe_load((pfad or config.PIPELINE_YAML).read_text())
|
|
stufen = daten["stufen"]
|
|
knoten = {}
|
|
for k in daten["knoten"]:
|
|
kn = Knoten(**{**k, "skills": k.get("skills", [])})
|
|
if kn.id in knoten:
|
|
raise GraphFehler(f"Knoten-id doppelt: {kn.id}")
|
|
if kn.stufe not in stufen:
|
|
raise GraphFehler(f"{kn.id}: unbekannte Stufe {kn.stufe}")
|
|
if kn.typ not in ("system", "llm", "gate"):
|
|
raise GraphFehler(f"{kn.id}: unbekannter Typ {kn.typ}")
|
|
if kn.klasse not in config.BUENDEL:
|
|
raise GraphFehler(f"{kn.id}: unbekannte Klasse {kn.klasse}")
|
|
knoten[kn.id] = kn
|
|
kanten = daten["kanten"]
|
|
|
|
for kante in kanten:
|
|
for ende in ("von", "nach"):
|
|
if kante[ende] not in knoten:
|
|
raise GraphFehler(f"Kante {kante}: unbekannter Knoten {kante[ende]}")
|
|
if kante["art"] not in ("normal", "reparatur", "ruecklauf"):
|
|
raise GraphFehler(f"Kante {kante}: unbekannte Art")
|
|
if kante["art"] != "normal":
|
|
von = knoten[kante["von"]]
|
|
if von.typ != "gate":
|
|
raise GraphFehler(f"{kante['art']}-Kante nur von Gates: {kante}")
|
|
if stufen.index(knoten[kante["nach"]].stufe) > stufen.index(von.stufe):
|
|
raise GraphFehler(f"{kante['art']}-Kante muss stufen-abwärts zeigen: {kante}")
|
|
|
|
# Normal-Kanten: DAG-Pflicht (graphlib erwartet Vorgänger-Format)
|
|
vorgaenger: dict[str, set] = {k: set() for k in knoten}
|
|
for kante in kanten:
|
|
if kante["art"] == "normal":
|
|
vorgaenger[kante["nach"]].add(kante["von"])
|
|
try:
|
|
graphlib.TopologicalSorter(vorgaenger).prepare()
|
|
except graphlib.CycleError as e:
|
|
raise GraphFehler(f"Zyklus in Normal-Kanten: {e}") from e
|
|
|
|
for stufe in stufen[:-1]:
|
|
gates = [k for k in knoten.values() if k.typ == "gate" and k.stufe == stufe]
|
|
if len(gates) != 1:
|
|
raise GraphFehler(f"Stufe {stufe}: genau 1 Gate nötig, {len(gates)} gefunden")
|
|
|
|
for kn in knoten.values():
|
|
if kn.typ == "llm" and not kn.skills:
|
|
raise GraphFehler(f"{kn.id}: llm-Knoten ohne Skills")
|
|
for s in kn.skills + [s for liste in kn.extra_skills.values()
|
|
for s in liste]:
|
|
skills.lade(s) # wirft SkillFehler, wenn Datei fehlt/kaputt
|
|
if worker_namen is not None and kn.worker not in worker_namen:
|
|
raise GraphFehler(f"{kn.id}: Worker '{kn.worker}' nicht registriert")
|
|
|
|
_graph = Graph(stufen=stufen, knoten=knoten, kanten=kanten)
|
|
return _graph
|
|
|
|
|
|
def get() -> Graph:
|
|
if _graph is None:
|
|
raise GraphFehler("Graph nicht geladen")
|
|
return _graph
|
|
|
|
|
|
def skills_von(knoten_id: str, extra: str | None = None) -> list:
|
|
"""EINE Wahrheit für Skill-Kompositionen: pipeline.yaml. Worker fragen
|
|
hier nach, statt Listen zu hardcoden (Drift bricht sonst erst zur
|
|
Laufzeit)."""
|
|
kn = get().knoten[knoten_id]
|
|
if extra is None:
|
|
return kn.skills
|
|
if extra not in kn.extra_skills:
|
|
raise GraphFehler(f"{knoten_id}: extra_skills '{extra}' nicht deklariert")
|
|
return kn.extra_skills[extra]
|
|
|
|
|
|
def timeout_fuer(knoten_id: str, n: int = 0) -> float:
|
|
kn = _graph.knoten.get(knoten_id) if _graph else None
|
|
basis, pro = (kn.timeout if kn and kn.timeout else config.TIMEOUT_DEFAULT)
|
|
return basis + pro * n
|