"""Model Manager for listing and dynamically changing LLM models at runtime. Supports OpenAI models, OpenCode Cloud models, Claude Code models, and local MLX models, persisting user model preferences to disk in model_settings.json. """ import json import logging import os import shutil import subprocess from pathlib import Path try: from loguru import logger except ImportError: logger = logging.getLogger("model_manager") DEFAULT_MODEL = "default" OPENAI_MODELS = { "openai/gpt-5.6-luna": "OpenAI GPT-5.6 Luna", "openai/gpt-5.6-luna-fast": "OpenAI GPT-5.6 Luna (Fast)", "openai/gpt-5.6-sol": "OpenAI GPT-5.6 Sol", "openai/gpt-5.6-terra": "OpenAI GPT-5.6 Terra", "openai/gpt-5.5": "OpenAI GPT-5.5", "openai/gpt-5.4": "OpenAI GPT-5.4", "openai/gpt-5.4-mini": "OpenAI GPT-4.4 Mini", "openai/gpt-4o": "OpenAI GPT-4o", "openai/gpt-4o-mini": "OpenAI GPT-4o Mini", } POPULAR_HERMES_MODELS = { "default": "Hermes Configured Default Model", "hermes-agent": "Hermes Agent", } CLAUDE_MODELS = { "claude-sonnet-4-6": "Claude 3.7 / Sonnet (Fast, High Capability)", "claude-opus-4-6": "Claude 3 Opus (Deep Reasoning)", "claude-haiku-4-6": "Claude 3.5 Haiku (Ultra Fast)", } MACOS_MODELS = { "mlx-community/Qwen2.5-7B-Instruct-4bit": "MLX Qwen 2.5 7B Instruct 4-bit (On-Device Apple Silicon)", } MODEL_ALIASES = { "luna": "openai/gpt-5.6-luna", "luna-fast": "openai/gpt-5.6-luna-fast", "sol": "openai/gpt-5.6-sol", "terra": "openai/gpt-5.6-terra", "gpt5.5": "openai/gpt-5.5", "gpt5.4": "openai/gpt-5.4", "gpt4o": "openai/gpt-4o", "gpt-4o": "openai/gpt-4o", "hermes": "hermes-3", "hermes3": "hermes-3", "hermes-agent": "hermes-agent", "deepseek": "ollama-cloud/deepseek-v4-flash", "sonnet": "claude-sonnet-4-6", "claude": "claude-sonnet-4-6", "opus": "claude-opus-4-6", "haiku": "claude-haiku-4-6", "qwen-local": "mlx-community/Qwen2.5-7B-Instruct-4bit", } def find_hermes_binary() -> str | None: candidates = [ shutil.which("hermes"), os.path.expanduser("~/.hermes/bin/hermes"), os.path.expanduser("~/.local/bin/hermes"), "/opt/homebrew/bin/hermes", "/usr/local/bin/hermes", ] for candidate in candidates: if candidate and os.path.exists(candidate) and os.access(candidate, os.X_OK): return candidate return shutil.which("hermes") class ModelManager: """Manages active LLM model configuration and dynamic model switching.""" def __init__(self, workspace_dir: Path | None = None, llm_processor=None): self._workspace_dir = Path(workspace_dir) if workspace_dir else Path(__file__).parent self._llm_processor = llm_processor self._config_file = self._workspace_dir / "model_settings.json" self._active_model = DEFAULT_MODEL self.load_saved_model() def set_llm_processor(self, llm_processor): self._llm_processor = llm_processor if self._active_model: self.apply_model(self._active_model) def load_saved_model(self) -> str: if self._config_file.exists(): try: data = json.loads(self._config_file.read_text()) if "model" in data and isinstance(data["model"], str) and data["model"].strip(): self._active_model = data["model"].strip() logger.info(f"Loaded saved model preference: {self._active_model}") except Exception as e: logger.warning(f"Could not load saved model settings: {e}") return self._active_model def sync_model(self) -> str: """Check if model_settings.json was updated on disk and update LLM processor live.""" current_disk_model = self.load_saved_model() if self._llm_processor and current_disk_model: if hasattr(self._llm_processor, "_model"): if getattr(self._llm_processor, "_model") != current_disk_model: setattr(self._llm_processor, "_model", current_disk_model) if hasattr(self._llm_processor, "_server_session_id"): self._llm_processor._server_session_id = None logger.info(f"Live LLM model synced to: {current_disk_model}") elif hasattr(self._llm_processor, "set_model"): self._llm_processor.set_model(current_disk_model) return current_disk_model def save_model(self, model_name: str): try: self._workspace_dir.mkdir(parents=True, exist_ok=True) self._config_file.write_text(json.dumps({"model": model_name}, indent=2)) except Exception as e: logger.warning(f"Could not save model setting: {e}") def fetch_all_models(self) -> list[str]: cli = find_hermes_binary() if cli: try: res = subprocess.run([cli, "models"], capture_output=True, text=True, timeout=5.0) if res.returncode == 0: models = [line.strip() for line in res.stdout.splitlines() if line.strip()] if models: return models except Exception as e: logger.debug(f"Could not query hermes models: {e}") return list(OPENAI_MODELS.keys()) + list(POPULAR_HERMES_MODELS.keys()) def get_models_dict(self) -> dict: """Return structured model categories for the Web UI.""" fetched = self.fetch_all_models() openai_list = [m for m in fetched if m.startswith("openai/")] hermes_list = [m for m in fetched if not m.startswith("openai/")] if not openai_list: openai_list = list(OPENAI_MODELS.keys()) if not hermes_list: hermes_list = list(POPULAR_HERMES_MODELS.keys()) return { "OpenAI Models": [{"id": m, "name": OPENAI_MODELS.get(m, m)} for m in openai_list], "Hermes Models": [{"id": m, "name": POPULAR_HERMES_MODELS.get(m, m)} for m in hermes_list], "Claude Code Models": [{"id": m, "name": name} for m, name in CLAUDE_MODELS.items()], "macOS On-Device MLX": [{"id": m, "name": name} for m, name in MACOS_MODELS.items()], } def list_available_models(self) -> str: """Fetch models and format as readable CLI catalog.""" models_dict = self.get_models_dict() lines = ["Available Models:\n", f"Active Model: {self._active_model}\n"] for category, items in models_dict.items(): lines.append(f"{category}:") for item in items: m_id = item["id"] name = item["name"] active = " (ACTIVE)" if m_id == self._active_model else "" lines.append(f" - {m_id}: {name}{active}") lines.append("") return "\n".join(lines) def apply_model(self, model_name: str) -> tuple[bool, str]: model_name = model_name.strip() matched_model = None clean_name = model_name.lower() if clean_name in MODEL_ALIASES: matched_model = MODEL_ALIASES[clean_name] all_known = { **OPENAI_MODELS, **POPULAR_HERMES_MODELS, **CLAUDE_MODELS, **MACOS_MODELS, } if not matched_model: for m in all_known: if clean_name == m.lower(): matched_model = m break if not matched_model: for m in all_known: if clean_name in m.lower(): matched_model = m break # Fall back to literal model string if explicit format if not matched_model and ("/" in model_name or ":" in model_name or "claude" in clean_name): matched_model = model_name if not matched_model: return False, f"Model '{model_name}' not found. Run list to view available models." self._active_model = matched_model self.save_model(matched_model) if self._llm_processor: try: if hasattr(self._llm_processor, "_model"): setattr(self._llm_processor, "_model", matched_model) if hasattr(self._llm_processor, "_server_session_id"): self._llm_processor._server_session_id = None logger.info(f"Dynamic model updated to: {matched_model}") return True, f"Model changed to {matched_model}." elif hasattr(self._llm_processor, "set_model"): self._llm_processor.set_model(matched_model) logger.info(f"Dynamic model updated to: {matched_model}") return True, f"Model changed to {matched_model}." except Exception as e: logger.error(f"Failed to apply model to LLM processor: {e}") return False, f"Could not change model: {e}" return True, f"Model set to {matched_model}."