From 4cbe313c624841a24137d255a04c92a5bda55001 Mon Sep 17 00:00:00 2001 From: Robert Rapp Date: Mon, 12 Oct 2026 00:47:13 +0200 Subject: [PATCH] core: Inventar/Placement, Konfig-Validierung, CLI, HTTP-API, LiteLLM-Konfiguration, Beispiel-Skills Co-Authored-By: Claude Opus 5.5 --- catalog/models.yaml | 51 ++++++ catalog/roles.yaml | 34 ++++ core/src/knaster/api.py | 204 ++++++++++++++++++++++++ core/src/knaster/cli.py | 82 ++++++++++ core/src/knaster/inventory/__init__.py | 1 + core/src/knaster/inventory/placement.py | 153 ++++++++++++++++++ core/src/knaster/inventory/profile.py | 108 +++++++++++++ core/src/knaster/lifecycle.py | 3 +- core/src/knaster/router.py | 14 +- core/src/knaster/skills.py | 4 + core/src/knaster/validate.py | 96 +++++++++++ core/tests/test_api.py | 110 +++++++++++++ core/tests/test_execution.py | 19 +-- core/tests/test_inventory_validate.py | 90 +++++++++++ core/tests/test_lifecycle.py | 7 +- core/tests/test_maintenance.py | 3 +- core/tests/test_router.py | 2 +- deploy/litellm/config.yaml | 75 +++++++++ sites/example-single-gpu/inventory.yaml | 10 ++ sites/knaster/inventory.yaml | 38 +++++ 20 files changed, 1085 insertions(+), 19 deletions(-) create mode 100644 catalog/models.yaml create mode 100644 catalog/roles.yaml create mode 100644 core/src/knaster/api.py create mode 100644 core/src/knaster/cli.py create mode 100644 core/src/knaster/inventory/__init__.py create mode 100644 core/src/knaster/inventory/placement.py create mode 100644 core/src/knaster/inventory/profile.py create mode 100644 core/src/knaster/validate.py create mode 100644 core/tests/test_api.py create mode 100644 core/tests/test_inventory_validate.py create mode 100644 deploy/litellm/config.yaml create mode 100644 sites/example-single-gpu/inventory.yaml create mode 100644 sites/knaster/inventory.yaml diff --git a/catalog/models.yaml b/catalog/models.yaml new file mode 100644 index 0000000..7ea47c3 --- /dev/null +++ b/catalog/models.yaml @@ -0,0 +1,51 @@ +# Modell-Katalog (SPEC-inventory §2). vram_gb = Gewichte + Mindest-KV-Cache (Richtwerte, per Kalibrierung prüfen). +# Ist ein Modell nicht mehr aktuell, wird ein neuer Eintrag ergänzt. Der Wechsel läuft über die Lifecycle-Pipeline. +models: + qwen3-14b-awq: + kind: llm + vram_gb: 14 + min_cc: 7.5 # AWQ-Kernels in vLLM ab Turing + engines: [vllm] + context: 32768 + lora: true + qwen3-8b-awq: + kind: llm + vram_gb: 9 + min_cc: 7.5 + engines: [vllm] + context: 32768 + lora: true + qwen3-8b-q4km: + kind: llm + vram_gb: 7 + min_cc: 5.0 + engines: [llamacpp] + context: 16384 + qwen3-4b-q4km: + kind: llm + vram_gb: 4 + min_cc: 5.0 + engines: [llamacpp] + context: 8192 + cpu_ok: true # langsam, aber als Notfall-Verifier nutzbar + qwen3-embedding-0.6b-q8: + kind: embedding + vram_gb: 1.5 + min_cc: 5.0 + engines: [llamacpp] + context: 8192 + cpu_ok: true + bge-m3-f16: + kind: embedding + vram_gb: 2.5 + min_cc: 5.0 + engines: [llamacpp] + context: 8192 + cpu_ok: true + bge-reranker-v2-m3-q8: + kind: reranker + vram_gb: 1.5 + min_cc: 5.0 + engines: [llamacpp] + context: 8192 + cpu_ok: true diff --git a/catalog/roles.yaml b/catalog/roles.yaml new file mode 100644 index 0000000..bc3b357 --- /dev/null +++ b/catalog/roles.yaml @@ -0,0 +1,34 @@ +# Rollen (SPEC-inventory §2/§3). Placement arbeitet sie nach priority ab. +roles: + gpu-main: + priority: 1 + needs: {gpu: true, engine: vllm, min_vram_gb: 16, min_cc: 7.5} + prefer: largest + kinds: [llm] + slots: [tier1-general] + gpu-flex: + priority: 2 + needs: {gpu: true, engine: vllm, min_vram_gb: 16, min_cc: 7.5} + prefer: largest + kinds: [llm] + optional: true # ohne zweite große GPU: kein Fallback-Slot, Training nur nachts auf gpu-main + slots: [tier1-flex] + gpu-embed: + priority: 3 + count: 2 # redundant, damit Routing einen GPU-Ausfall übersteht + needs: {gpu: true, engine: llamacpp, min_vram_gb: 3, min_cc: 5.0} + prefer: smallest_fitting + kinds: [embedding, reranker] + cpu_fallback: true + slots: [embed-default, rerank-default] + gpu-small: + priority: 4 + needs: {gpu: true, engine: llamacpp, min_vram_gb: 4, min_cc: 5.0} + prefer: smallest_fitting + kinds: [llm] + cpu_fallback: true + slots: [verifier-small, signature-extractor] + infra: + priority: 5 + needs: {gpu: false, min_ram_gb: 16} + slots: [] diff --git a/core/src/knaster/api.py b/core/src/knaster/api.py new file mode 100644 index 0000000..189812e --- /dev/null +++ b/core/src/knaster/api.py @@ -0,0 +1,204 @@ +"""HTTP-API des Router-Service (SPEC-core §8).""" + +from __future__ import annotations + +from typing import Any + +from fastapi import Depends, FastAPI, Header, HTTPException, Response +from pydantic import BaseModel, Field + +from .graduation import graduate +from .lifecycle import LifecycleError +from .settings import Settings +from .system import System, build_system + + +class RouteIn(BaseModel): + input: str + skill: str | None = None + + +class SlotStateIn(BaseModel): + status: str = Field(pattern="^(ok|degraded|unavailable)$") + reason: str = "" + set_by: str = "human" + + +class CommandIn(BaseModel): + text: str + user: str = "human" + session: str = "default" + + +class DecisionIn(BaseModel): + user: str = "human" + reason: str = "" + + +class CandidateIn(BaseModel): + kind: str = Field(pattern="^(base_model|adapter|quantization|embedding_model|engine)$") + slot: str + target: dict[str, Any] + requested_by: str = "api" + + +class ReportIn(BaseModel): + step: str + payload: dict[str, Any] = Field(default_factory=dict) + + +class GraduateIn(BaseModel): + variants: list[str] + threshold: float = 0.9 + min_variants: int = 10 + + +def create_app(system: System | None = None) -> FastAPI: + system = system or build_system(Settings.from_env()) + app = FastAPI(title="knaster", version="0.1.0") + app.state.system = system + + def auth(authorization: str | None = Header(default=None)) -> None: + token = system.settings.api_token + if token and authorization != f"Bearer {token}": + raise HTTPException(status_code=401, detail="unauthorized") + + v1 = [Depends(auth)] + + @app.get("/health") + def health() -> dict[str, Any]: + return {"status": "ok", "skills": len(system.library.all()), "dry_run": system.settings.dry_run} + + @app.get("/metrics") + def metrics() -> Response: + for slot in system.registry.slots(): + system.metrics.slot_status.labels(slot.name).set(1 if slot.available else 0) + system.metrics.open_proposals.set(len(system.maintenance.proposals("open"))) + return Response(system.metrics.render(), media_type="text/plain; version=0.0.4") + + # --- Routing ---------------------------------------------------------------------------- + @app.post("/v1/route", dependencies=v1) + def route(body: RouteIn) -> dict[str, Any]: + try: + return system.router.route(body.input, hint_skill=body.skill).model_dump(mode="json") + except KeyError as exc: + raise HTTPException(404, str(exc)) from exc + + @app.post("/v1/run", dependencies=v1) + def run(body: RouteIn) -> dict[str, Any]: + try: + result = system.runner.run(body.input, hint_skill=body.skill) + except KeyError as exc: + raise HTTPException(404, str(exc)) from exc + data = result.model_dump(mode="json") + data["escalations"] = result.escalations + return data + + # --- Registry --------------------------------------------------------------------------- + @app.get("/v1/registry", dependencies=v1) + def registry() -> dict[str, Any]: + return { + "slots": [{**s.spec.model_dump(), "status": s.status, "reason": s.reason, "set_by": s.set_by, + "effective_context": s.effective_context, "base_model": s.base_model, + "canary_model": s.canary_model, "canary_share": s.canary_share} + for s in system.registry.slots()], + "adapters": [a.model_dump() for a in system.registry.adapters()], + "skills": [{"name": s.name, "status": s.meta.status, "stage": s.meta.stage, "slot": s.meta.slot, + "cluster": s.cluster, "version": s.meta.version} for s in system.library.all()], + } + + @app.post("/v1/registry/slots/{slot}/state", dependencies=v1) + def slot_state(slot: str, body: SlotStateIn) -> dict[str, Any]: + if not system.registry.has_slot(slot): + raise HTTPException(404, f"Slot {slot} unbekannt") + if body.status == "ok" and body.set_by != "human": + # Wieder-Freischalten nur über Vorschlag/Freigabe, außer ein Mensch setzt es direkt + p = system.maintenance.propose("mark_slot", {"slot": slot, "status": "ok", "reason": body.reason}, + f"Slot {slot} wieder auf ok setzen", source="api") + return {"proposal": p.id} + system.registry.set_status(slot, body.status, reason=body.reason, set_by=body.set_by) # type: ignore + return {"slot": slot, "status": body.status} + + # --- Pflege ----------------------------------------------------------------------------- + @app.post("/v1/maintenance/routine", dependencies=v1) + def routine() -> dict[str, Any]: + report = system.maintenance.run_routine() + return {**report.model_dump(mode="json"), "proposal_ids": [p.id for p in report.proposals]} + + @app.post("/v1/maintenance/command", dependencies=v1) + def command(body: CommandIn) -> dict[str, Any]: + reply = system.maintenance.handle_command(body.text, user=body.user, session=body.session) + return {"text": reply.text, "proposals": [{**p.model_dump(mode="json"), "id": p.id} + for p in reply.proposals]} + + @app.get("/v1/maintenance/proposals", dependencies=v1) + def proposals(state: str | None = "open") -> list[dict[str, Any]]: + return [{**p.model_dump(mode="json"), "id": p.id} for p in system.maintenance.proposals(state)] + + @app.post("/v1/maintenance/proposals/{seq}/{decision}", dependencies=v1) + def decide(seq: int, decision: str, body: DecisionIn) -> dict[str, Any]: + try: + if decision == "approve": + p = system.maintenance.approve(seq, by=body.user) + elif decision == "reject": + p = system.maintenance.reject(seq, by=body.user, reason=body.reason) + else: + raise HTTPException(404, "approve oder reject") + except KeyError as exc: + raise HTTPException(404, str(exc)) from exc + except ValueError as exc: + raise HTTPException(409, str(exc)) from exc + return {**p.model_dump(mode="json"), "id": p.id} + + # --- Healer ----------------------------------------------------------------------------- + @app.post("/v1/healer/alertmanager", dependencies=v1) + def alertmanager(payload: dict[str, Any]) -> dict[str, Any]: + outcomes = system.healer.handle_alertmanager(payload) + return {"outcomes": [o.model_dump() for o in outcomes]} + + # --- Lifecycle -------------------------------------------------------------------------- + @app.get("/v1/lifecycle/candidates", dependencies=v1) + def candidates(state: str | None = None) -> list[dict[str, Any]]: + return system.store.list_candidates(state=state) + + @app.post("/v1/lifecycle/candidates", dependencies=v1) + def new_candidate(body: CandidateIn) -> dict[str, Any]: + if not system.registry.has_slot(body.slot): + raise HTTPException(404, f"Slot {body.slot} unbekannt") + p = system.maintenance.propose("create_candidate", + {"kind": body.kind, "slot": body.slot, "target": body.target}, + f"Kandidat {body.kind} für {body.slot}: {body.target}", + reason=f"angefragt von {body.requested_by}", source="api") + return {**p.model_dump(mode="json"), "id": p.id} + + @app.post("/v1/lifecycle/candidates/{cid}/report", dependencies=v1) + def report(cid: str, body: ReportIn) -> dict[str, Any]: + try: + return system.lifecycle.report(cid, body.step, body.payload) + except LifecycleError as exc: + raise HTTPException(409, str(exc)) from exc + + # --- Skills ----------------------------------------------------------------------------- + @app.post("/v1/skills/{name}/graduate", dependencies=v1) + def graduate_skill(name: str, body: GraduateIn) -> dict[str, Any]: + if not system.library.has(name): + raise HTTPException(404, f"Skill {name} unbekannt") + rep = graduate(name, body.variants, router=system.router, executor=system.executor, + verifier=system.verifier, library=system.library, store=system.store, + threshold=body.threshold, min_variants=body.min_variants) + out: dict[str, Any] = {"report": rep.model_dump()} + if rep.graduated: + p = system.maintenance.propose("promote_skill", {"skill": name, "stage": 1}, + f"Skill {name} für Stufe 1 freigeben", + reason=f"Graduierung {rep.passed_n}/{rep.n} ({rep.pass_rate:.0%})", + source="graduation") + out["proposal"] = p.id + return out + + @app.post("/v1/skills/reload", dependencies=v1) + def reload_skills() -> dict[str, Any]: + system.library.reload(system.settings.skills_dir) + system.router.refresh() + return {"skills": len(system.library.all())} + + return app diff --git a/core/src/knaster/cli.py b/core/src/knaster/cli.py new file mode 100644 index 0000000..2b2c963 --- /dev/null +++ b/core/src/knaster/cli.py @@ -0,0 +1,82 @@ +"""Kommandozeile: serve, validate, inventory, placement, registry export.""" + +from __future__ import annotations + +import argparse +import sys +from pathlib import Path + +import yaml + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(prog="knaster") + sub = parser.add_subparsers(dest="cmd", required=True) + + p_serve = sub.add_parser("serve", help="Router-Service starten") + p_serve.add_argument("--host", default="0.0.0.0") + p_serve.add_argument("--port", type=int, default=8080) + + p_val = sub.add_parser("validate", help="Konfiguration prüfen") + p_val.add_argument("--root", default=".") + p_val.add_argument("--site", default="knaster") + + p_inv = sub.add_parser("inventory", help="Inventar") + inv_sub = p_inv.add_subparsers(dest="inv_cmd", required=True) + p_col = inv_sub.add_parser("collect", help="lokales System auslesen") + p_col.add_argument("--name") + + p_plan = sub.add_parser("placement", help="Placement-Plan") + p_plan.add_argument("--inventory", required=True) + p_plan.add_argument("--catalog", default="catalog") + + p_reg = sub.add_parser("registry", help="Registry") + reg_sub = p_reg.add_subparsers(dest="reg_cmd", required=True) + reg_sub.add_parser("export", help="effektiven Stand (inkl. Laufzeitzustand) als YAML ausgeben") + + args = parser.parse_args(argv) + + if args.cmd == "serve": + import uvicorn + + uvicorn.run("knaster.api:create_app", factory=True, host=args.host, port=args.port) + return 0 + + if args.cmd == "validate": + from .validate import validate_repo + + issues = validate_repo(Path(args.root), args.site) + for level, msg in issues: + print(f"{level.upper():8} {msg}") + errors = [i for i in issues if i[0] == "error"] + print(f"{len(errors)} Fehler, {len(issues) - len(errors)} Warnungen") + return 1 if errors else 0 + + if args.cmd == "inventory": + from .inventory.profile import collect + + print(yaml.safe_dump({"nodes": [collect(args.name).model_dump()]}, sort_keys=False, allow_unicode=True)) + return 0 + + if args.cmd == "placement": + from .inventory.placement import load_catalog, load_nodes, plan + + roles, models = load_catalog(args.catalog) + result = plan(load_nodes(args.inventory), roles, models) + print(yaml.safe_dump(result.model_dump(), sort_keys=False, allow_unicode=True)) + return 1 if any(not u.optional for u in result.unplaced) else 0 + + if args.cmd == "registry": + from .registry import Registry + from .settings import Settings + from .store import Store + + settings = Settings.from_env() + registry = Registry.from_file(settings.site_dir / "registry.yaml", Store(settings.db_url)) + print(yaml.safe_dump(registry.export(), sort_keys=False, allow_unicode=True)) + return 0 + return 2 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/core/src/knaster/inventory/__init__.py b/core/src/knaster/inventory/__init__.py new file mode 100644 index 0000000..f8eded2 --- /dev/null +++ b/core/src/knaster/inventory/__init__.py @@ -0,0 +1 @@ +"""Inventar, Modell-Katalog und Placement (Übertragbarkeit auf andere Cluster).""" diff --git a/core/src/knaster/inventory/placement.py b/core/src/knaster/inventory/placement.py new file mode 100644 index 0000000..98ab6df --- /dev/null +++ b/core/src/knaster/inventory/placement.py @@ -0,0 +1,153 @@ +"""Deterministisches Placement: Rollen → GPUs/Nodes anhand von Katalog-Anforderungen.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Literal + +from pydantic import BaseModel, Field + +from ..util import load_yaml +from .profile import GPU, NodeProfile + + +class ModelSpec(BaseModel): + name: str + kind: Literal["llm", "embedding", "reranker"] + vram_gb: float + min_cc: float = 5.0 + engines: list[str] + context: int = 8192 + lora: bool = False + cpu_ok: bool = False + note: str = "" + + +class Needs(BaseModel): + gpu: bool = True + engine: str | None = None + min_vram_gb: float = 0 + min_cc: float = 0 + max_cc: float | None = None + min_ram_gb: float = 0 + + +class RoleSpec(BaseModel): + name: str + priority: int = 100 + count: int = 1 + needs: Needs = Field(default_factory=Needs) + prefer: Literal["largest", "smallest_fitting"] = "largest" + kinds: list[str] = Field(default_factory=list) + cpu_fallback: bool = False + optional: bool = False + slots: list[str] = Field(default_factory=list) + + +class Assignment(BaseModel): + role: str + node: str + gpu_index: int | None = None + gpu_name: str | None = None + vram_gb: float | None = None + mode: Literal["gpu", "cpu", "cpu_fallback"] + models_fitting: list[str] = Field(default_factory=list) + slots: list[str] = Field(default_factory=list) + + +class Unplaced(BaseModel): + role: str + reason: str + optional: bool = False + + +class PlacementPlan(BaseModel): + assignments: list[Assignment] = Field(default_factory=list) + unplaced: list[Unplaced] = Field(default_factory=list) + free_gpus: list[str] = Field(default_factory=list) + + def roles_on(self, node: str) -> list[str]: + return [a.role for a in self.assignments if a.node == node] + + +def load_catalog(directory: str | Path) -> tuple[list[RoleSpec], list[ModelSpec]]: + d = Path(directory) + roles = [RoleSpec(name=k, **v) for k, v in (load_yaml(d / "roles.yaml").get("roles") or {}).items()] + models = [ModelSpec(name=k, **v) for k, v in (load_yaml(d / "models.yaml").get("models") or {}).items()] + return roles, models + + +def load_nodes(path: str | Path) -> list[NodeProfile]: + return [NodeProfile(**n) for n in load_yaml(path).get("nodes") or []] + + +def _gpu_ok(gpu: GPU, needs: Needs) -> bool: + if gpu.memory_gb < needs.min_vram_gb or gpu.compute_cap < needs.min_cc: + return False + return needs.max_cc is None or gpu.compute_cap <= needs.max_cc + + +def _fitting_models(role: RoleSpec, models: list[ModelSpec], gpu: GPU | None) -> list[str]: + out = [] + for m in models: + if m.kind not in role.kinds: + continue + if gpu is None: + if m.cpu_ok: + out.append(m.name) + continue + if role.needs.engine and role.needs.engine not in m.engines: + continue + if m.vram_gb <= gpu.memory_gb and m.min_cc <= gpu.compute_cap: + out.append(m.name) + return out + + +def plan(nodes: list[NodeProfile], roles: list[RoleSpec], models: list[ModelSpec]) -> PlacementPlan: + result = PlacementPlan() + free: list[tuple[NodeProfile, GPU]] = [(n, g) for n in nodes for g in n.gpus] + cpu_load: dict[str, int] = {n.name: 0 for n in nodes} + + for role in sorted(roles, key=lambda r: (r.priority, r.name)): + placed = 0 + if role.needs.gpu: + used_nodes: set[str] = set() + for _ in range(role.count): + options = [(n, g) for n, g in free if _gpu_ok(g, role.needs) + and _fitting_models(role, models, g)] + if not options: + break + # Anti-Affinität: Replikate einer Rolle möglichst auf verschiedene Hosts + if role.prefer == "largest": + options.sort(key=lambda x: (x[0].name in used_nodes, -x[1].memory_gb, -x[1].compute_cap, + x[0].name, x[1].index)) + else: + options.sort(key=lambda x: (x[0].name in used_nodes, x[1].memory_gb, -x[1].compute_cap, + x[0].name, x[1].index)) + node, gpu = options[0] + free.remove((node, gpu)) + used_nodes.add(node.name) + result.assignments.append(Assignment( + role=role.name, node=node.name, gpu_index=gpu.index, gpu_name=gpu.name, + vram_gb=gpu.memory_gb, mode="gpu", models_fitting=_fitting_models(role, models, gpu), + slots=role.slots)) + placed += 1 + missing = role.count - placed + if missing and (not role.needs.gpu or role.cpu_fallback): + mode: Literal["cpu", "cpu_fallback"] = "cpu" if not role.needs.gpu else "cpu_fallback" + candidates = [n for n in nodes if n.memory_gb >= role.needs.min_ram_gb] + # bevorzugt Nodes ohne GPU-Rollen, dann wenig belastete, dann viel RAM + gpu_roles = {a.node for a in result.assignments if a.mode == "gpu"} + candidates.sort(key=lambda n: (n.name in gpu_roles, cpu_load[n.name], -n.memory_gb, n.name)) + for node in candidates[:missing]: + cpu_load[node.name] += 1 + result.assignments.append(Assignment(role=role.name, node=node.name, mode=mode, + models_fitting=_fitting_models(role, models, None), + slots=role.slots)) + missing -= 1 + if missing: + reason = "keine passende GPU frei" if role.needs.gpu else "kein Node mit genug RAM" + result.unplaced.append(Unplaced(role=role.name, reason=f"{reason} ({missing} fehlen)", + optional=role.optional)) + result.free_gpus = [f"{n.name}/gpu{g.index} {g.name} ({g.memory_gb} GB)" for n, g in free] + return result diff --git a/core/src/knaster/inventory/profile.py b/core/src/knaster/inventory/profile.py new file mode 100644 index 0000000..c22fba6 --- /dev/null +++ b/core/src/knaster/inventory/profile.py @@ -0,0 +1,108 @@ +"""Node-Profile und Parser für Systemausgaben (reine Funktionen, testbar).""" + +from __future__ import annotations + +import csv +import io +import json +import re +import shutil +import socket +import subprocess +from pathlib import Path + +from pydantic import BaseModel, Field + + +class GPU(BaseModel): + index: int + name: str + memory_gb: float + compute_cap: float + driver: str | None = None + + +class Disk(BaseModel): + name: str + size_gb: float + rotational: bool = False + + +class CPU(BaseModel): + model: str = "" + cores: int = 0 + threads: int = 0 + + +class NodeProfile(BaseModel): + name: str + cpu: CPU = Field(default_factory=CPU) + memory_gb: float = 0 + disks: list[Disk] = Field(default_factory=list) + gpus: list[GPU] = Field(default_factory=list) + labels: dict[str, str] = Field(default_factory=dict) + + +NVIDIA_SMI_QUERY = ["nvidia-smi", "--query-gpu=index,name,memory.total,compute_cap,driver_version", + "--format=csv,noheader,nounits"] + + +def parse_nvidia_smi(text: str) -> list[GPU]: + gpus = [] + for row in csv.reader(io.StringIO(text.strip())): + if not row or not row[0].strip(): + continue + idx, name, mem_mib, cc, driver = (c.strip() for c in row[:5]) + gpus.append(GPU(index=int(idx), name=name, memory_gb=round(float(mem_mib) / 1024, 1), + compute_cap=float(cc), driver=driver)) + return gpus + + +def parse_meminfo(text: str) -> float: + m = re.search(r"^MemTotal:\s+(\d+)\s+kB", text, re.MULTILINE) + return round(int(m.group(1)) / 1024 / 1024, 1) if m else 0.0 + + +def parse_lscpu(text: str) -> CPU: + fields = {} + for line in text.splitlines(): + if ":" in line: + k, v = line.split(":", 1) + fields[k.strip()] = v.strip() + threads = int(fields.get("CPU(s)", "0") or 0) + cores = int(fields.get("Core(s) per socket", "0") or 0) * int(fields.get("Socket(s)", "1") or 1) + return CPU(model=fields.get("Model name", ""), cores=cores, threads=threads) + + +def parse_lsblk(text: str) -> list[Disk]: + data = json.loads(text) + disks = [] + for dev in data.get("blockdevices", []): + rota = dev.get("rota") + disks.append(Disk(name=dev["name"], size_gb=round(int(dev["size"]) / 1e9, 1), + rotational=rota in (True, "1", 1))) + return disks + + +def _run(cmd: list[str]) -> str | None: + if shutil.which(cmd[0]) is None: + return None + try: + return subprocess.run(cmd, capture_output=True, text=True, timeout=20, check=True).stdout + except (subprocess.SubprocessError, OSError): + return None + + +def collect(name: str | None = None) -> NodeProfile: + """Liest das lokale System aus (Linux). Fehlende Werkzeuge ergeben leere Felder.""" + smi = _run(NVIDIA_SMI_QUERY) + lscpu = _run(["lscpu"]) + lsblk = _run(["lsblk", "-J", "-b", "-d", "-o", "NAME,SIZE,ROTA"]) + meminfo = Path("/proc/meminfo").read_text() if Path("/proc/meminfo").exists() else "" + return NodeProfile( + name=name or socket.gethostname(), + cpu=parse_lscpu(lscpu) if lscpu else CPU(), + memory_gb=parse_meminfo(meminfo), + disks=parse_lsblk(lsblk) if lsblk else [], + gpus=parse_nvidia_smi(smi) if smi else [], + ) diff --git a/core/src/knaster/lifecycle.py b/core/src/knaster/lifecycle.py index 26d5683..c86c3f2 100644 --- a/core/src/knaster/lifecycle.py +++ b/core/src/knaster/lifecycle.py @@ -176,7 +176,8 @@ class Lifecycle: 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] + 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 ----------------------------------------------------------------- diff --git a/core/src/knaster/router.py b/core/src/knaster/router.py index 2243f1a..846e42e 100644 --- a/core/src/knaster/router.py +++ b/core/src/knaster/router.py @@ -46,6 +46,11 @@ class RouterConfig: default_max_output_tokens: int = 1024 +def adapter_model(slot_alias: str, adapter: str) -> str: + """LiteLLM-Name eines Adapters: '/' (Wildcard-Route auf den vLLM des Slots).""" + return f"{slot_alias}/{adapter}" + + def system_prompt(skill: Skill | None, variant: PromptVariant) -> str: parts: list[str] = [] if skill is not None and variant != "none": @@ -211,15 +216,16 @@ class Router: if slot.spec.supports_lora and skill.meta.adapter == "auto": found = self.registry.active_adapter(slot.base_model, skill.cluster) if found: - model, adapter = found.name, found.name + model, adapter = adapter_model(slot.spec.litellm_model, found.name), found.name reasons.append(f"Adapter {found.name}") canary = False if (slot.canary_model and slot.canary_cluster in (None, skill.cluster) and self.rng.random() < slot.canary_share): canary = True - model = slot.canary_model - if slot.canary_cluster: - adapter = slot.canary_model + if slot.canary_cluster: # Adapter-Kandidat + model, adapter = adapter_model(slot.spec.litellm_model, slot.canary_model), slot.canary_model + else: # Basis-/Engine-Kandidat unter eigenem LiteLLM-Alias + model = slot.canary_model reasons.append(f"Canary {slot.canary_model}") return self._decision(stage=1, slot=slot.name, model=model, adapter=adapter, skill=skill, canary=canary, prompt_variant=variant, similarity=similarity, diff --git a/core/src/knaster/skills.py b/core/src/knaster/skills.py index 7a30dfc..2aa1a3d 100644 --- a/core/src/knaster/skills.py +++ b/core/src/knaster/skills.py @@ -116,6 +116,10 @@ class SkillLibrary: skills.append(parse_skill(file.read_text(encoding="utf-8"), str(file))) return cls(skills, store) + def reload(self, directory: str | Path) -> None: + """Lädt die Dateien neu; das Objekt bleibt, damit alle Referenzen (Router, Agent) gültig bleiben.""" + self._skills = SkillLibrary.from_dir(directory)._skills + def _states(self) -> dict[str, dict[str, Any]]: if not self.store: return {} diff --git a/core/src/knaster/validate.py b/core/src/knaster/validate.py new file mode 100644 index 0000000..c328134 --- /dev/null +++ b/core/src/knaster/validate.py @@ -0,0 +1,96 @@ +"""Konsistenzprüfung der Konfiguration (CI und `knaster validate`).""" + +from __future__ import annotations + +from pathlib import Path + +from pydantic import ValidationError + +from . import programs +from .healer.engine import HealerConfig +from .inventory.placement import load_catalog +from .maintenance.actions import ActionExecutor +from .policy import Policy +from .registry import Registry +from .skills import SkillLibrary +from .store import Store +from .util import load_yaml + + +def _known_actions() -> set[str]: + return {name.removeprefix("_do_") for name in dir(ActionExecutor) if name.startswith("_do_")} + + +def validate_repo(root: str | Path, site: str = "knaster") -> list[tuple[str, str]]: + root = Path(root) + issues: list[tuple[str, str]] = [] + + def err(msg: str) -> None: + issues.append(("error", msg)) + + def warn(msg: str) -> None: + issues.append(("warning", msg)) + + store = Store("sqlite://") + try: + registry = Registry.from_file(root / "sites" / site / "registry.yaml", store) + except (ValidationError, ValueError, OSError) as exc: + return [("error", f"Registry: {exc}")] + slots = {s.name: s for s in registry.slots()} + if not any(s.spec.tier == 0 and s.spec.kind == "chat" for s in slots.values()): + err("Registry: kein Stufe-0-Slot (tier 0, kind chat)") + + try: + library = SkillLibrary.from_dir(root / "skills") + except (ValidationError, ValueError) as exc: + return issues + [("error", f"Skills: {exc}")] + for skill in library.all(): + m = skill.meta + if m.slot not in slots: + err(f"Skill {skill.name}: Slot {m.slot} unbekannt") + elif slots[m.slot].spec.kind != "chat": + err(f"Skill {skill.name}: Slot {m.slot} ist kein Chat-Slot") + for check in m.verifier: + if check.get("type") == "judge" and check.get("slot", "verifier-small") not in slots: + err(f"Skill {skill.name}: Judge-Slot {check.get('slot', 'verifier-small')} unbekannt") + if m.execution == "program" and m.program and m.program not in programs.names(): + (err if m.stage >= 2 else warn)(f"Skill {skill.name}: Programm {m.program} nicht registriert") + if m.execution == "classifier" and m.classifier: + path = root / "config" / "classifiers" / f"{m.classifier}.json" + if not path.exists(): + (err if m.stage >= 3 else warn)(f"Skill {skill.name}: Classifier {path.name} fehlt") + if m.output.format == "json" and not m.output.schema_: + warn(f"Skill {skill.name}: JSON-Ausgabe ohne Schema") + + litellm = root / "deploy" / "litellm" / "config.yaml" + if litellm.exists(): + names = {m["model_name"] for m in load_yaml(litellm).get("model_list") or []} + for s in slots.values(): + if s.spec.litellm_model not in names: + err(f"LiteLLM: Alias {s.spec.litellm_model} (Slot {s.name}) fehlt in {litellm.name}") + else: + warn("LiteLLM-Konfiguration fehlt") + + known = _known_actions() + cfg = root / "config" + if (cfg / "healer-rules.yaml").exists(): + healer = HealerConfig.from_file(cfg / "healer-rules.yaml") + for rule in healer.rules: + for step in [*rule.ladder, *rule.on_resolved]: + if step.action not in known: + err(f"Healer-Regel {rule.id}: unbekannte Aktion {step.action}") + for a in healer.enabled_actions: + if a not in {"restart_container", "restart_vm", "reboot_host"}: + err(f"Healer: enabled_actions enthält {a} (nur Infrastruktur-Aktionen erlaubt)") + if (cfg / "policy.yaml").exists(): + for kind in Policy.from_file(cfg / "policy.yaml").actions: + if kind not in known: + err(f"Policy: unbekannte Aktion {kind}") + + if (root / "catalog").exists(): + roles, _ = load_catalog(root / "catalog") + for role in roles: + for slot in role.slots: + if slot not in slots: + err(f"Rolle {role.name}: Slot {slot} fehlt in der Registry") + return issues diff --git a/core/tests/test_api.py b/core/tests/test_api.py new file mode 100644 index 0000000..bae7736 --- /dev/null +++ b/core/tests/test_api.py @@ -0,0 +1,110 @@ +import json + +import pytest +from fastapi.testclient import TestClient + +from knaster.api import create_app + +from .test_router import RECHNUNG + +GOOD = json.dumps({"nummer": "4711", "betrag": 99}) +H = {"Authorization": "Bearer secret"} + + +@pytest.fixture +def client(make_system): + sysm = make_system() + sysm.chat.handlers.update({"tier1-general/rechnungen-v1": lambda m: '{"nummer": "1"}', + "tier0-strong": lambda m: GOOD}) + return TestClient(create_app(sysm)) + + +def test_auth(client): + assert client.post("/v1/route", json={"input": "x"}).status_code == 401 + assert client.get("/health").status_code == 200 + + +def test_route_and_run(client): + r = client.post("/v1/route", json={"input": RECHNUNG}, headers=H).json() + assert r["stage"] == 1 and r["adapter"] == "rechnungen-v1" + r = client.post("/v1/run", json={"input": RECHNUNG}, headers=H).json() + assert r["status"] == "ok" and r["stage"] == 0 and r["escalations"] == 1 + assert r["attempts"][0]["verification"]["passed"] is False + assert client.post("/v1/route", json={"input": "x", "skill": "gibtsnicht"}, headers=H).status_code == 404 + + +def test_metrics(client): + client.post("/v1/run", json={"input": RECHNUNG}, headers=H) + text = client.get("/metrics").text + assert 'knaster_runs_total{stage="0",status="ok"} 1.0' in text + assert 'knaster_escalations_total{signal="verifier_failed"} 1.0' in text + assert 'knaster_slot_available{slot="tier1-general"} 1.0' in text + + +def test_registry_and_slot_state(client): + reg = client.get("/v1/registry", headers=H).json() + assert {s["name"] for s in reg["slots"]} >= {"tier1-general", "tier0-strong"} + r = client.post("/v1/registry/slots/tier1-general/state", json={"status": "degraded", "reason": "x"}, + headers=H) + assert r.json()["status"] == "degraded" + r = client.post("/v1/registry/slots/tier1-general/state", json={"status": "ok", "set_by": "n8n"}, headers=H) + assert "proposal" in r.json() + assert client.post("/v1/registry/slots/nope/state", json={"status": "ok"}, headers=H).status_code == 404 + + +def test_maintenance_flow(client): + r = client.post("/v1/maintenance/command", json={"text": "markiere tier1-flex als degraded", "user": "robert"}, + headers=H).json() + pid = r["proposals"][0]["id"] + seq = r["proposals"][0]["seq"] + open_ = client.get("/v1/maintenance/proposals", headers=H).json() + assert [p["id"] for p in open_] == [pid] + r = client.post(f"/v1/maintenance/proposals/{seq}/approve", json={"user": "robert"}, headers=H).json() + assert r["state"] == "executed" + assert client.post(f"/v1/maintenance/proposals/{seq}/approve", json={}, headers=H).status_code == 409 + rep = client.post("/v1/maintenance/routine", headers=H).json() + assert "Service-Routine" in rep["text"] + + +def test_healer_endpoint(client): + payload = {"version": "4", "status": "firing", "alerts": [ + {"status": "firing", "labels": {"alertname": "GpuTemperatureHigh", "slot": "tier1-flex"}, + "fingerprint": "abc"}]} + r = client.post("/v1/healer/alertmanager", json=payload, headers=H).json() + assert r["outcomes"][0]["status"] == "executed" + reg = client.get("/v1/registry", headers=H).json() + assert next(s for s in reg["slots"] if s["name"] == "tier1-flex")["status"] == "degraded" + + +def test_candidate_via_api_needs_approval_then_reports(client): + r = client.post("/v1/lifecycle/candidates", json={"kind": "adapter", "slot": "tier1-general", + "target": {"cluster": "rechnungen"}}, headers=H).json() + assert r["state"] == "open" + client.post(f"/v1/maintenance/proposals/{r['seq']}/approve", json={}, headers=H) + cands = client.get("/v1/lifecycle/candidates", headers=H).json() + cid = cands[0]["id"] + assert cands[0]["state"] == "building" + r = client.post(f"/v1/lifecycle/candidates/{cid}/report", json={"step": "build", "payload": {"model": "rv2"}}, + headers=H).json() + assert r["state"] == "evaluating" + assert client.post(f"/v1/lifecycle/candidates/{cid}/report", json={"step": "canary"}, + headers=H).status_code == 409 + + +def test_graduation_endpoint(client): + app_sys = client.app.state.system + app_sys.chat.handlers["tier1-general/rechnungen-v1"] = lambda m: GOOD + variants = [f"{RECHNUNG} Variante {i}" for i in range(10)] + r = client.post("/v1/skills/rechnung-extrahieren/graduate", json={"variants": variants}, headers=H).json() + assert r["report"]["graduated"] and "proposal" in r + cases = app_sys.store.list_cases() + assert len(cases) == 10 and all(c["origin"] == "graduation" for c in cases) + + app_sys.chat.handlers["tier1-general/rechnungen-v1"] = lambda m: "kaputt" + r = client.post("/v1/skills/rechnung-extrahieren/graduate", json={"variants": variants}, headers=H).json() + assert not r["report"]["graduated"] and r["report"]["failures"] + assert client.post("/v1/skills/nope/graduate", json={"variants": []}, headers=H).status_code == 404 + + +def test_reload_skills(client): + assert client.post("/v1/skills/reload", headers=H).json()["skills"] == 5 diff --git a/core/tests/test_execution.py b/core/tests/test_execution.py index 516c97d..469ac66 100644 --- a/core/tests/test_execution.py +++ b/core/tests/test_execution.py @@ -41,7 +41,7 @@ def make_runner(registry, library, embedder, store, fake_chat, classifiers): def test_weak_success_records_case(make_runner, store): - runner = make_runner(**{"rechnungen-v1": lambda m: GOOD}) + runner = make_runner(**{"tier1-general/rechnungen-v1": lambda m: GOOD}) res = runner.run(RECHNUNG) assert res.status == "ok" and res.stage == 1 and res.escalations == 0 cases = store.list_cases() @@ -51,7 +51,7 @@ def test_weak_success_records_case(make_runner, store): def test_schema_failure_escalates_and_creates_pair(make_runner, store): runner = make_runner(**{ - "rechnungen-v1": lambda m: json.dumps({"nummer": "4711"}), # betrag fehlt + "tier1-general/rechnungen-v1": lambda m: json.dumps({"nummer": "4711"}), # betrag fehlt "tier0-strong": lambda m: GOOD, }) res = runner.run(RECHNUNG) @@ -63,7 +63,7 @@ def test_schema_failure_escalates_and_creates_pair(make_runner, store): def test_invalid_json_signal(make_runner, store): - runner = make_runner(**{"rechnungen-v1": lambda m: "kein json", "tier0-strong": lambda m: GOOD}) + runner = make_runner(**{"tier1-general/rechnungen-v1": lambda m: "kein json", "tier0-strong": lambda m: GOOD}) res = runner.run(RECHNUNG) assert Signal.INVALID_FORMAT in res.attempts[0].execution.signals assert res.stage == 0 @@ -74,7 +74,7 @@ def test_endpoint_error_uses_fallback_slot_without_case(make_runner, store): def boom(m): raise EndpointError("down") - runner = make_runner(**{"rechnungen-v1": boom}) + runner = make_runner(**{"tier1-general/rechnungen-v1": boom}) # Fallback-Slot tier1-flex hat dieselbe Basis → gleicher Adapter-Name; daher Fehler nur einmal calls = {"n": 0} @@ -84,7 +84,8 @@ def test_endpoint_error_uses_fallback_slot_without_case(make_runner, store): raise EndpointError("down") return GOOD - runner.executor.chat.handlers["rechnungen-v1"] = flaky + runner.executor.chat.handlers["tier1-general/rechnungen-v1"] = flaky + runner.executor.chat.handlers["tier1-flex/rechnungen-v1"] = lambda m: GOOD res = runner.run(RECHNUNG) assert res.status == "ok" assert [a.decision.slot for a in res.attempts] == ["tier1-general", "tier1-flex"] @@ -96,7 +97,7 @@ def test_context_overflow_detected(make_runner): def overflow(m): raise ContextOverflowError("maximum context length") - runner = make_runner(**{"rechnungen-v1": overflow, "tier0-strong": lambda m: GOOD}) + runner = make_runner(**{"tier1-general/rechnungen-v1": overflow, "tier0-strong": lambda m: GOOD}) res = runner.run(RECHNUNG) assert res.attempts[0].execution.signals == [Signal.CONTEXT_OVERFLOW] assert res.status == "ok" @@ -105,7 +106,7 @@ def test_context_overflow_detected(make_runner): def test_truncated_and_degenerate(make_runner): loop = " ".join(["immer wieder das gleiche"] * 30) runner = make_runner(**{ - "rechnungen-v1": lambda m: ChatResult(content=loop, finish_reason="length"), + "tier1-general/rechnungen-v1": lambda m: ChatResult(content=loop, finish_reason="length"), "tier0-strong": lambda m: GOOD, }) res = runner.run(RECHNUNG) @@ -114,7 +115,7 @@ def test_truncated_and_degenerate(make_runner): def test_total_failure(make_runner): - runner = make_runner(**{"rechnungen-v1": lambda m: "x", "tier0-strong": lambda m: "y"}) + runner = make_runner(**{"tier1-general/rechnungen-v1": lambda m: "x", "tier0-strong": lambda m: "y"}) res = runner.run(RECHNUNG) assert res.status == "failed" @@ -193,7 +194,7 @@ def test_split_is_deterministic(): def test_exports(make_runner, store): - runner = make_runner(**{"rechnungen-v1": lambda m: '{"nummer": "1"}', "tier0-strong": lambda m: GOOD}) + runner = make_runner(**{"tier1-general/rechnungen-v1": lambda m: '{"nummer": "1"}', "tier0-strong": lambda m: GOOD}) for i in range(30): runner.run(f"{RECHNUNG} Nr {i}") train = list(export_sft(store, cluster="rechnungen", split="train")) diff --git a/core/tests/test_inventory_validate.py b/core/tests/test_inventory_validate.py new file mode 100644 index 0000000..195765c --- /dev/null +++ b/core/tests/test_inventory_validate.py @@ -0,0 +1,90 @@ +import json + +from knaster.cli import main +from knaster.inventory.placement import load_catalog, load_nodes, plan +from knaster.inventory.profile import NodeProfile, parse_lsblk, parse_lscpu, parse_meminfo, parse_nvidia_smi +from knaster.settings import REPO_ROOT +from knaster.validate import validate_repo + +SMI = """0, NVIDIA RTX 4000 Ada Generation, 20475, 8.9, 570.86 +1, Quadro P2000, 5120, 6.1, 535.183 +""" +LSCPU = """Architecture: x86_64 +CPU(s): 8 +Model name: Intel(R) Core(TM) i7-7700 CPU @ 3.60GHz +Core(s) per socket: 4 +Socket(s): 1 +""" + + +def test_parsers(): + gpus = parse_nvidia_smi(SMI) + assert [(g.name, g.memory_gb, g.compute_cap) for g in gpus] == [ + ("NVIDIA RTX 4000 Ada Generation", 20.0, 8.9), ("Quadro P2000", 5.0, 6.1)] + assert parse_meminfo("MemTotal: 32768000 kB\nMemFree: 1 kB") == 31.2 + cpu = parse_lscpu(LSCPU) + assert (cpu.cores, cpu.threads) == (4, 8) and "i7-7700" in cpu.model + disks = parse_lsblk(json.dumps({"blockdevices": [{"name": "sda", "size": 256060514304, "rota": False}]})) + assert disks[0].size_gb == 256.1 and not disks[0].rotational + + +def test_placement_knaster(): + roles, models = load_catalog(REPO_ROOT / "catalog") + p = plan(load_nodes(REPO_ROOT / "sites/knaster/inventory.yaml"), roles, models) + by_role = {} + for a in p.assignments: + by_role.setdefault(a.role, []).append(a) + assert by_role["gpu-main"][0].gpu_name.startswith("NVIDIA RTX 4000 Ada") + assert "qwen3-14b-awq" in by_role["gpu-main"][0].models_fitting + embed_nodes = {a.node for a in by_role["gpu-embed"]} + assert len(embed_nodes) == 2 # Anti-Affinität + assert all("P2000" in a.gpu_name for a in by_role["gpu-embed"] + by_role["gpu-small"]) + assert by_role["infra"][0].mode == "cpu" + assert not p.unplaced + assert len(p.free_gpus) == 4 and all("M4000" in g for g in p.free_gpus) + + +def test_placement_other_cluster(): + roles, models = load_catalog(REPO_ROOT / "catalog") + p = plan(load_nodes(REPO_ROOT / "sites/example-single-gpu/inventory.yaml"), roles, models) + modes = {(a.role, a.mode) for a in p.assignments} + assert ("gpu-main", "gpu") in modes and ("gpu-embed", "cpu_fallback") in modes + assert [u.role for u in p.unplaced] == ["gpu-flex"] and p.unplaced[0].optional + + +def test_placement_maxwell_never_gets_vllm(): + roles, models = load_catalog(REPO_ROOT / "catalog") + nodes = [NodeProfile(name="old", memory_gb=32, gpus=[ + {"index": 0, "name": "Quadro M6000", "memory_gb": 24, "compute_cap": 5.2}])] + p = plan(nodes, roles, models) + assert "gpu-main" in [u.role for u in p.unplaced] + small = next(a for a in p.assignments if a.role == "gpu-embed" and a.mode == "gpu") + assert small.node == "old" + + +def test_repo_validates_without_errors(): + issues = validate_repo(REPO_ROOT) + assert [i for i in issues if i[0] == "error"] == [], issues + + +def test_validate_detects_problems(tmp_path): + import shutil + + root = tmp_path / "repo" + for d in ("sites", "skills", "config", "catalog", "deploy"): + shutil.copytree(REPO_ROOT / d, root / d) + skill = root / "skills" / "zusammenfassung" / "SKILL.md" + skill.write_text(skill.read_text().replace("slot: tier1-general", "slot: gibts-nicht")) + lite = root / "deploy" / "litellm" / "config.yaml" + lite.write_text(lite.read_text().replace("model_name: verifier-small", "model_name: verifier-x")) + errors = [m for lvl, m in validate_repo(root) if lvl == "error"] + assert any("gibts-nicht" in e for e in errors) + assert any("verifier-small" in e and "LiteLLM" in e for e in errors) + + +def test_cli(capsys): + assert main(["validate", "--root", str(REPO_ROOT)]) == 0 + assert main(["placement", "--inventory", str(REPO_ROOT / "sites/knaster/inventory.yaml"), + "--catalog", str(REPO_ROOT / "catalog")]) == 0 + out = capsys.readouterr().out + assert "gpu-main" in out diff --git a/core/tests/test_lifecycle.py b/core/tests/test_lifecycle.py index 527795f..7c924f2 100644 --- a/core/tests/test_lifecycle.py +++ b/core/tests/test_lifecycle.py @@ -96,9 +96,10 @@ def test_canary_metrics_from_cases(life, store): life.report(cid, "eval", {"pass_rate": 0.95, "n": 40}) life.report(cid, "shadow", {"agreement": 0.9, "n": 20}) for i in range(4): - store.add_case(input=f"x{i}", stage=1, slot="tier1-general", model="rv2", passed=i < 3, split="train", + store.add_case(input=f"x{i}", stage=1, slot="tier1-general", adapter="rv2", + model="tier1-general/rv2", passed=i < 3, split="train", origin="route", ts=life.get(cid)["updated_at"] + 1) - store.add_case(input="y", stage=1, slot="tier1-general", model="other", passed=False, split="train", + store.add_case(input="y", stage=1, slot="tier1-general", adapter="other", passed=False, split="train", origin="route") assert life.canary_metrics(cid) == {"pass_rate": 0.75, "n": 4} @@ -113,7 +114,7 @@ def test_router_uses_canary_model(registry, library, embedder): registry.set_canary("tier1-general", "rechnungen-v2", 1.0, "rechnungen") r = Router(registry, library, embedder, rng=random.Random(1)) d = r.route(RECHNUNG) - assert d.canary and d.model == d.adapter == "rechnungen-v2" + assert d.canary and d.model == "tier1-general/rechnungen-v2" and d.adapter == "rechnungen-v2" registry.set_canary("tier1-general", "rechnungen-v2", 0.0, "rechnungen") assert not r.route(RECHNUNG).canary registry.set_adapter(AdapterSpec(name="x", base_model="base-14b", cluster="andere")) diff --git a/core/tests/test_maintenance.py b/core/tests/test_maintenance.py index 65d857e..76ab106 100644 --- a/core/tests/test_maintenance.py +++ b/core/tests/test_maintenance.py @@ -190,7 +190,8 @@ def test_routine_evaluates_canary_and_proposes_promotion(make_system): life.report(cid, "eval", {"pass_rate": 0.95, "n": 40}) life.report(cid, "shadow", {"agreement": 0.9, "n": 20}) for i in range(6): - sysm.store.add_case(input=f"c{i}", stage=1, slot="tier1-general", model="rv2", passed=True, + sysm.store.add_case(input=f"c{i}", stage=1, slot="tier1-general", adapter="rv2", + model="tier1-general/rv2", passed=True, split="train", origin="route") report = sysm.maintenance.run_routine() assert any("Canary" in a for a in report.auto_executed) diff --git a/core/tests/test_router.py b/core/tests/test_router.py index a546998..b6d1ae6 100644 --- a/core/tests/test_router.py +++ b/core/tests/test_router.py @@ -15,7 +15,7 @@ def test_routes_to_weak_slot_with_adapter(router): d = router.route(RECHNUNG) assert d.status == "routed" assert (d.stage, d.slot, d.skill) == (1, "tier1-general", "rechnung-extrahieren") - assert d.model == d.adapter == "rechnungen-v1" + assert d.model == "tier1-general/rechnungen-v1" and d.adapter == "rechnungen-v1" assert d.prompt_variant == "full" diff --git a/deploy/litellm/config.yaml b/deploy/litellm/config.yaml new file mode 100644 index 0000000..2327f3a --- /dev/null +++ b/deploy/litellm/config.yaml @@ -0,0 +1,75 @@ +# LiteLLM-Gateway: einzige Stelle mit URLs/Keys der Modelle. Clients kennen nur die Aliase (= Slots). +# Hostnamen sind die DNS-Namen bzw. IPs der GPU-VMs im VLAN "svc" (siehe platform/proxmox). +model_list: + # --- Stufe 0: starkes Modell über OpenRouter (Modell hier austauschbar) --- + - model_name: tier0-strong + litellm_params: + model: openrouter/anthropic/claude-opus-5.5 # TODO: beim Deployment gewünschtes Modell eintragen + api_key: os.environ/OPENROUTER_API_KEY + model_info: {mode: chat} + + # --- Stufe 1: vLLM auf den RTX 4000 Ada --- + - model_name: tier1-general + litellm_params: + model: hosted_vllm/tier1-general # vLLM --served-model-name + api_base: http://gpu-main:8000/v1 + model_info: {mode: chat} + - model_name: tier1-flex + litellm_params: + model: hosted_vllm/tier1-flex + api_base: http://gpu-flex:8000/v1 + model_info: {mode: chat} + + # LoRA-Adapter: vLLM stellt jeden Adapter unter eigenem Namen bereit. Der Router ruft sie als + # "/" auf; die Wildcard leitet an den vLLM des jeweiligen Slots weiter. + # (Wildcard-Routing beim Deployment gegen die eingesetzte LiteLLM-Version prüfen.) + - model_name: "tier1-general/*" + litellm_params: + model: "hosted_vllm/*" + api_base: http://gpu-main:8000/v1 + - model_name: "tier1-flex/*" + litellm_params: + model: "hosted_vllm/*" + api_base: http://gpu-flex:8000/v1 + + # --- llama.cpp auf P2000 / M4000 --- + - model_name: verifier-small + litellm_params: + model: openai/verifier-small + api_base: http://gpu-small:8080/v1 + api_key: none + model_info: {mode: chat} + - model_name: signature-extractor + litellm_params: + model: openai/signature-extractor + api_base: http://gpu-small:8080/v1 + api_key: none + model_info: {mode: chat} + - model_name: embed-default + litellm_params: + model: openai/embed-default + api_base: http://gpu-embed:8081/v1 + api_key: none + model_info: {mode: embedding} + - model_name: rerank-default # Router ruft Rerank direkt (RERANK_URL); Eintrag für Monitoring + litellm_params: + model: openai/rerank-default + api_base: http://gpu-embed:8082/v1 + api_key: none + model_info: {mode: rerank} + +router_settings: + # Infrastruktur-Fallbacks innerhalb einer Stufe; Qualitäts-Eskalation macht der Router-Service + fallbacks: + - tier1-general: [tier1-flex] + allowed_fails: 2 + cooldown_time: 60 + num_retries: 1 + timeout: 120 + +litellm_settings: + drop_params: true + request_timeout: 120 + +general_settings: + master_key: os.environ/LITELLM_MASTER_KEY diff --git a/sites/example-single-gpu/inventory.yaml b/sites/example-single-gpu/inventory.yaml new file mode 100644 index 0000000..33b72b3 --- /dev/null +++ b/sites/example-single-gpu/inventory.yaml @@ -0,0 +1,10 @@ +# Beispiel für Übertragbarkeit: ein Workstation-Cluster mit einer großen GPU und einem CPU-Node. +nodes: + - name: ws1 + cpu: {model: "AMD Ryzen 9", cores: 16, threads: 32} + memory_gb: 128 + gpus: + - {index: 0, name: "NVIDIA GeForce RTX 3090", memory_gb: 24, compute_cap: 8.6} + - name: nas + cpu: {model: "Intel Celeron", cores: 4, threads: 4} + memory_gb: 16 diff --git a/sites/knaster/inventory.yaml b/sites/knaster/inventory.yaml new file mode 100644 index 0000000..8b8bdca --- /dev/null +++ b/sites/knaster/inventory.yaml @@ -0,0 +1,38 @@ +# Inventar des Clusters "knaster". +# ANNAHME: Die Verteilung der GPUs und SSDs auf die Hosts ist noch nicht erfasst. Bitte beim +# gemeinsamen Deployment mit `knaster inventory collect` auf jedem Node ersetzen. +nodes: + - name: pve-ada + cpu: {model: "unbekannt", cores: 6, threads: 12} + memory_gb: 64 + disks: [{name: sda, size_gb: 256}, {name: sdb, size_gb: 256}] + gpus: + - {index: 0, name: "NVIDIA RTX 4000 Ada Generation", memory_gb: 20, compute_cap: 8.9} + - {index: 1, name: "NVIDIA RTX 4000 Ada Generation", memory_gb: 20, compute_cap: 8.9} + - name: pve1 + cpu: {model: "Intel Core i7 (6. Gen) 3.4 GHz", cores: 4, threads: 8} + memory_gb: 32 + disks: [{name: sda, size_gb: 128}, {name: sdb, size_gb: 256}] + gpus: + - {index: 0, name: "Quadro M4000", memory_gb: 8, compute_cap: 5.2} + - {index: 1, name: "Quadro M4000", memory_gb: 8, compute_cap: 5.2} + - name: pve2 + cpu: {model: "Intel Core i7 (7. Gen) 3.6 GHz", cores: 4, threads: 8} + memory_gb: 32 + disks: [{name: sda, size_gb: 128}, {name: sdb, size_gb: 256}, {name: sdc, size_gb: 256}] + gpus: + - {index: 0, name: "Quadro M4000", memory_gb: 8, compute_cap: 5.2} + - {index: 1, name: "Quadro M4000", memory_gb: 8, compute_cap: 5.2} + - name: pve3 + cpu: {model: "Intel Core i7 (7. Gen) 3.6 GHz", cores: 4, threads: 8} + memory_gb: 32 + disks: [{name: sda, size_gb: 128}, {name: sdb, size_gb: 256}] + gpus: + - {index: 0, name: "Quadro P2000", memory_gb: 5, compute_cap: 6.1} + - {index: 1, name: "Quadro P2000", memory_gb: 5, compute_cap: 6.1} + - name: pve4 + cpu: {model: "Intel Core i7 (7. Gen) 3.6 GHz", cores: 4, threads: 8} + memory_gb: 32 + disks: [{name: sda, size_gb: 128}, {name: sdb, size_gb: 256}, {name: sdc, size_gb: 256}] + gpus: + - {index: 0, name: "Quadro P2000", memory_gb: 5, compute_cap: 6.1}