#!/usr/bin/env -S uv run --script # /// script # requires-python = ">=3.11" # dependencies = [ # "textual>=0.75.0", # "rich>=13.0.0", # ] # /// from __future__ import annotations import os import signal import subprocess import urllib.request import json from dataclasses import dataclass from pathlib import Path from typing import Optional from rich.text import Text from textual.app import App, ComposeResult from textual.containers import Container, Horizontal, Vertical, VerticalScroll from textual.binding import Binding from textual.widgets import ( Button, Checkbox, Footer, Header, Input, Label, Log, RadioSet, RadioButton, Select, Static, TabbedContent, TabPane, ) SCRIPT_DIR = Path(__file__).resolve().parent.parent MODELS_DIR = SCRIPT_DIR / "models" TEMPLATES_DIR = SCRIPT_DIR / "templates" LOGS_DIR = SCRIPT_DIR / "logs" RUNNER_BIN = SCRIPT_DIR / "bin" / "llm-runner" TOTAL_USABLE_VRAM_GB = 30.0 @dataclass class ModelProfile: name: str path: Path size_gb: float directory: Path paired_mmproj: Optional[Path] paired_template: Optional[Path] preset_key: str PRESETS = { "cluster": { "name": "🌟 Multi-Agent Cluster (SuperQwen 27B + 2x Qwen 4B)", "desc": "Planner on 11434 (SuperQwen 27B + mmproj), Worker 1 on 11435, Worker 2 on 11436 (Qwen 4B + mmproj). Optimized for Pi & Hermes agents.", "port": 11434, "ctx": 65536, "temp": 0.6, "top_p": 0.95, "top_k": 40, "min_p": 0.05, "ngl": 99, "flash_attn": True, "kv_quant": "q8_0", "command": "cluster", "target_model_match": "SuperQwen", "is_cluster": True, }, "muse": { "name": "🔮 Muse-Glimmer 30B Solo (128K Context)", "desc": "Dedicated deep-reasoning agent with max 128K context on port 11434. Auto-pairs Muse-Glimmer Q8_0 vision projector.", "port": 11434, "ctx": 131072, "temp": 0.6, "top_p": 0.95, "top_k": 40, "min_p": 0.05, "ngl": 99, "flash_attn": True, "kv_quant": "q8_0", "command": "muse", "target_model_match": "Muse-Glimmer", "is_cluster": False, }, "superqwen": { "name": "🧠 SuperQwen 3.8 27B Solo (256K Context)", "desc": "Dedicated ultra-long reasoning agent with max 256K context on port 11434. Auto-pairs SuperQwen Q8_0 vision projector.", "port": 11434, "ctx": 262144, "temp": 0.6, "top_p": 0.95, "top_k": 40, "min_p": 0.05, "ngl": 99, "flash_attn": True, "kv_quant": "q8_0", "command": "superqwen", "target_model_match": "SuperQwen", "is_cluster": False, }, "qwen4b": { "name": "⚡ Qwen 3.5 4B Solo (Fast Instruct)", "desc": "Lightweight high-speed instruct model on port 11434 with ~3.9GB VRAM footprint. Auto-pairs BF16 vision projector.", "port": 11434, "ctx": 32768, "temp": 0.7, "top_p": 0.8, "top_k": 20, "min_p": 0.05, "ngl": 99, "flash_attn": True, "kv_quant": "q8_0", "command": "qwen4b", "target_model_match": "Qwen3.5-4B", "is_cluster": False, }, "custom": { "name": "🛠️ Custom Model & Parameters", "desc": "Specify exact GGUF weight files, context window, and sampling hyperparameters.", "port": 11434, "ctx": 65536, "temp": 0.6, "top_p": 0.95, "top_k": 40, "min_p": 0.05, "ngl": 99, "flash_attn": True, "kv_quant": "q8_0", "command": "custom", "target_model_match": None, "is_cluster": False, }, } def scan_models() -> tuple[list[ModelProfile], list[tuple[str, str]], list[tuple[str, str]]]: profiles: list[ModelProfile] = [] model_options: list[tuple[str, str]] = [] mmproj_options: list[tuple[str, str]] = [("None (Text Only / Vision Disabled)", "none")] if not MODELS_DIR.exists(): return profiles, [("No GGUF models found in models/", "")], mmproj_options # Gather all mmproj projector files first all_mmproj = sorted(MODELS_DIR.rglob("mmproj*.gguf")) for p in all_mmproj: try: sz = p.stat().st_size / 1e9 except Exception: sz = 0.0 mmproj_options.append((f"{p.name} ({sz:.1f} GB) [{p.parent.name}]", str(p))) # Gather base model GGUF files for p in sorted(MODELS_DIR.rglob("*.gguf")): if "mmproj" in p.name.lower(): continue try: sz = p.stat().st_size / 1e9 except Exception: sz = 0.0 # Find paired mmproj in same folder paired_mmproj = None for m in all_mmproj: if m.parent == p.parent: paired_mmproj = m break # Find paired chat template in directory tmpl = p.parent / "chat_template.jinja" paired_template = tmpl if tmpl.exists() else None # Determine preset mapping p_lower = (p.name + " " + p.parent.name).lower() if "muse" in p_lower: preset_key = "muse" tag = "Solo Reasoning (128K)" elif "superqwen" in p_lower or "qwen3.8" in p_lower: preset_key = "superqwen" tag = "Solo 256K / Planner" elif "qwen3.5" in p_lower or "4b" in p_lower: preset_key = "qwen4b" tag = "Fast Instruct / Worker" else: preset_key = "custom" tag = "Custom" prof = ModelProfile( name=p.name, path=p, size_gb=sz, directory=p.parent, paired_mmproj=paired_mmproj, paired_template=paired_template, preset_key=preset_key, ) profiles.append(prof) model_options.append((f"{p.name} ({sz:.1f} GB) — {tag}", str(p))) if not model_options: model_options.append(("No GGUF models found in models/", "")) return profiles, model_options, mmproj_options def check_port_status(port: int) -> tuple[bool, str]: url = f"http://127.0.0.1:{port}/health" try: req = urllib.request.Request(url, headers={"User-Agent": "LLM-TUI"}) with urllib.request.urlopen(req, timeout=0.8) as response: data = json.loads(response.read().decode()) status = data.get("status", "unknown") return True, f"Online ({status})" except urllib.error.HTTPError as e: if e.code == 503: try: data = json.loads(e.read().decode()) msg = data.get("error", {}).get("message", "Loading model") return True, f"Loading ({msg})" except Exception: return True, "Loading model..." return False, f"HTTP Error {e.code}" except Exception: return False, "Offline" class LLMConfigApp(App): CSS = """ Screen { background: #14171d; color: #d8dee9; } Header { background: #1e222b; color: #88c0d0; text-style: bold; } TabbedContent { height: 1fr; } TabPane { padding: 1 2; } .card { background: #1a1e27; border: solid #3b4252; padding: 1 2; margin-bottom: 1; height: auto; } .title { color: #88c0d0; text-style: bold; margin-bottom: 1; } .field-row { height: auto; margin-bottom: 1; } .field-label { width: 18; color: #81a1c1; padding-top: 1; } Input { background: #232834; border: solid #434c5e; color: #eceff4; width: 1fr; } Input:focus { border: solid #88c0d0; } Select { background: #232834; border: solid #434c5e; color: #eceff4; width: 1fr; } Button { margin-right: 1; background: #2e3440; color: #eceff4; border: solid #4c566a; } Button:hover { background: #434c5e; } Button.-primary { background: #5e81ac; color: #eceff4; text-style: bold; border: solid #81a1c1; } Button.-error { background: #bf616a; color: #eceff4; text-style: bold; border: solid #d08770; } #binding-indicator { background: #1e2531; border-left: thick #88c0d0; padding: 1 2; margin-bottom: 1; height: auto; } #vram-meter { background: #232834; border: solid #4c566a; padding: 1; margin-top: 1; height: auto; } #status-card { background: #1e222b; border: solid #4c566a; padding: 1; margin-bottom: 1; height: auto; } Log { background: #0f1117; border: solid #3b4252; height: 1fr; } """ BINDINGS = [ Binding("f5", "launch", "Launch Server"), Binding("f8", "stop", "Stop Selected"), Binding("f9", "stop_all", "Stop All"), Binding("f2", "refresh_status", "Refresh Status"), Binding("ctrl+c", "quit", "Quit"), Binding("q", "quit", "Quit"), ] selected_preset: str = "cluster" profiles: list[ModelProfile] = [] profiles_by_path: dict[str, ModelProfile] = {} models_options: list[tuple[str, str]] = [] mmproj_options: list[tuple[str, str]] = [] _sync_lock: bool = False _last_selected_model: str = "" _last_selected_mmproj: str = "" _last_log_size: int = -1 _last_log_name: str = "" def compose(self) -> ComposeResult: yield Header(show_clock=True) self.profiles, self.models_options, self.mmproj_options = scan_models() self.profiles_by_path = {str(p.path): p for p in self.profiles} # Choose default model and mmproj based on initial cluster preset default_model = "" default_mmproj = "none" for p in self.profiles: if "superqwen" in p.name.lower() or "qwen3.8" in p.name.lower(): default_model = str(p.path) if p.paired_mmproj: default_mmproj = str(p.paired_mmproj) break if not default_model and self.profiles: default_model = str(self.profiles[0].path) if self.profiles[0].paired_mmproj: default_mmproj = str(self.profiles[0].paired_mmproj) self._last_selected_model = default_model self._last_selected_mmproj = default_mmproj with TabbedContent(): with TabPane("🚀 Model Launcher & Form", id="tab-launch"): with VerticalScroll(): # 1. Preset Selector with Container(classes="card"): yield Label("Select Inference Preset", classes="title") with RadioSet(id="preset-radios"): for key, preset in PRESETS.items(): yield RadioButton( preset["name"], id=f"preset-{key}", value=(key == "cluster"), ) yield Label( PRESETS["cluster"]["desc"], id="preset-description", classes="text-muted", ) # 2. Active Tied Binding Indicator Card yield Static(id="binding-indicator") # 3. Parameters Form (GGUF, MMProj, and Tuning) with Container(classes="card", id="form-container"): yield Label("Hyperparameters & Hardware Tuning", classes="title") with Horizontal(classes="field-row"): yield Label("GGUF Model:", classes="field-label") yield Select( options=[(lbl, val) for lbl, val in self.models_options], value=default_model if default_model else None, id="select-model", ) with Horizontal(classes="field-row"): yield Label("Vision MMProj:", classes="field-label") yield Select( options=[(lbl, val) for lbl, val in self.mmproj_options], value=default_mmproj, id="select-mmproj", ) with Horizontal(classes="field-row"): yield Label("Context Size:", classes="field-label") yield Input(value="65536", id="input-ctx") with Horizontal(classes="field-row"): yield Label("Port:", classes="field-label") yield Input(value="11434", id="input-port") with Horizontal(classes="field-row"): yield Label("Temperature:", classes="field-label") yield Input(value="0.6", id="input-temp") with Horizontal(classes="field-row"): yield Label("Top-P:", classes="field-label") yield Input(value="0.95", id="input-top-p") with Horizontal(classes="field-row"): yield Label("Min-P:", classes="field-label") yield Input(value="0.05", id="input-min-p") with Horizontal(classes="field-row"): yield Label("GPU Layers (-ngl):", classes="field-label") yield Input(value="99", id="input-ngl") with Horizontal(classes="field-row"): yield Label("KV Cache Quant:", classes="field-label") yield Select( options=[("q8_0 (High Quality, 50% savings)", "q8_0"), ("f16 (Full Precision)", "f16"), ("q4_0 (Max Memory Savings)", "q4_0")], value="q8_0", id="select-kv", ) with Horizontal(classes="field-row"): yield Label("Flash Attention:", classes="field-label") yield Checkbox("Enabled (-fa on)", value=True, id="check-fa") # VRAM Estimation Banner yield Static(id="vram-meter") # 4. Execution Action Bar with Horizontal(classes="card"): yield Button("🚀 Launch Server (F5)", variant="primary", id="btn-launch") yield Button("🛑 Stop Server (F8)", variant="error", id="btn-stop") yield Button("⚠️ Stop All Instances (F9)", variant="error", id="btn-stop-all") yield Button("🔄 Refresh Health (F2)", id="btn-refresh") with TabPane("📊 Server Status & Live Logs", id="tab-status"): with Vertical(): yield Static(id="status-card") yield Label("Live Logs Streaming:", classes="title") yield Log(id="log-stream", highlight=True) yield Footer() def on_mount(self) -> None: self.apply_preset(self.selected_preset) self.action_refresh_status() self.set_interval(3.0, self.action_refresh_status) self.set_interval(1.5, self.tail_logs) def _set_active_radio(self, preset_key: str) -> None: rs = self.query_one("#preset-radios", RadioSet) target = self.query_one(f"#preset-{preset_key}", RadioButton) with rs.prevent(RadioButton.Changed): if rs._pressed_button and rs._pressed_button != target: rs._pressed_button.value = False target.value = True rs._pressed_button = target def apply_preset(self, preset_key: str) -> None: if preset_key not in PRESETS: return self._sync_lock = True try: self.selected_preset = preset_key preset = PRESETS[preset_key] self.query_one("#preset-description", Label).update(preset["desc"]) self.query_one("#input-port", Input).value = str(preset["port"]) self.query_one("#input-ctx", Input).value = str(preset["ctx"]) self.query_one("#input-temp", Input).value = str(preset["temp"]) self.query_one("#input-top-p", Input).value = str(preset["top_p"]) self.query_one("#input-min-p", Input).value = str(preset["min_p"]) self.query_one("#input-ngl", Input).value = str(preset["ngl"]) self.query_one("#check-fa", Checkbox).value = preset["flash_attn"] self.query_one("#select-kv", Select).value = preset["kv_quant"] # Automatically tie GGUF model and MMProj to preset prof = self._find_profile_for_preset(preset_key) if prof: self._last_selected_model = str(prof.path) self.query_one("#select-model", Select).value = str(prof.path) target_mmproj = str(prof.paired_mmproj) if prof.paired_mmproj else "none" self._last_selected_mmproj = target_mmproj self.query_one("#select-mmproj", Select).value = target_mmproj # Sync active RadioButton in RadioSet self._set_active_radio(preset_key) self.update_binding_indicator() self.update_vram_estimation() finally: self._sync_lock = False def _find_profile_for_preset(self, preset_key: str) -> Optional[ModelProfile]: preset = PRESETS.get(preset_key) if not preset or not preset.get("target_model_match"): return None target = preset["target_model_match"].lower() for prof in self.profiles: if target in prof.name.lower() or target in prof.directory.name.lower(): return prof return None def on_radio_set_changed(self, event: RadioSet.Changed) -> None: if self._sync_lock: return radio_id = event.pressed.id or "" preset_key = radio_id.replace("preset-", "") self.apply_preset(preset_key) def on_select_changed(self, event: Select.Changed) -> None: if self._sync_lock: return if event.select.id == "select-model": model_path_str = str(event.value or "") if model_path_str == self._last_selected_model: return self._last_selected_model = model_path_str self._sync_lock = True try: prof = self.profiles_by_path.get(model_path_str) if prof: # 1. Automatically tie matching MMProj if prof.paired_mmproj: self._last_selected_mmproj = str(prof.paired_mmproj) self.query_one("#select-mmproj", Select).value = str(prof.paired_mmproj) else: self._last_selected_mmproj = "none" self.query_one("#select-mmproj", Select).value = "none" # 2. Automatically sync inference preset and sampling defaults p_key = prof.preset_key if self.selected_preset == "cluster" and p_key in ("superqwen", "qwen4b"): # SuperQwen and Qwen 4B are the Planner and Workers of Cluster mode pass elif p_key != "custom" and p_key in PRESETS: self.selected_preset = p_key self._set_active_radio(p_key) preset = PRESETS[p_key] self.query_one("#preset-description", Label).update(preset["desc"]) self.query_one("#input-port", Input).value = str(preset["port"]) self.query_one("#input-ctx", Input).value = str(preset["ctx"]) self.query_one("#input-temp", Input).value = str(preset["temp"]) self.query_one("#input-top-p", Input).value = str(preset["top_p"]) self.query_one("#input-min-p", Input).value = str(preset["min_p"]) self.query_one("#input-ngl", Input).value = str(preset["ngl"]) self.query_one("#check-fa", Checkbox).value = preset["flash_attn"] self.query_one("#select-kv", Select).value = preset["kv_quant"] else: self.selected_preset = "custom" self._set_active_radio("custom") self.query_one("#preset-description", Label).update(PRESETS["custom"]["desc"]) self.update_binding_indicator() self.update_vram_estimation() finally: self._sync_lock = False elif event.select.id == "select-mmproj": mmproj_str = str(event.value or "") if mmproj_str == self._last_selected_mmproj: return self._last_selected_mmproj = mmproj_str self.update_binding_indicator() self.update_vram_estimation() elif event.select.id == "select-kv": self.update_binding_indicator() self.update_vram_estimation() def on_input_changed(self, event: Input.Changed) -> None: self.update_vram_estimation() def update_binding_indicator(self) -> None: preset_key = self.selected_preset preset = PRESETS.get(preset_key, PRESETS["custom"]) model_val = str(self.query_one("#select-model", Select).value or "") mmproj_val = str(self.query_one("#select-mmproj", Select).value or "") model_name = Path(model_val).name if (model_val and Path(model_val).exists()) else "None" if mmproj_val and mmproj_val != "none" and Path(mmproj_val).exists(): mmproj_text = f"[green]✔ {Path(mmproj_val).name} (Vision Enabled)[/]" else: mmproj_text = "[yellow]○ None (Text Only / Vision Disabled)[/]" if preset.get("is_cluster"): text = ( f"[bold cyan]🔗 Tied Configuration:[/] [bold white]{preset['name']}[/]\n" f" • [bold]Planner (Port 11434):[/] SuperQwen 3.8 27B + mmproj-SuperQwen (64K Ctx)\n" f" • [bold]Workers (Ports 11435/11436):[/] 2x Qwen 3.5 4B + mmproj-BF16 (16K Ctx)\n" f" • [bold green]● GGUF weights, projectors, and inference presets are coordinated across all 3 nodes[/]" ) else: text = ( f"[bold cyan]🔗 Tied Configuration:[/] [bold white]{preset['name']}[/]\n" f" • [bold]GGUF Model:[/] [bold magenta]{model_name}[/]\n" f" • [bold]Vision Projector:[/] {mmproj_text}\n" f" • [bold]Chat Template:[/] Jinja template auto-paired from model directory" ) self.query_one("#binding-indicator", Static).update(text) def update_vram_estimation(self) -> None: try: ctx = int(self.query_one("#input-ctx", Input).value or "0") except ValueError: ctx = 65536 preset = self.selected_preset kv_choice = str(self.query_one("#select-kv", Select).value or "q8_0") kv_mult = 1.0 if kv_choice == "q8_0" else (2.0 if kv_choice == "f16" else 0.5) if preset == "cluster": weights_gb = 25.9 kv_gb = 3.2 * kv_mult else: model_val = str(self.query_one("#select-model", Select).value or "") if model_val and Path(model_val).exists(): model_gb = Path(model_val).stat().st_size / 1e9 else: model_gb = 16.0 mmproj_val = str(self.query_one("#select-mmproj", Select).value or "") if mmproj_val and mmproj_val != "none" and Path(mmproj_val).exists(): mmproj_gb = Path(mmproj_val).stat().st_size / 1e9 else: mmproj_gb = 0.0 weights_gb = model_gb + mmproj_gb if weights_gb > 10.0: kv_gb = (ctx / 131072.0) * 8.5 * kv_mult else: kv_gb = (ctx / 32768.0) * 1.5 * kv_mult total_est = weights_gb + kv_gb remaining = TOTAL_USABLE_VRAM_GB - total_est if total_est <= TOTAL_USABLE_VRAM_GB: status_color = "green" verdict = "SAFE (Fits in VRAM)" else: status_color = "red" verdict = "OVERFLOW WARNING (May spill to system RAM / ROCm crash)" meter_text = ( f"VRAM Budget (~{TOTAL_USABLE_VRAM_GB:.0f}GB Usable): " f"Weights ~{weights_gb:.1f}GB + KV Cache ~{kv_gb:.1f}GB = " f"[{status_color}]{total_est:.1f}GB total[/] | Remaining Headroom: ~{remaining:.1f}GB [{verdict}]" ) self.query_one("#vram-meter", Static).update(meter_text) def action_refresh_status(self) -> None: lines = [] for port, label in [(11434, "Planner / Solo"), (11435, "Worker 1"), (11436, "Worker 2")]: online, msg = check_port_status(port) if not online: badge = "[yellow]○ OFFLINE[/]" elif "loading" in msg.lower(): badge = "[cyan]⏳ LOADING[/]" else: badge = "[green]● ONLINE[/]" lines.append(f"Port {port} ({label}): {badge} - {msg}") card = self.query_one("#status-card", Static) card.update("\n".join(lines)) def tail_logs(self) -> None: log_widget = self.query_one("#log-stream", Log) logs = sorted(LOGS_DIR.glob("*.log"), key=os.path.getmtime, reverse=True) if logs: latest = logs[0] try: cur_size = latest.stat().st_size if cur_size == self._last_log_size and latest.name == self._last_log_name: return self._last_log_size = cur_size self._last_log_name = latest.name with open(latest, "r", errors="ignore") as f: f.seek(max(0, cur_size - 16384)) tail = f.read() log_widget.clear() log_widget.write(f"--- Streaming {latest.name} ---\n" + tail) except Exception: pass def action_launch(self) -> None: preset = self.selected_preset log_widget = self.query_one("#log-stream", Log) model_val = str(self.query_one("#select-model", Select).value or "") mmproj_val = str(self.query_one("#select-mmproj", Select).value or "") env = os.environ.copy() env["CTX"] = self.query_one("#input-ctx", Input).value or "65536" env["PORT"] = self.query_one("#input-port", Input).value or "11434" env["TEMP"] = self.query_one("#input-temp", Input).value or "0.6" env["TOP_P"] = self.query_one("#input-top-p", Input).value or "0.95" env["MIN_P"] = self.query_one("#input-min-p", Input).value or "0.05" env["NGL"] = self.query_one("#input-ngl", Input).value or "99" env["FLASH_ATTN"] = "on" if self.query_one("#check-fa", Checkbox).value else "off" env["CTK"] = str(self.query_one("#select-kv", Select).value or "q8_0") env["CTV"] = env["CTK"] if model_val and Path(model_val).exists(): env["MODEL"] = model_val tmpl = Path(model_val).parent / "chat_template.jinja" if tmpl.exists(): env["TEMPLATE"] = str(tmpl) if mmproj_val and mmproj_val != "none" and Path(mmproj_val).exists(): env["MMPROJ"] = mmproj_val else: env["MMPROJ"] = "none" runner_cmd = PRESETS.get(preset, {}).get("command", preset) cmd = [str(RUNNER_BIN), runner_cmd] log_widget.write( f"\n[bold cyan]>>> Launching preset '{preset}' (runner command: {runner_cmd})...[/]\n" f" GGUF Model: {Path(model_val).name if model_val else 'Default'}\n" f" Vision MMProj: {Path(mmproj_val).name if (mmproj_val and mmproj_val != 'none') else 'None (Text Only)'}\n" f" Port: {env['PORT']} | Context: {env['CTX']} | KV Quant: {env['CTK']}\n" ) subprocess.Popen(cmd, env=env, cwd=str(SCRIPT_DIR), start_new_session=True) self.notify(f"Launching {preset} server. Switching to Status & Logs...", severity="information") self.query_one(TabbedContent).active = "tab-status" self.action_refresh_status() def action_stop(self) -> None: port = self.query_one("#input-port", Input).value or "11434" subprocess.run([str(RUNNER_BIN), "stop", port], cwd=str(SCRIPT_DIR)) self.notify(f"Stopped server on port {port}", severity="warning") self.action_refresh_status() def action_stop_all(self) -> None: subprocess.run([str(RUNNER_BIN), "stop", "all"], cwd=str(SCRIPT_DIR)) self.notify("Stopped all running LLM servers", severity="warning") self.action_refresh_status() def on_button_pressed(self, event: Button.Pressed) -> None: if event.button.id == "btn-launch": self.action_launch() elif event.button.id == "btn-stop": self.action_stop() elif event.button.id == "btn-stop-all": self.action_stop_all() elif event.button.id == "btn-refresh": self.action_refresh_status() if __name__ == "__main__": app = LLMConfigApp() app.run()