duffyduckandClaude Opus 5.5
68686de9dc
Modell-Katalog → ⚙ llama.cpp zeigt je LLM-Box die installierte Version, sucht nach Updates (Pruefsumme des Image-Tags in der Registry + letztes llama-swap-Release) und aktualisiert auf Knopfdruck: altes Image sichern, neues ziehen, Container per Compose neu anlegen, Selbsttest (llama.cpp sieht die Karten, vorgeladene Modelle wieder bereit) — sonst automatisch zurueck. Zuruecksetzen auf die Vorversion per Knopf. Docker-Zugriff hat nur der neue llm-updater (feste Aktionen fuer aria-llama-swap, kein Port nach aussen, Token in /models). Der llm-adapter reicht llm_runtime weiter und meldet den Fortschritt. RVS-Allow-List um llm_runtime/_result/_progress ergaenzt. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2002 lines
89 KiB
Python
2002 lines
89 KiB
Python
"""
|
|
ARIA Local-LLM-Adapter (AI-Box) — Plan B, Phase B0.
|
|
|
|
Bruecke zwischen RVS und dem lokalen llama.cpp-Server. Spiegelt das Muster der
|
|
whisper-bridge: verbindet sich per WebSocket mit dem RVS (Token-Room, TLS mit
|
|
ws-Fallback, Reconnect-Backoff), lauscht auf `llm_request` und ruft den lokalen
|
|
llama.cpp-`/v1/chat/completions`-Endpoint (OpenAI-kompatibel), antwortet mit
|
|
`llm_response` (korreliert per requestId).
|
|
|
|
Topologie: AI-Box steht zuhause, ARIA im RZ — die Kommunikation laeuft ueber
|
|
den RVS (wie TTS/STT), keine IPs zu pflegen. Nur URL + Token.
|
|
|
|
Env:
|
|
RVS_HOST, RVS_PORT, RVS_TLS, RVS_TLS_FALLBACK, RVS_TOKEN (wie f5tts/whisper)
|
|
LLAMA_URL Default http://llama:8081 (llama.cpp im selben Compose-Netz)
|
|
LLM_MODEL optionaler Modell-Name fuer llama (llama.cpp ignoriert ihn
|
|
meist, dient nur der Transparenz im Log)
|
|
LLM_TIMEOUT_SEC Default 60
|
|
|
|
Bewusst NICHT-streamend in B0 (volle llm_response). Token-Streaming (llm_partial)
|
|
kommt in B2 zusammen mit TTS-on-first-sentence.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import hashlib
|
|
import queue
|
|
import shlex
|
|
import shutil
|
|
import threading
|
|
import time
|
|
|
|
import httpx
|
|
import websockets
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
|
)
|
|
logger = logging.getLogger("llm-adapter")
|
|
|
|
RVS_HOST = os.getenv("RVS_HOST", "").strip()
|
|
RVS_PORT = os.getenv("RVS_PORT", "443").strip()
|
|
RVS_TLS = os.getenv("RVS_TLS", "true").lower() == "true"
|
|
RVS_TLS_FALLBACK = os.getenv("RVS_TLS_FALLBACK", "true").lower() == "true"
|
|
RVS_TOKEN = os.getenv("RVS_TOKEN", "").strip()
|
|
# TLS-Hostname (SNI + Cert), falls RVS_HOST eine IP ist (Box im selben Netz wie
|
|
# der RVS, direkt auf die interne IP). Leer = SNI = RVS_HOST.
|
|
RVS_SNI = os.getenv("RVS_SNI", "").strip()
|
|
|
|
# In-Process-DNS-Override: wenn RVS_SNI gesetzt ist, verbindet die URI ueber den
|
|
# HOSTNAMEN (Host-Header + SNI + Cert stimmen), waehrend getaddrinfo den Namen auf
|
|
# die echte IP in RVS_HOST aufloest. Versionsunabhaengig — host=/port= kollidiert
|
|
# in der Legacy-websockets-API mit dem aus der URI abgeleiteten Host.
|
|
if RVS_TLS and RVS_SNI:
|
|
import socket as _socket
|
|
_orig_getaddrinfo = _socket.getaddrinfo
|
|
def _sni_getaddrinfo(host, *a, **k):
|
|
return _orig_getaddrinfo(RVS_HOST if host == RVS_SNI else host, *a, **k)
|
|
_socket.getaddrinfo = _sni_getaddrinfo
|
|
|
|
|
|
LLAMA_URL = os.getenv("LLAMA_URL", "http://llama:8081").rstrip("/")
|
|
LLM_MODEL = os.getenv("LLM_MODEL", "qwen3-8b")
|
|
LLM_TIMEOUT_SEC = float(os.getenv("LLM_TIMEOUT_SEC", "60"))
|
|
|
|
# ── Compute-Fleet: Worker-Identitaet & Registrierung ──────────────
|
|
# Meldet sich bei der aria-bridge (worker_hello) + periodischer worker_ping.
|
|
NODE_NAME = os.getenv("NODE_NAME", "node").strip() or "node"
|
|
GPU_IDS = os.getenv("NVIDIA_VISIBLE_DEVICES", "").strip()
|
|
WORKER_SERVICE = "llm"
|
|
INSTANCE_ID = f"{WORKER_SERVICE}@{NODE_NAME}"
|
|
WORKER_PING_INTERVAL_S = int(os.getenv("WORKER_PING_INTERVAL_S", "10"))
|
|
# Empfangs-Watchdog: kommt in RX_STALE_S kein Broadcast rein (ein echter Raum hat
|
|
# staendig Traffic, z.B. sat_hello alle 25s / Brain-Polling), gilt die Verbindung
|
|
# als halb-tot (Caddy pongt die WS-Pings selbst) -> Zwangs-Reconnect.
|
|
RX_STALE_S = int(os.getenv("RX_STALE_S", "60"))
|
|
_inflight = 0 # laufende llm_requests (busy-Report im ping)
|
|
# Qwen3 hat Thinking-Mode default AN — dann verbraet es Tokens in einem
|
|
# <think>-Block und liefert (bei kleinem max_tokens) leeren/abgeschnittenen
|
|
# content, ausserdem 3x langsamer. ARIAs schnelles Tier will KEIN Grübeln
|
|
# (grübeln = harter Turn = Claude). Wir schalten Thinking daher per
|
|
# chat_template_kwargs ab (Qwen3-Template versteht enable_thinking=false;
|
|
# andere Templates ignorieren das kwarg). Bei einem Modell, das darauf
|
|
# empfindlich reagiert: LLM_DISABLE_THINKING=false setzen.
|
|
LLM_DISABLE_THINKING = os.getenv("LLM_DISABLE_THINKING", "true").lower() == "true"
|
|
|
|
# ── Modell-Verwaltung (Stage D): Adapter besitzt llama-swaps Config ──
|
|
# llama-swap liest die GENERIERTE Config (beschreibbar, im /models-Bind). Wir
|
|
# erzeugen sie aus dem Basis-Template (kuratierte Defaults) + der persistenten
|
|
# Box-Registry (per Diagnostic hinzugefuegte Modelle). So werden neue Modelle
|
|
# ohne Image-Rebuild waehlbar.
|
|
import yaml # pyyaml
|
|
BASE_CONFIG_PATH = os.getenv("LLAMA_BASE_CONFIG", "/llamaswap/config.yaml")
|
|
GEN_CONFIG_PATH = os.getenv("LLAMA_GEN_CONFIG", "/models/llama-swap.config.yaml")
|
|
REGISTRY_PATH = os.getenv("LLM_REGISTRY", "/models/aria_models.json")
|
|
|
|
# ── Karten-Belegung ─────────────────────────────────────────────────
|
|
# Grundregel: Modelle LADEN nur ueber die Diagnostic, und zwar pro Karte. Jede
|
|
# LLM-Karte haelt genau ein Modell, dauerhaft (ttl 0, beim Start vorgeladen).
|
|
# Clients — ARIA, Fremdprogramme ueber RVS, OpenHands ueber den OpenAI-Port —
|
|
# sehen nur die geladenen Modelle und waehlen daraus. Eine Anfrage fuer ein
|
|
# anderes Modell wird abgelehnt und laedt NIE etwas nach. So kann kein Client
|
|
# einem anderen das Modell unter den Fuessen wegtauschen.
|
|
#
|
|
# Welche Karten dem LLM gehoeren, steht in der .env der Box:
|
|
# LLM_GPUS=0,1 (mehrere) — oder wie bisher — LLM_GPU=0 (eine)
|
|
# Die Belegung (Karte → Modell) liegt in SLOTS_PATH und wird nur von
|
|
# llm_provision_model / llm_unload_model geaendert.
|
|
SLOTS_PATH = os.getenv("LLM_SLOTS", "/models/aria_slots.json")
|
|
|
|
|
|
def _llm_cards() -> list:
|
|
raw = os.getenv("LLM_GPUS", "").strip() or os.getenv("LLM_GPU", "0").strip() or "0"
|
|
cards = [c.strip() for c in raw.split(",") if c.strip()]
|
|
return [c for i, c in enumerate(cards) if c not in cards[:i]] # Reihenfolge, ohne Doppelte
|
|
|
|
|
|
def _load_slots() -> dict:
|
|
"""{karte: modell}. Fehlt die Datei (erste Version mit Karten), wird das
|
|
bisherige Standardmodell (LLM_MODEL) auf die erste Karte gelegt — so laeuft
|
|
eine bestehende Box nach dem Update genau so weiter wie vorher."""
|
|
try:
|
|
with open(SLOTS_PATH) as f:
|
|
data = json.load(f)
|
|
if isinstance(data, dict):
|
|
cards = set(_llm_cards())
|
|
return {str(k): str(v) for k, v in data.items() if str(k) in cards and v}
|
|
except FileNotFoundError:
|
|
cards = _llm_cards()
|
|
if cards and _model_cmd(LLM_MODEL):
|
|
slots = {cards[0]: LLM_MODEL}
|
|
_save_slots(slots)
|
|
logger.info("Karten-Belegung neu angelegt: Karte %s → %s", cards[0], LLM_MODEL)
|
|
return slots
|
|
except Exception as e:
|
|
logger.warning("Karten-Belegung nicht lesbar (%s) — keine Modelle", e)
|
|
return {}
|
|
|
|
|
|
def _save_slots(slots: dict) -> None:
|
|
tmp = SLOTS_PATH + ".tmp"
|
|
with open(tmp, "w") as f:
|
|
json.dump(slots, f, indent=2)
|
|
os.replace(tmp, SLOTS_PATH)
|
|
|
|
|
|
def _card_of(model: str) -> str:
|
|
for card, m in _load_slots().items():
|
|
if m == model:
|
|
return card
|
|
return ""
|
|
|
|
# ── OpenAI-kompatibler Endpunkt (fuer OpenHands & Co.) ─────
|
|
# Der Port ist ein Docker-Port-Mapping und steht beim Container-Start fest —
|
|
# er kommt IMMER aus der .env. Der API-Key kann zusaetzlich zur Laufzeit aus der
|
|
# Diagnostic gesetzt werden, ABER: steht er in der .env, hat die .env Vorrang
|
|
# und das Feld ist in der Diagnostic schreibgeschuetzt.
|
|
OPENAI_PORT = os.getenv("LLM_OPENAI_PORT", "8081").strip() or "8081"
|
|
OPENAI_BIND = os.getenv("LLM_OPENAI_BIND", "127.0.0.1").strip() or "127.0.0.1"
|
|
_OPENAI_KEY_ENV = os.getenv("LLM_OPENAI_KEY", "").strip()
|
|
OPENAI_KEY_FILE = os.getenv("LLM_OPENAI_KEY_FILE", "/models/openai_key.json")
|
|
|
|
|
|
def _openai_key() -> str:
|
|
"""Gueltiger API-Key: .env gewinnt, sonst der in der Diagnostic gesetzte."""
|
|
if _OPENAI_KEY_ENV:
|
|
return _OPENAI_KEY_ENV
|
|
try:
|
|
with open(OPENAI_KEY_FILE) as f:
|
|
return str((json.load(f) or {}).get("key") or "").strip()
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def _openai_key_save(key: str) -> bool:
|
|
"""Speichert den in der Diagnostic gesetzten Key. Bei .env-Vorrang: nein."""
|
|
if _OPENAI_KEY_ENV:
|
|
return False
|
|
try:
|
|
os.makedirs(os.path.dirname(OPENAI_KEY_FILE), exist_ok=True)
|
|
tmp = OPENAI_KEY_FILE + ".tmp"
|
|
with open(tmp, "w") as f:
|
|
json.dump({"key": key.strip()}, f)
|
|
os.replace(tmp, OPENAI_KEY_FILE)
|
|
os.chmod(OPENAI_KEY_FILE, 0o600)
|
|
return True
|
|
except Exception as exc:
|
|
logger.error("OpenAI-Key speichern fehlgeschlagen: %s", exc)
|
|
return False
|
|
|
|
# ── Auslastungs-Monitor (Stage E) ──────────────────────────
|
|
import node_stats
|
|
STATS_PATH = os.getenv("STATS_PATH", "/models/aria_stats.json")
|
|
_total_tokens = 0 # kumulativ, fuer den Token-Graph
|
|
_stats = node_stats.NodeStats(INSTANCE_ID, NODE_NAME, STATS_PATH,
|
|
token_getter=lambda: _total_tokens, logger=logger)
|
|
|
|
|
|
def _load_registry() -> list:
|
|
try:
|
|
with open(REGISTRY_PATH) as f:
|
|
data = json.load(f)
|
|
return data if isinstance(data, list) else []
|
|
except Exception:
|
|
return []
|
|
|
|
|
|
def _save_registry(reg: list) -> None:
|
|
try:
|
|
tmp = REGISTRY_PATH + ".tmp"
|
|
with open(tmp, "w") as f:
|
|
json.dump(reg, f, indent=2)
|
|
os.replace(tmp, REGISTRY_PATH)
|
|
except Exception as e:
|
|
logger.warning("Registry speichern fehlgeschlagen: %s", e)
|
|
|
|
|
|
def _base_config() -> dict:
|
|
try:
|
|
with open(BASE_CONFIG_PATH) as f:
|
|
return yaml.safe_load(f) or {}
|
|
except Exception as e:
|
|
logger.warning("Basis-Template %s nicht lesbar (%s)", BASE_CONFIG_PATH, e)
|
|
return {}
|
|
|
|
|
|
def _cpu_moe_flag(v) -> str:
|
|
"""MoE-Modelle: Experten in den Arbeitsspeicher. -1 = alle (--cpu-moe),
|
|
N > 0 = die der ersten N Schichten (--n-cpu-moe N), 0/leer = alles auf die Karte.
|
|
Der Rest (Attention, gemeinsame Teile, KV-Cache) bleibt auf der Karte — pro
|
|
Wort rechnet ein MoE nur wenige Experten, darum bleibt es brauchbar schnell."""
|
|
try:
|
|
n = int(v or 0)
|
|
except (TypeError, ValueError):
|
|
return ""
|
|
if n < 0:
|
|
return " --cpu-moe"
|
|
return f" --n-cpu-moe {n}" if n > 0 else ""
|
|
|
|
|
|
def _model_cpu_moe(key: str) -> int:
|
|
for e in _load_registry():
|
|
if (e.get("key") or "").strip() == key:
|
|
try:
|
|
return int(e.get("cpuMoe") or 0)
|
|
except (TypeError, ValueError):
|
|
return 0
|
|
m = re.search(r"--n-cpu-moe\s+(\d+)", _model_cmd(key) or "")
|
|
return int(m.group(1)) if m else (-1 if "--cpu-moe" in (_model_cmd(key) or "") else 0)
|
|
|
|
|
|
def _model_cmd(key: str) -> str:
|
|
"""llama-server-Aufruf fuer ein Modell: aus der Registry (ueber die
|
|
Diagnostic geladen) oder aus dem Basis-Template (qwen3-8b & Co.)."""
|
|
for e in _load_registry():
|
|
if (e.get("key") or "").strip() == key and (e.get("hfRepo") or "").strip():
|
|
quant = (e.get("quant") or "Q4_K_M").strip()
|
|
return (f"llama-server --port ${{PORT}} --host 127.0.0.1\n"
|
|
f"-hf {e['hfRepo'].strip()}:{quant}\n"
|
|
f"-ngl {int(e.get('ngl') or 99)} -c {int(e.get('ctx') or 8192)} --jinja"
|
|
+ _cpu_moe_flag(e.get("cpuMoe")))
|
|
spec = (_base_config().get("models") or {}).get(key) or {}
|
|
return str(spec.get("cmd") or "")
|
|
|
|
|
|
def _known_models() -> list:
|
|
"""Alles, was diese Box laden KOENNTE (Basis-Template + Registry)."""
|
|
keys = list((_base_config().get("models") or {}).keys())
|
|
keys += [e["key"] for e in _load_registry() if e.get("key") and e["key"] not in keys]
|
|
return keys
|
|
|
|
|
|
def _generate_config() -> int:
|
|
"""Schreibt die llama-swap-Config: NUR die Modelle, die einer Karte
|
|
zugeordnet sind. Was nicht drinsteht, kennt llama-swap nicht — auch der
|
|
OpenAI-Port lehnt es dann ab (404 "no router for requested model"). Das ist
|
|
der Kern der Regel "Clients laden nichts".
|
|
|
|
env CUDA_VISIBLE_DEVICES=<karte> pinnt das Modell auf seine Karte
|
|
ttl 0 nie wegen Leerlauf entladen
|
|
groups.swap false alle gleichzeitig geladen, keins
|
|
verdraengt ein anderes
|
|
hooks.on_startup.preload beim Start (und nach jedem Reload)
|
|
sofort laden, nicht erst beim 1. Aufruf
|
|
Getestet gegen llama-swap: genau dieses Verhalten.
|
|
"""
|
|
base = _base_config()
|
|
slots = _load_slots()
|
|
key = _openai_key()
|
|
models = {}
|
|
for card in _llm_cards():
|
|
m = slots.get(card)
|
|
if not m:
|
|
continue
|
|
cmd = _model_cmd(m)
|
|
if not cmd:
|
|
logger.error("Karte %s: Modell %r unbekannt (weder Registry noch Basis) — uebersprungen", card, m)
|
|
continue
|
|
cmd = re.sub(r"\s*--api-key\s+\S+", "", cmd)
|
|
if key:
|
|
# llama-server prueft dann den Authorization-Header (OpenAI-Port).
|
|
cmd = cmd.rstrip() + f" --api-key {shlex.quote(key)}"
|
|
models[m] = {"cmd": cmd, "env": [f"CUDA_VISIBLE_DEVICES={card}"], "ttl": 0}
|
|
|
|
out = {k: v for k, v in base.items() if k not in ("models", "groups", "hooks")}
|
|
out["models"] = models
|
|
if models:
|
|
out["groups"] = {"aria-karten": {"swap": False, "exclusive": True,
|
|
"persistent": True, "members": list(models)}}
|
|
out["hooks"] = {"on_startup": {"preload": list(models)}}
|
|
try:
|
|
os.makedirs(os.path.dirname(GEN_CONFIG_PATH), exist_ok=True)
|
|
tmp = GEN_CONFIG_PATH + ".tmp"
|
|
with open(tmp, "w") as f:
|
|
yaml.safe_dump(out, f, sort_keys=False, default_flow_style=False)
|
|
os.replace(tmp, GEN_CONFIG_PATH)
|
|
logger.info("llama-swap-Config: %s",
|
|
", ".join(f"Karte {_card_of(m) or '?'} → {m}" for m in models) or "keine Karte belegt")
|
|
except Exception as e:
|
|
logger.error("Config schreiben fehlgeschlagen: %s", e)
|
|
return len(models)
|
|
|
|
|
|
async def _reload_llama() -> None:
|
|
"""llama-swap laeuft mit --watch-config und liest die Datei selbst neu
|
|
(Poll alle 2 s). Die frueher hier aufgerufenen Endpunkte /reload und
|
|
/api/config/reload gibt es nicht (404, ausprobiert). Beim Neuladen startet
|
|
llama-swap ALLE Modelle neu und laedt sie per Preload wieder — die anderen
|
|
Karten sind also kurz nicht erreichbar."""
|
|
await asyncio.sleep(3) # Poll-Intervall + Puffer, bevor jemand fragt
|
|
|
|
|
|
# Antworten, die nicht verloren gehen duerfen: Bricht die RVS-Verbindung ab,
|
|
# waehrend eine Anfrage noch rechnet (z.B. RVS-/Proxy-Neustart), wuerde die
|
|
# fertige Antwort sonst ueber die tote Verbindung geschickt und verworfen — der
|
|
# Fragende (Schmiede, ARIA) wartet dann bis zu seiner Frist ins Leere (am
|
|
# 28.10. 11 Minuten). Jetzt: ueber die NEUE Verbindung senden, notfalls bis zum
|
|
# Wiederverbinden aufheben.
|
|
_RELIABLE_TYPES = {"llm_response", "llm_runtime_result"}
|
|
_OUTBOX_TTL_S = 300
|
|
_outbox: list = [] # [(ts, mtype, payload)]
|
|
_current_ws = None # die aktuell offene RVS-Verbindung
|
|
|
|
|
|
async def _send(ws, mtype: str, payload: dict) -> None:
|
|
data = json.dumps({"type": mtype, "payload": payload, "timestamp": int(time.time() * 1000)})
|
|
targets = [ws] if ws is _current_ws or _current_ws is None else [ws, _current_ws]
|
|
last_err = None
|
|
for t in targets:
|
|
try:
|
|
await t.send(data)
|
|
return
|
|
except Exception as e:
|
|
last_err = e
|
|
if mtype in _RELIABLE_TYPES:
|
|
_outbox.append((time.time(), mtype, payload))
|
|
logger.warning("Send fehlgeschlagen (%s): %s — aufgehoben, geht nach dem Wiederverbinden raus",
|
|
mtype, last_err)
|
|
else:
|
|
logger.warning("Send fehlgeschlagen (%s): %s", mtype, last_err)
|
|
|
|
|
|
async def _flush_outbox(ws) -> None:
|
|
"""Nach dem (Wieder-)Verbinden: aufgehobene Antworten nachschicken."""
|
|
now = time.time()
|
|
keep = [(t, m, p) for (t, m, p) in _outbox if now - t <= _OUTBOX_TTL_S]
|
|
_outbox.clear()
|
|
for t, m, p in keep:
|
|
try:
|
|
await ws.send(json.dumps({"type": m, "payload": p, "timestamp": int(time.time() * 1000)}))
|
|
logger.info("Aufgehobene Antwort nachgeschickt (%s, %.0f s alt)", m, now - t)
|
|
except Exception as e:
|
|
_outbox.append((t, m, p))
|
|
logger.warning("Nachschicken fehlgeschlagen (%s): %s", m, e)
|
|
return
|
|
|
|
|
|
def _sanitize_messages(messages: list) -> list:
|
|
"""Verlauf so aufbereiten, dass llama.cpp ihn nicht als Ganzes ablehnt.
|
|
|
|
Steht im Verlauf ein frueherer Werkzeugaufruf, dessen arguments KEIN
|
|
gueltiges JSON sind, antwortet llama.cpp auf die GANZE Anfrage mit 500
|
|
("Failed to parse tool call arguments as JSON") — und jede weitere Anfrage
|
|
mit demselben Verlauf auch. Bei Programmen, die Code schreiben, passiert
|
|
das leicht (Dateiinhalt mit Anfuehrungszeichen/Umbruechen). Statt den
|
|
ganzen Auftrag scheitern zu lassen: solche Argumente als {"_raw": "…"}
|
|
verpacken. Ein dict als arguments (statt String) wird zu JSON.
|
|
Nur Kopien — die Nachrichten des Aufrufers bleiben unveraendert."""
|
|
out = []
|
|
for m in messages or []:
|
|
tcs = m.get("tool_calls") if isinstance(m, dict) else None
|
|
if not tcs:
|
|
out.append(m)
|
|
continue
|
|
fixed = []
|
|
for tc in tcs:
|
|
fn = dict((tc or {}).get("function") or {})
|
|
args = fn.get("arguments")
|
|
if isinstance(args, (dict, list)):
|
|
fn["arguments"] = json.dumps(args, ensure_ascii=False)
|
|
elif isinstance(args, str):
|
|
try:
|
|
json.loads(args or "{}")
|
|
if not args.strip():
|
|
fn["arguments"] = "{}"
|
|
except Exception:
|
|
logger.warning("Werkzeugaufruf %r im Verlauf hat kein gueltiges JSON als "
|
|
"arguments (%d Zeichen) — eingepackt", fn.get("name"), len(args))
|
|
fn["arguments"] = json.dumps({"_raw": args}, ensure_ascii=False)
|
|
elif args is None:
|
|
fn["arguments"] = "{}"
|
|
fixed.append(dict(tc, function=fn))
|
|
out.append(dict(m, tool_calls=fixed))
|
|
return out
|
|
|
|
|
|
def _llama_error_info(exc: Exception) -> dict:
|
|
"""Maschinenlesbare Fehlerart. Wichtigster Fall: der Verlauf ist laenger
|
|
als das Kontextfenster — dann soll ein Programm automatisch kuerzen und
|
|
nochmal fragen koennen, statt einen Text auszuwerten."""
|
|
resp = getattr(exc, "response", None)
|
|
if resp is None:
|
|
return {}
|
|
try:
|
|
e = (resp.json() or {}).get("error") or {}
|
|
except Exception:
|
|
return {}
|
|
if isinstance(e, dict) and e.get("type") == "exceed_context_size_error":
|
|
return {"errorCode": "context_exceeded",
|
|
"promptTokens": int(e.get("n_prompt_tokens") or 0),
|
|
"ctx": int(e.get("n_ctx") or 0)}
|
|
return {}
|
|
|
|
|
|
def _llama_error_text(exc: Exception) -> str:
|
|
"""Bei HTTP-Fehlern den GRUND von llama.cpp mitnehmen. Bisher kam nur
|
|
"Server error '500 …'" an — die eigentliche Erklaerung steht im Body."""
|
|
resp = getattr(exc, "response", None)
|
|
if resp is None:
|
|
return re.sub(r"\s*For more information check:\s*\S+", "", str(exc)).strip()
|
|
detail = ""
|
|
try:
|
|
j = resp.json()
|
|
e = j.get("error") if isinstance(j, dict) else None
|
|
detail = (e.get("message") if isinstance(e, dict) else e) or json.dumps(j)[:600]
|
|
except Exception:
|
|
detail = (resp.text or "")[:600]
|
|
logger.warning("llama.cpp HTTP %s — Antwort: %s", resp.status_code, (resp.text or "")[:2000])
|
|
return f"llama.cpp HTTP {resp.status_code}: {str(detail).strip()[:600]}"
|
|
|
|
|
|
async def _call_llama(messages: list, *, max_tokens: int, temperature: float,
|
|
stop, tools=None, model=None, timeout: float | None = None) -> dict:
|
|
"""Ruft llama.cpp/llama-swap /v1/chat/completions (OpenAI-Format). Gibt
|
|
{ok, content, tool_calls, error} zurueck — wirft nie.
|
|
|
|
model: welches Modell llama-swap laden soll (B0.5). Kommt aus dem Request
|
|
(Brain -> local_llm.json). Faellt auf LLM_MODEL (env) zurueck.
|
|
tools: optionale OpenAI-Tool-Definitionen (B1b). Qwen3 (--jinja) kann
|
|
natives Tool-Calling und liefert dann message.tool_calls."""
|
|
body = {
|
|
"model": model or LLM_MODEL,
|
|
"messages": _sanitize_messages(messages),
|
|
"max_tokens": max_tokens,
|
|
"temperature": temperature,
|
|
"stream": False,
|
|
}
|
|
if stop:
|
|
body["stop"] = stop
|
|
if tools:
|
|
body["tools"] = tools
|
|
body["tool_choice"] = "auto"
|
|
if LLM_DISABLE_THINKING:
|
|
# llama.cpp (--jinja) reicht chat_template_kwargs an die Chat-Vorlage
|
|
# weiter. Qwen3 unterdrueckt damit den <think>-Block.
|
|
body["chat_template_kwargs"] = {"enable_thinking": False}
|
|
try:
|
|
async with httpx.AsyncClient(timeout=timeout or LLM_TIMEOUT_SEC) as client:
|
|
r = await client.post(f"{LLAMA_URL}/v1/chat/completions", json=body)
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
msg = (data.get("choices") or [{}])[0].get("message", {}) or {}
|
|
usage = data.get("usage") or {}
|
|
try:
|
|
global _total_tokens
|
|
_total_tokens += int(usage.get("total_tokens") or 0)
|
|
except Exception:
|
|
pass
|
|
return {
|
|
"ok": True,
|
|
"content": msg.get("content") or "",
|
|
"tool_calls": msg.get("tool_calls") or None,
|
|
"usage": usage,
|
|
}
|
|
except Exception as e:
|
|
msg = _llama_error_text(e)
|
|
logger.warning("llama.cpp-Call fehlgeschlagen: %s", msg[:300])
|
|
return {"ok": False, "content": "", "error": msg[:700], **_llama_error_info(e)}
|
|
|
|
|
|
# B0.5-2: Lade-Status ans Diagnostic (service_status, service="llm"). Wir kennen
|
|
# den Download-Fortschritt nicht (llama-swap gibt ihn nicht her), aber wir melden
|
|
# den Zustand bei Modellwechsel: loading -> ready/error. _last_model = aktuell
|
|
# geladenes; _ready_models = in dieser Session schon einmal bereit gewesene
|
|
# (fuer den "frisch geladen"-Hinweis 🎉 bei langem Erst-Load).
|
|
_last_model = None
|
|
_ready_models: set = set()
|
|
|
|
|
|
async def _emit_llm_status(ws, state: str, model: str, **extra) -> None:
|
|
await _send(ws, "service_status",
|
|
{"service": "llm", "state": state, "model": model, **extra})
|
|
|
|
|
|
async def _fetch_available_models() -> list:
|
|
"""Fragt llama-swap ab, welche Modelle diese Box fahren kann (GET /v1/models,
|
|
OpenAI-kompatibel → {data:[{id},...]}). Das sind die config.yaml-Keys.
|
|
Defensiv: bei Fehler Fallback auf [LLM_MODEL]."""
|
|
try:
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
r = await client.get(f"{LLAMA_URL}/v1/models")
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
ids = [m.get("id") for m in (data.get("data") or []) if m.get("id")]
|
|
return ids or [LLM_MODEL]
|
|
except Exception as e:
|
|
logger.warning("llama-swap /v1/models nicht abfragbar (%s) — Fallback [%s]", e, LLM_MODEL)
|
|
return [LLM_MODEL]
|
|
|
|
|
|
async def _fetch_loaded() -> list:
|
|
"""Welche Modelle liegen GERADE in der Karte? llama-swap /running →
|
|
[{"model", "state"}]. Ohne groups-Config ist das hoechstens eins — ein
|
|
anderes anzufragen entlaedt es. None, wenn llama-swap nicht antwortet
|
|
("unbekannt" ist etwas anderes als "nichts geladen")."""
|
|
try:
|
|
async with httpx.AsyncClient(timeout=5) as client:
|
|
r = await client.get(f"{LLAMA_URL}/running")
|
|
r.raise_for_status()
|
|
return [{"model": m.get("model", ""), "state": m.get("state", "")}
|
|
for m in (r.json().get("running") or []) if m.get("model")]
|
|
except Exception as e:
|
|
logger.debug("llama-swap /running nicht abfragbar: %s", e)
|
|
return None
|
|
|
|
|
|
_gpu_name_cache: dict = {}
|
|
|
|
|
|
async def _gpu_names() -> dict:
|
|
"""{index: {"name", "vramMB"}} aller Karten, einmal per nvidia-smi geholt.
|
|
Der VRAM steht fest — die Diagnostic warnt damit vor Modellen, die nicht
|
|
auf die Karte passen."""
|
|
if _gpu_name_cache:
|
|
return _gpu_name_cache
|
|
try:
|
|
proc = await asyncio.create_subprocess_exec(
|
|
"nvidia-smi", "--query-gpu=index,name,memory.total", "--format=csv,noheader,nounits",
|
|
stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL)
|
|
out, _ = await asyncio.wait_for(proc.communicate(), 10)
|
|
for line in out.decode().splitlines():
|
|
parts = [x.strip() for x in line.split(",")]
|
|
if len(parts) >= 3:
|
|
_gpu_name_cache[parts[0]] = {"name": parts[1], "vramMB": int(float(parts[2]))}
|
|
except Exception:
|
|
pass
|
|
return _gpu_name_cache
|
|
|
|
|
|
def _model_ctx(key: str) -> int:
|
|
"""Kontextfenster, mit dem llama-server das Modell startet (-c). Clients
|
|
MUESSEN das wissen: rechnen sie mit mehr, scheitern lange Auftraege oder
|
|
werden falsch gekuerzt."""
|
|
m = re.search(r"(?:^|\s)(?:-c|--ctx-size)\s+(\d+)", _model_cmd(key) or "")
|
|
return int(m.group(1)) if m else 0
|
|
|
|
|
|
# ── Kontextfenster: was das Modell kann + was es an Grafikspeicher kostet ──
|
|
#
|
|
# Aus dem Kopf der GGUF-Datei (nur Metadaten, die Gewichte werden nicht
|
|
# gelesen): trainiertes Maximum (context_length) und die Groesse des KV-Caches
|
|
# pro Token. Modelle mit Sliding-Window-Attention (Gemma) halten in den
|
|
# "lokalen" Schichten nur ein festes Fenster — nur die globalen Schichten
|
|
# wachsen mit dem Kontext. Annahme: KV-Cache in f16 (llama.cpp-Standard).
|
|
|
|
_GGUF_T = {0: "B", 1: "b", 2: "H", 3: "h", 4: "I", 5: "i", 6: "f", 7: "?", 10: "Q", 11: "q", 12: "d"}
|
|
_gguf_cache: dict = {} # path → (mtime, meta)
|
|
CTX_STEPS = [4096, 8192, 16384, 32768, 65536, 131072, 262144, 524288, 1048576]
|
|
# Fester Aufschlag (CUDA-Kontext + Rechenpuffer von llama.cpp) — gemessen auf
|
|
# einer RTX 3060 mit Gemma 4 12B bei 32k: ~1,4 GB ueber Modell + KV-Cache.
|
|
VRAM_OVERHEAD_MB = 1400
|
|
UBATCH = 512
|
|
|
|
|
|
def _gguf_meta(path: str, fsize: int = 0) -> dict:
|
|
"""Metadaten-Block einer GGUF-Datei (gecacht). Lange String-Listen (der
|
|
Tokenizer) werden uebersprungen, nicht gespeichert. fsize: wahre Groesse,
|
|
wenn path nur der Anfang der Datei ist (Kopf per HTTP-Range geholt)."""
|
|
try:
|
|
mt = os.path.getmtime(path)
|
|
except OSError:
|
|
return {}
|
|
hit = _gguf_cache.get(path)
|
|
if hit and hit[0] == mt and not fsize:
|
|
return hit[1]
|
|
import struct
|
|
meta: dict = {}
|
|
|
|
def rd(f, fmt):
|
|
return struct.unpack("<" + fmt, f.read(struct.calcsize("<" + fmt)))[0]
|
|
|
|
def rstr(f):
|
|
return f.read(rd(f, "Q")).decode("utf-8", "replace")
|
|
|
|
def rval(f, t):
|
|
if t == 8:
|
|
return rstr(f)
|
|
if t == 9:
|
|
et, n = rd(f, "I"), rd(f, "Q")
|
|
if et == 8:
|
|
for _ in range(n):
|
|
f.seek(rd(f, "Q"), 1)
|
|
return None
|
|
if et == 9:
|
|
raise ValueError("verschachtelte Arrays")
|
|
fmt = _GGUF_T[et]
|
|
if n > 4096:
|
|
f.seek(struct.calcsize("<" + fmt) * n, 1)
|
|
return None
|
|
return [rd(f, fmt) for _ in range(n)]
|
|
return rd(f, _GGUF_T[t])
|
|
|
|
try:
|
|
with open(path, "rb") as f:
|
|
if f.read(4) != b"GGUF":
|
|
return {}
|
|
rd(f, "I")
|
|
n_tensors = rd(f, "Q")
|
|
for _ in range(rd(f, "Q")):
|
|
k = rstr(f)
|
|
v = rval(f, rd(f, "I"))
|
|
if v is not None:
|
|
meta[k] = v
|
|
# Tensor-Verzeichnis: Groesse der MoE-Experten (…_exps) je Schicht —
|
|
# fuer die Schaetzung "was bleibt auf der Karte, was geht in den RAM"
|
|
infos = []
|
|
for _ in range(n_tensors):
|
|
name = rstr(f)
|
|
nd = rd(f, "I")
|
|
for _ in range(nd):
|
|
rd(f, "Q")
|
|
rd(f, "I")
|
|
infos.append((rd(f, "Q"), name))
|
|
align = int(meta.get("general.alignment") or 32)
|
|
data_start = (f.tell() + align - 1) // align * align
|
|
fsize = fsize or os.path.getsize(path)
|
|
infos.sort()
|
|
exp_by_layer: dict = {}
|
|
for i, (off, name) in enumerate(infos):
|
|
nxt = infos[i + 1][0] if i + 1 < len(infos) else fsize - data_start
|
|
if "_exps" in name:
|
|
mm = re.match(r"blk\.(\d+)\.", name)
|
|
lay = int(mm.group(1)) if mm else -1
|
|
exp_by_layer[lay] = exp_by_layer.get(lay, 0) + max(0, nxt - off)
|
|
meta["_expert_bytes_by_layer"] = exp_by_layer
|
|
except Exception as e:
|
|
logger.debug("GGUF %s nicht lesbar: %s", path, e)
|
|
meta = {}
|
|
_gguf_cache[path] = (mt, meta)
|
|
return meta
|
|
|
|
|
|
def _kv_bytes(meta: dict, ctx: int) -> int:
|
|
"""KV-Cache (f16) fuer ctx Tokens laut GGUF-Metadaten — Sliding-Window-
|
|
Schichten (Gemma) halten nur ihr Fenster."""
|
|
arch = meta.get("general.architecture")
|
|
if not arch:
|
|
return 0
|
|
g = lambda k, d=None: meta.get(f"{arch}.{k}", d)
|
|
n_layer = int(g("block_count", 0) or 0)
|
|
if not n_layer:
|
|
return 0
|
|
|
|
def per_layer(v, d=0):
|
|
return list(v) if isinstance(v, list) else [v if v is not None else d] * n_layer
|
|
kv_heads = per_layer(g("attention.head_count_kv", g("attention.head_count", 0)))
|
|
k_len = int(g("attention.key_length", 0) or 0)
|
|
v_len = int(g("attention.value_length", 0) or 0)
|
|
if not (k_len and v_len):
|
|
emb, heads = int(g("embedding_length", 0) or 0), int(g("attention.head_count", 0) or 0)
|
|
k_len = v_len = emb // heads if heads else 0
|
|
k_swa = int(g("attention.key_length_swa", k_len) or k_len)
|
|
v_swa = int(g("attention.value_length_swa", v_len) or v_len)
|
|
window = int(g("attention.sliding_window", 0) or 0)
|
|
pattern = g("attention.sliding_window_pattern")
|
|
swa = per_layer(pattern, False) if isinstance(pattern, list) and window else [False] * n_layer
|
|
per_tok = sum(int(kv_heads[i]) * (k_len + v_len) * 2 for i in range(n_layer) if not swa[i])
|
|
swa_tok = sum(int(kv_heads[i]) * (k_swa + v_swa) * 2 for i in range(n_layer) if swa[i])
|
|
return per_tok * ctx + swa_tok * min(ctx, window + UBATCH)
|
|
|
|
|
|
async def _auto_cpu_moe(repo: str, file: dict, vram_mb: int, ctx: int) -> int:
|
|
"""Wie viele Schichten mit Experten muessen in den RAM, damit der Rest auf
|
|
die Karte passt? Holt nur den Kopf der GGUF (Metadaten + Tensor-Verzeichnis,
|
|
wenige MB per HTTP-Range) — die Datei selbst laedt erst llama.cpp.
|
|
0 = passt komplett, -1 = alle Experten in den RAM (oder unbekannt)."""
|
|
import tempfile
|
|
url = f"https://huggingface.co/{repo}/resolve/main/{file['path']}"
|
|
headers = {"Range": "bytes=0-41943039"}
|
|
if HF_TOKEN:
|
|
headers["Authorization"] = f"Bearer {HF_TOKEN}"
|
|
try:
|
|
async with httpx.AsyncClient(timeout=60, follow_redirects=True) as c:
|
|
r = await c.get(url, headers=headers)
|
|
r.raise_for_status()
|
|
with tempfile.NamedTemporaryFile(suffix=".gguf", delete=False) as tf:
|
|
tf.write(r.content)
|
|
tmp = tf.name
|
|
try:
|
|
meta = _gguf_meta(tmp, fsize=int(file.get("size") or 0))
|
|
finally:
|
|
_gguf_cache.pop(tmp, None)
|
|
os.unlink(tmp)
|
|
except Exception as e:
|
|
logger.warning("MoE-Aufteilung fuer %s nicht berechenbar (%s) — alle Experten in den RAM", repo, e)
|
|
return -1
|
|
exp = meta.get("_expert_bytes_by_layer") or {}
|
|
if not exp:
|
|
return 0 # kein MoE
|
|
total_mb = int(file.get("size") or 0) / 1048576
|
|
exp_mb = sum(exp.values()) / 1048576
|
|
budget = vram_mb * 0.92 - VRAM_OVERHEAD_MB - _kv_bytes(meta, ctx) / 1048576 - (total_mb - exp_mb)
|
|
per_layer = exp_mb / len(exp)
|
|
on_gpu = max(0, int(budget // per_layer)) if per_layer else 0
|
|
n = max(0, len(exp) - on_gpu)
|
|
logger.info("MoE %s: %d von %d Schichten mit Experten in den RAM (Karte %d MB, ctx %d)",
|
|
repo, n, len(exp), vram_mb, ctx)
|
|
return -1 if n >= len(exp) else n
|
|
|
|
|
|
def _model_files(key: str) -> tuple:
|
|
"""(Gewichte.gguf, mmproj.gguf|None) eines Modells im HF-Cache unter /models
|
|
— aus "-hf org/repo:QUANT" bzw. "-m pfad" im llama-server-Aufruf."""
|
|
import glob
|
|
cmd = _model_cmd(key) or ""
|
|
m = re.search(r"(?:^|\s)-m\s+(\S+)", cmd)
|
|
if m and os.path.isfile(m.group(1)):
|
|
return m.group(1), None
|
|
m = re.search(r"(?:^|\s)-hf\s+([^\s:]+)(?::(\S+))?", cmd)
|
|
if not m:
|
|
return None, None
|
|
repo, quant = m.group(1), (m.group(2) or "").lower()
|
|
base = os.path.join(MODELS_DIR, "models--" + repo.replace("/", "--"), "snapshots")
|
|
files = [p for p in glob.glob(os.path.join(base, "*", "*.gguf")) if os.path.isfile(p)]
|
|
weights = [p for p in files if "mmproj" not in os.path.basename(p).lower()]
|
|
if quant:
|
|
weights = [p for p in weights if quant in os.path.basename(p).lower()] or weights
|
|
mmproj = [p for p in files if "mmproj" in os.path.basename(p).lower()]
|
|
return (weights[0] if weights else None), (mmproj[0] if mmproj else None)
|
|
|
|
|
|
def _ctx_profile(key: str) -> dict:
|
|
"""{ctxMax, options: [{ctx, estMB}]} fuer die Kontext-Auswahl der Diagnostic.
|
|
Leer, wenn die Modelldatei (noch) nicht da oder nicht lesbar ist."""
|
|
weights, mmproj = _model_files(key)
|
|
if not weights:
|
|
return {}
|
|
return _profile_of_files(weights, mmproj, lambda: _model_cpu_moe(key))
|
|
|
|
|
|
def _profile_of_files(weights: str, mmproj, cpu_moe_of=lambda: 0) -> dict:
|
|
"""Wie _ctx_profile, aber fuer beliebige Dateien — auch heruntergeladene
|
|
Modelle, die (noch) auf keiner Karte liegen. cpu_moe_of: Aufteilung der
|
|
Experten (0 = Schaetzung "alles auf der Karte")."""
|
|
meta = _gguf_meta(weights)
|
|
arch = meta.get("general.architecture")
|
|
if not arch:
|
|
return {}
|
|
g = lambda k, d=None: meta.get(f"{arch}.{k}", d)
|
|
n_layer = int(g("block_count", 0) or 0)
|
|
if not n_layer:
|
|
return {}
|
|
|
|
files_mb = (os.path.getsize(weights) + (os.path.getsize(mmproj) if mmproj else 0)) / 1048576
|
|
ctx_max = int(g("context_length", 0) or 0)
|
|
exp_by_layer = meta.get("_expert_bytes_by_layer") or {}
|
|
moe = bool(exp_by_layer) or int(g("expert_count", 0) or 0) > 0
|
|
cpu_moe = cpu_moe_of() if moe else 0
|
|
if cpu_moe:
|
|
layers = sorted(exp_by_layer) if cpu_moe < 0 else sorted(exp_by_layer)[:cpu_moe]
|
|
ram_mb = sum(exp_by_layer[l] for l in layers) / 1048576
|
|
else:
|
|
ram_mb = 0.0
|
|
gpu_files_mb = files_mb - ram_mb
|
|
steps = [c for c in CTX_STEPS if not ctx_max or c <= ctx_max]
|
|
if ctx_max and ctx_max not in steps:
|
|
steps.append(ctx_max) # krummes Maximum (z.B. 40960) mit anbieten
|
|
opts = []
|
|
for c in steps:
|
|
kv_mb = _kv_bytes(meta, c) / 1048576
|
|
opts.append({"ctx": c, "estMB": int(gpu_files_mb + kv_mb + VRAM_OVERHEAD_MB)})
|
|
out = {"ctxMax": ctx_max, "options": opts, "moe": moe, "cpuMoe": cpu_moe}
|
|
if moe:
|
|
out["expertsMB"] = int(sum(exp_by_layer.values()) / 1048576)
|
|
out["ramMB"] = int(ram_mb)
|
|
return out
|
|
|
|
|
|
async def _vram_used() -> dict:
|
|
"""{index: belegte MB} — frisch bei jedem Aufruf (aendert sich laufend)."""
|
|
try:
|
|
proc = await asyncio.create_subprocess_exec(
|
|
"nvidia-smi", "--query-gpu=index,memory.used", "--format=csv,noheader,nounits",
|
|
stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL)
|
|
out, _ = await asyncio.wait_for(proc.communicate(), 10)
|
|
return {a.strip(): int(float(b)) for a, b in
|
|
(l.split(",", 1) for l in out.decode().splitlines() if "," in l)}
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
# Ein geladenes Modell belegt GB auf seiner Karte. Bleibt die Karte trotz
|
|
# "ready" praktisch leer, rechnet llama.cpp auf der CPU — typisch, wenn der
|
|
# Treiber zu alt fuer das CUDA im Image ist ("forward compatibility …").
|
|
# llama.cpp meldet das nicht nach aussen; wir sehen es nur am Speicher.
|
|
ON_CPU_BELOW_MB = 400
|
|
|
|
|
|
async def _slots_info() -> list:
|
|
"""Pro LLM-Karte: [{gpu, gpuName, model, state}]. state aus llama-swap
|
|
/running: ready | starting | stopping | stopped; "unknown", wenn llama-swap
|
|
nicht antwortet; "" bei leerer Karte."""
|
|
slots = _load_slots()
|
|
running = await _fetch_loaded()
|
|
rmap = {r["model"]: r["state"] for r in running} if running is not None else None
|
|
gpus = await _gpu_names()
|
|
used = await _vram_used()
|
|
out = []
|
|
for card in _llm_cards():
|
|
m = slots.get(card, "")
|
|
state = "" if not m else ("unknown" if rmap is None else rmap.get(m, "stopped"))
|
|
g = gpus.get(card) or {}
|
|
u = used.get(card)
|
|
on_cpu = bool(m and state == "ready" and u is not None and u < ON_CPU_BELOW_MB)
|
|
out.append({"gpu": card, "gpuName": g.get("name", ""), "vramMB": g.get("vramMB", 0),
|
|
"vramUsedMB": u if u is not None else -1, "onCpu": on_cpu,
|
|
"model": m, "state": state, "ctx": _model_ctx(m) if m else 0,
|
|
**(_ctx_profile(m) if m else {})})
|
|
return out
|
|
|
|
|
|
def _selectable(slots_info: list) -> list:
|
|
"""Was Clients auswaehlen duerfen: belegte Karten, deren Modell laeuft oder
|
|
gerade (neu) startet — llama-swap haelt Anfragen bis "ready" fest. Auch
|
|
"unknown": wenn llama-swap /running nicht kennt (aeltere Version), lieber
|
|
die Belegung zeigen als die Box unbenutzbar machen."""
|
|
return [s["model"] for s in slots_info
|
|
if s["model"] and s["state"] in ("ready", "starting", "unknown")]
|
|
|
|
|
|
async def _ping(ws) -> None:
|
|
"""Lebenszeichen: busy + Karten-Belegung + auswaehlbare Modelle."""
|
|
info = await _slots_info()
|
|
await _send(ws, "worker_ping",
|
|
{"instanceId": INSTANCE_ID, "busy": _inflight > 0,
|
|
"slots": info, "models": _selectable(info)})
|
|
|
|
|
|
async def _announce(ws) -> None:
|
|
"""Sendet ein frisches worker_hello mit der aktuellen Modell-Liste (nach
|
|
Provision/Remove aufrufen, damit Bridge+Diagnostic das neue Modell lernen)."""
|
|
info = await _slots_info()
|
|
models = _selectable(info)
|
|
await _send(ws, "worker_hello", {
|
|
"instanceId": INSTANCE_ID, "service": WORKER_SERVICE,
|
|
# Die LLM-Karten (LLM_GPUS/LLM_GPU). Die eigene NVIDIA_VISIBLE_DEVICES
|
|
# des Adapters ist "all" bzw. wird vom Container-Toolkit ersetzt
|
|
# (Anzeige "GPU void") — sagt nichts darueber, wo die Modelle laufen.
|
|
"node": NODE_NAME, "gpus": ",".join(_llm_cards()),
|
|
"model": models[0] if models else "",
|
|
# NUR geladene Modelle — das ist die Auswahl fuer Clients (ARIA, Fremd-
|
|
# programme). Laden geht ausschliesslich ueber die Diagnostic.
|
|
"models": models,
|
|
"slots": info, # pro LLM-Karte: Modell + Zustand (Diagnostic)
|
|
"knownModels": _known_models(), # was in eine Karte geladen werden KOENNTE
|
|
"disk": await asyncio.to_thread(_disk_info), # frei + heruntergeladene Modelle
|
|
# OpenAI-Endpunkt fuer Fremd-Tools (OpenHands & Co.). Der Key selbst wird
|
|
# NICHT gesendet (Broadcast!) — nur ob einer gesetzt ist und ob die .env
|
|
# ihn vorgibt (dann ist das Feld in der Diagnostic schreibgeschuetzt).
|
|
"openaiPort": OPENAI_PORT,
|
|
"openaiBind": OPENAI_BIND,
|
|
"openaiKeySet": bool(_openai_key()),
|
|
"openaiEnvLocked": bool(_OPENAI_KEY_ENV),
|
|
})
|
|
logger.info("worker_hello: %s", ", ".join(
|
|
f"Karte {x['gpu']}={x['model'] or '-'}({x['state'] or 'leer'}{', CPU!' if x.get('onCpu') else ''})" for x in info))
|
|
|
|
|
|
async def _worker_register(ws) -> None:
|
|
"""Meldet diesen Worker bei der aria-bridge an (worker_hello) und haelt die
|
|
Flotten-Registry per periodischem worker_ping (mit busy-Status) frisch."""
|
|
try:
|
|
await _announce(ws)
|
|
n = 0
|
|
while True:
|
|
await asyncio.sleep(WORKER_PING_INTERVAL_S)
|
|
n += 1
|
|
await _ping(ws)
|
|
if n % 3 == 0: # ~30s worker_hello wiederholen (wie der Satellit) →
|
|
await _announce(ws) # auch neu gestartetes Diagnostic/Bridge lernt uns
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception:
|
|
return # Socket tot → still beenden; _run reconnectet + startet neu
|
|
|
|
|
|
async def _handle_llm_request(ws, payload: dict) -> None:
|
|
global _last_model, _inflight
|
|
_inflight += 1
|
|
try:
|
|
await _do_llm_request(ws, payload)
|
|
finally:
|
|
_inflight -= 1
|
|
|
|
|
|
async def _do_llm_request(ws, payload: dict) -> None:
|
|
global _last_model
|
|
req_id = payload.get("requestId", "")
|
|
messages = payload.get("messages") or []
|
|
if not isinstance(messages, list) or not messages:
|
|
await _send(ws, "llm_response", {
|
|
"requestId": req_id, "ok": False, "error": "leere/ungueltige messages",
|
|
})
|
|
return
|
|
max_tokens = int(payload.get("max_tokens", 512) or 512)
|
|
temperature = float(payload.get("temperature", 0.7) or 0.7)
|
|
stop = payload.get("stop")
|
|
tools = payload.get("tools") or None
|
|
model = (payload.get("model") or "").strip() or None
|
|
slots = _load_slots()
|
|
loaded = [slots[c] for c in _llm_cards() if slots.get(c)]
|
|
targeted = payload.get("targetInstance") == INSTANCE_ID
|
|
|
|
async def _refuse(code: str, msg: str) -> None:
|
|
await _send(ws, "llm_response", {
|
|
"requestId": req_id, "ok": False, "error": msg, "errorCode": code,
|
|
"instanceId": INSTANCE_ID, "node": NODE_NAME, "loadedModels": loaded,
|
|
})
|
|
|
|
# Clients laden nichts: ein Modell, das auf keiner Karte liegt, gibt es
|
|
# fuer sie nicht. Gezielt an uns adressiert → klare Absage mit der Liste,
|
|
# was geladen IST. Als Broadcast → schweigen: eine andere Box hat es
|
|
# vielleicht (sonst antworteten alle Boxen ohne das Modell mit Fehlern).
|
|
if model and model not in loaded:
|
|
if targeted:
|
|
await _refuse("model_not_loaded",
|
|
f"Modell {model!r} ist auf {NODE_NAME} nicht geladen. Geladen: "
|
|
+ (", ".join(loaded) or "nichts")
|
|
+ ". Laden geht nur ueber die ARIA-Diagnostic.")
|
|
return
|
|
if not model:
|
|
if not loaded:
|
|
if targeted:
|
|
await _refuse("no_model_loaded",
|
|
f"Auf {NODE_NAME} ist keine Karte mit einem Modell belegt.")
|
|
return
|
|
# Ohne Modellangabe: das der ersten LLM-Karte (Verhalten wie frueher,
|
|
# als es nur eins gab). Clients sollten das Modell immer angeben.
|
|
model = loaded[0]
|
|
eff_model = model
|
|
|
|
t0 = time.time()
|
|
res = await _call_llama(messages, max_tokens=max_tokens,
|
|
temperature=temperature, stop=stop, tools=tools,
|
|
model=model)
|
|
dt = time.time() - t0
|
|
if not res.get("ok"):
|
|
await _emit_llm_status(ws, "error", eff_model, error=(res.get("error") or "")[:120])
|
|
tc = res.get("tool_calls")
|
|
logger.info("llm_request id=%s model=%s -> ok=%s %.2fs content_len=%d tool_calls=%d",
|
|
(req_id[:8] if req_id else "?"), model, res.get("ok"), dt,
|
|
len(res.get("content") or ""), len(tc) if tc else 0)
|
|
await _send(ws, "llm_response", {
|
|
"requestId": req_id,
|
|
"ok": res.get("ok", False),
|
|
"content": res.get("content", ""),
|
|
"tool_calls": tc,
|
|
"error": res.get("error"),
|
|
"model": model,
|
|
# Fehlerart fuer Programme (z.B. context_exceeded + promptTokens + ctx)
|
|
**{k: res[k] for k in ("errorCode", "promptTokens", "ctx") if res.get(k)},
|
|
# Wer hat geantwortet: Box + Karte. Hilft Clients mit mehreren Boxen.
|
|
"instanceId": INSTANCE_ID, "node": NODE_NAME, "gpu": _card_of(model),
|
|
"usage": res.get("usage") or {},
|
|
"elapsedMs": int(dt * 1000),
|
|
})
|
|
|
|
|
|
# ── llama-swap/llama.cpp aktualisieren (ueber den llm-updater) ──
|
|
# Der Adapter hat bewusst KEINEN Docker-Zugriff. Pruefen, Aktualisieren und
|
|
# Zuruecksetzen macht der llm-updater (eigener Container, nur im Compose-Netz);
|
|
# der Adapter reicht die Diagnostic-Anfrage weiter und meldet den Fortschritt.
|
|
UPDATER_URL = os.getenv("UPDATER_URL", "http://llm-updater:8090").rstrip("/")
|
|
_UPDATER_TOKEN_FILE = "/models/.updater_token"
|
|
|
|
|
|
def _updater_token() -> str:
|
|
"""Gemeinsamer Token mit dem llm-updater. Liegt in /models (der Updater
|
|
haengt es nur lesend ein); fehlt er, legt der Adapter ihn an."""
|
|
try:
|
|
tok = open(_UPDATER_TOKEN_FILE).read().strip()
|
|
if tok:
|
|
return tok
|
|
except OSError:
|
|
pass
|
|
import secrets
|
|
tok = secrets.token_hex(24)
|
|
try:
|
|
fd = os.open(_UPDATER_TOKEN_FILE, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
|
|
with os.fdopen(fd, "w") as f:
|
|
f.write(tok)
|
|
except OSError as e:
|
|
logger.warning("Updater-Token nicht schreibbar: %s", e)
|
|
return tok
|
|
|
|
|
|
async def _updater(method: str, path: str, timeout: float = 60) -> dict:
|
|
try:
|
|
async with httpx.AsyncClient(timeout=timeout) as c:
|
|
r = await c.request(method, UPDATER_URL + path,
|
|
headers={"X-Updater-Token": _updater_token()})
|
|
return r.json()
|
|
except httpx.ConnectError:
|
|
return {"ok": False, "error": "Der Update-Dienst (llm-updater) laeuft auf dieser Box "
|
|
"nicht. Mit `docker compose up -d llm-updater` starten."}
|
|
except Exception as e:
|
|
return {"ok": False, "error": f"Update-Dienst: {str(e)[:150]}"}
|
|
|
|
|
|
async def _handle_runtime(ws, payload: dict) -> None:
|
|
"""op = status | check | update | rollback (aus der Diagnostic)."""
|
|
if (payload.get("instanceId") or INSTANCE_ID) != INSTANCE_ID:
|
|
return
|
|
op = payload.get("op", "status")
|
|
base = {"requestId": payload.get("requestId", ""), "instanceId": INSTANCE_ID,
|
|
"node": NODE_NAME, "op": op}
|
|
if op in ("status", "check"):
|
|
res = await _updater("GET", "/" + op, timeout=60)
|
|
await _send(ws, "llm_runtime_result", {**base, **res})
|
|
return
|
|
if op not in ("update", "rollback"):
|
|
await _send(ws, "llm_runtime_result", {**base, "ok": False, "error": "unbekannte Aktion"})
|
|
return
|
|
start = await _updater("POST", "/" + op)
|
|
if not start.get("ok"):
|
|
await _send(ws, "llm_runtime_result", {**base, **start})
|
|
return
|
|
logger.info("llama-swap %s gestartet (%s)", op, start.get("job"))
|
|
seen = 0
|
|
while True:
|
|
await asyncio.sleep(3)
|
|
job = await _updater("GET", "/job", timeout=20)
|
|
if job.get("error") and "state" not in job:
|
|
continue # Updater kurz nicht erreichbar — weiter warten
|
|
lines = job.get("log") or []
|
|
if len(lines) > seen:
|
|
await _send(ws, "llm_runtime_progress", {**base, "lines": lines[seen:]})
|
|
seen = len(lines)
|
|
if job.get("state") == "done":
|
|
await _send(ws, "llm_runtime_result", {**base, **(job.get("result") or {})})
|
|
logger.info("llama-swap %s fertig: %s", op, (job.get("result") or {}).get("ok"))
|
|
await _announce(ws)
|
|
return
|
|
|
|
|
|
async def _handle_openai_config(ws, payload: dict) -> None:
|
|
"""Setzt den API-Key des OpenAI-Endpunkts (aus der Diagnostic).
|
|
|
|
Der PORT wird hier bewusst NICHT geaendert: er ist ein Docker-Port-Mapping
|
|
und steht beim Container-Start fest — dafuer ist die .env zustaendig.
|
|
Steht der Key in der .env, gewinnt sie und wir lehnen ab.
|
|
"""
|
|
if (payload.get("instanceId") or INSTANCE_ID) != INSTANCE_ID:
|
|
return
|
|
req_id = payload.get("requestId", "")
|
|
key = str(payload.get("apiKey") or "").strip()
|
|
if _OPENAI_KEY_ENV:
|
|
await _send(ws, "llm_openai_config_result", {
|
|
"requestId": req_id, "instanceId": INSTANCE_ID, "ok": False,
|
|
"error": "Der API-Key steht in der .env dieser Box — die hat Vorrang. "
|
|
"Zum Aendern dort anpassen und die Box neu starten.",
|
|
})
|
|
return
|
|
if not _openai_key_save(key):
|
|
await _send(ws, "llm_openai_config_result", {
|
|
"requestId": req_id, "instanceId": INSTANCE_ID, "ok": False,
|
|
"error": "Key konnte nicht gespeichert werden (Schreibrechte auf /models?)",
|
|
})
|
|
return
|
|
# Config neu schreiben, damit llama-server den Key beim naechsten Laden nutzt.
|
|
_generate_config()
|
|
logger.info("OpenAI-API-Key %s (llama-swap laedt die Modelle neu)",
|
|
"gesetzt" if key else "entfernt")
|
|
await _send(ws, "llm_openai_config_result", {
|
|
"requestId": req_id, "instanceId": INSTANCE_ID, "ok": True,
|
|
"keySet": bool(key),
|
|
"message": ("API-Key gesetzt — greift in wenigen Sekunden. llama-swap startet "
|
|
"die Modelle dafuer kurz neu."
|
|
if key else "API-Key entfernt — der Endpunkt ist jetzt offen."),
|
|
})
|
|
await _announce(ws)
|
|
|
|
|
|
# ── Download-Fortschritt beim Modell-Laden ─────────────────────────────
|
|
# llama.cpp zieht das GGUF per -hf selbst und meldet keinen Fortschritt. Wir
|
|
# messen von aussen: Hugging Face sagt vorab, welche Dateien kommen (Name,
|
|
# Groesse, sha256). llama.cpp legt sie im HF-Cache-Layout ab —
|
|
# /models/models--org--repo/blobs/<sha256>.downloadInProgress (waechst)
|
|
# /models/models--org--repo/blobs/<sha256> (fertig)
|
|
# Genau diese Dateien werden gemessen. Das klappt auch bei gesplitteten
|
|
# Modellen und bei zwei Downloads gleichzeitig. Fallback fuer aeltere
|
|
# llama.cpp-Layouts: alle *.downloadInProgress unter /models.
|
|
MODELS_DIR = os.getenv("LLAMA_CACHE_DIR", "/models")
|
|
HF_TOKEN = os.getenv("HF_TOKEN", "").strip()
|
|
WARMUP_TIMEOUT_S = float(os.getenv("LLM_WARMUP_TIMEOUT_S", "1900")) # > healthCheckTimeout
|
|
_INPROGRESS = ".downloadInProgress"
|
|
_SPLIT_RE = re.compile(r"-(\d{5})-of-(\d{5})\.gguf$", re.IGNORECASE)
|
|
|
|
|
|
# Quantisierung aus einem Dateinamen: "gemma-4-12B-it-qat-UD-Q4_K_XL.gguf" → "UD-Q4_K_XL"
|
|
_QUANT_RE = re.compile(r"(?i)((?:UD-)?(?:IQ[1-4]_(?:XXS|XS|S|M|NL)|Q[2-8]_K(?:_(?:XL|L|M|S))?|Q[4-8]_[01]|BF16|F16|F32))")
|
|
# Wenn die gewuenschte fehlt: die erste vorhandene aus dieser Liste. 4-Bit zuerst
|
|
# (beste Groesse/Qualitaet fuer 12-GB-Karten), dann groessere.
|
|
QUANT_PREFERENCE = ["Q4_K_M", "UD-Q4_K_XL", "Q4_K_XL", "Q4_K_S", "IQ4_XS", "IQ4_NL", "Q4_0",
|
|
"Q5_K_M", "UD-Q5_K_XL", "Q6_K", "Q8_0"]
|
|
|
|
|
|
def _quant_of(path: str) -> str:
|
|
base = os.path.basename(path)
|
|
base = _SPLIT_RE.sub(".gguf", base)
|
|
m = _QUANT_RE.findall(base[:-5] if base.lower().endswith(".gguf") else base)
|
|
return m[-1].upper() if m else ""
|
|
|
|
|
|
def _is_main_gguf(path: str) -> bool:
|
|
"""Das eigentliche Modell — nicht Bild-Projektor (mmproj), nicht die
|
|
Zusatzdatei fuers spekulative Dekodieren (mtp), keine imatrix."""
|
|
low = path.lower()
|
|
base = os.path.basename(low)
|
|
return (low.endswith(".gguf") and "mmproj" not in base and not base.startswith("mtp")
|
|
and "/mtp/" not in ("/" + low) and "imatrix" not in base)
|
|
|
|
|
|
async def _hf_repo_files(repo: str):
|
|
"""Alle Modell-GGUFs eines Repos: [{path, oid, size, quant}] — oder None,
|
|
wenn Hugging Face nicht erreichbar ist (dann pruefen wir eben nicht)."""
|
|
url = f"https://huggingface.co/api/models/{repo}/tree/main?recursive=1"
|
|
headers = {"Authorization": f"Bearer {HF_TOKEN}"} if HF_TOKEN else {}
|
|
try:
|
|
async with httpx.AsyncClient(timeout=15, follow_redirects=True) as c:
|
|
r = await c.get(url, headers=headers)
|
|
r.raise_for_status()
|
|
entries = r.json()
|
|
except Exception as e:
|
|
logger.warning("HF-Dateiliste fuer %s nicht abrufbar (%s)", repo, e)
|
|
return None
|
|
out = []
|
|
for e in entries if isinstance(entries, list) else []:
|
|
path = e.get("path") or ""
|
|
if e.get("type") != "file" or not _is_main_gguf(path):
|
|
continue
|
|
lfs = e.get("lfs") or {}
|
|
out.append({"path": path, "oid": lfs.get("oid") or e.get("oid") or "",
|
|
"size": int(lfs.get("size") or e.get("size") or 0), "quant": _quant_of(path)})
|
|
return out
|
|
|
|
|
|
def _pick_quant(files: list, wanted: str) -> str:
|
|
"""Die gewuenschte Quantisierung, wenn es sie gibt — sonst die erste aus
|
|
QUANT_PREFERENCE, die das Repo hat. "" wenn gar nichts Brauchbares da ist."""
|
|
have = {f["quant"] for f in files if f["quant"]}
|
|
if wanted and wanted.upper() in have:
|
|
return wanted.upper()
|
|
return next((q for q in QUANT_PREFERENCE if q in have), "")
|
|
|
|
|
|
def _files_for_quant(files: list, quant: str) -> list:
|
|
"""Die Datei(en), die llama.cpp fuer repo:quant zieht. Gesplittet → alle
|
|
Teile; sonst die eine (bei mehreren die mit dem kuerzesten Namen)."""
|
|
match = [f for f in files if f["quant"] == quant.upper()]
|
|
split = [f for f in match if _SPLIT_RE.search(f["path"])]
|
|
if split:
|
|
return split
|
|
return [min(match, key=lambda f: len(f["path"]))] if match else []
|
|
|
|
|
|
async def _hf_expected_files(repo: str, quant: str) -> list:
|
|
"""Welche GGUF-Dateien zieht 'repo:quant'? [{path, oid, size}]. Fail-soft: []."""
|
|
files = await _hf_repo_files(repo)
|
|
return _files_for_quant(files, quant) if files else []
|
|
|
|
|
|
def _scan_download(oids: set) -> tuple:
|
|
"""(bytes_da, laeuft_noch, gefunden) fuer die erwarteten Blobs."""
|
|
done, active, found = 0, False, 0
|
|
fallback, fallback_active = 0, False
|
|
for root, _dirs, fns in os.walk(MODELS_DIR):
|
|
for fn in fns:
|
|
inprog = fn.endswith(_INPROGRESS)
|
|
base = fn[:-len(_INPROGRESS)] if inprog else fn
|
|
try:
|
|
st = os.stat(os.path.join(root, fn))
|
|
except OSError:
|
|
continue
|
|
# Der eigene Parallel-Download legt die Datei in voller Groesse an
|
|
# (sparse) — dann zaehlt nur, was wirklich geschrieben ist.
|
|
size = min(st.st_size, st.st_blocks * 512) if inprog else st.st_size
|
|
if base in oids:
|
|
done += size
|
|
found += 1
|
|
active |= inprog
|
|
elif inprog:
|
|
fallback += size
|
|
fallback_active = True
|
|
if not found and fallback_active:
|
|
return fallback, True, 0
|
|
return done, active, found
|
|
|
|
|
|
# Freier Platz vor dem Download: llama.cpp bricht bei voller Platte mitten
|
|
# im Download ab, und der Grund kommt ueber llama-swap nicht an. Vorher
|
|
# rechnen: was fehlt noch (fertige Blobs zaehlen nicht) gegen den freien
|
|
# Platz minus Puffer. Halbe Downloads, an denen seit STALE_PARTIAL_S niemand
|
|
# mehr schreibt, sind Reste abgebrochener Versuche — die duerfen weg.
|
|
DISK_RESERVE_BYTES = int(float(os.getenv("LLM_DISK_RESERVE_GB", "2")) * 1024**3)
|
|
STALE_PARTIAL_S = 30 * 60
|
|
|
|
|
|
def _blob_state(oids: set) -> tuple:
|
|
"""({oid: Bytes fertig}, [(pfad, bytes, mtime)] aller halben Downloads)."""
|
|
done, partial = {}, []
|
|
for root, _dirs, fns in os.walk(MODELS_DIR):
|
|
for fn in fns:
|
|
path = os.path.join(root, fn)
|
|
try:
|
|
st = os.stat(path)
|
|
except OSError:
|
|
continue
|
|
if fn.endswith(_INPROGRESS):
|
|
partial.append((path, st.st_size, st.st_mtime))
|
|
elif fn in oids:
|
|
done[fn] = st.st_size
|
|
return done, partial
|
|
|
|
|
|
def _check_disk(files: list) -> tuple:
|
|
"""(ok, Meldung). Ohne HF-Dateiliste (files leer) wird nicht geprueft."""
|
|
if not files:
|
|
return True, ""
|
|
done, partial = _blob_state({f["oid"] for f in files if f.get("oid")})
|
|
need = sum(f["size"] for f in files if f.get("oid") not in done)
|
|
if not need:
|
|
return True, ""
|
|
free = shutil.disk_usage(MODELS_DIR).free
|
|
note = ""
|
|
if need + DISK_RESERVE_BYTES > free:
|
|
stale = [(p, b) for p, b, mt in partial if time.time() - mt > STALE_PARTIAL_S]
|
|
for path, _b in stale:
|
|
try:
|
|
os.remove(path)
|
|
except OSError:
|
|
pass
|
|
if stale:
|
|
freed = sum(b for _p, b in stale)
|
|
note = f"Reste abgebrochener Downloads geloescht ({len(stale)} Datei(en), {freed / 1024**3:.1f} GB)"
|
|
logger.info("Speicherplatz: %s", note)
|
|
free = shutil.disk_usage(MODELS_DIR).free
|
|
if need + DISK_RESERVE_BYTES > free:
|
|
gb = lambda n: f"{n / 1024**3:.1f}".replace(".", ",")
|
|
return False, (f"Zu wenig Speicherplatz auf {NODE_NAME}: das Modell braucht ca. {gb(need)} GB, "
|
|
f"frei sind {gb(free)} GB (dazu {gb(DISK_RESERVE_BYTES)} GB Puffer). "
|
|
f"Nicht mehr gebrauchte Modelle im Katalog entfernen oder Platz auf der Box schaffen."
|
|
+ (f" ({note})" if note else ""))
|
|
return True, note
|
|
|
|
|
|
# Paralleler Download: llama.cpp zieht eine Datei ueber EINE Verbindung, das
|
|
# bleibt oft bei 10-50 MB/s haengen, auch bei 1-Gbit-Leitung. Wir holen die
|
|
# Datei vorher selbst in Stuecken ueber mehrere Verbindungen — direkt ins
|
|
# HF-Cache-Layout, das llama.cpp selbst anlegt (blobs/<sha256>, snapshots/
|
|
# <commit>/<datei> → Symlink, refs/main). Danach findet llama.cpp sie als
|
|
# "liegt schon da". Klappt es nicht, laedt llama.cpp wie bisher selbst.
|
|
DOWNLOAD_CONNECTIONS = int(os.getenv("LLM_DOWNLOAD_CONNECTIONS", "8"))
|
|
DOWNLOAD_PIECE = 64 * 1024**2
|
|
_provisions: dict = {} # key → {"cancel": threading.Event, "task", "phase"}
|
|
|
|
|
|
def _repo_dir(repo: str) -> str:
|
|
return os.path.join(MODELS_DIR, "models--" + repo.replace("/", "--"))
|
|
|
|
|
|
async def _hf_commit(repo: str) -> str:
|
|
headers = {"Authorization": f"Bearer {HF_TOKEN}"} if HF_TOKEN else {}
|
|
try:
|
|
async with httpx.AsyncClient(timeout=15, follow_redirects=True) as c:
|
|
r = await c.get(f"https://huggingface.co/api/models/{repo}", headers=headers)
|
|
r.raise_for_status()
|
|
return str(r.json().get("sha") or "")
|
|
except Exception as e:
|
|
logger.warning("HF-Commit fuer %s nicht abrufbar (%s)", repo, e)
|
|
return ""
|
|
|
|
|
|
def _link_snapshot(repo: str, commit: str, f: dict) -> None:
|
|
rdir = _repo_dir(repo)
|
|
link = os.path.join(rdir, "snapshots", commit, f["path"])
|
|
os.makedirs(os.path.dirname(link), exist_ok=True)
|
|
target = os.path.relpath(os.path.join(rdir, "blobs", f["oid"]), os.path.dirname(link))
|
|
if os.path.islink(link) or os.path.exists(link):
|
|
os.remove(link)
|
|
os.symlink(target, link)
|
|
os.makedirs(os.path.join(rdir, "refs"), exist_ok=True)
|
|
with open(os.path.join(rdir, "refs", "main"), "w") as fh:
|
|
fh.write(commit)
|
|
|
|
|
|
def _download_parallel(repo: str, commit: str, f: dict, cancel: threading.Event) -> str:
|
|
"""Eine Datei in Stuecken ueber mehrere Verbindungen. "" = fertig, sonst Grund."""
|
|
blobs = os.path.join(_repo_dir(repo), "blobs")
|
|
os.makedirs(blobs, exist_ok=True)
|
|
final = os.path.join(blobs, f["oid"])
|
|
part = final + _INPROGRESS
|
|
if os.path.exists(final):
|
|
return ""
|
|
if os.path.exists(part) and time.time() - os.path.getmtime(part) < 120:
|
|
return "es laeuft schon ein Download dieser Datei"
|
|
size = f["size"]
|
|
url = f"https://huggingface.co/{repo}/resolve/{commit}/{f['path']}"
|
|
headers = {"Authorization": f"Bearer {HF_TOKEN}"} if HF_TOKEN else {}
|
|
with open(part, "wb") as fh:
|
|
fh.truncate(size) # sparse: belegt erst beim Schreiben
|
|
pieces: queue.Queue = queue.Queue()
|
|
for start in range(0, size, DOWNLOAD_PIECE):
|
|
pieces.put((start, min(size, start + DOWNLOAD_PIECE) - 1))
|
|
errors: list = []
|
|
|
|
def worker() -> None:
|
|
fd = os.open(part, os.O_WRONLY)
|
|
try:
|
|
with httpx.Client(timeout=httpx.Timeout(30, read=60), follow_redirects=True,
|
|
headers=headers) as c:
|
|
while not cancel.is_set() and not errors:
|
|
try:
|
|
start, end = pieces.get_nowait()
|
|
except queue.Empty:
|
|
return
|
|
pos = start
|
|
for attempt in range(5):
|
|
try:
|
|
with c.stream("GET", url, headers={"Range": f"bytes={pos}-{end}"}) as r:
|
|
if r.status_code != 206:
|
|
raise RuntimeError(f"HTTP {r.status_code} statt Teil-Download")
|
|
for chunk in r.iter_bytes(1024 * 1024):
|
|
if cancel.is_set():
|
|
return
|
|
os.pwrite(fd, chunk, pos)
|
|
pos += len(chunk)
|
|
if pos != end + 1:
|
|
raise RuntimeError("Stueck unvollstaendig")
|
|
break
|
|
except Exception as e:
|
|
if attempt == 4 or cancel.is_set():
|
|
errors.append(str(e)[:200])
|
|
return
|
|
time.sleep(2 * (attempt + 1)) # weiter ab pos
|
|
finally:
|
|
os.close(fd)
|
|
|
|
threads = [threading.Thread(target=worker, daemon=True) for _ in range(max(1, DOWNLOAD_CONNECTIONS))]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
if cancel.is_set() or errors:
|
|
try:
|
|
os.remove(part)
|
|
except OSError:
|
|
pass
|
|
return "abgebrochen" if cancel.is_set() else errors[0]
|
|
h = hashlib.sha256()
|
|
with open(part, "rb") as fh:
|
|
while chunk := fh.read(8 * 1024 * 1024):
|
|
if cancel.is_set():
|
|
os.remove(part)
|
|
return "abgebrochen"
|
|
h.update(chunk)
|
|
if h.hexdigest() != f["oid"]:
|
|
os.remove(part)
|
|
return "Pruefsumme stimmt nicht — Datei verworfen"
|
|
os.replace(part, final)
|
|
return ""
|
|
|
|
|
|
async def _prefetch(ws, key: str, repo: str, files: list, card: str, cancel: threading.Event,
|
|
strict: bool = False) -> str:
|
|
"""Laedt fehlende Dateien parallel vor. "" = alles da (oder Fallback auf
|
|
llama.cpp), "abgebrochen" = Benutzer hat abgebrochen. strict (nur
|
|
herunterladen): kein Fallback — jeder Fehler kommt als Grund zurueck."""
|
|
files = [f for f in files if f.get("oid") and f.get("size")]
|
|
if not files:
|
|
return "keine Dateiliste von HuggingFace" if strict else ""
|
|
if DOWNLOAD_CONNECTIONS < 2 and not strict:
|
|
return ""
|
|
missing = [f for f in files if not os.path.exists(os.path.join(_repo_dir(repo), "blobs", f["oid"]))]
|
|
if not missing:
|
|
return ""
|
|
commit = await _hf_commit(repo)
|
|
if not commit:
|
|
return "HuggingFace nicht erreichbar" if strict else ""
|
|
stop = asyncio.Event()
|
|
await _emit_progress(ws, key, "start", total=sum(f["size"] for f in files), files=len(files), gpu=card,
|
|
note=f"Lade mit {DOWNLOAD_CONNECTIONS} Verbindungen")
|
|
watcher = asyncio.create_task(_watch_download(ws, key, files, stop, final_phase=False))
|
|
err = ""
|
|
try:
|
|
for f in missing:
|
|
err = await asyncio.to_thread(_download_parallel, repo, commit, f, cancel)
|
|
if err:
|
|
break
|
|
if not err:
|
|
for f in files:
|
|
await asyncio.to_thread(_link_snapshot, repo, commit, f)
|
|
finally:
|
|
stop.set()
|
|
try:
|
|
await watcher
|
|
except Exception:
|
|
pass
|
|
if err and err != "abgebrochen" and not strict:
|
|
logger.warning("Paralleler Download von %s fehlgeschlagen (%s) — llama.cpp laedt selbst", repo, err)
|
|
return ""
|
|
return err
|
|
|
|
|
|
def _disk_profiles(repo_dir: str) -> list:
|
|
"""Kontext-Profil (Maximum + GB je Stufe, Annahme: alles auf der Karte) je
|
|
heruntergeladener Quantisierung — damit die Diagnostic auch fuer Modelle,
|
|
die auf keiner Karte liegen, zeigen kann, was sie koennen und kosten."""
|
|
import glob
|
|
files = [p for p in glob.glob(os.path.join(repo_dir, "snapshots", "*", "**", "*.gguf"), recursive=True)
|
|
if os.path.isfile(p)]
|
|
mmproj = next((p for p in files if "mmproj" in os.path.basename(p).lower()), None)
|
|
out = []
|
|
for w in files:
|
|
if not _is_main_gguf(w) or _SPLIT_RE.search(w):
|
|
continue # gesplittete Modelle: Groesse waere nur die des ersten Teils
|
|
prof = _profile_of_files(w, mmproj)
|
|
if prof:
|
|
out.append({"quant": _quant_of(w), **{k: prof[k] for k in ("ctxMax", "options", "moe") if k in prof},
|
|
**({"expertsMB": prof["expertsMB"]} if "expertsMB" in prof else {})})
|
|
return out
|
|
|
|
|
|
def _disk_info() -> dict:
|
|
"""Freier Platz + fertig heruntergeladene Modelle (je Repo) fuer die Diagnostic."""
|
|
out = {"freeBytes": 0, "models": []}
|
|
try:
|
|
out["freeBytes"] = shutil.disk_usage(MODELS_DIR).free
|
|
for name in sorted(os.listdir(MODELS_DIR)):
|
|
if not name.startswith("models--"):
|
|
continue
|
|
blobs = os.path.join(MODELS_DIR, name, "blobs")
|
|
size = 0
|
|
for fn in (os.listdir(blobs) if os.path.isdir(blobs) else []):
|
|
if not fn.endswith(_INPROGRESS):
|
|
try:
|
|
size += os.path.getsize(os.path.join(blobs, fn))
|
|
except OSError:
|
|
pass
|
|
if size:
|
|
out["models"].append({"repo": name[len("models--"):].replace("--", "/", 1), "bytes": size,
|
|
"profiles": _disk_profiles(os.path.join(MODELS_DIR, name))})
|
|
except OSError as e:
|
|
logger.info("Plattenbelegung nicht lesbar: %s", e)
|
|
return out
|
|
|
|
|
|
async def _gpu_info(card: str) -> dict:
|
|
"""Name + VRAM einer Karte nach dem Laden (fuer 'in Karte N geladen')."""
|
|
if not card:
|
|
return {}
|
|
try:
|
|
proc = await asyncio.create_subprocess_exec(
|
|
"nvidia-smi", f"--id={card}",
|
|
"--query-gpu=index,name,memory.used,memory.total", "--format=csv,noheader,nounits",
|
|
stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL)
|
|
out, _ = await asyncio.wait_for(proc.communicate(), 10)
|
|
idx, name, used, total = [x.strip() for x in out.decode().strip().split(",")[:4]]
|
|
return {"gpu": idx, "gpuName": name, "vramUsedMB": int(used), "vramTotalMB": int(total)}
|
|
except Exception as e:
|
|
logger.info("GPU-Info nicht verfuegbar: %s", e)
|
|
return {"gpu": card}
|
|
|
|
|
|
# Tempo der Netzwerkkarte(n) der Box — der Container sieht nur sein eigenes
|
|
# eth0, darum liest er das /sys des Rechners (docker-compose: /sys:/hostsys:ro).
|
|
# Nur echte Karten (mit device-Link), die "up" sind. Fehlt der Mount: [].
|
|
HOST_SYS = os.getenv("HOST_SYS", "/hostsys")
|
|
_link_cache: dict = {"ts": 0.0, "links": []}
|
|
|
|
|
|
def _host_links() -> list:
|
|
if time.time() - _link_cache["ts"] < 60:
|
|
return _link_cache["links"]
|
|
links = []
|
|
base = os.path.join(HOST_SYS, "class", "net")
|
|
try:
|
|
for name in sorted(os.listdir(base)):
|
|
d = os.path.join(base, name)
|
|
try:
|
|
if not os.path.exists(os.path.join(d, "device")):
|
|
continue
|
|
with open(os.path.join(d, "operstate")) as fh:
|
|
if fh.read().strip() != "up":
|
|
continue
|
|
with open(os.path.join(d, "speed")) as fh:
|
|
mbit = int(fh.read().strip())
|
|
except (OSError, ValueError):
|
|
continue
|
|
if mbit > 0:
|
|
links.append({"iface": name, "mbit": mbit})
|
|
except OSError:
|
|
pass
|
|
_link_cache.update(ts=time.time(), links=links)
|
|
return links
|
|
|
|
|
|
async def _emit_progress(ws, key: str, phase: str, **extra) -> None:
|
|
await _send(ws, "llm_provision_progress",
|
|
{"instanceId": INSTANCE_ID, "node": NODE_NAME, "key": key,
|
|
"phase": phase, "ts": int(time.time() * 1000), "links": _host_links(), **extra})
|
|
|
|
|
|
async def _watch_download(ws, key: str, files: list, stop: asyncio.Event, final_phase: bool = True) -> None:
|
|
"""Meldet jede Sekunde den Download-Stand, bis der Warmup fertig ist.
|
|
Ist alles da, wechselt die Phase auf 'load' (llama.cpp laedt in die Karte)."""
|
|
oids = {f["oid"] for f in files if f.get("oid")}
|
|
total = sum(f["size"] for f in files)
|
|
phase = ""
|
|
downloaded = False # gab es einen echten Download? sonst: lag schon im Cache
|
|
last_b, last_t, speed = None, time.time(), 0.0
|
|
while not stop.is_set():
|
|
b, active, found = await asyncio.to_thread(_scan_download, oids)
|
|
now = time.time()
|
|
if last_b is not None and now > last_t and b >= last_b:
|
|
inst = (b - last_b) / (now - last_t)
|
|
speed = inst if speed == 0 else 0.7 * speed + 0.3 * inst
|
|
last_b, last_t = b, now
|
|
if active:
|
|
downloaded = True
|
|
phase = "download"
|
|
pct = round(min(100.0, 100.0 * b / total), 1) if total else None
|
|
eta = int(max(0, total - b) / speed) if total and speed > 1 else None
|
|
await _emit_progress(ws, key, "download", bytes=b, total=total, percent=pct,
|
|
speedBps=int(speed), etaSeconds=eta)
|
|
else:
|
|
# Nichts laeuft mehr: entweder fertig geladen (alle Blobs da) oder
|
|
# es lag schon im Cache. Im Notfallmodus (keine HF-Liste) zaehlt
|
|
# nur: es lief ein Download, und jetzt nicht mehr.
|
|
all_there = bool(oids) and found >= len(oids)
|
|
if final_phase and phase != "load" and (all_there or downloaded):
|
|
phase = "load"
|
|
await _emit_progress(ws, key, "load", bytes=total or b, total=total,
|
|
percent=100, cached=not downloaded)
|
|
try:
|
|
await asyncio.wait_for(stop.wait(), 1.0)
|
|
except asyncio.TimeoutError:
|
|
pass
|
|
|
|
|
|
async def _provision_result(ws, key: str, ok: bool, **extra) -> None:
|
|
await _send(ws, "llm_provision_result",
|
|
{"instanceId": INSTANCE_ID, "key": key, "ok": ok, **extra})
|
|
|
|
|
|
async def _handle_provision(ws, payload: dict) -> None:
|
|
key = (payload.get("key") or "").strip()
|
|
if key in _provisions:
|
|
return await _provision_result(ws, key, False, gpu=str(payload.get("gpu") or ""),
|
|
error="Fuer dieses Modell laeuft schon ein Auftrag.")
|
|
_provisions[key] = {"cancel": threading.Event(), "task": asyncio.current_task(), "phase": "prep"}
|
|
try:
|
|
await _do_provision(ws, payload, _provisions[key])
|
|
finally:
|
|
_provisions.pop(key, None)
|
|
|
|
|
|
async def _download_only(ws, key: str, repo: str, wanted: str, job: dict) -> None:
|
|
"""Nur auf die Platte der Box laden — keine Karte, keine Registry. Spaeter
|
|
"Laden" findet die Datei dann als "liegt schon da"."""
|
|
async def fail(msg: str) -> None:
|
|
await _emit_progress(ws, key, "error", error=msg, downloadOnly=True)
|
|
await _provision_result(ws, key, False, error=msg, downloadOnly=True)
|
|
if not repo:
|
|
return await fail("hfRepo fehlt")
|
|
files = await _hf_repo_files(repo)
|
|
if files is None:
|
|
return await fail(f"{repo}: nicht gefunden oder HuggingFace nicht erreichbar")
|
|
quant = _pick_quant(files, wanted) if files else ""
|
|
if not quant:
|
|
return await fail(f"In {repo} gibt es keine ladbare GGUF-Datei.")
|
|
qfiles = _files_for_quant(files, quant)
|
|
disk_ok, disk_note = await asyncio.to_thread(_check_disk, qfiles)
|
|
if not disk_ok:
|
|
return await fail(disk_note)
|
|
job["phase"] = "download"
|
|
err = await _prefetch(ws, key, repo, qfiles, "", job["cancel"], strict=True)
|
|
if err or job["cancel"].is_set():
|
|
return await fail("Abgebrochen" if (err == "abgebrochen" or job["cancel"].is_set()) else err)
|
|
await _announce(ws)
|
|
total = sum(f["size"] for f in qfiles)
|
|
await _emit_progress(ws, key, "ready", percent=100, bytes=total, total=total, downloadOnly=True, quant=quant,
|
|
note="; ".join(x for x in (disk_note, f"{quant} liegt auf der Box — in keine Karte geladen") if x))
|
|
await _provision_result(ws, key, True, downloadOnly=True)
|
|
logger.info("Nur heruntergeladen: %s (%s, %s)", key, repo, quant)
|
|
|
|
|
|
async def _do_provision(ws, payload: dict, job: dict) -> None:
|
|
"""Laedt ein Modell in eine LLM-Karte (nur aus der Diagnostic). Ersetzt,
|
|
was dort lag. Ablauf: Registry + Belegung → Config (llama-swap laedt per
|
|
--watch-config + Preload) → Warmup, der wartet bis es bereit ist; daneben
|
|
misst _watch_download den Download."""
|
|
key = (payload.get("key") or "").strip()
|
|
repo = (payload.get("hfRepo") or "").strip()
|
|
if payload.get("downloadOnly"):
|
|
return await _download_only(ws, key, repo, (payload.get("quant") or "Q4_K_M").strip(), job)
|
|
cards = _llm_cards()
|
|
card = str(payload.get("gpu") if payload.get("gpu") is not None else "").strip()
|
|
if not card and len(cards) == 1:
|
|
card = cards[0] # aeltere Diagnostic ohne Kartenwahl, Box mit einer Karte
|
|
if not key:
|
|
return await _provision_result(ws, key, False, error="key fehlt")
|
|
if card not in cards:
|
|
return await _provision_result(
|
|
ws, key, False, gpu=card,
|
|
error=(f"Karte {card or '?'} ist auf {NODE_NAME} nicht fuers LLM vorgesehen "
|
|
f"(LLM-Karten: {', '.join(cards)} — LLM_GPUS in der .env der Box)."))
|
|
slots = _load_slots()
|
|
elsewhere = next((c for c, m in slots.items() if m == key and c != card), None)
|
|
if elsewhere is not None:
|
|
return await _provision_result(
|
|
ws, key, False, gpu=card,
|
|
error=f"{key} liegt schon auf Karte {elsewhere} dieser Box — ein Modell pro Box nur einmal.")
|
|
quant_note = ""
|
|
repo_files = None
|
|
# Fuer ein Zuruecksetzen: der bisherige Registry-Eintrag dieses Modells.
|
|
# Noetig, wenn nur das Kontextfenster geaendert wird (gleiches Modell, gleiche
|
|
# Karte) — sonst bliebe nach einem Fehlschlag der zu grosse Wert stehen.
|
|
old_entry = next((dict(e) for e in _load_registry() if e.get("key") == key), None)
|
|
if repo:
|
|
wanted = (payload.get("quant") or "Q4_K_M").strip()
|
|
quant = wanted
|
|
# Gibt es die Datei ueberhaupt? Fehlt sie, beendet sich llama-server
|
|
# sofort und llama-swap meldet nur "500" — der Grund geht verloren.
|
|
# Lieber vorher pruefen und ggf. die naechstbeste Quantisierung nehmen
|
|
# (unsloth z.B. hat statt Q4_K_M ein UD-Q4_K_XL).
|
|
repo_files = await _hf_repo_files(repo)
|
|
if repo_files is not None:
|
|
if not repo_files:
|
|
return await _provision_result(
|
|
ws, key, False, gpu=card,
|
|
error=f"In {repo} gibt es keine GGUF-Modelldatei.")
|
|
quant = _pick_quant(repo_files, wanted)
|
|
if not quant:
|
|
have = sorted({f["quant"] for f in repo_files if f["quant"]}) or ["?"]
|
|
return await _provision_result(
|
|
ws, key, False, gpu=card,
|
|
error=f"{repo}: keine brauchbare Quantisierung. Vorhanden: {', '.join(have)}")
|
|
if quant != wanted.upper():
|
|
quant_note = f"{wanted} gibt es in {repo} nicht — nehme {quant}"
|
|
logger.info("Karte %s: %s", card, quant_note)
|
|
# Vor jeder Aenderung an Registry/Belegung: passt es auf die Platte?
|
|
disk_ok, disk_note = await asyncio.to_thread(_check_disk, _files_for_quant(repo_files, quant))
|
|
if not disk_ok:
|
|
await _emit_progress(ws, key, "error", gpu=card, error=disk_note)
|
|
return await _provision_result(ws, key, False, gpu=card, error=disk_note)
|
|
if disk_note:
|
|
quant_note = "; ".join(x for x in (quant_note, disk_note) if x)
|
|
# Vor dem Umstellen der Karte: die alte bleibt nutzbar, solange geladen wird.
|
|
job["phase"] = "download"
|
|
pf = await _prefetch(ws, key, repo, _files_for_quant(repo_files, quant), card, job["cancel"])
|
|
if pf == "abgebrochen" or job["cancel"].is_set():
|
|
msg = "Abgebrochen — Karte unveraendert"
|
|
await _emit_progress(ws, key, "error", gpu=card, error=msg)
|
|
return await _provision_result(ws, key, False, gpu=card, error=msg)
|
|
ctx_req = int(payload.get("ctx") or 8192)
|
|
try:
|
|
cpu_moe = int(payload.get("cpuMoe") or 0)
|
|
except (TypeError, ValueError):
|
|
cpu_moe = 0
|
|
if cpu_moe == -2:
|
|
# "automatisch verteilen": so viele Experten auf die Karte wie passen
|
|
qfiles = _files_for_quant(repo_files, quant) if repo_files else []
|
|
vram = int(((await _gpu_names()).get(card) or {}).get("vramMB") or 0)
|
|
if len(qfiles) == 1 and vram:
|
|
cpu_moe = await _auto_cpu_moe(repo, qfiles[0], vram, ctx_req)
|
|
else:
|
|
cpu_moe = -1 # gesplittete Datei / Karte unbekannt: alle in den RAM
|
|
entry = {
|
|
"key": key, "hfRepo": repo, "quant": quant,
|
|
"ctx": ctx_req,
|
|
"ngl": int(payload.get("ngl") or 99),
|
|
"cpuMoe": cpu_moe,
|
|
}
|
|
_save_registry([e for e in _load_registry() if e.get("key") != key] + [entry])
|
|
elif not _model_cmd(key):
|
|
return await _provision_result(ws, key, False, gpu=card, error="hfRepo fehlt (unbekanntes Modell)")
|
|
|
|
previous = slots.get(card, "")
|
|
slots[card] = key
|
|
_save_slots(slots)
|
|
_generate_config()
|
|
|
|
quant = next((e.get("quant") for e in _load_registry() if e.get("key") == key), "") or "Q4_K_M"
|
|
files = _files_for_quant(repo_files, quant) if repo_files else []
|
|
await _emit_progress(ws, key, "start", total=sum(f["size"] for f in files),
|
|
files=len(files), gpu=card, replaces=previous, quant=quant,
|
|
note=quant_note)
|
|
await _reload_llama()
|
|
await _announce(ws)
|
|
stop = asyncio.Event()
|
|
watcher = asyncio.create_task(_watch_download(ws, key, files, stop))
|
|
t0 = time.time()
|
|
# Langer Timeout: ein GGUF mit mehreren GB braucht Minuten. Mit dem normalen
|
|
# LLM_TIMEOUT_SEC meldete das frueher "Fehler", waehrend llama-swap weiterlud.
|
|
job["phase"] = "llama"
|
|
try:
|
|
if job["cancel"].is_set():
|
|
raise asyncio.CancelledError
|
|
res = await _call_llama([{"role": "user", "content": "hi"}],
|
|
max_tokens=1, temperature=0.0, stop=None, model=key,
|
|
timeout=WARMUP_TIMEOUT_S)
|
|
except asyncio.CancelledError:
|
|
# Abbrechen-Knopf: wie ein Fehlschlag behandeln → Karte zuruecksetzen
|
|
res = {"ok": False, "error": "Abgebrochen"}
|
|
dt = time.time() - t0
|
|
stop.set()
|
|
try:
|
|
await watcher
|
|
except Exception:
|
|
pass
|
|
if res.get("ok"):
|
|
_ready_models.add(key)
|
|
await _emit_llm_status(ws, "ready", key, loadSeconds=round(dt, 1), freshlyDownloaded=dt > 25)
|
|
await _emit_progress(ws, key, "ready", percent=100, loadSeconds=round(dt, 1),
|
|
replaces=previous, **(await _gpu_info(card)))
|
|
await _ping(ws)
|
|
await _provision_result(ws, key, True, gpu=card, elapsedMs=int(dt * 1000))
|
|
logger.info("Karte %s: %s geladen (%.1fs)%s", card, key, dt,
|
|
f", ersetzt {previous}" if previous else "")
|
|
return
|
|
# Fehlgeschlagen (z.B. VRAM zu klein): Karte zuruecksetzen. Sonst stuende
|
|
# ein kaputtes Modell in der Config, und jedes Neuladen von llama-swap
|
|
# risse die anderen Karten mit.
|
|
err = (res.get("error") or "Laden fehlgeschlagen")[:200]
|
|
if "500" in err and dt < 120:
|
|
# llama-swap reicht den Grund nicht weiter ("upstream command exited
|
|
# prematurely"). Kam der Abbruch schnell, lief kein Download — dann ist
|
|
# es fast immer eins von dreien.
|
|
err = ("llama.cpp hat sich beim Start sofort beendet. Haeufigste Gruende: "
|
|
"die Modell-Architektur ist neuer als die llama.cpp-Version der Box, "
|
|
"zu wenig (V)RAM, oder die Datei passt nicht. (llama-swap gibt den "
|
|
"genauen Grund nicht weiter.)")
|
|
slots = _load_slots()
|
|
if previous:
|
|
slots[card] = previous
|
|
else:
|
|
slots.pop(card, None)
|
|
_save_slots(slots)
|
|
reg = [e for e in _load_registry() if e.get("key") != key]
|
|
if old_entry:
|
|
reg.append(old_entry)
|
|
_save_registry(reg)
|
|
_generate_config()
|
|
await _reload_llama()
|
|
if job["cancel"].is_set():
|
|
# llama.cpp ist gestoppt — seinen halben Download nicht liegen lassen
|
|
for f in files:
|
|
try:
|
|
os.remove(os.path.join(_repo_dir(repo), "blobs", f["oid"] + _INPROGRESS))
|
|
except OSError:
|
|
pass
|
|
await _announce(ws)
|
|
note = f" — Karte {card} wieder auf {previous}" if previous else f" — Karte {card} wieder leer"
|
|
await _emit_llm_status(ws, "error", key, error=err)
|
|
await _emit_progress(ws, key, "error", gpu=card, error=err + note)
|
|
await _provision_result(ws, key, False, gpu=card, error=err + note, elapsedMs=int(dt * 1000))
|
|
logger.warning("Karte %s: %s fehlgeschlagen (%s)%s", card, key, err, note)
|
|
|
|
|
|
async def _handle_unload(ws, payload: dict) -> None:
|
|
"""Karte leeren (nur Diagnostic). Das GGUF bleibt im Cache — erneutes Laden
|
|
geht dann ohne Download."""
|
|
card = str(payload.get("gpu") if payload.get("gpu") is not None else "").strip()
|
|
slots = _load_slots()
|
|
key = slots.pop(card, "")
|
|
if not key:
|
|
return await _provision_result(ws, "", False, gpu=card, error=f"Karte {card} ist schon leer")
|
|
_save_slots(slots)
|
|
_generate_config()
|
|
await _reload_llama()
|
|
await _announce(ws)
|
|
await _provision_result(ws, key, True, gpu=card, unloaded=True)
|
|
logger.info("Karte %s geleert (war %s)", card, key)
|
|
|
|
|
|
async def _handle_remove(ws, payload: dict) -> None:
|
|
"""Modell aus der Registry entfernen (und von seiner Karte, falls belegt).
|
|
Das GGUF bleibt im Cache — ausser deleteFiles. cancel = laufenden
|
|
Download/Ladevorgang abbrechen (gleicher Nachrichtentyp, damit die
|
|
RVS-Allow-List nicht erweitert werden muss)."""
|
|
key = (payload.get("key") or "").strip()
|
|
if not key:
|
|
return
|
|
if payload.get("cancel"):
|
|
job = _provisions.get(key)
|
|
if not job:
|
|
# Nichts laeuft (mehr) — haengenden Eintrag im Download-Fenster schliessen
|
|
return await _provision_result(ws, key, False, error="Abgebrochen (lief nicht mehr)")
|
|
job["cancel"].set()
|
|
if job["phase"] == "llama" and job.get("task"):
|
|
job["task"].cancel()
|
|
logger.info("Abbruch angefordert: %s (Phase %s)", key, job["phase"])
|
|
return
|
|
if payload.get("deleteFiles"):
|
|
return await _delete_model_files(ws, key, (payload.get("hfRepo") or "").strip())
|
|
_save_registry([e for e in _load_registry() if e.get("key") != key])
|
|
slots = _load_slots()
|
|
for c in [c for c, m in slots.items() if m == key]:
|
|
slots.pop(c)
|
|
_save_slots(slots)
|
|
_generate_config()
|
|
await _reload_llama()
|
|
await _announce(ws)
|
|
await _provision_result(ws, key, True, removed=True)
|
|
logger.info("removed model %s", key)
|
|
|
|
|
|
async def _delete_model_files(ws, key: str, repo: str) -> None:
|
|
"""Heruntergeladene Dateien eines Repos von der Platte loeschen. Nicht,
|
|
solange es auf einer Karte liegt oder gerade geladen wird."""
|
|
reg = _load_registry()
|
|
repo = repo or next((e.get("hfRepo") for e in reg if e.get("key") == key), "") or ""
|
|
if not re.fullmatch(r"\w[\w.-]*/\w[\w.-]*", repo):
|
|
return await _provision_result(ws, key, False, error=f"Unbekanntes Repo '{repo}' — nichts geloescht")
|
|
repo_of = {e.get("key"): e.get("hfRepo") for e in reg}
|
|
busy = [c for c, m in _load_slots().items() if m and repo_of.get(m) == repo]
|
|
if busy:
|
|
return await _provision_result(ws, key, False, error=(
|
|
f"Liegt noch auf Karte {', '.join(busy)} — erst die Karte leeren oder ein anderes Modell laden."))
|
|
if any(repo_of.get(k) == repo for k in _provisions):
|
|
return await _provision_result(ws, key, False, error="Wird gerade geladen — erst abbrechen.")
|
|
path = _repo_dir(repo)
|
|
freed = sum(m["bytes"] for m in _disk_info()["models"] if m["repo"] == repo)
|
|
try:
|
|
await asyncio.to_thread(shutil.rmtree, path)
|
|
except FileNotFoundError:
|
|
pass
|
|
except OSError as e:
|
|
return await _provision_result(ws, key, False, error=f"Loeschen fehlgeschlagen: {e}")
|
|
_save_registry([e for e in reg if e.get("hfRepo") != repo])
|
|
_generate_config()
|
|
await _announce(ws)
|
|
await _provision_result(ws, key, True, removed=True, deleted=True, freedBytes=freed)
|
|
logger.info("Dateien von %s geloescht (%.1f GB)", repo, freed / 1024**3)
|
|
|
|
|
|
async def _handle_models_request(ws, payload: dict) -> None:
|
|
"""Erkundung fuer Clients: welche Modelle kann ich bei DIESER Box fragen?
|
|
Jede Box antwortet einzeln (RVS ist ein Broadcast-Raum) — der Client
|
|
sammelt die Antworten ein. Nur geladene Modelle, mit Karte."""
|
|
info = await _slots_info()
|
|
sel = set(_selectable(info))
|
|
await _send(ws, "llm_models", {
|
|
"requestId": payload.get("requestId", ""),
|
|
"instanceId": INSTANCE_ID, "node": NODE_NAME,
|
|
"models": [{"model": x["model"], "gpu": x["gpu"], "gpuName": x["gpuName"], "ctx": x["ctx"],
|
|
"state": x["state"]} for x in info if x["model"] in sel],
|
|
})
|
|
|
|
|
|
async def _run() -> None:
|
|
global _current_ws
|
|
if not RVS_HOST:
|
|
logger.error("RVS_HOST nicht gesetzt — Abbruch")
|
|
return
|
|
if not RVS_TOKEN:
|
|
logger.error("RVS_TOKEN nicht gesetzt — Abbruch")
|
|
return
|
|
|
|
# llama-swap-Config aus Basis-Template + Registry erzeugen, BEVOR llama-swap
|
|
# sie braucht (llama-swap restart: unless-stopped faengt die Erst-Boot-
|
|
# Reihenfolge ab, falls es kurz vor uns startet).
|
|
_generate_config()
|
|
|
|
# Auslastungs-Sampler (GPU + Tokens) laeuft unabhaengig vom RVS.
|
|
asyncio.create_task(_stats.run_sampler())
|
|
|
|
use_tls = RVS_TLS
|
|
retry_s = 2
|
|
tls_fallback_tried = False
|
|
|
|
while True:
|
|
scheme = "wss" if use_tls else "ws"
|
|
uri_host = RVS_SNI if (use_tls and RVS_SNI) else RVS_HOST
|
|
url = f"{scheme}://{uri_host}:{RVS_PORT}?token={RVS_TOKEN}"
|
|
masked = url.replace(RVS_TOKEN, "***") if RVS_TOKEN else url
|
|
# max_size MUSS zum RVS/zur Bridge passen (1500 MB). Der RVS ist ein
|
|
# BROADCAST-Relay: eine grosse Nachricht (Datei-Transfer, Voice-Upload,
|
|
# APK-Update) geht an JEDEN Client im Raum. Wer ein kleineres Limit hat,
|
|
# schliesst die Verbindung mit Code 1009 ("message too big") und landet in
|
|
# einer Reconnect-Schleife — real gesehen mit einem 209-MB-Frame, der
|
|
# whisper und f5tts im Minutentakt neu starten liess.
|
|
_MAX_WS_BYTES = int(os.getenv("RVS_MAX_SIZE_MB", "1500")) * 1024 * 1024
|
|
connect_kwargs = {"ping_interval": 20, "ping_timeout": 10, "max_size": _MAX_WS_BYTES}
|
|
try:
|
|
logger.info("Verbinde zu RVS: %s%s (llama=%s)", masked,
|
|
f" (TCP {RVS_HOST})" if (use_tls and RVS_SNI) else "", LLAMA_URL)
|
|
async with websockets.connect(url, **connect_kwargs) as ws:
|
|
logger.info("RVS verbunden — llm-adapter online")
|
|
_current_ws = ws
|
|
if _outbox:
|
|
await _flush_outbox(ws)
|
|
retry_s = 2
|
|
tls_fallback_tried = False
|
|
ping_task = asyncio.create_task(_worker_register(ws))
|
|
while True:
|
|
try:
|
|
raw = await asyncio.wait_for(ws.recv(), timeout=RX_STALE_S)
|
|
except asyncio.TimeoutError:
|
|
logger.warning("Kein RVS-Traffic seit %ds — Verbindung halb-tot, reconnect", RX_STALE_S)
|
|
raise ConnectionError("rvs-stale")
|
|
try:
|
|
msg = json.loads(raw)
|
|
except Exception:
|
|
continue
|
|
mtype = msg.get("type")
|
|
payload = msg.get("payload", {}) or {}
|
|
# Redundanz-Routing: gezielt an eine andere Instanz adressiert
|
|
# → ignorieren. Ohne targetInstance → wie bisher (jeder nimmt).
|
|
tgt = payload.get("targetInstance")
|
|
if tgt and tgt != INSTANCE_ID:
|
|
continue
|
|
# Auslastungs-Monitor (node_stats_*) abfangen.
|
|
if await _stats.handle(ws, mtype, payload, _send):
|
|
continue
|
|
if mtype == "llm_openai_config":
|
|
asyncio.create_task(_handle_openai_config(ws, payload))
|
|
continue
|
|
if mtype == "llm_runtime":
|
|
asyncio.create_task(_handle_runtime(ws, payload))
|
|
continue
|
|
if mtype not in ("llm_request", "llm_provision_model", "llm_remove_model",
|
|
"llm_unload_model", "llm_models_request"):
|
|
continue
|
|
if mtype == "llm_provision_model":
|
|
asyncio.create_task(_handle_provision(ws, payload))
|
|
elif mtype == "llm_unload_model":
|
|
asyncio.create_task(_handle_unload(ws, payload))
|
|
elif mtype == "llm_remove_model":
|
|
asyncio.create_task(_handle_remove(ws, payload))
|
|
elif mtype == "llm_models_request":
|
|
asyncio.create_task(_handle_models_request(ws, payload))
|
|
else:
|
|
# Jede Anfrage nebenlaeufig — llama.cpp serialisiert intern,
|
|
# aber wir blockieren so nicht den Empfang weiterer Messages.
|
|
asyncio.create_task(_handle_llm_request(ws, payload))
|
|
except Exception as e:
|
|
logger.warning("RVS-Verbindung verloren/fehlgeschlagen: %s", e)
|
|
try:
|
|
ping_task.cancel()
|
|
except NameError:
|
|
pass
|
|
if use_tls and RVS_TLS_FALLBACK and not tls_fallback_tried:
|
|
tls_fallback_tried = True
|
|
use_tls = False
|
|
logger.info("TLS fehlgeschlagen — Fallback auf ws://")
|
|
continue
|
|
await asyncio.sleep(min(retry_s, 30))
|
|
retry_s = min(retry_s * 2, 30)
|
|
use_tls = RVS_TLS
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(_run())
|