""" Kuratierter Modell-Katalog ("Cookbook", inspiriert von Odysseus): EINE Quelle der Wahrheit für KORREKTE Metadaten (total/active params, moe, generation) statt Namens-Raterei. Macht Empfehlung + Upgrade-Erkennung präzise und MoE-bewusst für die bandbreiten-limitierte Strix-Halo-Box. Daten: backend/models_catalog.json. Fällt sanft aus (leerer Katalog), wenn die Datei fehlt → discover nutzt dann nur die HF-Dynamik. """ import json import math import os import re from services.fit import estimate_memory_gb, estimate_speed _CATALOG_PATH = os.path.join(os.path.dirname(__file__), "..", "models_catalog.json") _cache: dict = {"mtime": 0.0, "models": []} def _load() -> list[dict]: try: mt = os.path.getmtime(_CATALOG_PATH) if mt != _cache["mtime"]: with open(_CATALOG_PATH, encoding="utf-8") as f: data = json.load(f) or {} _cache.update(mtime=mt, models=[m for m in data.get("models", []) if m.get("name")]) except (OSError, ValueError): _cache.update(mtime=0.0, models=[]) return _cache["models"] def _norm(name: str) -> str: """Vergleichs-Stamm: kleingeschrieben, Org-Prefix/Quant/GGUF/Split entfernt.""" s = (name or "").lower().split("/")[-1] s = re.sub(r"\.gguf$", "", s) s = re.sub(r"-\d+-of-\d+$", "", s) s = re.sub(r"[-_](ud-)?(i?q\d[\w]*|f16|bf16|fp16|f32|mxfp4)$", "", s) return s.strip("-_ ") def entries() -> list[dict]: return list(_load()) def entries_for_role(role: str) -> list[dict]: return [e for e in _load() if e.get("role") == role] def meta_for_name(name: str) -> dict | None: """Katalog-Metadaten zu einem Modell(namen) — matcht lokalen Namen ODER HF-Repo.""" n = _norm(name) if not n: return None for e in _load(): cand = {_norm(e.get("name", "")), _norm(e.get("repo", ""))} if n in cand or any(c and (c in n or n in c) for c in cand): return e return None def fit_of(e: dict, ram_gb: float) -> dict: """Hardware-Fit eines Katalog-Eintrags (MoE-bewusst über active_params_b).""" total = float(e.get("total_params_b") or 7) active = float(e.get("active_params_b") or total) quant = e.get("quant") or "Q4_K_M" ctx = int(e.get("ctx") or 32768) req_gb = estimate_memory_gb(total, quant, ctx) tps = estimate_speed(req_gb, ram_gb, (active / total) if total else 1.0) usable = max(ram_gb - 4.0, 0) if req_gb > usable: level, text = "too_tight", "Zu groß (OOM)" elif req_gb > usable * 0.8: level, text = "marginal", "Könnte knapp werden" else: level, text = "perfect", "Passt perfekt" return {"level": level, "text": text, "req_gb": round(req_gb, 1), "tps": round(tps, 0)} def stack_score(e: dict, ram_gb: float) -> float: """Score für DIESE Hardware: muss passen, dann Wissen (total params) + Tempo (tps — belohnt MoE durch niedrige aktive Params automatisch). Bandbreiten-Box → MoE gewinnt bei vergleichbarem Wissen gegen dense.""" fit = fit_of(e, ram_gb) if fit["level"] == "too_tight": return -100.0 total = float(e.get("total_params_b") or 7) fit_bonus = 3.0 if fit["level"] == "perfect" else 1.0 knowledge = math.log2(total + 1) / 8.0 # ~0..1 (bis ~256B) speed = min((fit["tps"] or 0) / 80.0, 1.0) # normalisiert; MoE = hohe tps return fit_bonus + 1.2 * knowledge + 1.0 * speed def to_model_dict(e: dict, ram_gb: float) -> dict: """Katalog-Eintrag → discover-kompatibles Modell-Dict (echte Metadaten).""" total = float(e.get("total_params_b") or 7) active = e.get("active_params_b") repo = e.get("repo") or e.get("name") role = e.get("role") caps = { "moe": bool(e.get("moe")), "active_b": active, "tools": "yes" if e.get("tools") else "no", "vision": bool(e.get("vision")), "coder": role == "coder", "reasoning": role == "heavy", "embedding": False, "ctx": e.get("ctx"), "params_b": total, "arch": e.get("family"), } return { "name": e.get("name"), "author": repo.split("/")[0] if "/" in repo else "catalog", "repo": repo, "role": role, "params_b": total, "active_b": active, "moe": bool(e.get("moe")), "generation": e.get("generation"), "family": e.get("family"), "quant": e.get("quant") or "Q4_K_M", "tags": ["catalog"], "downloads": 0, "fit": fit_of(e, ram_gb), "optimal_ctx": int(e.get("ctx") or 32768), "caps": caps, "curated": True, }