duffyduckandClaude Opus 5.5
79a901089b
Anlass: die Software-Schmiede bekam "request (8274 tokens) exceeds the available context size (8192 tokens)". Sie holt ctx vom Cluster zwar ab, kuerzt aber nach einem eigenen, von Hand eingetragenen Wert (32768). ARIA-Seite: - Neue Fehlerart context_exceeded (statt model_error) mit promptTokens und ctx — Programme koennen automatisch kuerzen und nochmal senden, statt Text auszuwerten. llm-adapter liest es aus llama.cpps exceed_context_size_error, der Cluster-Router reicht es durch und wiederholt es nicht anderswo. - Diagnostic → LLM-Karten: Kontextfenster pro Karte (4k … 128k). Laedt das Modell mit neuem -c; Programme sehen den neuen Wert ueber ctx. Passt es nicht in den Grafikspeicher, setzt die Box zurueck — jetzt inkl. Registry (vorher blieb bei einem Fehlschlag der zu grosse ctx in aria_models.json stehen und waere beim naechsten Start wieder gescheitert). - Doku: context_exceeded, ctx kann sich aendern, "danach kuerzen, nicht nur anzeigen". Getestet gegen llama-swap: Kontext 512 → 1024 (gleiche Karte), Fehlschlag bei 65536 → Registry + Config wieder 1024, zu langer Verlauf → context_exceeded mit promptTokens=6030, ctx=1024. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
220 lines
11 KiB
Python
220 lines
11 KiB
Python
"""LLM-Cluster: ein Eingang fuer alle lokalen Sprachmodelle aller KI-Boxen.
|
|
|
|
Clients waehlen ein MODELL, nicht Box oder Karte. Wo es laeuft, entscheidet
|
|
dieser Router:
|
|
|
|
* Auswahl — unter allen online Boxen, die das Modell GELADEN haben
|
|
(Karten-Belegung aus worker_hello/worker_ping), die mit den
|
|
wenigsten laufenden Anfragen; bei Gleichstand reihum.
|
|
Ist nur eine Karte da oder alles frei, landet alles dort —
|
|
"verteilen, wenn noetig, sonst nicht".
|
|
* Ausfall — verschwindet eine Box mitten in der Anfrage (kein Ping mehr)
|
|
oder antwortet nicht, geht die Anfrage EINMAL an eine andere
|
|
Box mit demselben Modell. Fehler des Modells selbst (z.B.
|
|
Kontext zu lang) werden nicht wiederholt.
|
|
* Fehler — kommen als FEHLER zurueck (ok=false / HTTP-Status), nie als
|
|
normale Antwort. Ein Client wuerde einen Fehlertext sonst
|
|
als Aussage der KI in seinen Verlauf schreiben — genau das
|
|
hat bei ARIA schon mal einen ganzen Chat vergiftet.
|
|
|
|
Eingang ist der RVS:
|
|
|
|
llm_cluster_models_request → llm_cluster_models
|
|
llm_cluster_request → llm_cluster_response
|
|
|
|
Programme brauchen nur eine ausgehende Verbindung zum RVS, egal in welchem
|
|
Netz sie stehen. Werkzeuge, die nur "OpenAI-kompatibel" koennen (OpenHands
|
|
& Co.), bekommen den OpenAI-RVS-Wrapper neben sich gestellt
|
|
(xtts/openai_rvs_wrapper) — fuer sie ist der Cluster dann 127.0.0.1.
|
|
|
|
Schnittstelle fuer Drittprogramme: docs/LLM-SCHNITTSTELLE.md
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import os
|
|
import time
|
|
import uuid
|
|
from typing import Optional
|
|
|
|
logger = logging.getLogger("aria-bridge.llm-cluster")
|
|
|
|
# Diese errorCodes heissen "die BOX war das Problem" → anderswo nochmal versuchen.
|
|
# Alles andere (Modell-Fehler, z.B. Kontext zu lang) waere ueberall gleich.
|
|
RETRY_CODES = {"timeout", "box_offline", "send_failed", "model_not_loaded"}
|
|
MAX_ATTEMPTS = 2
|
|
|
|
|
|
class LlmCluster:
|
|
def __init__(self, bridge):
|
|
self.b = bridge
|
|
self.inflight: dict[str, int] = {} # instanceId → laufende Anfragen (von uns)
|
|
self._rr: dict[str, int] = {}
|
|
self.timeout = float(os.getenv("CLUSTER_LLM_TIMEOUT_S", "600"))
|
|
|
|
# ── Flotte ──────────────────────────────────────────────────────────
|
|
|
|
def _online_llm(self) -> list:
|
|
now = time.time()
|
|
return [w for w in self.b._workers.values()
|
|
if w.get("service") == "llm"
|
|
and (now - w.get("last_seen", 0)) < self.b.WORKER_OFFLINE_S]
|
|
|
|
def _is_online(self, instance_id: str) -> bool:
|
|
return any(w.get("instanceId") == instance_id for w in self._online_llm())
|
|
|
|
def models(self) -> list:
|
|
"""Jedes geladene Modell GENAU EINMAL, egal auf wie vielen Karten.
|
|
ctx = das kleinste Kontextfenster unter seinen Karten (darauf muss sich
|
|
ein Client verlassen koennen, egal wohin der Router verteilt)."""
|
|
out: dict[str, dict] = {}
|
|
for w in self._online_llm():
|
|
slots = w.get("slots")
|
|
if isinstance(slots, list):
|
|
for sl in slots:
|
|
m = sl.get("model")
|
|
if not m or m not in (w.get("models") or []):
|
|
continue
|
|
e = out.setdefault(m, {"model": m, "ctx": 0, "cards": []})
|
|
e["cards"].append({"node": w.get("node", ""), "instanceId": w.get("instanceId", ""),
|
|
"gpu": sl.get("gpu", ""), "gpuName": sl.get("gpuName", ""),
|
|
"state": sl.get("state", "")})
|
|
c = int(sl.get("ctx") or 0)
|
|
if c and (not e["ctx"] or c < e["ctx"]):
|
|
e["ctx"] = c
|
|
else:
|
|
# Box mit altem Adapter (ohne Karten): nur die Modell-Liste
|
|
for m in (w.get("models") or []):
|
|
e = out.setdefault(m, {"model": m, "ctx": 0, "cards": []})
|
|
e["cards"].append({"node": w.get("node", ""), "instanceId": w.get("instanceId", ""),
|
|
"gpu": "", "gpuName": "", "state": ""})
|
|
for e in out.values():
|
|
e["available"] = len(e["cards"])
|
|
return sorted(out.values(), key=lambda e: e["model"])
|
|
|
|
def candidates(self, model: str) -> list:
|
|
return [w for w in self._online_llm() if model in (w.get("models") or [])]
|
|
|
|
def pick(self, model: str, exclude=()) -> Optional[str]:
|
|
"""Die Box mit den wenigsten laufenden Anfragen; Gleichstand reihum."""
|
|
cands = [w["instanceId"] for w in self.candidates(model) if w["instanceId"] not in exclude]
|
|
if not cands:
|
|
return None
|
|
low = min(self.inflight.get(c, 0) for c in cands)
|
|
tied = sorted(c for c in cands if self.inflight.get(c, 0) == low)
|
|
i = self._rr.get(model, 0) % len(tied)
|
|
self._rr[model] = i + 1
|
|
return tied[i]
|
|
|
|
# ── Anfrage an eine bestimmte Box ───────────────────────────────────
|
|
|
|
async def _send_to(self, target: str, req: dict, deadline: float) -> dict:
|
|
"""llm_request an GENAU diese Box, Antwort abwarten. Bricht frueh ab,
|
|
wenn die Box unterwegs offline geht — dann kann route() woanders hin."""
|
|
rid = str(uuid.uuid4())
|
|
fut: asyncio.Future = asyncio.get_running_loop().create_future()
|
|
self.b._pending_llm[rid] = fut
|
|
self.inflight[target] = self.inflight.get(target, 0) + 1
|
|
try:
|
|
payload = {k: v for k, v in req.items() if v is not None}
|
|
payload.update(requestId=rid, targetInstance=target)
|
|
ok = await self.b._send_to_rvs({"type": "llm_request", "payload": payload,
|
|
"timestamp": int(time.time() * 1000)})
|
|
if not ok:
|
|
return {"ok": False, "errorCode": "send_failed", "error": "RVS: Anfrage nicht gesendet"}
|
|
while True:
|
|
left = deadline - time.time()
|
|
if left <= 0:
|
|
return {"ok": False, "errorCode": "timeout",
|
|
"error": f"Keine Antwort von {target} in der Frist"}
|
|
try:
|
|
return await asyncio.wait_for(asyncio.shield(fut), timeout=min(5.0, left))
|
|
except asyncio.TimeoutError:
|
|
if not self._is_online(target):
|
|
return {"ok": False, "errorCode": "box_offline",
|
|
"error": f"{target} ist waehrend der Anfrage offline gegangen"}
|
|
finally:
|
|
self.inflight[target] = max(0, self.inflight.get(target, 1) - 1)
|
|
self.b._pending_llm.pop(rid, None)
|
|
|
|
async def route(self, req: dict, timeout: Optional[float] = None) -> dict:
|
|
"""Eine Anfrage an den Cluster. req: model, messages, max_tokens,
|
|
temperature, stop, tools. Antwort: wie llm_response der Box, plus
|
|
attempts; bei Fehler errorCode + error (+ available).
|
|
timeout: Gesamtfrist inkl. Ausfall-Ersatz. ARIA nutzt eine kurze (lokal
|
|
ist ihr Schnellweg — dauert es zu lange, nimmt sie Claude)."""
|
|
model = (req.get("model") or "").strip()
|
|
messages = req.get("messages")
|
|
if not isinstance(messages, list) or not messages:
|
|
return {"ok": False, "errorCode": "bad_request", "error": "messages fehlt oder ist leer"}
|
|
available = [m["model"] for m in self.models()]
|
|
if not model:
|
|
if len(available) == 1:
|
|
model = available[0] # eindeutig — dann darf es fehlen
|
|
else:
|
|
return {"ok": False, "errorCode": "bad_request", "available": available,
|
|
"error": "model fehlt. Geladen: " + (", ".join(available) or "nichts")}
|
|
body = {"model": model, "messages": messages,
|
|
"max_tokens": int(req.get("max_tokens") or 512),
|
|
"temperature": float(req["temperature"]) if req.get("temperature") is not None else 0.7,
|
|
"stop": req.get("stop") or None, "tools": req.get("tools") or None}
|
|
limit = timeout or self.timeout
|
|
deadline = time.time() + limit
|
|
tried: list = []
|
|
last: dict = {}
|
|
for _ in range(MAX_ATTEMPTS):
|
|
target = self.pick(model, exclude=tried)
|
|
if not target:
|
|
break
|
|
tried.append(target)
|
|
t0 = time.time()
|
|
last = await self._send_to(target, body, deadline)
|
|
if last.get("ok"):
|
|
last["attempts"] = len(tried)
|
|
return last
|
|
logger.warning("[cluster] %s auf %s: %s (%s)%s", model, target, last.get("errorCode") or "Fehler",
|
|
(last.get("error") or "")[:120], " — versuche andere Box" if
|
|
last.get("errorCode") in RETRY_CODES else "")
|
|
if last.get("errorCode") not in RETRY_CODES or time.time() >= deadline:
|
|
break
|
|
if not tried:
|
|
return {"ok": False, "errorCode": "model_unavailable", "available": available,
|
|
"error": (f"Modell {model!r} ist gerade auf keiner KI-Box im Cluster geladen. "
|
|
f"Geladen: {', '.join(available) or 'nichts'}.")}
|
|
last.setdefault("errorCode", "model_error")
|
|
last["attempts"] = len(tried)
|
|
if last["errorCode"] in RETRY_CODES and len(tried) >= 1:
|
|
others = [c for c in self.candidates(model) if c["instanceId"] not in tried]
|
|
if not others:
|
|
last["error"] = (last.get("error") or "") + (
|
|
" — keine weitere Box mit diesem Modell erreichbar.")
|
|
return last
|
|
|
|
# ── RVS-Eingang ─────────────────────────────────────────────────────
|
|
|
|
async def handle_rvs(self, mtype: str, payload: dict) -> bool:
|
|
if mtype == "llm_cluster_models_request":
|
|
await self.b._send_to_rvs({"type": "llm_cluster_models", "payload": {
|
|
"requestId": payload.get("requestId", ""), "models": self.models()},
|
|
"timestamp": int(time.time() * 1000)})
|
|
return True
|
|
if mtype == "llm_cluster_request":
|
|
asyncio.create_task(self._rvs_request(payload))
|
|
return True
|
|
return False
|
|
|
|
async def _rvs_request(self, payload: dict) -> None:
|
|
rid = payload.get("requestId", "")
|
|
res = await self.route(payload)
|
|
out = {"requestId": rid, "ok": bool(res.get("ok"))}
|
|
for k in ("content", "tool_calls", "model", "node", "gpu", "instanceId", "usage",
|
|
"elapsedMs", "attempts", "error", "errorCode", "available",
|
|
"promptTokens", "ctx"):
|
|
if res.get(k) is not None:
|
|
out[k] = res[k]
|
|
await self.b._send_to_rvs({"type": "llm_cluster_response", "payload": out,
|
|
"timestamp": int(time.time() * 1000)})
|