#!/usr/bin/env python3 """ Hermes Local Model Switcher v4 ------------------------------- Lists available models and hot-swaps them (stop old, start new, update config). Usage: python3 ~/.hermes/scripts/model-switcher (interactive menu) python3 ~/.hermes/scripts/model-switcher 35b (direct switch) python3 ~/.hermes/scripts/model-switcher 9b (direct switch) v4: Clean rewrite — fixed corrupted imports and duplicate function defs. """ import os import sys import time import subprocess import tempfile import urllib.request import yaml # Import model definitions from shared source of truth sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from models import MODELS HERMES_CONFIG = os.path.expanduser("~/.hermes/config.yaml") def write_config_atomic(provider: str, model_name: str, base_url: str): """Atomically update model.provider, model.default, and model.base_url.""" with open(HERMES_CONFIG, encoding="utf-8") as f: cfg = yaml.safe_load(f) or {} if not isinstance(cfg, dict): cfg = {} model_cfg = cfg.setdefault("model", {}) if not isinstance(model_cfg, dict): model_cfg = {} cfg["model"] = model_cfg model_cfg["provider"] = provider model_cfg["default"] = model_name if base_url: model_cfg["base_url"] = base_url directory = os.path.dirname(HERMES_CONFIG) or "." fd, tmp_path = tempfile.mkstemp(prefix=".config.yaml.", suffix=".tmp", dir=directory, text=True) try: with os.fdopen(fd, "w", encoding="utf-8") as f: yaml.safe_dump(cfg, f, default_flow_style=False, sort_keys=False, allow_unicode=True) f.flush() os.fsync(f.fileno()) os.replace(tmp_path, HERMES_CONFIG) finally: if os.path.exists(tmp_path): os.unlink(tmp_path) def get_active_service(): """Find which model service is currently active, skipping preserved ones.""" for model in MODELS: if model.get("preserve"): continue r = subprocess.run( ["systemctl", "is-active", model["service"]], capture_output=True, text=True, ) if r.stdout.strip() == "active": return model return None def swap(target): """Stop active model, start target, update config. Writes all config changes atomically FIRST, then swaps services. When switching to a cloud model, keeps local models warm in VRAM. VRAM drain delay only applies to local-to-local switching. """ active = get_active_service() base_url = f"http://127.0.0.1:{target['port']}/v1" write_config_atomic(target["provider"], target["model_name"], base_url) print(f" ⚙ Config → {target['provider']} / {target['model_name']}", flush=True) # Only stop active service if switching to another local model # and the active model is not a preserved/background service if active and active["provider"] == "local" and target["provider"] == "local": if active.get("preserve"): print(f" 🛡 Preserving {active['name']} (background service)", flush=True) else: print(f" ⏚ Stopping {active['name']}", end="", flush=True) r = subprocess.run( ["sudo", "systemctl", "stop", active["service"]], capture_output=True, text=True, ) print(f" {'✅' if r.returncode == 0 else '❌'}", flush=True) # Give GPU VRAM time to drain between local models print(" 💤 Waiting 3s for VRAM free...", flush=True) time.sleep(3) elif not active: print(" ℹī¸ No active service to stop", flush=True) # Only start target if it's a local model if target["provider"] == "local": print(f" â–ļī¸ Starting {target['name']}", end="", flush=True) r = subprocess.run( ["sudo", "systemctl", "start", target["service"]], capture_output=True, text=True, ) print(f" {'✅' if r.returncode == 0 else '❌'}", flush=True) # Wait for model to be ready print(f" âŗ Waiting for {target['name']}...", end="", flush=True) for _ in range(30): try: req = urllib.request.Request(f"http://127.0.0.1:{target['port']}/v1/models") resp = urllib.request.urlopen(req, timeout=2) if resp.status == 200: print(" ✅", flush=True) break except Exception: pass print(".", end="", flush=True) time.sleep(2) else: print(" ❌ (timeout)", flush=True) return False print(f"\n ✅ Now on {target['name']} - {target['context']} token context") print(f" Desc: {target['desc']}") print(f" Run /reset in Hermes to start fresh.", flush=True) return True def show_all_models(): """Print formatted model list.""" print() print(f" {'#':<3} {'Model':22s} {'Status':12s} {'Port':6s} {'Context':>8s} Description") print(f" {'─'*3} {'─'*22} {'─'*12} {'─'*6} {'─'*8} {'─'*58}") for i, model in enumerate(MODELS): r = subprocess.run( ["systemctl", "is-active", model["service"]], capture_output=True, text=True, ) active_status = r.stdout.strip() status = "đŸŸĸ ACTIVE" if active_status == "active" else "âšĒ" ctx = f"{model['context']:,}" print(f" [{i+1}] {model['name']:22s} {status:12s} {str(model['port']):6s} {ctx:>8s} {model['desc']}") print() def main(): # Non-interactive: direct arg if len(sys.argv) > 1: arg = sys.argv[1].lower().replace("-", "").replace(" ", "") for model in MODELS: if arg == model["short"].lower() or arg in model["name"].lower().replace(".", "").replace("-", ""): return 0 if swap(model) else 1 print(f"Unknown model. Try: {', '.join(m['short'] for m in MODELS)}") print(f"Example: {sys.argv[0]} 9b") return 1 # Interactive menu while True: os.system('clear' if os.name == 'posix' else 'cls') active = get_active_service() print("╔══════════════════════════════════════════════════════╗") print("║ 🤖 Hermes Local Model Switcher v4 ║") print("╚══════════════════════════════════════════════════════╝") if active: print(f"\n đŸŸĸ Currently running: {active['name']}") print(f" {active['detail']}") else: print(f"\n âšĒ No local model currently running") show_all_models() print(" Quick shortcuts:") print(f" {' | '.join(m['short'] for m in MODELS[:8])}") print() print(" [q] Quit") print() choice = input(" Select model (# or name): ").strip().lower() if choice in ('q', 'quit', 'exit'): print("\n 👋 Bye!") break # Try direct shortcut first matched = [m for m in MODELS if choice == m["short"].lower()] if not matched: try: idx = int(choice) - 1 if 0 <= idx < len(MODELS): matched = [MODELS[idx]] except ValueError: for m in MODELS: if choice in m["name"].lower().replace("-", "").replace(".", ""): matched = [m] break if matched: swap(matched[0]) input("\n Press Enter to continue...") else: print(f"\n ❌ Invalid choice. Try a number (1-{len(MODELS)}) or shortcut name.") input(" Press Enter...") if __name__ == "__main__": try: main() except KeyboardInterrupt: print("\n\n 👋 Bye!")