""" Helfer rund um llama-swap und dessen config.yaml. - _swap_get: liest llama-swap-Endpoints (/running, /v1/models, ...) - read_config / write_config: lesen/schreiben der config.yaml ueber ruamel.yaml, sodass Kommentare und Quotes erhalten bleiben. """ import os import re import httpx from config import CONFIG_PATH, LLAMA_SWAP_URL, yaml # Kanonische Rollen (Tags auf den Modellen). Bewusst eine kleine, feste Palette fuer die # UI-Schnellwahl — frei waehlbare Rollen sind trotzdem erlaubt. ROLE_IDS = {"vision", "coder", "scout", "reviewer", "manager"} def model_id_from_path(model_path: str) -> str: """Sprechende Modell-ID (= llama-swap-Modellname, den man in der API angibt) aus dem GGUF-Pfad ableiten: der Repo-Ordnername ohne '-GGUF'. Fallback: Dateiname ohne Quant-Suffix/.gguf. So steht in der Modell-Liste der echte Name statt 'coder'.""" d = os.path.basename(os.path.dirname(model_path)) name = re.sub(r"[-_]?GGUF$", "", d, flags=re.I).strip("-_") if not name: fn = re.sub(r"\.gguf$", "", os.path.basename(model_path), flags=re.I) fn = re.sub(r"-\d+-of-\d+$", "", fn) name = re.sub(r"[-_](Q\d[\w]*|IQ\d[\w]*|F16|BF16|FP16|F32)$", "", fn, flags=re.I) return name or "modell" def set_role_alias(cfg: dict, model_id: str, role: str | None) -> None: """Rolle als llama-swap-`aliases` auf ein Modell legen (damit der Rollenname in der API ebenfalls funktioniert). Haelt den Alias EINDEUTIG: derselbe Rollen-Alias wird vorher bei allen anderen Modellen entfernt. role=None/leer entfernt den Alias.""" models = cfg.get("models") or {} role = (role or "").strip().lower() if role: for mid, spec in models.items(): if mid == model_id or not isinstance(spec, dict): continue al = [a for a in (spec.get("aliases") or []) if str(a).lower() != role] if al: spec["aliases"] = al else: spec.pop("aliases", None) spec = models.get(model_id) if isinstance(spec, dict): if role and role != model_id.lower(): spec["aliases"] = [role] else: spec.pop("aliases", None) def _swap_get(path: str): with httpx.Client(timeout=5.0) as c: r = c.get(f"{LLAMA_SWAP_URL}{path}") r.raise_for_status() return r.json() def read_config() -> dict: if not CONFIG_PATH.exists(): return {"models": {}} with CONFIG_PATH.open("r", encoding="utf-8") as f: data = yaml.load(f) or {} if "models" not in data or data["models"] is None: data["models"] = {} return data def write_config(cfg: dict) -> None: """Schreibt die config.yaml atomar (tmp-Datei + os.replace), damit llama-swap mit -watch-config nie eine halb geschriebene Datei sieht. Fehlende Schreibrechte werden in eine klare Meldung uebersetzt statt als roher Traceback zu landen.""" try: CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True) tmp = CONFIG_PATH.with_name(CONFIG_PATH.name + ".tmp") with tmp.open("w", encoding="utf-8") as f: yaml.dump(cfg, f) os.replace(tmp, CONFIG_PATH) except PermissionError as exc: raise PermissionError( f"Mission Control darf '{CONFIG_PATH}' nicht schreiben. " f"Einmalig Besitz uebergeben: sudo chown -R hitonabi:hitonabi {CONFIG_PATH.parent}" ) from exc