feat(llm): ein Modell pro Karte, Clients waehlen nur aus — und Doku fuer Drittprogramme
Bisher konnte jeder Client per Modellname llama-swap zum Wechseln bringen — ein Fremdprogramm haette ARIA das Modell unter den Fuessen weggetauscht. Jetzt: - Jede LLM-Karte einer Box (LLM_GPUS=1,2, alt LLM_GPU=1) haelt GENAU ein Modell, dauerhaft. Welches, legt nur die Diagnostic fest. Karten, die nicht fuers LLM vorgesehen sind, tauchen beim LLM nicht auf und werden als Ziel abgelehnt. - Clients (ARIA, Fremdprogramme ueber RVS, OpenHands ueber den OpenAI-Port) sehen nur die geladenen Modelle und laden nie: Anfrage fuer ein anderes → model_not_loaded mit Liste des Geladenen (RVS) bzw. HTTP 404 (OpenAI-Port). Nichts wird verdraengt. - Mehrere Clients koennen dasselbe Modell gleichzeitig nutzen. llm-adapter: Belegung in /models/aria_slots.json. Die llama-swap-Config enthaelt nur belegte Karten: pro Modell CUDA_VISIBLE_DEVICES=<Karte>, ttl 0, groups.swap=false, hooks.on_startup.preload. Alles gegen llama-swap ausprobiert (zwei Modelle gleichzeitig ready, keins verdraengt, Unbekanntes 404, env kommt beim jeweiligen llama-server an). llama-swap laeuft mit --watch-config; die frueher aufgerufenen Reload-Endpunkte (/reload, /api/config/reload) gibt es nicht (404). Neu: llm_unload_model (Karte leeren), llm_models_request → llm_models (Erkundung, jede Box antwortet einzeln). Fehlgeschlagenes Laden setzt die Karte auf ihr vorheriges Modell zurueck, sonst risse jedes Neuladen die anderen Karten mit. Uebernahme: ohne Belegungsdatei landet LLM_MODEL auf der ersten LLM-Karte. Compose: llama-swap sieht alle Karten (NVIDIA_VISIBLE_DEVICES=all) und zaehlt sie wie nvidia-smi (CUDA_DEVICE_ORDER=PCI_BUS_ID); das Pinnen pro Modell macht der Adapter. Bridge: uebernimmt models/slots auch aus worker_ping (nach "Entladen" sonst 30 s veraltet) und sagt sofort ab, wenn das Modell auf keiner Box liegt, statt per Broadcast in den Timeout zu laufen. Diagnostic: Flotte zeigt pro LLM-Karte Modell + Zustand + "Entladen"; Katalog waehlt eine Ziel-KARTE (leere vorausgewaehlt, Ersetzen mit Rueckfrage); Download-Fenster nennt Karte und was ersetzt wird. RVS: llm_unload_model, llm_models_request, llm_models freigegeben. docs/LLM-SCHNITTSTELLE.md: Schnittstelle fuer Drittprogramme (RVS + OpenAI- Port), Fehlercodes, Gleichzeitigkeit, Python-Beispiel. Tests: 23 gegen llama-swap (Adapter direkt) + Ende-zu-Ende ueber den echten RVS mit dem Adapter als Prozess: Laden per RVS mit Fortschritt, das Doku-Beispiel unveraendert (nur Adresse) findet beide Karten und bekommt eine Antwort, Absage/Schweigen wie dokumentiert, zwei Clients parallel. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
+8
-1
@@ -37,7 +37,14 @@ NODE_NAME=ai-box
|
||||
WHISPER_GPU=0 # kleines STT → passt neben F5-TTS auf EINE Karte
|
||||
F5TTS_GPU=0 # TTS ist klein → gleiche Karte wie Whisper
|
||||
VOXTRAL_GPU=1 # STT-3B ~9 GB → nur auf einer groesseren Karte (z.B. 12 GB)
|
||||
LLM_GPU=0 # lokales LLM (teilt sich ggf. die Karte mit F5-TTS)
|
||||
LLM_GPU=0 # lokales LLM auf EINER Karte (teilt sich ggf. die Karte mit F5-TTS)
|
||||
# Mehrere Karten fuers LLM: Liste statt LLM_GPU. Jede Karte haelt genau EIN
|
||||
# Modell, dauerhaft geladen — welches, legst du in der Diagnostic fest
|
||||
# (Modell-Katalog → Laden → Karte). Clients (ARIA, Fremdprogramme, OpenHands)
|
||||
# koennen nur die geladenen nutzen und nie selbst etwas laden.
|
||||
# Beispiel 3 Karten: 0 = Whisper + F5-TTS, 1 und 2 nur LLM:
|
||||
# WHISPER_GPU=0 F5TTS_GPU=0 LLM_GPUS=1,2
|
||||
#LLM_GPUS=
|
||||
|
||||
# ─── RVS-Verbindung ───────────────────────────────────────
|
||||
# WICHTIG: Host, Port UND Token muessen EXAKT mit dem ARIA-Stack (Bridge/
|
||||
|
||||
+16
-6
@@ -133,12 +133,20 @@ services:
|
||||
volumes:
|
||||
- ./models:/models # HF-Cache + generierte Config
|
||||
environment:
|
||||
- NVIDIA_VISIBLE_DEVICES=${LLM_GPU:-0}
|
||||
# Alle Karten sichtbar — welches Modell auf welcher Karte laeuft, legt der
|
||||
# llm-adapter pro Modell fest (CUDA_VISIBLE_DEVICES=<Karte> in der
|
||||
# generierten Config). Erlaubt sind nur die LLM-Karten aus LLM_GPUS.
|
||||
- NVIDIA_VISIBLE_DEVICES=all
|
||||
- NVIDIA_DRIVER_CAPABILITIES=compute,utility
|
||||
# CUDA zaehlt Karten sonst "schnellste zuerst" — dann waere CUDA-Karte 0
|
||||
# nicht unbedingt nvidia-smi-Karte 0. PCI-Reihenfolge = wie nvidia-smi.
|
||||
- CUDA_DEVICE_ORDER=PCI_BUS_ID
|
||||
- LLAMA_CACHE=/models # llama-server legt -hf-Downloads hier ab
|
||||
# Liest die vom llm-adapter generierte Config. Beim allerersten Boot faengt
|
||||
# restart: unless-stopped die Reihenfolge ab, bis der Adapter sie geschrieben hat.
|
||||
command: ["--config", "/models/llama-swap.config.yaml", "--listen", "0.0.0.0:8080"]
|
||||
# Liest die vom llm-adapter generierte Config. --watch-config: Aenderungen
|
||||
# (Modell in Karte geladen/entladen) greifen von selbst — einen Reload-
|
||||
# Endpunkt hat llama-swap nicht. Beim allerersten Boot faengt restart:
|
||||
# unless-stopped die Reihenfolge ab, bis der Adapter die Config geschrieben hat.
|
||||
command: ["--config", "/models/llama-swap.config.yaml", "--listen", "0.0.0.0:8080", "--watch-config"]
|
||||
# OpenAI-kompatibler Endpunkt (/v1/chat/completions, /v1/models) — damit
|
||||
# koennen Fremd-Tools wie OpenHands, Continue oder Open WebUI direkt auf das
|
||||
# lokale LLM. Standardmaessig NUR auf localhost gebunden (sicher); fuer
|
||||
@@ -178,8 +186,10 @@ services:
|
||||
- LLM_REGISTRY=/models/aria_models.json
|
||||
# Erster Load eines Modells kann ein GGUF ziehen (mehrere GB) — grosszuegig.
|
||||
- LLM_TIMEOUT_SEC=${LLM_TIMEOUT_SEC:-600}
|
||||
# Auf welche Karte llama-swap gepinnt ist — fuer die Meldung „in Karte N
|
||||
# geladen" samt VRAM nach dem Laden. Dieselbe Variable wie bei llama-swap.
|
||||
# Welche Karten dem LLM gehoeren. Jede haelt genau ein Modell, geladen
|
||||
# wird nur ueber die Diagnostic. Mehrere: LLM_GPUS=1,2 — eine: LLM_GPU=1
|
||||
# (alte Schreibweise, gilt weiter, wenn LLM_GPUS leer ist).
|
||||
- LLM_GPUS=${LLM_GPUS:-}
|
||||
- LLM_GPU=${LLM_GPU:-0}
|
||||
# Fuer gesperrte (gated) HF-Repos: die Dateiliste fuer den Fortschritt.
|
||||
- HF_TOKEN=${HF_TOKEN:-}
|
||||
|
||||
+352
-132
@@ -96,6 +96,62 @@ 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
|
||||
@@ -161,69 +217,93 @@ def _save_registry(reg: list) -> None:
|
||||
logger.warning("Registry speichern fehlgeschlagen: %s", e)
|
||||
|
||||
|
||||
def _generate_config() -> int:
|
||||
"""Schreibt die llama-swap-Config aus Basis-Template + Registry. Gibt die
|
||||
Anzahl Modelle zurueck. Idempotent, bei jeder Aenderung + beim Start."""
|
||||
base = {}
|
||||
def _base_config() -> dict:
|
||||
try:
|
||||
with open(BASE_CONFIG_PATH) as f:
|
||||
base = yaml.safe_load(f) or {}
|
||||
return yaml.safe_load(f) or {}
|
||||
except Exception as e:
|
||||
logger.warning("Basis-Template %s nicht lesbar (%s)", BASE_CONFIG_PATH, e)
|
||||
models = dict(base.get("models") or {})
|
||||
for e in _load_registry():
|
||||
key = (e.get("key") or "").strip()
|
||||
repo = (e.get("hfRepo") or "").strip()
|
||||
if not key or not repo:
|
||||
continue
|
||||
quant = (e.get("quant") or "Q4_K_M").strip()
|
||||
ctx = int(e.get("ctx") or 8192)
|
||||
ngl = int(e.get("ngl") or 99)
|
||||
models[key] = {
|
||||
"cmd": (f"llama-server --port ${{PORT}} --host 127.0.0.1\n"
|
||||
f"-hf {repo}:{quant}\n-ngl {ngl} -c {ctx} --jinja"),
|
||||
"ttl": 3600,
|
||||
}
|
||||
# API-Key auf JEDES Modell anwenden (auch die aus dem Basis-Template) —
|
||||
# llama-server prueft dann den Authorization-Header. Ohne Key bleibt der
|
||||
# Endpunkt offen; deshalb ist er per Default nur an localhost gebunden.
|
||||
key = _openai_key()
|
||||
for k, spec in list(models.items()):
|
||||
cmd = str((spec or {}).get("cmd") or "")
|
||||
if not cmd:
|
||||
continue
|
||||
cmd = re.sub(r"\s*--api-key\s+\S+", "", cmd) # alten Key entfernen
|
||||
if key:
|
||||
cmd = cmd.rstrip() + f" --api-key {shlex.quote(key)}"
|
||||
models[k] = dict(spec, cmd=cmd)
|
||||
return {}
|
||||
|
||||
out = dict(base)
|
||||
|
||||
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")
|
||||
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 generiert: %d Modelle → %s", len(models), 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:
|
||||
"""Stoesst llama-swap-Reload an. Viele Builds watchen die Config-Datei ohnehin;
|
||||
zusaetzlich versuchen wir bekannte Reload-Endpunkte (Fehler ignoriert)."""
|
||||
for path in ("/api/config/reload", "/reload"):
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=10) as c:
|
||||
r = await c.post(f"{LLAMA_URL}{path}")
|
||||
if r.status_code < 400:
|
||||
logger.info("llama-swap reload via %s", path)
|
||||
return
|
||||
except Exception:
|
||||
pass
|
||||
logger.info("llama-swap reload: kein Endpoint — verlasse mich auf File-Watch")
|
||||
"""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
|
||||
|
||||
|
||||
async def _send(ws, mtype: str, payload: dict) -> None:
|
||||
@@ -331,27 +411,77 @@ async def _fetch_loaded() -> list:
|
||||
return None
|
||||
|
||||
|
||||
_gpu_name_cache: dict = {}
|
||||
|
||||
|
||||
async def _gpu_names() -> dict:
|
||||
"""{index: name} aller Karten, einmal per nvidia-smi geholt."""
|
||||
if _gpu_name_cache:
|
||||
return _gpu_name_cache
|
||||
try:
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
"nvidia-smi", "--query-gpu=index,name", "--format=csv,noheader",
|
||||
stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL)
|
||||
out, _ = await asyncio.wait_for(proc.communicate(), 10)
|
||||
for line in out.decode().splitlines():
|
||||
if "," in line:
|
||||
i, n = line.split(",", 1)
|
||||
_gpu_name_cache[i.strip()] = n.strip()
|
||||
except Exception:
|
||||
pass
|
||||
return _gpu_name_cache
|
||||
|
||||
|
||||
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
|
||||
names = await _gpu_names()
|
||||
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"))
|
||||
out.append({"gpu": card, "gpuName": names.get(card, ""), "model": m, "state": state})
|
||||
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 mit busy + geladenem Modell. loaded=None heisst "konnte
|
||||
llama-swap nicht fragen" — das ist etwas anderes als "nichts geladen"."""
|
||||
"""Lebenszeichen: busy + Karten-Belegung + auswaehlbare Modelle."""
|
||||
info = await _slots_info()
|
||||
await _send(ws, "worker_ping",
|
||||
{"instanceId": INSTANCE_ID, "busy": _inflight > 0,
|
||||
"loaded": await _fetch_loaded()})
|
||||
"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)."""
|
||||
models = await _fetch_available_models()
|
||||
info = await _slots_info()
|
||||
models = _selectable(info)
|
||||
await _send(ws, "worker_hello", {
|
||||
"instanceId": INSTANCE_ID, "service": WORKER_SERVICE,
|
||||
# Die LLM-Karte ist die von llama-swap (LLM_GPU). Die eigene
|
||||
# NVIDIA_VISIBLE_DEVICES des Adapters ist "all" bzw. wird vom
|
||||
# Container-Toolkit ersetzt (Anzeige "GPU void") — sagt nichts darueber,
|
||||
# wo das Modell laeuft.
|
||||
"node": NODE_NAME, "gpus": LLM_GPU or GPU_IDS, "model": LLM_MODEL,
|
||||
"models": models, # welche Modelle diese Box fahren kann (llama-swap-Keys)
|
||||
"loaded": await _fetch_loaded(), # welches davon GERADE in der Karte liegt
|
||||
# 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
|
||||
# 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).
|
||||
@@ -360,7 +490,7 @@ async def _announce(ws) -> None:
|
||||
"openaiKeySet": bool(_openai_key()),
|
||||
"openaiEnvLocked": bool(_OPENAI_KEY_ENV),
|
||||
})
|
||||
logger.info("worker_hello: models=%s", models)
|
||||
logger.info("worker_hello: %s", ", ".join(f"Karte {x['gpu']}={x['model'] or '-'}({x['state'] or 'leer'})" for x in info))
|
||||
|
||||
|
||||
async def _worker_register(ws) -> None:
|
||||
@@ -404,34 +534,48 @@ async def _do_llm_request(ws, payload: dict) -> None:
|
||||
stop = payload.get("stop")
|
||||
tools = payload.get("tools") or None
|
||||
model = (payload.get("model") or "").strip() or None
|
||||
eff_model = model or LLM_MODEL
|
||||
slots = _load_slots()
|
||||
loaded = [slots[c] for c in _llm_cards() if slots.get(c)]
|
||||
targeted = payload.get("targetInstance") == INSTANCE_ID
|
||||
|
||||
# Modellwechsel (oder erster Request) → llama-swap laedt/swappt: Status melden.
|
||||
switching = eff_model != _last_model
|
||||
if switching:
|
||||
await _emit_llm_status(ws, "loading", eff_model)
|
||||
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 switching:
|
||||
if res.get("ok"):
|
||||
fresh = (eff_model not in _ready_models) and dt > 25
|
||||
_ready_models.add(eff_model)
|
||||
_last_model = eff_model
|
||||
await _emit_llm_status(ws, "ready", eff_model,
|
||||
loadSeconds=round(dt, 1), freshlyDownloaded=fresh)
|
||||
await _ping(ws) # Diagnostic zeigt den Wechsel sofort, nicht erst in 10 s
|
||||
else:
|
||||
# bei Fehler _last_model NICHT setzen → naechster Versuch meldet erneut loading
|
||||
await _emit_llm_status(ws, "error", eff_model,
|
||||
error=(res.get("error") or "")[:120])
|
||||
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 or LLM_MODEL, res.get("ok"), dt,
|
||||
(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,
|
||||
@@ -439,7 +583,10 @@ async def _do_llm_request(ws, payload: dict) -> None:
|
||||
"content": res.get("content", ""),
|
||||
"tool_calls": tc,
|
||||
"error": res.get("error"),
|
||||
"model": model or LLM_MODEL,
|
||||
"model": model,
|
||||
# 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),
|
||||
})
|
||||
|
||||
@@ -470,14 +617,13 @@ async def _handle_openai_config(ws, payload: dict) -> None:
|
||||
return
|
||||
# Config neu schreiben, damit llama-server den Key beim naechsten Laden nutzt.
|
||||
_generate_config()
|
||||
logger.info("OpenAI-API-Key %s (wirkt beim naechsten Modell-Laden)",
|
||||
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. Er greift, sobald das Modell das naechste Mal "
|
||||
"geladen wird — bereits laufende Modelle einmal entladen lassen "
|
||||
"(TTL) oder ein anderes Modell anfragen."
|
||||
"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)
|
||||
@@ -494,7 +640,6 @@ async def _handle_openai_config(ws, payload: dict) -> None:
|
||||
# llama.cpp-Layouts: alle *.downloadInProgress unter /models.
|
||||
MODELS_DIR = os.getenv("LLAMA_CACHE_DIR", "/models")
|
||||
HF_TOKEN = os.getenv("HF_TOKEN", "").strip()
|
||||
LLM_GPU = os.getenv("LLM_GPU", "").strip() # Karte, auf die llama-swap gepinnt ist
|
||||
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)
|
||||
@@ -558,13 +703,13 @@ def _scan_download(oids: set) -> tuple:
|
||||
return done, active, found
|
||||
|
||||
|
||||
async def _gpu_info() -> dict:
|
||||
"""Name + VRAM der LLM-Karte nach dem Laden (fuer 'in Karte N geladen')."""
|
||||
if not LLM_GPU:
|
||||
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={LLM_GPU}",
|
||||
"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)
|
||||
@@ -572,7 +717,7 @@ async def _gpu_info() -> dict:
|
||||
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": LLM_GPU}
|
||||
return {"gpu": card}
|
||||
|
||||
|
||||
async def _emit_progress(ws, key: str, phase: str, **extra) -> None:
|
||||
@@ -618,39 +763,62 @@ async def _watch_download(ws, key: str, files: list, stop: asyncio.Event) -> Non
|
||||
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:
|
||||
"""Fuegt ein Modell hinzu: Registry+Config schreiben, reload, dann Warmup
|
||||
(zieht das GGUF via -hf beim ersten Load). Meldet die neue Modell-Liste."""
|
||||
"""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 not key or not repo:
|
||||
await _send(ws, "llm_provision_result",
|
||||
{"instanceId": INSTANCE_ID, "key": key, "ok": False, "error": "key/hfRepo fehlt"})
|
||||
return
|
||||
entry = {
|
||||
"key": key, "hfRepo": repo,
|
||||
"quant": (payload.get("quant") or "Q4_K_M").strip(),
|
||||
"ctx": int(payload.get("ctx") or 8192),
|
||||
"ngl": int(payload.get("ngl") or 99),
|
||||
}
|
||||
reg = [e for e in _load_registry() if e.get("key") != key]
|
||||
reg.append(entry)
|
||||
_save_registry(reg)
|
||||
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.")
|
||||
if repo:
|
||||
entry = {
|
||||
"key": key, "hfRepo": repo,
|
||||
"quant": (payload.get("quant") or "Q4_K_M").strip(),
|
||||
"ctx": int(payload.get("ctx") or 8192),
|
||||
"ngl": int(payload.get("ngl") or 99),
|
||||
}
|
||||
_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()
|
||||
await _reload_llama()
|
||||
await _announce(ws) # Bridge/Diagnostic lernen das neue Modell
|
||||
# Warmup: Mini-Request → llama-swap zieht das GGUF (falls noch nicht da)
|
||||
# und laedt es in die Karte. Daneben misst _watch_download den Download.
|
||||
files = await _hf_expected_files(repo, entry["quant"])
|
||||
|
||||
quant = next((e.get("quant") for e in _load_registry() if e.get("key") == key), "") or "Q4_K_M"
|
||||
files = await _hf_expected_files(repo, quant) if repo else []
|
||||
await _emit_progress(ws, key, "start", total=sum(f["size"] for f in files),
|
||||
files=len(files), gpu=LLM_GPU)
|
||||
await _emit_llm_status(ws, "loading", key)
|
||||
files=len(files), gpu=card, replaces=previous)
|
||||
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 (60 s) meldete das hier "Fehler", waehrend llama-swap im
|
||||
# Hintergrund munter weiterlud.
|
||||
# LLM_TIMEOUT_SEC meldete das frueher "Fehler", waehrend llama-swap weiterlud.
|
||||
res = await _call_llama([{"role": "user", "content": "hi"}],
|
||||
max_tokens=1, temperature=0.0, stop=None, model=key,
|
||||
timeout=WARMUP_TIMEOUT_S)
|
||||
@@ -662,35 +830,82 @@ async def _handle_provision(ws, payload: dict) -> None:
|
||||
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_llm_status(ws, "ready", key, loadSeconds=round(dt, 1), freshlyDownloaded=dt > 25)
|
||||
await _emit_progress(ws, key, "ready", percent=100, loadSeconds=round(dt, 1),
|
||||
**(await _gpu_info()))
|
||||
replaces=previous, **(await _gpu_info(card)))
|
||||
await _ping(ws)
|
||||
else:
|
||||
await _emit_llm_status(ws, "error", key, error=(res.get("error") or "")[:160])
|
||||
await _emit_progress(ws, key, "error", error=(res.get("error") or "")[:200])
|
||||
await _send(ws, "llm_provision_result",
|
||||
{"instanceId": INSTANCE_ID, "key": key, "ok": res.get("ok", False),
|
||||
"error": res.get("error"), "elapsedMs": int(dt * 1000)})
|
||||
logger.info("provision %s (%s) → ok=%s %.1fs", key, repo, res.get("ok"), dt)
|
||||
|
||||
|
||||
async def _handle_remove(ws, payload: dict) -> None:
|
||||
"""Entfernt ein Modell aus Registry+Config (GGUF bleibt im Cache)."""
|
||||
key = (payload.get("key") or "").strip()
|
||||
if not key:
|
||||
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
|
||||
reg = [e for e in _load_registry() if e.get("key") != key]
|
||||
_save_registry(reg)
|
||||
# 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]
|
||||
slots = _load_slots()
|
||||
if previous:
|
||||
slots[card] = previous
|
||||
else:
|
||||
slots.pop(card, None)
|
||||
_save_slots(slots)
|
||||
_generate_config()
|
||||
await _reload_llama()
|
||||
await _announce(ws)
|
||||
await _send(ws, "llm_provision_result",
|
||||
{"instanceId": INSTANCE_ID, "key": key, "ok": True, "removed": True})
|
||||
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."""
|
||||
key = (payload.get("key") or "").strip()
|
||||
if not key:
|
||||
return
|
||||
_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 _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"],
|
||||
"state": x["state"]} for x in info if x["model"] in sel],
|
||||
})
|
||||
|
||||
|
||||
async def _run() -> None:
|
||||
if not RVS_HOST:
|
||||
logger.error("RVS_HOST nicht gesetzt — Abbruch")
|
||||
@@ -755,12 +970,17 @@ async def _run() -> None:
|
||||
if mtype == "llm_openai_config":
|
||||
asyncio.create_task(_handle_openai_config(ws, payload))
|
||||
continue
|
||||
if mtype not in ("llm_request", "llm_provision_model", "llm_remove_model"):
|
||||
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.
|
||||
|
||||
Reference in New Issue
Block a user