feat: /model command — full provider+model system overhaul

New foundation files:
- hermes_cli/providers.py: single source of truth for provider identity,
  aliases, labels, transport types, api_mode determination
- hermes_cli/model_normalize.py: per-provider model name normalization
  (anthropic uses hyphens, openrouter uses vendor/ prefix, etc.)
- agent/models_dev.py: extended with ModelCapabilities, get_model_capabilities(),
  list_provider_models(), search_models_dev()

Rebuilt model_switch.py:
- Dynamic alias resolution from catalog (no hardcoded versions)
- Aggregator-aware resolution (stays on OpenRouter, doesn't hijack to opencode-zen)
- Vendor:model conversion on aggregators (openai:gpt-5.4 -> openai/gpt-5.4)
- Per-provider model name normalization
- Capability metadata from models.dev
- Fuzzy suggestions on error

AIAgent.switch_model(): in-place model swap following _try_activate_fallback()
pattern. Updates primary runtime, invalidates system prompt, rebuilds client
for cross-api-mode switches. Uses determine_api_mode() from providers.py.

/model command:
- Session-only by default (no config.yaml write)
- --global flag to persist permanently
- Confirmation shows model, provider, context, capabilities, cache status
- Running-agent guard on gateway
- Gateway stores session overrides in _session_model_overrides dict
- Works across CLI, Telegram, Discord, Slack, Matrix, all platforms
This commit is contained in:
Teknium
2026-04-04 19:56:42 -07:00
parent cb63b5f381
commit afcbab5323
8 changed files with 1743 additions and 47 deletions
+206 -1
View File
@@ -8,12 +8,14 @@ Data is cached in memory (1hr TTL) and on disk (~/.hermes/models_dev_cache.json)
to avoid cold-start network latency.
"""
import difflib
import json
import logging
import os
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, Optional
from typing import Any, Dict, List, Optional
from utils import atomic_json_write
@@ -170,3 +172,206 @@ def _extract_context(entry: Dict[str, Any]) -> Optional[int]:
if isinstance(ctx, (int, float)) and ctx > 0:
return int(ctx)
return None
# ---------------------------------------------------------------------------
# Model capability metadata
# ---------------------------------------------------------------------------
@dataclass
class ModelCapabilities:
"""Structured capability metadata for a model from models.dev."""
supports_tools: bool = True
supports_vision: bool = False
supports_reasoning: bool = False
context_window: int = 200000
max_output_tokens: int = 8192
model_family: str = ""
def _get_provider_models(provider: str) -> Optional[Dict[str, Any]]:
"""Resolve a Hermes provider ID to its models dict from models.dev.
Returns the models dict or None if the provider is unknown or has no data.
"""
mdev_provider_id = PROVIDER_TO_MODELS_DEV.get(provider)
if not mdev_provider_id:
return None
data = fetch_models_dev()
provider_data = data.get(mdev_provider_id)
if not isinstance(provider_data, dict):
return None
models = provider_data.get("models", {})
if not isinstance(models, dict):
return None
return models
def _find_model_entry(models: Dict[str, Any], model: str) -> Optional[Dict[str, Any]]:
"""Find a model entry by exact match, then case-insensitive fallback."""
# Exact match
entry = models.get(model)
if isinstance(entry, dict):
return entry
# Case-insensitive match
model_lower = model.lower()
for mid, mdata in models.items():
if mid.lower() == model_lower and isinstance(mdata, dict):
return mdata
return None
def get_model_capabilities(provider: str, model: str) -> Optional[ModelCapabilities]:
"""Look up full capability metadata from models.dev cache.
Uses the existing fetch_models_dev() and PROVIDER_TO_MODELS_DEV mapping.
Returns None if model not found.
Extracts from model entry fields:
- reasoning (bool) → supports_reasoning
- tool_call (bool) → supports_tools
- attachment (bool) → supports_vision
- limit.context (int) → context_window
- limit.output (int) → max_output_tokens
- family (str) → model_family
"""
models = _get_provider_models(provider)
if models is None:
return None
entry = _find_model_entry(models, model)
if entry is None:
return None
# Extract capability flags (default to False if missing)
supports_tools = bool(entry.get("tool_call", False))
supports_vision = bool(entry.get("attachment", False))
supports_reasoning = bool(entry.get("reasoning", False))
# Extract limits
limit = entry.get("limit", {})
if not isinstance(limit, dict):
limit = {}
ctx = limit.get("context")
context_window = int(ctx) if isinstance(ctx, (int, float)) and ctx > 0 else 200000
out = limit.get("output")
max_output_tokens = int(out) if isinstance(out, (int, float)) and out > 0 else 8192
model_family = entry.get("family", "") or ""
return ModelCapabilities(
supports_tools=supports_tools,
supports_vision=supports_vision,
supports_reasoning=supports_reasoning,
context_window=context_window,
max_output_tokens=max_output_tokens,
model_family=model_family,
)
def list_provider_models(provider: str) -> List[str]:
"""Return all model IDs for a provider from models.dev.
Returns an empty list if the provider is unknown or has no data.
"""
models = _get_provider_models(provider)
if models is None:
return []
return list(models.keys())
def search_models_dev(
query: str, provider: str = None, limit: int = 5
) -> List[Dict[str, Any]]:
"""Fuzzy search across models.dev catalog. Returns matching model entries.
Args:
query: Search string to match against model IDs.
provider: Optional Hermes provider ID to restrict search scope.
If None, searches across all providers in PROVIDER_TO_MODELS_DEV.
limit: Maximum number of results to return.
Returns:
List of dicts, each containing 'provider', 'model_id', and the full
model 'entry' from models.dev.
"""
data = fetch_models_dev()
if not data:
return []
# Build list of (provider_id, model_id, entry) candidates
candidates: List[tuple] = []
if provider is not None:
# Search only the specified provider
mdev_provider_id = PROVIDER_TO_MODELS_DEV.get(provider)
if not mdev_provider_id:
return []
provider_data = data.get(mdev_provider_id, {})
if isinstance(provider_data, dict):
models = provider_data.get("models", {})
if isinstance(models, dict):
for mid, mdata in models.items():
candidates.append((provider, mid, mdata))
else:
# Search across all mapped providers
for hermes_prov, mdev_prov in PROVIDER_TO_MODELS_DEV.items():
provider_data = data.get(mdev_prov, {})
if isinstance(provider_data, dict):
models = provider_data.get("models", {})
if isinstance(models, dict):
for mid, mdata in models.items():
candidates.append((hermes_prov, mid, mdata))
if not candidates:
return []
# Use difflib for fuzzy matching — case-insensitive comparison
model_ids_lower = [c[1].lower() for c in candidates]
query_lower = query.lower()
# First try exact substring matches (more intuitive than pure edit-distance)
substring_matches = []
for prov, mid, mdata in candidates:
if query_lower in mid.lower():
substring_matches.append({"provider": prov, "model_id": mid, "entry": mdata})
# Then add difflib fuzzy matches for any remaining slots
fuzzy_ids = difflib.get_close_matches(
query_lower, model_ids_lower, n=limit * 2, cutoff=0.4
)
seen_ids: set = set()
results: List[Dict[str, Any]] = []
# Prioritize substring matches
for match in substring_matches:
key = (match["provider"], match["model_id"])
if key not in seen_ids:
seen_ids.add(key)
results.append(match)
if len(results) >= limit:
return results
# Add fuzzy matches
for fid in fuzzy_ids:
# Find original-case candidates matching this lowered ID
for prov, mid, mdata in candidates:
if mid.lower() == fid:
key = (prov, mid)
if key not in seen_ids:
seen_ids.add(key)
results.append({"provider": prov, "model_id": mid, "entry": mdata})
if len(results) >= limit:
return results
return results
+120
View File
@@ -3519,6 +3519,124 @@ class HermesCLI:
remaining = len(self.conversation_history)
print(f" {remaining} message(s) remaining in history.")
def _handle_model_switch(self, cmd_original: str):
"""Handle /model command — switch model for this session.
Supports:
/model show current model + usage hints
/model <name> switch for this session only
/model <name> --global switch and persist to config.yaml
/model provider:model switch provider + model
"""
from hermes_cli.model_switch import switch_model, ModelSwitchResult
from hermes_cli.models import _PROVIDER_LABELS
# Parse args from the original command
parts = cmd_original.split(None, 1) # split off '/model'
raw_args = parts[1].strip() if len(parts) > 1 else ""
# Check for --global flag
persist_global = "--global" in raw_args
if persist_global:
raw_args = raw_args.replace("--global", "").strip()
# No args: show current model info and usage
if not raw_args:
model_display = self.model or "unknown"
provider_display = _PROVIDER_LABELS.get(self.provider, self.provider) if self.provider else "unknown"
_cprint(f" Current model: {model_display}")
_cprint(f" Provider: {provider_display} ({self.provider})")
if self.base_url:
_cprint(f" Endpoint: {self.base_url}")
_cprint("")
_cprint(" Aliases: /model sonnet")
_cprint(" /model opus")
_cprint(" /model gpt5")
_cprint(" /model gemini")
_cprint("")
_cprint(" Full name: /model anthropic/claude-sonnet-4.5")
_cprint(" Provider: /model anthropic:sonnet (switch to native Anthropic)")
_cprint(" /model deepseek:deepseek-chat (switch to native DeepSeek)")
_cprint(" Persist: /model sonnet --global (save to config)")
_cprint(" Custom: /model custom:my-local-model")
return
# Perform the switch
result = switch_model(
raw_input=raw_args,
current_provider=self.provider or "",
current_model=self.model or "",
current_base_url=self.base_url or "",
current_api_key=self.api_key or "",
)
if not result.success:
_cprint(f"{result.error_message}")
return
# Apply to CLI state
old_model = self.model
self.model = result.new_model
self.provider = result.target_provider
if result.api_key:
self.api_key = result.api_key
if result.base_url:
self.base_url = result.base_url
if result.api_mode:
self.api_mode = result.api_mode
# Apply to running agent (in-place swap)
if self.agent is not None:
try:
self.agent.switch_model(
new_model=result.new_model,
new_provider=result.target_provider,
api_key=result.api_key,
base_url=result.base_url,
api_mode=result.api_mode,
)
except Exception as exc:
_cprint(f" ⚠ Agent swap failed ({exc}); change applied to next session.")
# Display confirmation
provider_label = result.provider_label or result.target_provider
_cprint(f" ✓ Model switched: {result.new_model}")
_cprint(f" Provider: {provider_label}")
# Show context window info
try:
from agent.model_metadata import get_model_context_length
ctx = get_model_context_length(
result.new_model,
base_url=result.base_url or self.base_url,
api_key=result.api_key or self.api_key,
provider=result.target_provider,
)
_cprint(f" Context: {ctx:,} tokens")
except Exception:
pass
# Cache notice
cache_enabled = (
("openrouter" in (result.base_url or "").lower() and "claude" in result.new_model.lower())
or result.api_mode == "anthropic_messages"
)
if cache_enabled:
_cprint(" Prompt caching: enabled")
# Warning from validation
if result.warning_message:
_cprint(f"{result.warning_message}")
# Persistence
if persist_global:
save_config_value("model.name", result.new_model)
if result.provider_changed:
save_config_value("model.provider", result.target_provider)
_cprint(" Saved to config.yaml (--global)")
else:
_cprint(" (session only — add --global to persist)")
def _show_model_and_providers(self):
"""Show current model + provider and list all authenticated providers.
@@ -4134,6 +4252,8 @@ class HermesCLI:
self.new_session()
elif canonical == "resume":
self._handle_resume_command(cmd_original)
elif canonical == "model":
self._handle_model_switch(cmd_original)
elif canonical == "provider":
self._show_model_and_providers()
elif canonical == "prompt":
+174
View File
@@ -509,6 +509,9 @@ class GatewayRunner:
self._effective_model: Optional[str] = None
self._effective_provider: Optional[str] = None
# Per-session model overrides from /model command.
# Key: session_key, Value: dict with model/provider/api_key/base_url/api_mode
self._session_model_overrides: Dict[str, Dict[str, str]] = {}
# Track pending exec approvals per session
# Key: session_key, Value: {"command": str, "pattern_key": str, ...}
self._pending_approvals: Dict[str, Dict[str, Any]] = {}
@@ -1859,6 +1862,10 @@ class GatewayRunner:
adapter._pending_messages[_quick_key] = queued_event
return "Queued for the next turn."
# /model must not be used while the agent is running.
if _cmd_def_inner and _cmd_def_inner.name == "model":
return "Agent is running — wait or /stop first, then switch models."
# /approve and /deny must bypass the running-agent interrupt path.
# The agent thread is blocked on a threading.Event inside
# tools/approval.py — sending an interrupt won't unblock it.
@@ -1958,6 +1965,9 @@ class GatewayRunner:
if canonical == "yolo":
return await self._handle_yolo_command(event)
if canonical == "model":
return await self._handle_model_command(event)
if canonical == "provider":
return await self._handle_provider_command(event)
@@ -3268,6 +3278,170 @@ class GatewayRunner:
lines.append(f"_(Requested page {requested_page} was out of range, showing page {page}.)_")
return "\n".join(lines)
async def _handle_model_command(self, event: MessageEvent) -> str:
"""Handle /model command — switch model for this session.
Supports:
/model show current model info
/model <name> switch for this session only
/model <name> --global switch and persist to config.yaml
/model provider:model switch provider + model
"""
import yaml
from hermes_cli.model_switch import switch_model as _switch_model
from hermes_cli.models import _PROVIDER_LABELS
raw_args = event.get_command_args().strip()
# Check for --global flag
persist_global = "--global" in raw_args
if persist_global:
raw_args = raw_args.replace("--global", "").strip()
# Read current model/provider from config
current_model = ""
current_provider = "openrouter"
current_base_url = ""
current_api_key = ""
config_path = _hermes_home / "config.yaml"
try:
if config_path.exists():
with open(config_path, encoding="utf-8") as f:
cfg = yaml.safe_load(f) or {}
model_cfg = cfg.get("model", {})
if isinstance(model_cfg, dict):
current_model = model_cfg.get("name", "")
current_provider = model_cfg.get("provider", current_provider)
current_base_url = model_cfg.get("base_url", "")
except Exception:
pass
# Check for session override
source = event.source
session_key = self._session_key_for_source(source)
override = self._session_model_overrides.get(session_key, {})
if override:
current_model = override.get("model", current_model)
current_provider = override.get("provider", current_provider)
current_base_url = override.get("base_url", current_base_url)
current_api_key = override.get("api_key", current_api_key)
# No args: show current model info
if not raw_args:
provider_label = _PROVIDER_LABELS.get(current_provider, current_provider)
lines = [
f"🤖 **Current model:** `{current_model or 'unknown'}`",
f"🔌 **Provider:** {provider_label} (`{current_provider}`)",
]
if current_base_url:
lines.append(f"🌐 **Endpoint:** `{current_base_url}`")
lines.append("")
lines.append("**Usage:**")
lines.append("`/model <model-name>` — switch for this session")
lines.append("`/model provider:model` — switch provider + model")
lines.append("`/model <name> --global` — persist to config")
lines.append("")
lines.append("**Examples:**")
lines.append("`/model claude-sonnet-4`")
lines.append("`/model openai:gpt-4.1`")
lines.append("`/model anthropic:claude-sonnet-4 --global`")
return "\n".join(lines)
# Perform the switch
result = _switch_model(
raw_input=raw_args,
current_provider=current_provider,
current_model=current_model,
current_base_url=current_base_url,
current_api_key=current_api_key,
)
if not result.success:
return f"{result.error_message}"
# If there's a cached agent, update it in-place
cached_entry = None
_cache_lock = getattr(self, "_agent_cache_lock", None)
_cache = getattr(self, "_agent_cache", None)
if _cache_lock and _cache is not None:
with _cache_lock:
cached_entry = _cache.get(session_key)
if cached_entry and cached_entry[0] is not None:
try:
cached_entry[0].switch_model(
new_model=result.new_model,
new_provider=result.target_provider,
api_key=result.api_key,
base_url=result.base_url,
api_mode=result.api_mode,
)
except Exception as exc:
logger.warning("In-place model switch failed for cached agent: %s", exc)
# Store session override so next agent creation uses the new model
self._session_model_overrides[session_key] = {
"model": result.new_model,
"provider": result.target_provider,
"api_key": result.api_key,
"base_url": result.base_url,
"api_mode": result.api_mode,
}
# Persist to config if --global
if persist_global:
try:
if config_path.exists():
with open(config_path, encoding="utf-8") as f:
cfg = yaml.safe_load(f) or {}
else:
cfg = {}
model_cfg = cfg.setdefault("model", {})
model_cfg["name"] = result.new_model
model_cfg["provider"] = result.target_provider
if result.base_url:
model_cfg["base_url"] = result.base_url
from hermes_cli.config import save_config
save_config(cfg)
except Exception as e:
logger.warning("Failed to persist model switch: %s", e)
# Build confirmation message
provider_label = result.provider_label or result.target_provider
lines = [f"✅ Model switched to `{result.new_model}`"]
lines.append(f"Provider: {provider_label}")
# Context window info
try:
from agent.model_metadata import get_model_context_length
ctx = get_model_context_length(
result.new_model,
base_url=result.base_url or current_base_url,
api_key=result.api_key or current_api_key,
provider=result.target_provider,
)
lines.append(f"📏 Context: {ctx:,} tokens")
except Exception:
pass
# Cache notice
cache_enabled = (
("openrouter" in (result.base_url or "").lower() and "claude" in result.new_model.lower())
or result.api_mode == "anthropic_messages"
)
if cache_enabled:
lines.append("💾 Prompt caching: enabled")
if result.warning_message:
lines.append(f"⚠️ {result.warning_message}")
if persist_global:
lines.append("💾 Saved to config.yaml (`--global`)")
else:
lines.append("_(session only — add `--global` to persist)_")
return "\n".join(lines)
async def _handle_provider_command(self, event: MessageEvent) -> str:
"""Handle /provider command - show available providers."""
import yaml
+1
View File
@@ -84,6 +84,7 @@ COMMAND_REGISTRY: list[CommandDef] = [
# Configuration
CommandDef("config", "Show current configuration", "Configuration",
cli_only=True),
CommandDef("model", "Switch model for this session", "Configuration", args_hint="[model] [--global]"),
CommandDef("provider", "Show available providers and current provider",
"Configuration"),
CommandDef("prompt", "View/set custom system prompt", "Configuration",
+359
View File
@@ -0,0 +1,359 @@
"""Per-provider model name normalization.
Different LLM providers expect model identifiers in different formats:
- **Aggregators** (OpenRouter, Nous, AI Gateway, Kilo Code) need
``vendor/model`` slugs like ``anthropic/claude-sonnet-4.6``.
- **Anthropic** native API expects bare names with dots replaced by
hyphens: ``claude-sonnet-4-6``.
- **Copilot** expects bare names *with* dots preserved:
``claude-sonnet-4.6``.
- **OpenCode** (Zen & Go) follows the same dot-to-hyphen convention as
Anthropic: ``claude-sonnet-4-6``.
- **DeepSeek** only accepts two model identifiers:
``deepseek-chat`` and ``deepseek-reasoner``.
- **Custom** and remaining providers pass the name through as-is.
This module centralises that translation so callers can simply write::
api_model = normalize_model_for_provider(user_input, provider)
Inspired by Clawdbot's ``normalizeAnthropicModelId`` pattern.
"""
from __future__ import annotations
from typing import Optional
# ---------------------------------------------------------------------------
# Vendor prefix mapping
# ---------------------------------------------------------------------------
# Maps the first hyphen-delimited token of a bare model name to the vendor
# slug used by aggregator APIs (OpenRouter, Nous, etc.).
#
# Example: "claude-sonnet-4.6" -> first token "claude" -> vendor "anthropic"
# -> aggregator slug: "anthropic/claude-sonnet-4.6"
_VENDOR_PREFIXES: dict[str, str] = {
"claude": "anthropic",
"gpt": "openai",
"o1": "openai",
"o3": "openai",
"o4": "openai",
"gemini": "google",
"deepseek": "deepseek",
"glm": "z-ai",
"kimi": "moonshotai",
"minimax": "minimax",
"grok": "x-ai",
"qwen": "qwen",
"mimo": "xiaomi",
"nemotron": "nvidia",
"llama": "meta-llama",
"step": "stepfun",
"trinity": "arcee-ai",
}
# Providers whose APIs consume vendor/model slugs.
_AGGREGATOR_PROVIDERS: frozenset[str] = frozenset({
"openrouter",
"nous",
"ai-gateway",
"kilocode",
})
# Providers that want bare names with dots replaced by hyphens.
_DOT_TO_HYPHEN_PROVIDERS: frozenset[str] = frozenset({
"anthropic",
"opencode-zen",
"opencode-go",
})
# Providers that want bare names with dots preserved.
_STRIP_VENDOR_ONLY_PROVIDERS: frozenset[str] = frozenset({
"copilot",
"copilot-acp",
})
# Providers whose own naming is authoritative -- pass through unchanged.
_PASSTHROUGH_PROVIDERS: frozenset[str] = frozenset({
"zai",
"kimi-coding",
"minimax",
"minimax-cn",
"alibaba",
"huggingface",
"openai-codex",
"custom",
})
# ---------------------------------------------------------------------------
# DeepSeek special handling
# ---------------------------------------------------------------------------
# DeepSeek's API only recognises exactly two model identifiers. We map
# common aliases and patterns to the canonical names.
_DEEPSEEK_REASONER_KEYWORDS: frozenset[str] = frozenset({
"reasoner",
"r1",
"think",
"reasoning",
"cot",
})
_DEEPSEEK_CANONICAL_MODELS: frozenset[str] = frozenset({
"deepseek-chat",
"deepseek-reasoner",
})
def _normalize_for_deepseek(model_name: str) -> str:
"""Map any model input to one of DeepSeek's two accepted identifiers.
Rules:
- Already ``deepseek-chat`` or ``deepseek-reasoner`` -> pass through.
- Contains any reasoner keyword (r1, think, reasoning, cot, reasoner)
-> ``deepseek-reasoner``.
- Everything else -> ``deepseek-chat``.
Args:
model_name: The bare model name (vendor prefix already stripped).
Returns:
One of ``"deepseek-chat"`` or ``"deepseek-reasoner"``.
"""
bare = _strip_vendor_prefix(model_name).lower()
if bare in _DEEPSEEK_CANONICAL_MODELS:
return bare
# Check for reasoner-like keywords anywhere in the name
for keyword in _DEEPSEEK_REASONER_KEYWORDS:
if keyword in bare:
return "deepseek-reasoner"
return "deepseek-chat"
# ---------------------------------------------------------------------------
# Helper utilities
# ---------------------------------------------------------------------------
def _strip_vendor_prefix(model_name: str) -> str:
"""Remove a ``vendor/`` prefix if present.
Examples::
>>> _strip_vendor_prefix("anthropic/claude-sonnet-4.6")
'claude-sonnet-4.6'
>>> _strip_vendor_prefix("claude-sonnet-4.6")
'claude-sonnet-4.6'
>>> _strip_vendor_prefix("meta-llama/llama-4-scout")
'llama-4-scout'
"""
if "/" in model_name:
return model_name.split("/", 1)[1]
return model_name
def _dots_to_hyphens(model_name: str) -> str:
"""Replace dots with hyphens in a model name.
Anthropic's native API uses hyphens where marketing names use dots:
``claude-sonnet-4.6`` -> ``claude-sonnet-4-6``.
"""
return model_name.replace(".", "-")
def detect_vendor(model_name: str) -> Optional[str]:
"""Detect the vendor slug from a bare model name.
Uses the first hyphen-delimited token of the model name to look up
the corresponding vendor in ``_VENDOR_PREFIXES``. Also handles
case-insensitive matching and special patterns.
Args:
model_name: A model name, optionally already including a
``vendor/`` prefix. If a prefix is present it is used
directly.
Returns:
The vendor slug (e.g. ``"anthropic"``, ``"openai"``) or ``None``
if no vendor can be confidently detected.
Examples::
>>> detect_vendor("claude-sonnet-4.6")
'anthropic'
>>> detect_vendor("gpt-5.4-mini")
'openai'
>>> detect_vendor("anthropic/claude-sonnet-4.6")
'anthropic'
>>> detect_vendor("my-custom-model")
"""
name = model_name.strip()
if not name:
return None
# If there's already a vendor/ prefix, extract it
if "/" in name:
return name.split("/", 1)[0].lower() or None
name_lower = name.lower()
# Try first hyphen-delimited token (exact match)
first_token = name_lower.split("-")[0]
if first_token in _VENDOR_PREFIXES:
return _VENDOR_PREFIXES[first_token]
# Handle patterns where the first token includes version digits,
# e.g. "qwen3.5-plus" -> first token "qwen3.5", but prefix is "qwen"
for prefix, vendor in _VENDOR_PREFIXES.items():
if name_lower.startswith(prefix):
return vendor
return None
def _prepend_vendor(model_name: str) -> str:
"""Prepend the detected ``vendor/`` prefix if missing.
Used for aggregator providers that require ``vendor/model`` format.
If the name already contains a ``/``, it is returned as-is.
If no vendor can be detected, the name is returned unchanged
(aggregators may still accept it or return an error).
Examples::
>>> _prepend_vendor("claude-sonnet-4.6")
'anthropic/claude-sonnet-4.6'
>>> _prepend_vendor("anthropic/claude-sonnet-4.6")
'anthropic/claude-sonnet-4.6'
>>> _prepend_vendor("my-custom-thing")
'my-custom-thing'
"""
if "/" in model_name:
return model_name
vendor = detect_vendor(model_name)
if vendor:
return f"{vendor}/{model_name}"
return model_name
# ---------------------------------------------------------------------------
# Main normalisation entry point
# ---------------------------------------------------------------------------
def normalize_model_for_provider(model_input: str, target_provider: str) -> str:
"""Translate a model name into the format the target provider's API expects.
This is the primary entry point for model name normalisation. It
accepts any user-facing model identifier and transforms it for the
specific provider that will receive the API call.
Args:
model_input: The model name as provided by the user or config.
Can be bare (``"claude-sonnet-4.6"``), vendor-prefixed
(``"anthropic/claude-sonnet-4.6"``), or already in native
format (``"claude-sonnet-4-6"``).
target_provider: The canonical Hermes provider id, e.g.
``"openrouter"``, ``"anthropic"``, ``"copilot"``,
``"deepseek"``, ``"custom"``. Should already be normalised
via ``hermes_cli.models.normalize_provider()``.
Returns:
The model identifier string that the target provider's API
expects.
Raises:
No exceptions -- always returns a best-effort string.
Examples::
>>> normalize_model_for_provider("claude-sonnet-4.6", "openrouter")
'anthropic/claude-sonnet-4.6'
>>> normalize_model_for_provider("anthropic/claude-sonnet-4.6", "anthropic")
'claude-sonnet-4-6'
>>> normalize_model_for_provider("anthropic/claude-sonnet-4.6", "copilot")
'claude-sonnet-4.6'
>>> normalize_model_for_provider("openai/gpt-5.4", "copilot")
'gpt-5.4'
>>> normalize_model_for_provider("claude-sonnet-4.6", "opencode-zen")
'claude-sonnet-4-6'
>>> normalize_model_for_provider("deepseek-v3", "deepseek")
'deepseek-chat'
>>> normalize_model_for_provider("deepseek-r1", "deepseek")
'deepseek-reasoner'
>>> normalize_model_for_provider("my-model", "custom")
'my-model'
>>> normalize_model_for_provider("claude-sonnet-4.6", "zai")
'claude-sonnet-4.6'
"""
name = (model_input or "").strip()
if not name:
return name
provider = (target_provider or "").strip().lower()
# --- Aggregators: need vendor/model format ---
if provider in _AGGREGATOR_PROVIDERS:
return _prepend_vendor(name)
# --- Anthropic / OpenCode: strip vendor, dots -> hyphens ---
if provider in _DOT_TO_HYPHEN_PROVIDERS:
bare = _strip_vendor_prefix(name)
return _dots_to_hyphens(bare)
# --- Copilot: strip vendor, keep dots ---
if provider in _STRIP_VENDOR_ONLY_PROVIDERS:
return _strip_vendor_prefix(name)
# --- DeepSeek: map to one of two canonical names ---
if provider == "deepseek":
return _normalize_for_deepseek(name)
# --- Custom & all others: pass through as-is ---
return name
# ---------------------------------------------------------------------------
# Batch / convenience helpers
# ---------------------------------------------------------------------------
def model_display_name(model_id: str) -> str:
"""Return a short, human-readable display name for a model id.
Strips the vendor prefix (if any) for a cleaner display in menus
and status bars, while preserving dots for readability.
Examples::
>>> model_display_name("anthropic/claude-sonnet-4.6")
'claude-sonnet-4.6'
>>> model_display_name("claude-sonnet-4-6")
'claude-sonnet-4-6'
"""
return _strip_vendor_prefix((model_id or "").strip())
def is_aggregator_provider(provider: str) -> bool:
"""Check if a provider is an aggregator that needs vendor/model format."""
return (provider or "").strip().lower() in _AGGREGATOR_PROVIDERS
def vendor_for_model(model_name: str) -> str:
"""Return the vendor slug for a model, or ``""`` if unknown.
Convenience wrapper around :func:`detect_vendor` that never returns
``None``.
"""
return detect_vendor(model_name) or ""
+421 -46
View File
@@ -3,18 +3,115 @@
Both the CLI (cli.py) and gateway (gateway/run.py) /model handlers
share the same core pipeline:
parse_model_input → is_custom detection → auto-detect provider
credential resolution → validate model → return result
alias resolution -> parse_model_input -> aggregator slug fixup ->
detect_provider -> credential resolution -> normalize model name ->
capability lookup -> build result
This module extracts that shared pipeline into pure functions that
return result objects. The callers handle all platform-specific
concerns: state mutation, config persistence, output formatting.
This module ties together the foundation layers:
- ``hermes_cli.providers`` -- canonical provider identity
- ``hermes_cli.model_normalize`` -- per-provider name formatting
- ``agent.models_dev`` -- models.dev catalog & capabilities
Callers handle all platform-specific concerns: state mutation, config
persistence, output formatting.
"""
from __future__ import annotations
from dataclasses import dataclass
import logging
from dataclasses import dataclass, field
from typing import List, NamedTuple, Optional
from hermes_cli.providers import (
ALIASES,
LABELS,
PROVIDERS,
determine_api_mode,
get_provider,
is_aggregator,
normalize_provider,
)
from hermes_cli.model_normalize import (
detect_vendor,
normalize_model_for_provider,
)
from agent.models_dev import (
ModelCapabilities,
get_model_capabilities,
list_provider_models,
search_models_dev,
)
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Model aliases -- short names -> (vendor, family) with NO version numbers.
# Resolved dynamically against the live models.dev catalog.
# ---------------------------------------------------------------------------
class ModelIdentity(NamedTuple):
"""Vendor slug and family prefix used for catalog resolution."""
vendor: str
family: str
MODEL_ALIASES: dict[str, ModelIdentity] = {
# Anthropic
"sonnet": ModelIdentity("anthropic", "claude-sonnet"),
"opus": ModelIdentity("anthropic", "claude-opus"),
"haiku": ModelIdentity("anthropic", "claude-haiku"),
"claude": ModelIdentity("anthropic", "claude"),
# OpenAI
"gpt5": ModelIdentity("openai", "gpt-5"),
"gpt": ModelIdentity("openai", "gpt"),
"codex": ModelIdentity("openai", "codex"),
"o3": ModelIdentity("openai", "o3"),
"o4": ModelIdentity("openai", "o4"),
# Google
"gemini": ModelIdentity("google", "gemini"),
# DeepSeek
"deepseek": ModelIdentity("deepseek", "deepseek-chat"),
# X.AI
"grok": ModelIdentity("x-ai", "grok"),
# Meta
"llama": ModelIdentity("meta-llama", "llama"),
# Qwen / Alibaba
"qwen": ModelIdentity("qwen", "qwen"),
# MiniMax
"minimax": ModelIdentity("minimax", "minimax"),
# Nvidia
"nemotron": ModelIdentity("nvidia", "nemotron"),
# Moonshot / Kimi
"kimi": ModelIdentity("moonshotai", "kimi"),
# Z.AI / GLM
"glm": ModelIdentity("z-ai", "glm"),
# StepFun
"step": ModelIdentity("stepfun", "step"),
# Xiaomi
"mimo": ModelIdentity("xiaomi", "mimo"),
# Arcee
"trinity": ModelIdentity("arcee-ai", "trinity"),
}
# ---------------------------------------------------------------------------
# Result dataclasses
# ---------------------------------------------------------------------------
@dataclass
class ModelSwitchResult:
@@ -27,11 +124,12 @@ class ModelSwitchResult:
api_key: str = ""
base_url: str = ""
api_mode: str = ""
persist: bool = False
error_message: str = ""
warning_message: str = ""
is_custom_target: bool = False
provider_label: str = ""
resolved_via_alias: str = ""
capabilities: Optional[ModelCapabilities] = None
is_global: bool = False
@dataclass
@@ -45,62 +143,291 @@ class CustomAutoResult:
error_message: str = ""
# ---------------------------------------------------------------------------
# Known provider names (for parse_model_input colon-splitting)
# ---------------------------------------------------------------------------
_KNOWN_PROVIDER_NAMES: set[str] = (
set(PROVIDERS.keys())
| set(ALIASES.keys())
| {"custom"}
)
# ---------------------------------------------------------------------------
# Alias resolution
# ---------------------------------------------------------------------------
def resolve_alias(
raw_input: str,
current_provider: str,
) -> Optional[tuple[str, str, str]]:
"""Resolve a short alias against the current provider's catalog.
Looks up *raw_input* in :data:`MODEL_ALIASES`, then searches the
current provider's models.dev catalog for the first model whose ID
starts with ``vendor/family`` (or just ``family`` for non-aggregator
providers).
Args:
raw_input: The user's raw model input (e.g. ``"sonnet"``).
current_provider: The currently active Hermes provider id.
Returns:
``(provider, resolved_model_id, alias_name)`` if a match is
found on the current provider, or ``None`` if the alias doesn't
exist or no matching model is available.
"""
key = raw_input.strip().lower()
identity = MODEL_ALIASES.get(key)
if identity is None:
return None
vendor, family = identity
# Search the provider's catalog from models.dev
catalog = list_provider_models(current_provider)
if not catalog:
return None
# For aggregators, models are vendor/model-name format
aggregator = is_aggregator(current_provider)
for model_id in catalog:
mid_lower = model_id.lower()
if aggregator:
# Match vendor/family prefix -- e.g. "anthropic/claude-sonnet"
prefix = f"{vendor}/{family}".lower()
if mid_lower.startswith(prefix):
return (current_provider, model_id, key)
else:
# Non-aggregator: bare names -- e.g. "claude-sonnet-4-6"
# Match family prefix (dots/hyphens may vary)
family_lower = family.lower()
if mid_lower.startswith(family_lower):
return (current_provider, model_id, key)
return None
def _resolve_alias_fallback(
raw_input: str,
fallback_providers: tuple[str, ...] = ("openrouter", "nous"),
) -> Optional[tuple[str, str, str]]:
"""Try to resolve an alias on fallback providers.
Called when the alias exists in MODEL_ALIASES but no matching model
was found on the current provider.
Returns:
``(provider, resolved_model_id, alias_name)`` or ``None``.
"""
for provider in fallback_providers:
result = resolve_alias(raw_input, provider)
if result is not None:
return result
return None
# ---------------------------------------------------------------------------
# parse_model_input -- shared with models.py but using our provider set
# ---------------------------------------------------------------------------
def parse_model_input(raw: str, current_provider: str) -> tuple[str, str]:
"""Parse ``/model`` input into ``(provider, model)``.
Supports ``provider:model`` syntax to switch providers at runtime::
openrouter:anthropic/claude-sonnet-4.5 -> ("openrouter", "anthropic/claude-sonnet-4.5")
nous:hermes-3 -> ("nous", "hermes-3")
anthropic/claude-sonnet-4.5 -> (current_provider, "anthropic/claude-sonnet-4.5")
gpt-5.4 -> (current_provider, "gpt-5.4")
The colon is only treated as a provider delimiter if the left side
is a recognized provider name or alias.
"""
stripped = raw.strip()
colon = stripped.find(":")
if colon > 0:
provider_part = stripped[:colon].strip().lower()
model_part = stripped[colon + 1:].strip()
if provider_part and model_part and provider_part in _KNOWN_PROVIDER_NAMES:
# Support custom:name:model triple syntax
if provider_part == "custom" and ":" in model_part:
second_colon = model_part.find(":")
custom_name = model_part[:second_colon].strip()
actual_model = model_part[second_colon + 1:].strip()
if custom_name and actual_model:
return (f"custom:{custom_name}", actual_model)
return (normalize_provider(provider_part), model_part)
return (current_provider, stripped)
# ---------------------------------------------------------------------------
# Core model-switching pipeline
# ---------------------------------------------------------------------------
def switch_model(
raw_input: str,
current_provider: str,
current_model: str,
current_base_url: str = "",
current_api_key: str = "",
is_global: bool = False,
) -> ModelSwitchResult:
"""Core model-switching pipeline shared between CLI and gateway.
Handles parsing, provider detection, credential resolution, and
model validation. Does NOT handle config persistence, state
mutation, or output formatting — those are caller responsibilities.
Resolution chain:
a. Try alias resolution on current provider
b. If alias exists but not on current provider -> try fallback
providers (openrouter, nous)
c. If on aggregator and input has vendor:model where vendor is NOT
a known hermes provider -> convert to vendor/model slug
d. Standard parse_model_input()
e. If on aggregator, try to resolve within aggregator catalog first
f. Fall back to detect_provider_for_model()
g. Resolve credentials via resolve_runtime_provider()
h. Normalize model name for target provider
i. Get capabilities from get_model_capabilities()
j. Build result
Args:
raw_input: The user's model input (e.g. "claude-sonnet-4",
raw_input: The user's model input (e.g. "sonnet", "claude-sonnet-4",
"zai:glm-5", "custom:local:qwen").
current_provider: The currently active provider.
current_base_url: The currently active base URL (used for
is_custom detection).
current_model: The currently active model name.
current_base_url: The currently active base URL.
current_api_key: The currently active API key.
is_global: Whether this switch should be persisted globally.
Returns:
ModelSwitchResult with all information the caller needs to
apply the switch and format output.
ModelSwitchResult with all information the caller needs.
"""
from hermes_cli.models import (
parse_model_input,
detect_provider_for_model,
validate_requested_model,
_PROVIDER_LABELS,
opencode_model_api_mode,
)
from hermes_cli.runtime_provider import resolve_runtime_provider
# Step 1: Parse provider:model syntax
target_provider, new_model = parse_model_input(raw_input, current_provider)
resolved_alias = ""
# Step 2: Detect if we're currently on a custom endpoint
# --- Step a: Try alias resolution on current provider ---
alias_result = resolve_alias(raw_input, current_provider)
if alias_result is not None:
target_provider, new_model, resolved_alias = alias_result
logger.debug(
"Alias '%s' resolved to %s on %s",
resolved_alias, new_model, target_provider,
)
else:
# --- Step b: Alias exists but not on current provider -> fallback ---
key = raw_input.strip().lower()
if key in MODEL_ALIASES:
fallback_result = _resolve_alias_fallback(raw_input)
if fallback_result is not None:
target_provider, new_model, resolved_alias = fallback_result
logger.debug(
"Alias '%s' resolved via fallback to %s on %s",
resolved_alias, new_model, target_provider,
)
else:
# Alias exists but no model found anywhere
identity = MODEL_ALIASES[key]
return ModelSwitchResult(
success=False,
is_global=is_global,
error_message=(
f"Alias '{key}' maps to {identity.vendor}/{identity.family} "
f"but no matching model was found in any provider catalog. "
f"Try specifying the full model name."
),
)
else:
# --- Step c: aggregator vendor:model fixup ---
# If on an aggregator and the input looks like vendor:model
# where vendor is NOT a known Hermes provider, treat the colon
# as a vendor/model separator instead of provider:model.
target_provider = current_provider
new_model = raw_input.strip()
colon_pos = raw_input.find(":")
if colon_pos > 0 and is_aggregator(current_provider):
left = raw_input[:colon_pos].strip().lower()
right = raw_input[colon_pos + 1:].strip()
if left and right and left not in _KNOWN_PROVIDER_NAMES:
# Convert vendor:model -> vendor/model for aggregator
new_model = f"{left}/{right}"
target_provider = current_provider
logger.debug(
"Converted vendor:model '%s' to aggregator slug '%s'",
raw_input, new_model,
)
# --- Step d: Standard parse_model_input() ---
if new_model == raw_input.strip():
target_provider, new_model = parse_model_input(
raw_input, current_provider,
)
# --- Step d2: If parsed model is a known alias, resolve it
# on the TARGET provider. This handles "anthropic:sonnet"
# -> resolve "sonnet" alias on anthropic provider.
model_as_alias = new_model.strip().lower()
if model_as_alias in MODEL_ALIASES and target_provider != current_provider:
alias_on_target = resolve_alias(new_model, target_provider)
if alias_on_target is not None:
_, new_model, resolved_alias = alias_on_target
logger.debug(
"Resolved alias '%s' on target provider %s -> %s",
model_as_alias, target_provider, new_model,
)
# --- Step e: If on aggregator, try to resolve within aggregator catalog ---
if is_aggregator(target_provider) and not resolved_alias:
catalog = list_provider_models(target_provider)
if catalog:
new_model_lower = new_model.lower()
# Exact match
for mid in catalog:
if mid.lower() == new_model_lower:
new_model = mid
break
else:
# Try matching bare name against catalog entries
for mid in catalog:
if "/" in mid:
_, bare = mid.split("/", 1)
if bare.lower() == new_model_lower:
new_model = mid
break
# --- Step f: Fall back to detect_provider_for_model() ---
# Only auto-detect when no explicit provider was given and we're not
# on a custom endpoint.
_base = current_base_url or ""
is_custom = current_provider == "custom" or (
"localhost" in _base or "127.0.0.1" in _base
)
# Step 3: Auto-detect provider when no explicit provider:model syntax
# was used. Skip for custom providers — the model name might
# coincidentally match a known provider's catalog.
if target_provider == current_provider and not is_custom:
if (
target_provider == current_provider
and not is_custom
and not resolved_alias
):
detected = detect_provider_for_model(new_model, current_provider)
if detected:
target_provider, new_model = detected
provider_changed = target_provider != current_provider
# Step 4: Resolve credentials for target provider
# --- Step g: Resolve credentials ---
api_key = current_api_key
base_url = current_base_url
api_mode = ""
provider_label = LABELS.get(target_provider, target_provider)
if provider_changed:
try:
runtime = resolve_runtime_provider(requested=target_provider)
@@ -108,28 +435,30 @@ def switch_model(
base_url = runtime.get("base_url", "")
api_mode = runtime.get("api_mode", "")
except Exception as e:
provider_label = _PROVIDER_LABELS.get(target_provider, target_provider)
if target_provider == "custom":
return ModelSwitchResult(
success=False,
target_provider=target_provider,
provider_label=provider_label,
is_global=is_global,
error_message=(
"No custom endpoint configured. Set model.base_url "
"in config.yaml, or set OPENAI_BASE_URL in .env, "
"or run: hermes setup Custom OpenAI-compatible endpoint"
"or run: hermes setup -> Custom OpenAI-compatible endpoint"
),
)
return ModelSwitchResult(
success=False,
target_provider=target_provider,
provider_label=provider_label,
is_global=is_global,
error_message=(
f"Could not resolve credentials for provider "
f"'{provider_label}': {e}"
),
)
else:
# Gateway also resolves for unchanged provider to get accurate
# base_url for validation probing.
# Re-resolve for current provider to get accurate base_url
try:
runtime = resolve_runtime_provider(requested=current_provider)
api_key = runtime.get("api_key", "")
@@ -138,7 +467,10 @@ def switch_model(
except Exception:
pass
# Step 5: Validate the model
# --- Step h: Normalize model name for target provider ---
new_model = normalize_model_for_provider(new_model, target_provider)
# --- Validate ---
try:
validation = validate_requested_model(
new_model,
@@ -160,23 +492,23 @@ def switch_model(
success=False,
new_model=new_model,
target_provider=target_provider,
provider_label=provider_label,
is_global=is_global,
error_message=msg,
)
# Step 6: Build result
provider_label = _PROVIDER_LABELS.get(target_provider, target_provider)
is_custom_target = target_provider == "custom" or (
base_url
and "openrouter.ai" not in (base_url or "")
and ("localhost" in (base_url or "") or "127.0.0.1" in (base_url or ""))
)
# --- OpenCode api_mode override ---
if target_provider in {"opencode-zen", "opencode-go"}:
# Recompute against the requested new model, not the currently-configured
# model used during runtime resolution. OpenCode mixes API surfaces by
# model family, so a same-provider model switch can change api_mode.
api_mode = opencode_model_api_mode(target_provider, new_model)
# --- Determine api_mode if not already set ---
if not api_mode:
api_mode = determine_api_mode(target_provider, base_url)
# --- Step i: Get capabilities from models.dev ---
capabilities = get_model_capabilities(target_provider, new_model)
# --- Step j: Build result ---
return ModelSwitchResult(
success=True,
new_model=new_model,
@@ -185,15 +517,58 @@ def switch_model(
api_key=api_key,
base_url=base_url,
api_mode=api_mode,
persist=bool(validation.get("persist")),
warning_message=validation.get("message") or "",
is_custom_target=is_custom_target,
provider_label=provider_label,
resolved_via_alias=resolved_alias,
capabilities=capabilities,
is_global=is_global,
)
# ---------------------------------------------------------------------------
# Fuzzy suggestions
# ---------------------------------------------------------------------------
def suggest_models(raw_input: str, limit: int = 3) -> List[str]:
"""Return fuzzy model suggestions for a (possibly misspelled) input.
Searches the models.dev catalog across all providers and returns up
to *limit* model IDs that best match *raw_input*.
Args:
raw_input: The user's raw model input.
limit: Maximum number of suggestions.
Returns:
List of model ID strings (may be empty).
"""
query = raw_input.strip()
if not query:
return []
results = search_models_dev(query, limit=limit)
suggestions: list[str] = []
for r in results:
mid = r.get("model_id", "")
prov = r.get("provider", "")
if mid:
# For aggregator providers, model_id already has vendor/ prefix
if prov and is_aggregator(prov):
suggestions.append(mid)
elif prov:
suggestions.append(f"{prov}:{mid}")
else:
suggestions.append(mid)
return suggestions[:limit]
# ---------------------------------------------------------------------------
# Custom provider switch
# ---------------------------------------------------------------------------
def switch_to_custom_provider() -> CustomAutoResult:
"""Handle bare '/model custom' resolve endpoint and auto-detect model.
"""Handle bare '/model custom' -- resolve endpoint and auto-detect model.
Returns a result object; the caller handles persistence and output.
"""
@@ -219,7 +594,7 @@ def switch_to_custom_provider() -> CustomAutoResult:
error_message=(
"No custom endpoint configured. "
"Set model.base_url in config.yaml, or set OPENAI_BASE_URL "
"in .env, or run: hermes setup Custom OpenAI-compatible endpoint"
"in .env, or run: hermes setup -> Custom OpenAI-compatible endpoint"
),
)
+339
View File
@@ -0,0 +1,339 @@
"""
Single source of truth for provider identity in Hermes Agent.
Every provider known to the system is defined here exactly once. Other
modules (auth, models, inference) import from this file rather than
maintaining their own parallel registries.
Concepts
--------
- **ProviderDef** -- immutable description of one inference provider.
- **PROVIDERS** -- canonical ``{id: ProviderDef}`` mapping.
- **ALIASES** -- maps human-friendly or legacy names to a canonical id.
- **LABELS** -- short display names for UI / CLI menus.
- **transport** -- wire protocol the provider speaks
(``openai_chat``, ``anthropic_messages``, ``codex_responses``).
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Optional
# -- Provider definition ------------------------------------------------------
@dataclass(frozen=True)
class ProviderDef:
"""Immutable description of a known inference provider."""
id: str
name: str
transport: str # 'openai_chat' | 'anthropic_messages' | 'codex_responses'
api_key_env_vars: tuple # env vars to check, in priority order
base_url: str = ""
base_url_env_var: str = ""
is_aggregator: bool = False
auth_type: str = "api_key" # api_key | oauth_device_code | oauth_external | external_process
# -- Canonical provider table -------------------------------------------------
PROVIDERS: dict[str, ProviderDef] = {
"openrouter": ProviderDef(
id="openrouter",
name="OpenRouter",
transport="openai_chat",
api_key_env_vars=("OPENROUTER_API_KEY", "OPENAI_API_KEY"),
base_url="https://openrouter.ai/api/v1",
base_url_env_var="OPENROUTER_BASE_URL",
is_aggregator=True,
),
"nous": ProviderDef(
id="nous",
name="Nous Portal",
transport="openai_chat",
api_key_env_vars=(),
base_url="https://inference-api.nousresearch.com/v1",
auth_type="oauth_device_code",
),
"openai-codex": ProviderDef(
id="openai-codex",
name="OpenAI Codex",
transport="codex_responses",
api_key_env_vars=(),
base_url="https://chatgpt.com/backend-api/codex",
auth_type="oauth_external",
),
"copilot": ProviderDef(
id="copilot",
name="GitHub Copilot",
transport="openai_chat",
api_key_env_vars=("COPILOT_GITHUB_TOKEN", "GH_TOKEN", "GITHUB_TOKEN"),
base_url="https://api.githubcopilot.com",
),
"copilot-acp": ProviderDef(
id="copilot-acp",
name="GitHub Copilot ACP",
transport="codex_responses",
api_key_env_vars=(),
base_url="acp://copilot",
base_url_env_var="COPILOT_ACP_BASE_URL",
auth_type="external_process",
),
"zai": ProviderDef(
id="zai",
name="Z.AI / GLM",
transport="openai_chat",
api_key_env_vars=("GLM_API_KEY", "ZAI_API_KEY", "Z_AI_API_KEY"),
base_url="https://api.z.ai/api/paas/v4",
base_url_env_var="GLM_BASE_URL",
),
"kimi-coding": ProviderDef(
id="kimi-coding",
name="Kimi / Moonshot",
transport="openai_chat",
api_key_env_vars=("KIMI_API_KEY",),
base_url="https://api.moonshot.ai/v1",
base_url_env_var="KIMI_BASE_URL",
),
"minimax": ProviderDef(
id="minimax",
name="MiniMax",
transport="openai_chat",
api_key_env_vars=("MINIMAX_API_KEY",),
base_url="https://api.minimax.io/anthropic",
base_url_env_var="MINIMAX_BASE_URL",
),
"minimax-cn": ProviderDef(
id="minimax-cn",
name="MiniMax (China)",
transport="openai_chat",
api_key_env_vars=("MINIMAX_CN_API_KEY",),
base_url="https://api.minimaxi.com/anthropic",
base_url_env_var="MINIMAX_CN_BASE_URL",
),
"anthropic": ProviderDef(
id="anthropic",
name="Anthropic",
transport="anthropic_messages",
api_key_env_vars=("ANTHROPIC_API_KEY", "ANTHROPIC_TOKEN", "CLAUDE_CODE_OAUTH_TOKEN"),
base_url="https://api.anthropic.com",
),
"deepseek": ProviderDef(
id="deepseek",
name="DeepSeek",
transport="openai_chat",
api_key_env_vars=("DEEPSEEK_API_KEY",),
base_url="https://api.deepseek.com/v1",
base_url_env_var="DEEPSEEK_BASE_URL",
),
"ai-gateway": ProviderDef(
id="ai-gateway",
name="AI Gateway",
transport="openai_chat",
api_key_env_vars=("AI_GATEWAY_API_KEY",),
base_url="https://ai-gateway.vercel.sh/v1",
base_url_env_var="AI_GATEWAY_BASE_URL",
is_aggregator=True,
),
"opencode-zen": ProviderDef(
id="opencode-zen",
name="OpenCode Zen",
transport="openai_chat",
api_key_env_vars=("OPENCODE_ZEN_API_KEY",),
base_url="https://opencode.ai/zen/v1",
base_url_env_var="OPENCODE_ZEN_BASE_URL",
is_aggregator=True,
),
"opencode-go": ProviderDef(
id="opencode-go",
name="OpenCode Go",
transport="openai_chat",
api_key_env_vars=("OPENCODE_GO_API_KEY",),
base_url="https://opencode.ai/zen/go/v1",
base_url_env_var="OPENCODE_GO_BASE_URL",
is_aggregator=True,
),
"kilocode": ProviderDef(
id="kilocode",
name="Kilo Code",
transport="openai_chat",
api_key_env_vars=("KILOCODE_API_KEY",),
base_url="https://api.kilo.ai/api/gateway",
base_url_env_var="KILOCODE_BASE_URL",
is_aggregator=True,
),
"alibaba": ProviderDef(
id="alibaba",
name="Alibaba Cloud (DashScope)",
transport="openai_chat",
api_key_env_vars=("DASHSCOPE_API_KEY",),
base_url="https://dashscope-intl.aliyuncs.com/compatible-mode/v1",
base_url_env_var="DASHSCOPE_BASE_URL",
),
"huggingface": ProviderDef(
id="huggingface",
name="Hugging Face",
transport="openai_chat",
api_key_env_vars=("HF_TOKEN",),
base_url="https://router.huggingface.co/v1",
base_url_env_var="HF_BASE_URL",
is_aggregator=True,
),
}
# -- Aliases ------------------------------------------------------------------
# Merged from auth.py _PROVIDER_ALIASES and models.py _PROVIDER_ALIASES.
# Every human-friendly / legacy name maps to exactly one canonical id.
ALIASES: dict[str, str] = {
# openrouter
"openai": "openrouter",
# zai
"glm": "zai",
"z-ai": "zai",
"z.ai": "zai",
"zhipu": "zai",
# kimi-coding
"kimi": "kimi-coding",
"moonshot": "kimi-coding",
# minimax-cn
"minimax-china": "minimax-cn",
"minimax_cn": "minimax-cn",
# anthropic
"claude": "anthropic",
"claude-code": "anthropic",
# copilot
"github": "copilot",
"github-copilot": "copilot",
"github-models": "copilot",
"github-model": "copilot",
# copilot-acp
"github-copilot-acp": "copilot-acp",
"copilot-acp-agent": "copilot-acp",
# ai-gateway
"aigateway": "ai-gateway",
"vercel": "ai-gateway",
"vercel-ai-gateway": "ai-gateway",
# opencode-zen
"opencode": "opencode-zen",
"zen": "opencode-zen",
# opencode-go
"go": "opencode-go",
"opencode-go-sub": "opencode-go",
# kilocode
"kilo": "kilocode",
"kilo-code": "kilocode",
"kilo-gateway": "kilocode",
# deepseek
"deep-seek": "deepseek",
# alibaba
"dashscope": "alibaba",
"aliyun": "alibaba",
"qwen": "alibaba",
"alibaba-cloud": "alibaba",
# huggingface
"hf": "huggingface",
"hugging-face": "huggingface",
"huggingface-hub": "huggingface",
# Local server aliases -- route to the virtual "custom" provider
"lmstudio": "custom",
"lm-studio": "custom",
"lm_studio": "custom",
"ollama": "custom",
"vllm": "custom",
"llamacpp": "custom",
"llama.cpp": "custom",
"llama-cpp": "custom",
}
# -- Display labels -----------------------------------------------------------
# Short human-readable names for CLI menus and status output.
LABELS: dict[str, str] = {pid: pdef.name for pid, pdef in PROVIDERS.items()}
# Virtual "custom" provider is not in PROVIDERS but needs a label.
LABELS["custom"] = "Custom endpoint"
# -- Transport -> API mode mapping --------------------------------------------
TRANSPORT_TO_API_MODE: dict[str, str] = {
"openai_chat": "chat_completions",
"anthropic_messages": "anthropic_messages",
"codex_responses": "codex_responses",
}
# -- Helper functions ---------------------------------------------------------
def normalize_provider(name: str) -> str:
"""Resolve aliases and normalise casing to a canonical provider id.
Returns the canonical id string. Does *not* validate that the id
corresponds to a known provider -- callers should check ``PROVIDERS``
if they need that guarantee.
"""
key = name.strip().lower()
return ALIASES.get(key, key)
def get_provider(name: str) -> Optional[ProviderDef]:
"""Look up a provider by id or alias.
Returns the ``ProviderDef`` or ``None`` if the provider is unknown.
"""
canonical = normalize_provider(name)
return PROVIDERS.get(canonical)
def is_aggregator(provider: str) -> bool:
"""Return True when the provider is a multi-model aggregator.
Aggregators (OpenRouter, AI Gateway, etc.) expose models from many
upstream providers through a single API key and endpoint.
"""
pdef = get_provider(provider)
return pdef.is_aggregator if pdef else False
def determine_api_mode(provider: str, base_url: str = "") -> str:
"""Determine the API mode (wire protocol) for a provider/endpoint.
Resolution order:
1. Known provider -> transport -> ``TRANSPORT_TO_API_MODE``.
2. URL heuristics for unknown / custom providers.
3. Default: ``'chat_completions'``.
"""
# 1. Known provider lookup
pdef = get_provider(provider)
if pdef is not None:
return TRANSPORT_TO_API_MODE.get(pdef.transport, "chat_completions")
# 2. URL-based heuristics for custom / unknown providers
if base_url:
url_lower = base_url.rstrip("/").lower()
if url_lower.endswith("/anthropic") or "api.anthropic.com" in url_lower:
return "anthropic_messages"
if "api.openai.com" in url_lower:
return "codex_responses"
# 3. Default
return "chat_completions"
+123
View File
@@ -1268,6 +1268,129 @@ class AIAgent:
# Iterative summary from previous session must not bleed into new one (#2635)
self.context_compressor._previous_summary = None
def switch_model(self, new_model, new_provider, api_key='', base_url='', api_mode=''):
"""Switch the model/provider in-place for a live agent.
Called by the /model command handlers (CLI and gateway) after
``model_switch.switch_model()`` has resolved credentials and
validated the model. This method performs the actual runtime
swap: rebuilding clients, updating caching flags, and refreshing
the context compressor.
The implementation mirrors ``_try_activate_fallback()`` for the
client-swap logic but also updates ``_primary_runtime`` so the
change persists across turns (unlike fallback which is
turn-scoped).
"""
import logging
from hermes_cli.providers import determine_api_mode
# ── Determine api_mode if not provided ──
if not api_mode:
api_mode = determine_api_mode(new_provider, base_url)
old_model = self.model
old_provider = self.provider
# ── Swap core runtime fields ──
self.model = new_model
self.provider = new_provider
self.base_url = base_url or self.base_url
self.api_mode = api_mode
if api_key:
self.api_key = api_key
# ── Build new client ──
if api_mode == "anthropic_messages":
from agent.anthropic_adapter import (
build_anthropic_client,
resolve_anthropic_token,
_is_oauth_token,
)
effective_key = api_key or self.api_key or resolve_anthropic_token() or ""
self.api_key = effective_key
self._anthropic_api_key = effective_key
self._anthropic_base_url = base_url or getattr(self, "_anthropic_base_url", None)
self._anthropic_client = build_anthropic_client(
effective_key, self._anthropic_base_url,
)
self._is_anthropic_oauth = _is_oauth_token(effective_key)
self.client = None
self._client_kwargs = {}
else:
effective_key = api_key or self.api_key
effective_base = base_url or self.base_url
self._client_kwargs = {
"api_key": effective_key,
"base_url": effective_base,
}
self.client = self._create_openai_client(
dict(self._client_kwargs),
reason="switch_model",
shared=True,
)
# ── Re-evaluate prompt caching ──
is_native_anthropic = api_mode == "anthropic_messages"
self._use_prompt_caching = (
("openrouter" in (self.base_url or "").lower() and "claude" in new_model.lower())
or is_native_anthropic
)
# ── Update context compressor ──
if hasattr(self, "context_compressor") and self.context_compressor:
from agent.model_metadata import get_model_context_length
new_context_length = get_model_context_length(
self.model,
base_url=self.base_url,
api_key=self.api_key,
provider=self.provider,
)
self.context_compressor.model = self.model
self.context_compressor.base_url = self.base_url
self.context_compressor.api_key = self.api_key
self.context_compressor.provider = self.provider
self.context_compressor.context_length = new_context_length
self.context_compressor.threshold_tokens = int(
new_context_length * self.context_compressor.threshold_percent
)
# ── Invalidate cached system prompt so it rebuilds next turn ──
self._cached_system_prompt = None
# ── Update _primary_runtime so the change persists across turns ──
_cc = self.context_compressor if hasattr(self, "context_compressor") and self.context_compressor else None
self._primary_runtime = {
"model": self.model,
"provider": self.provider,
"base_url": self.base_url,
"api_mode": self.api_mode,
"api_key": getattr(self, "api_key", ""),
"client_kwargs": dict(self._client_kwargs),
"use_prompt_caching": self._use_prompt_caching,
"compressor_model": _cc.model if _cc else self.model,
"compressor_base_url": _cc.base_url if _cc else self.base_url,
"compressor_api_key": getattr(_cc, "api_key", "") if _cc else "",
"compressor_provider": _cc.provider if _cc else self.provider,
"compressor_context_length": _cc.context_length if _cc else 0,
"compressor_threshold_tokens": _cc.threshold_tokens if _cc else 0,
}
if api_mode == "anthropic_messages":
self._primary_runtime.update({
"anthropic_api_key": self._anthropic_api_key,
"anthropic_base_url": self._anthropic_base_url,
"is_anthropic_oauth": self._is_anthropic_oauth,
})
# ── Reset fallback state ──
self._fallback_activated = False
self._fallback_index = 0
logging.info(
"Model switched in-place: %s (%s) -> %s (%s)",
old_model, old_provider, new_model, new_provider,
)
def _safe_print(self, *args, **kwargs):
"""Print that silently handles broken pipes / closed stdout.