Hacker-Software/ARIA-AGENT
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 236 Wiki Activity

feat(compute): Worker-Selbstanmeldung ueber RVS + Flotten-Anzeige (Stage 2)

Browse Source 
Jeder GPU-Dienst meldet sich beim Connect mit worker_hello {instanceId,
service, node, gpus, model} und haelt die Registry per periodischem
worker_ping {instanceId, busy} (~10s) frisch. So weiss ARIA, was wo laeuft.

- xtts/{voxtral,whisper,f5tts}/bridge.py + llm-adapter/adapter.py:
  INSTANCE_ID=service@NODE_NAME, _worker_register()-Coroutine (hello + ping),
  busy-Quelle je Worker (aktive STT-Sessions / TTS-Render / in-flight LLM);
  Task sauber gecancelt bei Reconnect.
- bridge/aria_bridge.py: self._workers-Registry + Handler worker_hello/
  worker_ping (spiegelt sat_hello), _worker_list() (35s-Offline-TTL),
  _pick_worker() (Round-Robin freie Instanz, fuer Stage-3-Routing),
  /internal/worker-list-Endpoint.
- diagnostic/server.js: workers-Map, worker_hello/worker_ping-Tracking,
  worker_update-Broadcast + worker_list-Action + on-connect-Snapshot.
- diagnostic/index.html: "Compute-Flotte"-Panel im Satelliten-Tab — pro Node
  gruppiert, mit Dienst/Modell/GPU und frei/beschaeftigt/offline-Status.

Stage 2 von 3. Reine Sichtbarkeit, kein Routing-Verhalten geaendert.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
duffyduck
2026-09-18 12:22:47 +02:00
co-authored by Claude Opus 4.8
parent e75f1eeb6a
commit f2ead1242f
7 changed files with 330 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
xtts/f5tts/bridge.py
+34
View File
@@ -60,6 +60,15 @@ RVS_TOKEN = os.getenv("RVS_TOKEN", "").strip()
# f5ttsCkptFile, f5ttsVocabFile, f5ttsCfgStrength, f5ttsNfeStep).
F5TTS_DEVICE = os.getenv("F5TTS_DEVICE", "cuda") # nur Bootstrap
# ── 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 = "f5tts"
INSTANCE_ID = f"{WORKER_SERVICE}@{NODE_NAME}"
WORKER_PING_INTERVAL_S = int(os.getenv("WORKER_PING_INTERVAL_S", "10"))
_tts_busy = False # True waehrend eine Synthese laeuft (busy-Report im ping)
DEFAULT_F5TTS_MODEL = "F5TTS_v1_Base"
DEFAULT_F5TTS_CKPT_FILE = "" # leer = Default-Checkpoint von HF
DEFAULT_F5TTS_VOCAB_FILE = "" # leer = Default-Vocab vom Modell
@@ -460,13 +469,16 @@ _tts_queue: asyncio.Queue[tuple] = asyncio.Queue()
async def _tts_worker(ws, runner: F5Runner) -> None:
"""Serialisiert Synthesen — GPU kann sonst OOM gehen."""
global _tts_busy
while True:
text, voice, request_id, message_id, language, speed = await _tts_queue.get()
_tts_busy = True
try:
await _do_tts(ws, runner, text, voice, request_id, message_id, language, speed)
except Exception:
logger.exception("TTS-Worker Fehler")
finally:
_tts_busy = False
_tts_queue.task_done()
@@ -808,6 +820,24 @@ async def _broadcast_status(ws, state: str, **extra) -> None:
await _send(ws, "service_status", payload)
async def _worker_register(ws, *, model: str = "", busy_fn=None) -> 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 _send(ws, "worker_hello", {
"instanceId": INSTANCE_ID, "service": WORKER_SERVICE,
"node": NODE_NAME, "gpus": GPU_IDS, "model": model,
})
while True:
await asyncio.sleep(WORKER_PING_INTERVAL_S)
busy = bool(busy_fn()) if busy_fn else False
await _send(ws, "worker_ping", {"instanceId": INSTANCE_ID, "busy": busy})
except asyncio.CancelledError:
raise
except Exception:
return # Socket tot → still beenden; run_loop reconnectet + startet neu
async def run_loop(runner: F5Runner) -> None:
use_tls = RVS_TLS
retry_s = 2
@@ -855,6 +885,9 @@ async def run_loop(runner: F5Runner) -> None:
# TTS-Worker fuer diese Verbindung starten
worker = asyncio.create_task(_tts_worker(ws, runner))
ping_task = asyncio.create_task(_worker_register(
ws, model=runner.model_id,
busy_fn=lambda: _tts_busy or not _tts_queue.empty()))
try:
async for raw in ws:
@@ -958,6 +991,7 @@ async def run_loop(runner: F5Runner) -> None:
_last_diag_voice = ""
finally:
worker.cancel()
ping_task.cancel()
try:
await worker
except asyncio.CancelledError:
xtts/llm-adapter/adapter.py
+41
View File
@@ -47,6 +47,15 @@ RVS_TOKEN = os.getenv("RVS_TOKEN", "").strip()
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"))
_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
@@ -124,7 +133,34 @@ async def _emit_llm_status(ws, state: str, model: str, **extra) -> None:
{"service": "llm", "state": state, "model": model, **extra})
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 _send(ws, "worker_hello", {
"instanceId": INSTANCE_ID, "service": WORKER_SERVICE,
"node": NODE_NAME, "gpus": GPU_IDS, "model": LLM_MODEL,
})
while True:
await asyncio.sleep(WORKER_PING_INTERVAL_S)
await _send(ws, "worker_ping",
{"instanceId": INSTANCE_ID, "busy": _inflight > 0})
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 []
@@ -201,6 +237,7 @@ async def _run() -> None:
logger.info("RVS verbunden — llm-adapter online")
retry_s = 2
tls_fallback_tried = False
ping_task = asyncio.create_task(_worker_register(ws))
async for raw in ws:
try:
msg = json.loads(raw)
@@ -214,6 +251,10 @@ async def _run() -> None:
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
xtts/voxtral/bridge.py
+35
View File
@@ -58,6 +58,16 @@ VOXTRAL_MODEL = os.getenv("VOXTRAL_MODEL", "mistralai/Voxtral-Mini-3B-2507")
VOXTRAL_LANGUAGE = os.getenv("VOXTRAL_LANGUAGE", "de")
VOXTRAL_DEVICE = os.getenv("VOXTRAL_DEVICE", "cuda")
# ── Compute-Fleet: Worker-Identitaet & Registrierung ──────────────
# Jeder Node meldet sich bei der aria-bridge (worker_hello) und haelt die
# Registry per periodischem worker_ping frisch. INSTANCE_ID adressiert diesen
# Worker bei Redundanz (targetInstance-Routing, Stage 3).
NODE_NAME = os.getenv("NODE_NAME", "node").strip() or "node"
GPU_IDS = os.getenv("NVIDIA_VISIBLE_DEVICES", "").strip()
WORKER_SERVICE = "voxtral"
INSTANCE_ID = f"{WORKER_SERVICE}@{NODE_NAME}"
WORKER_PING_INTERVAL_S = int(os.getenv("WORKER_PING_INTERVAL_S", "10"))
STREAM_TRANSCRIBE_INTERVAL_MS = int(os.getenv("STREAM_TRANSCRIBE_INTERVAL_MS", "1000"))
STREAM_DEFAULT_ENDPOINT_MS = 2400
STREAM_DEFAULT_HARD_CAP_MS = 300000
@@ -695,6 +705,24 @@ async def _broadcast_status(ws, state: str, **extra) -> None:
await _send(ws, "service_status", payload)
async def _worker_register(ws, *, model: str = "", busy_fn=None) -> 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 _send(ws, "worker_hello", {
"instanceId": INSTANCE_ID, "service": WORKER_SERVICE,
"node": NODE_NAME, "gpus": GPU_IDS, "model": model,
})
while True:
await asyncio.sleep(WORKER_PING_INTERVAL_S)
busy = bool(busy_fn()) if busy_fn else False
await _send(ws, "worker_ping", {"instanceId": INSTANCE_ID, "busy": busy})
except asyncio.CancelledError:
raise
except Exception:
return # Socket tot → still beenden; run_loop reconnectet + startet neu
async def run_loop(sessions: SessionManager) -> None:
use_tls = RVS_TLS
retry_s = 2
@@ -713,6 +741,9 @@ async def run_loop(sessions: SessionManager) -> None:
sessions.attach_ws(ws)
await _broadcast_status(ws, "ready", model=VOXTRAL_MODEL)
await _send(ws, "config_request", {"service": "voxtral"})
ping_task = asyncio.create_task(_worker_register(
ws, model=VOXTRAL_MODEL,
busy_fn=lambda: bool(sessions._sessions)))
async for raw in ws:
try:
msg = json.loads(raw)
@@ -797,6 +828,10 @@ async def run_loop(sessions: SessionManager) -> None:
"AN" if SPEAKER_ID_ENABLED else "AUS")
except Exception as e:
logger.warning("RVS-Verbindung verloren: %s — retry in %ds", e, retry_s)
try:
ping_task.cancel()
except NameError:
pass
if use_tls and RVS_TLS_FALLBACK and not tls_fallback_tried:
use_tls = False
tls_fallback_tried = True
xtts/whisper/bridge.py
+33
View File
@@ -59,6 +59,14 @@ WHISPER_DEVICE = os.getenv("WHISPER_DEVICE", "cuda")
WHISPER_COMPUTE_TYPE = os.getenv("WHISPER_COMPUTE_TYPE", "float16")
WHISPER_LANGUAGE = os.getenv("WHISPER_LANGUAGE", "de")
# ── 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 = "whisper"
INSTANCE_ID = f"{WORKER_SERVICE}@{NODE_NAME}"
WORKER_PING_INTERVAL_S = int(os.getenv("WORKER_PING_INTERVAL_S", "10"))
ALLOWED_MODELS = {"tiny", "base", "small", "medium", "large-v3"}
# Streaming-Parameter (Defaults — koennen pro Session vom App-Payload ueberschrieben werden)
@@ -822,6 +830,24 @@ async def _broadcast_status(ws, state: str, **extra) -> None:
await _send(ws, "service_status", payload)
async def _worker_register(ws, *, model: str = "", busy_fn=None) -> 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 _send(ws, "worker_hello", {
"instanceId": INSTANCE_ID, "service": WORKER_SERVICE,
"node": NODE_NAME, "gpus": GPU_IDS, "model": model,
})
while True:
await asyncio.sleep(WORKER_PING_INTERVAL_S)
busy = bool(busy_fn()) if busy_fn else False
await _send(ws, "worker_ping", {"instanceId": INSTANCE_ID, "busy": busy})
except asyncio.CancelledError:
raise
except Exception:
return # Socket tot → still beenden; run_loop reconnectet + startet neu
# ──────────────────────────────────────────────────────────────
# WS-LOOP
# ──────────────────────────────────────────────────────────────
@@ -862,6 +888,9 @@ async def run_loop(runner: WhisperRunner, sessions: SessionManager) -> None:
except Exception as e:
logger.exception("Initial-Handshake crashed: %s", e)
asyncio.create_task(_initial_handshake())
ping_task = asyncio.create_task(_worker_register(
ws, model=(runner.model_size or WHISPER_MODEL),
busy_fn=lambda: bool(sessions._sessions)))
async for raw in ws:
try:
@@ -1038,6 +1067,10 @@ async def run_loop(runner: WhisperRunner, sessions: SessionManager) -> None:
except Exception as e:
logger.warning("Verbindung verloren: %s", e)
sessions.detach_ws()
try:
ping_task.cancel()
except NameError:
pass
if use_tls and RVS_TLS_FALLBACK and not tls_fallback_tried:
logger.info("TLS-Verbindung fehlgeschlagen — Fallback auf ws://")
use_tls = False