ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 11 Packages Projects Releases Wiki Activity

Merge branch 'simp/r3-22-D' into simp/r3-22

Browse Source
This commit is contained in:
Teknium
2026-09-02 22:38:27 -07:00
parent 1235b36d22 315da6f7f6
commit 4234a27ada
5 changed files with 950 additions and 1960 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/models_catalog_static.py
+359 -1014
View File
File diff suppressed because it is too large Load Diff
hermes_cli/models_local.py
+201 -333
View File
@@ -1,15 +1,15 @@
"""Local / self-hosted model servers.
Ollama (native ``/api/tags`` probe, request headers, base-url resolution), LM Studio
(``/api/v1/models``, load-on-demand), and Ollama Cloud (live + models.dev merged catalog with a
disk cache).
Split out of ``hermes_cli.models``; every moved name is re-imported there, so
``hermes_cli.models.<name>`` keeps resolving (and monkeypatching) as before.
(``/api/v1/models``, load-on-demand) and Ollama Cloud (live + models.dev merged catalog with a
disk cache). Split out of ``hermes_cli.models``, which re-imports every name; origin helpers are
looked up on ``hermes_cli.models`` at call time so ``patch("hermes_cli.models.<name>")`` mocks
keep intercepting.
"""
from __future__ import annotations
import hashlib
import http.client
import json
import logging
@@ -25,18 +25,14 @@ from hermes_cli.urllib_security import url_origin
# Log-record parity with the origin module.
logger = logging.getLogger("hermes_cli.models")
_OLLAMA_DEFAULT_PORT = 11434
def _root_for_ollama_native_api(base_url: str) -> str:
"""Convert an OpenAI-style Ollama base URL to the native API root."""
root = str(base_url or "").strip().rstrip("/")
if root.startswith(":"):
root = "http://127.0.0.1" + root
elif root and "://" not in root:
root = "http://" + root
for suffix in ("/api/tags", "/v1/models", "/api", "/v1"):
def _strip_suffixes(root: str, suffixes: tuple[str, ...]) -> str:
"""Drop the first matching path suffix (and any trailing slash) from *root*."""
for suffix in suffixes:
if root.endswith(suffix):
root = root[: -len(suffix)].rstrip("/")
break
return root[: -len(suffix)].rstrip("/")
return root
@@ -50,6 +46,16 @@ def _normalize_openai_base_url(base_url: Optional[str]) -> str:
return value
def _root_for_ollama_native_api(base_url: str) -> str:
"""Convert an OpenAI-style Ollama base URL to the native API root."""
root = str(base_url or "").strip().rstrip("/")
if root.startswith(":"):
root = "http://127.0.0.1" + root
elif root and "://" not in root:
root = "http://" + root
return _strip_suffixes(root, ("/api/tags", "/v1/models", "/api", "/v1"))
def _configured_ollama_base_url() -> str:
"""``providers.ollama.base_url`` (legacy keys ``api`` / ``url``), or ``""``."""
from hermes_cli.models import _get_provider_config_dict
@@ -58,14 +64,38 @@ def _configured_ollama_base_url() -> str:
return str(cfg.get("base_url") or cfg.get("api") or cfg.get("url") or "").strip()
def _get_ollama_base_url() -> str:
"""Resolve the local Ollama-compatible endpoint URL.
def _ollama_host_from_env(env_host: str) -> str:
"""Apply Ollama's own ``OLLAMA_HOST`` defaulting rules (port 11434, IPv6 bracketing)."""
port = _OLLAMA_DEFAULT_PORT
if env_host.startswith(":") and not env_host.startswith("::"):
return "127.0.0.1" + env_host
if env_host.startswith("[") and env_host.endswith("]"):
return f"{env_host}:{port}"
if "://" in env_host:
try:
parsed = urllib.parse.urlsplit(env_host)
if parsed.hostname and parsed.port is None:
hostname = parsed.hostname
if ":" in hostname and not hostname.startswith("["):
hostname = f"[{hostname}]"
userinfo = parsed.netloc.rsplit("@", 1)[0] + "@" if "@" in parsed.netloc else ""
return parsed._replace(netloc=f"{userinfo}{hostname}:{port}").geturl()
except ValueError:
pass
return env_host
if env_host.count(":") > 1 and not env_host.startswith("["):
return f"[{env_host}]:{port}"
if ":" not in env_host:
return f"{env_host}:{port}"
return env_host
Prefer explicit config under ``providers.ollama.base_url`` because this is how local Ollama-
compatible endpoints can be wired without changing the active model provider. Fall back to
active ``model.base_url`` only when the active provider is ollama/custom, then to Ollama's local
default.
"""
def _get_ollama_base_url() -> str:
"""Resolve the local Ollama-compatible endpoint URL: explicit ``providers.ollama.base_url``
(wires local endpoints without changing the active provider) → active ``model.base_url`` when
the active provider is ollama, or custom AND the endpoint actually serves ``/api/tags``
(otherwise the picker would probe an unrelated endpoint and hide the local catalog) →
``OLLAMA_HOST`` → Ollama's local default."""
from hermes_cli.models import _get_model_config_dict, should_use_ollama_native_catalog
configured = _configured_ollama_base_url()
if configured:
@@ -77,73 +107,51 @@ def _get_ollama_base_url() -> str:
if model_provider == "ollama" and model_base:
return model_base
if model_provider == "custom" and model_base:
# Only reuse the active bare custom endpoint when it is actually Ollama-compatible;
# otherwise the Ollama picker would probe an unrelated endpoint's /api/tags and hide the
# local Ollama catalog.
try:
if should_use_ollama_native_catalog("custom", model_base):
return model_base
except (OSError, RuntimeError, TypeError, ValueError):
pass
env_host = os.getenv("OLLAMA_HOST", "").strip()
if env_host:
if env_host.startswith(":") and not env_host.startswith("::"):
env_host = "127.0.0.1" + env_host
elif env_host.startswith("[") and env_host.endswith("]"):
env_host = f"{env_host}:11434"
elif "://" in env_host:
try:
parsed = urllib.parse.urlsplit(env_host)
if parsed.hostname and parsed.port is None:
hostname = parsed.hostname
if ":" in hostname and not hostname.startswith("["):
hostname = f"[{hostname}]"
userinfo = (
parsed.netloc.rsplit("@", 1)[0] + "@"
if "@" in parsed.netloc
else ""
)
env_host = parsed._replace(
netloc=f"{userinfo}{hostname}:11434"
).geturl()
except ValueError:
pass
elif env_host.count(":") > 1 and not env_host.startswith("["):
env_host = f"[{env_host}]:11434"
elif ":" not in env_host:
env_host = f"{env_host}:11434"
return env_host
return "http://localhost:11434"
return _ollama_host_from_env(env_host) if env_host else "http://localhost:11434"
def _api_key_from_provider_config(entry: dict, *env_keys: str) -> str:
"""``api_key`` from a provider config block, else the env var named by the first set *env_keys*."""
api_key = str(entry.get("api_key") or "").strip()
if api_key:
return api_key
key_env = str(next((entry.get(k) for k in env_keys if entry.get(k)), "") or "").strip()
return os.getenv(key_env, "").strip() if key_env else ""
def _drop_authorization(headers: dict[str, str]) -> None:
for key in tuple(headers):
if key.lower() == "authorization":
del headers[key]
def _get_ollama_request_headers() -> dict[str, str]:
"""Return configured headers and credentials for native Ollama requests."""
from hermes_cli.models import _get_provider_config_dict
entry = _get_provider_config_dict("ollama")
raw = entry.get("extra_headers")
try:
from hermes_cli.config import normalize_extra_headers
result = normalize_extra_headers(raw)
result = normalize_extra_headers(entry.get("extra_headers"))
except (ImportError, OSError, RuntimeError, TypeError, ValueError):
result = {}
api_key = str(entry.get("api_key") or "").strip()
if not api_key:
key_env = str(entry.get("key_env") or entry.get("api_key_env") or "").strip()
api_key = os.getenv(key_env, "").strip() if key_env else ""
api_key = _api_key_from_provider_config(entry, "key_env", "api_key_env")
if api_key and not any(key.lower() == "authorization" for key in result):
result["Authorization"] = f"Bearer {api_key}"
return result
def _get_ollama_native_headers(
base_url: Optional[str],
*,
api_key: Optional[str] = None,
) -> dict[str, str]:
"""Resolve Ollama credentials and headers for one endpoint origin."""
def _get_ollama_native_headers(base_url: Optional[str], *, api_key: Optional[str] = None) -> dict[str, str]:
"""Ollama credentials and headers for one endpoint origin. Configured headers apply only when
*base_url* shares the configured Ollama root; an explicit *api_key* replaces any configured
Authorization variant rather than inheriting it."""
from hermes_cli.models import _get_ollama_request_headers
configured_base = _configured_ollama_base_url()
explicit_key = str(api_key or "").strip()
@@ -152,11 +160,7 @@ def _get_ollama_native_headers(
return {}
headers = _get_ollama_request_headers() if configured_matches else {}
if explicit_key:
# A provider-specific key must not inherit any configured Authorization
# variant from the Ollama origin when both share a native root.
for key in tuple(headers):
if key.lower() == "authorization":
del headers[key]
_drop_authorization(headers)
headers["Authorization"] = f"Bearer {explicit_key}"
return headers
@@ -168,8 +172,6 @@ _OLLAMA_LOCAL_MODELS_CACHE: dict[str, tuple[tuple[str, ...], float]] = {}
_OLLAMA_LOCAL_PROBE_FAILURE_CACHE: dict[str, float] = {}
_OLLAMA_LOCAL_PROBE_REACHABLE: dict[str, bool] = {}
_OLLAMA_LOCAL_PROBE_FAILURE_TTL: int = 30
_OLLAMA_LOCAL_CACHE_MAX_ENTRIES: int = 256
@@ -183,27 +185,16 @@ def _evict_related_ollama_cache_entries(key: str) -> None:
def _remember_ollama_cache(cache: dict[str, Any], key: str, value: Any) -> None:
if key not in cache and len(cache) >= _OLLAMA_LOCAL_CACHE_MAX_ENTRIES:
oldest_key = next(iter(cache))
_evict_related_ollama_cache_entries(
oldest_key.split("|timeout:", 1)[0]
)
_evict_related_ollama_cache_entries(next(iter(cache)).split("|timeout:", 1)[0])
cache[key] = value
def _ollama_probe_cache_key(root: str, headers: Optional[dict[str, str]]) -> str:
cache_key = root
if headers:
import hashlib
normalized_headers = sorted(
(str(key).lower(), str(value)) for key, value in headers.items()
)
header_blob = json.dumps(
normalized_headers, ensure_ascii=False, separators=(",", ":")
).encode("utf-8", errors="replace")
header_fingerprint = hashlib.blake2b(header_blob, digest_size=8).hexdigest()
cache_key = f"{root}|headers:{header_fingerprint}"
return cache_key
if not headers:
return root
normalized_headers = sorted((str(key).lower(), str(value)) for key, value in headers.items())
header_blob = json.dumps(normalized_headers, ensure_ascii=False, separators=(",", ":")).encode("utf-8", errors="replace")
return f"{root}|headers:{hashlib.blake2b(header_blob, digest_size=8).hexdigest()}"
def _parse_ollama_tags(payload: Any) -> Optional[list[str]]:
@@ -212,13 +203,11 @@ def _parse_ollama_tags(payload: Any) -> Optional[list[str]]:
if not isinstance(raw_models, list):
return None
models: list[str] = []
seen: set[str] = set()
for item in raw_models:
if not isinstance(item, dict):
return None
model_id = str(item.get("model") or item.get("name") or "").strip()
if model_id and model_id not in seen:
seen.add(model_id)
if model_id and model_id not in models:
models.append(model_id)
if raw_models and not models:
return None
@@ -230,13 +219,9 @@ def probe_ollama_local_models(
timeout: float = 2.0,
headers: Optional[dict[str, str]] = None,
) -> Optional[list[str]]:
"""Probe local Ollama-compatible models from native ``/api/tags``.
Returns ``None`` when the endpoint cannot be reached or returns malformed data, and a list
(possibly empty) when ``/api/tags`` was reachable. Stock Ollama exposes its authoritative local
model catalog at ``/api/tags``; OpenAI-compatible ``/v1/models`` is not required for local
Ollama servers.
"""
"""Probe local Ollama-compatible models from native ``/api/tags`` (Ollama's authoritative local
catalog; ``/v1/models`` is not required for local servers). ``None`` when the endpoint cannot be
reached or returns malformed data; a list (possibly empty) when it was reachable."""
from hermes_cli.models import _HERMES_USER_AGENT, _get_ollama_base_url, _urlopen_model_catalog_request
root = _root_for_ollama_native_api(base_url or _get_ollama_base_url())
if not root:
@@ -244,33 +229,24 @@ def probe_ollama_local_models(
cache_key = _ollama_probe_cache_key(root, headers)
failure_key = f"{cache_key}|timeout:{float(timeout):.3f}"
cached = _OLLAMA_LOCAL_MODELS_CACHE.get(cache_key)
if cached is not None:
cached_models, cached_at = cached
if time.monotonic() - cached_at < _OLLAMA_LOCAL_MODELS_CACHE_TTL:
return list(cached_models)
if cached is not None and time.monotonic() - cached[1] < _OLLAMA_LOCAL_MODELS_CACHE_TTL:
return list(cached[0])
failed_at = _OLLAMA_LOCAL_PROBE_FAILURE_CACHE.get(failure_key)
if failed_at is not None:
if time.monotonic() - failed_at < _OLLAMA_LOCAL_PROBE_FAILURE_TTL:
return None
_OLLAMA_LOCAL_PROBE_FAILURE_CACHE.pop(failure_key, None)
def _unreachable() -> None:
_remember_ollama_cache(_OLLAMA_LOCAL_PROBE_REACHABLE, cache_key, False)
_remember_ollama_cache(_OLLAMA_LOCAL_PROBE_FAILURE_CACHE, failure_key, time.monotonic())
try:
request_headers = {"User-Agent": _HERMES_USER_AGENT, **(headers or {})}
req = urllib.request.Request(root.rstrip("/") + "/api/tags", headers=request_headers)
with _urlopen_model_catalog_request(req, timeout=timeout) as resp:
payload = json.loads(resp.read().decode())
models = _parse_ollama_tags(json.loads(resp.read().decode()))
except (ValueError, OSError, TimeoutError, http.client.HTTPException, urllib.error.URLError,
json.JSONDecodeError, UnicodeDecodeError):
_unreachable()
return None
models = _parse_ollama_tags(payload)
models = None
if models is None:
_unreachable()
_remember_ollama_cache(_OLLAMA_LOCAL_PROBE_REACHABLE, cache_key, False)
_remember_ollama_cache(_OLLAMA_LOCAL_PROBE_FAILURE_CACHE, failure_key, time.monotonic())
return None
_remember_ollama_cache(_OLLAMA_LOCAL_PROBE_REACHABLE, cache_key, True)
_OLLAMA_LOCAL_PROBE_FAILURE_CACHE.pop(failure_key, None)
@@ -305,19 +281,20 @@ def _same_ollama_native_root(left: str, right: str) -> bool:
return False
_NEVER_OLLAMA_PROVIDERS = frozenset({"openrouter", "nous", "anthropic", "openai", "openai-codex", "gemini", "ollama-cloud"})
_LOCAL_LIKE_PROVIDERS = frozenset({"", "custom", "local", "llamacpp", "llama.cpp", "llama-cpp", "vllm"})
def should_use_ollama_native_catalog(
provider: Optional[str],
base_url: Optional[str],
headers: Optional[dict[str, str]] = None,
) -> bool:
"""Return True when model discovery should use local Ollama ``/api/tags``.
Bare ``ollama`` is normalized to ``custom`` elsewhere so runtime paths share the OpenAI-
compatible client, but local Ollama's authoritative model list is ``/api/tags``. Use it when
the caller asked for Ollama explicitly, the base URL matches ``providers.ollama.base_url``,
or an ambiguous custom URL on Ollama's default port actually serves ``/api/tags``; other
custom endpoints keep the ``/models`` probe.
"""
"""True when model discovery should use local Ollama ``/api/tags``: the caller asked for Ollama
explicitly, the base URL matches ``providers.ollama.base_url``, or an ambiguous custom URL on
Ollama's default port actually serves ``/api/tags``. (Bare ``ollama`` is normalized to
``custom`` elsewhere so runtime paths share the OpenAI client, but ``/api/tags`` is the
authoritative local list; other custom endpoints keep the ``/models`` probe.)"""
from hermes_cli.models import probe_ollama_local_models
requested = str(provider or "").strip().lower()
root = _root_for_ollama_native_api(base_url or "")
@@ -329,7 +306,7 @@ def should_use_ollama_native_catalog(
except ValueError:
pass
if requested in {"openrouter", "nous", "anthropic", "openai", "openai-codex", "gemini", "ollama-cloud"}:
if requested in _NEVER_OLLAMA_PROVIDERS:
return False
configured_base = _configured_ollama_base_url()
@@ -346,16 +323,14 @@ def should_use_ollama_native_catalog(
if not root:
return False
local_like_providers = {"", "custom", "local", "llamacpp", "llama.cpp", "llama-cpp", "vllm"}
if requested not in local_like_providers and not requested.startswith("custom:"):
if requested not in _LOCAL_LIKE_PROVIDERS and not requested.startswith("custom:"):
return False
if requested == "custom:ollama" or requested.endswith("-ollama"):
return True
try:
parsed = urllib.parse.urlparse(root)
if parsed.port != 11434:
if urllib.parse.urlparse(root).port != _OLLAMA_DEFAULT_PORT:
return False
except ValueError:
return False
@@ -365,7 +340,8 @@ def should_use_ollama_native_catalog(
def _ollama_local_catalog(force_refresh: bool) -> list[str]:
"""Catalog for the raw ``ollama`` provider: native ``/api/tags`` when the endpoint is a real
Ollama server, else the OpenAI-style ``/v1/models`` of the configured gateway."""
Ollama server, else the OpenAI-style ``/v1/models`` of the configured gateway (incl. Ollama
Cloud)."""
from hermes_cli.models import _get_ollama_base_url, _get_ollama_native_headers, _get_provider_config_dict, fetch_api_models, fetch_ollama_local_models, should_use_ollama_native_catalog
if force_refresh:
_OLLAMA_LOCAL_MODELS_CACHE.clear()
@@ -374,48 +350,28 @@ def _ollama_local_catalog(force_refresh: bool) -> list[str]:
base_url = _get_ollama_base_url()
headers = _get_ollama_native_headers(base_url)
if should_use_ollama_native_catalog("ollama", base_url, headers=headers):
if headers:
native_models = fetch_ollama_local_models(base_url, headers=headers)
else:
native_models = fetch_ollama_local_models(base_url)
native_models = fetch_ollama_local_models(base_url, headers=headers) if headers else fetch_ollama_local_models(base_url)
native_key = _ollama_probe_cache_key(_root_for_ollama_native_api(base_url), headers or None)
if native_models or _OLLAMA_LOCAL_PROBE_REACHABLE.get(native_key) is True:
return native_models or []
# Non-native Ollama-compatible endpoints (incl. Ollama Cloud) and gateways exposing only
# OpenAI-style /v1/models.
config = _get_provider_config_dict("ollama")
fallback_key = str(config.get("api_key") or "").strip()
if not fallback_key:
key_env = str(config.get("key_env") or "").strip()
fallback_key = os.getenv(key_env, "").strip() if key_env else ""
fallback_key = _api_key_from_provider_config(config, "key_env")
fallback_base = _normalize_openai_base_url(config.get("base_url") or base_url)
fallback_headers = _get_ollama_native_headers(fallback_base, api_key=fallback_key)
return fetch_api_models(fallback_key, fallback_base, headers=fallback_headers or None) or []
def _lmstudio_server_root(base_url: Optional[str]) -> Optional[str]:
"""Return the LM Studio server root for native ``/api/v1`` endpoints.
Users commonly copy either the OpenAI-compatible runtime URL (``.../v1``) or the native API
prefix (``.../api`` / ``.../api/v1``). Native probes append ``/api/v1/...`` themselves, so
normalize all accepted forms back to the bare server root to avoid ``/api/api/v1`` requests.
"""
root = (base_url or "").strip().rstrip("/")
for suffix in ("/api/v1", "/api", "/v1"):
if root.endswith(suffix):
root = root[: -len(suffix)].rstrip("/")
break
return root or None
"""LM Studio server root: users paste the OpenAI runtime URL (``.../v1``) or the native prefix
(``.../api``, ``.../api/v1``); native probes append ``/api/v1/...`` themselves."""
return _strip_suffixes((base_url or "").strip().rstrip("/"), ("/api/v1", "/api", "/v1")) or None
def _lmstudio_request_headers(api_key: Optional[str] = None) -> dict:
"""Build HTTP headers for LM Studio native API requests."""
"""HTTP headers for LM Studio native API requests."""
from hermes_cli.models import _HERMES_USER_AGENT
headers = {"User-Agent": _HERMES_USER_AGENT}
token = str(api_key or "").strip()
if token:
headers["Authorization"] = f"Bearer {token}"
return headers
return {"User-Agent": _HERMES_USER_AGENT, **({"Authorization": f"Bearer {token}"} if token else {})}
def _lmstudio_fetch_raw_models(
@@ -423,14 +379,14 @@ def _lmstudio_fetch_raw_models(
base_url: Optional[str] = None,
timeout: float = 5.0,
) -> Optional[list[dict]]:
"""Fetch the raw model list from LM Studio's ``/api/v1/models``."""
"""Raw model list from LM Studio's ``/api/v1/models``; None on network errors / malformed
payloads; raises ``AuthError`` on HTTP 401/403."""
from hermes_cli.models import _urlopen_model_catalog_request
server_root = _lmstudio_server_root(base_url)
if not server_root:
return None
headers = _lmstudio_request_headers(api_key)
request = urllib.request.Request(server_root + "/api/v1/models", headers=headers)
request = urllib.request.Request(server_root + "/api/v1/models", headers=_lmstudio_request_headers(api_key))
try:
with _urlopen_model_catalog_request(request, timeout=timeout) as resp:
payload = json.loads(resp.read().decode())
@@ -455,18 +411,30 @@ def _lmstudio_fetch_raw_models(
return raw_models
def _lmstudio_raw_models_or_none(api_key, base_url, timeout) -> Optional[list[dict]]:
"""``_lmstudio_fetch_raw_models`` with every failure (incl. AuthError) collapsed to None."""
from hermes_cli.models import _lmstudio_fetch_raw_models
try:
return _lmstudio_fetch_raw_models(api_key=api_key, base_url=base_url, timeout=timeout)
except Exception:
return None
def _lmstudio_entry_for(raw_models: list, model: str) -> Optional[dict]:
for raw in raw_models:
if isinstance(raw, dict) and (raw.get("key") == model or raw.get("id") == model):
return raw
return None
def probe_lmstudio_models(
api_key: Optional[str] = None,
base_url: Optional[str] = None,
timeout: float = 5.0,
) -> Optional[list[str]]:
"""Probe LM Studio's model listing.
Returns chat-capable model keys, including a valid empty list when the server is reachable
but has no non-embedding models; returns ``None`` on network errors, malformed responses, or
bad base URLs. Raises ``AuthError`` on HTTP 401/403 so token issues surface separately from
reachability.
"""
"""Chat-capable LM Studio model keys — a valid empty list when the server is reachable but has
no non-embedding models; ``None`` on network errors, malformed responses, or bad base URLs.
Raises ``AuthError`` on HTTP 401/403 so token issues surface separately from reachability."""
from hermes_cli.models import _lmstudio_fetch_raw_models
raw_models = _lmstudio_fetch_raw_models(api_key=api_key, base_url=base_url, timeout=timeout)
if raw_models is None:
@@ -474,9 +442,7 @@ def probe_lmstudio_models(
keys: list[str] = []
for raw in raw_models:
if not isinstance(raw, dict):
continue
if str(raw.get("type") or "").strip().lower() == "embedding":
if not isinstance(raw, dict) or str(raw.get("type") or "").strip().lower() == "embedding":
continue
key = str(raw.get("key") or raw.get("id") or "").strip()
if key and key not in keys:
@@ -489,16 +455,10 @@ def fetch_lmstudio_models(
base_url: Optional[str] = None,
timeout: float = 5.0,
) -> list[str]:
"""Fetch LM Studio chat-capable model keys from native ``/api/v1/models``.
Embedding models are filtered out; network errors, malformed responses, and bad base URLs
yield an empty list. Raises ``AuthError`` on HTTP 401/403 so callers can distinguish a
missing or wrong ``LM_API_KEY`` from an unreachable server — the most common LM Studio
support case.
"""
"""LM Studio chat-capable model keys; ``[]`` when unreachable/malformed. Raises ``AuthError`` on
HTTP 401/403 so callers can tell a wrong ``LM_API_KEY`` from an unreachable server."""
from hermes_cli.models import probe_lmstudio_models
models = probe_lmstudio_models(api_key=api_key, base_url=base_url, timeout=timeout)
return models or []
return probe_lmstudio_models(api_key=api_key, base_url=base_url, timeout=timeout) or []
class LMStudioLoadResult(NamedTuple):
@@ -509,6 +469,25 @@ class LMStudioLoadResult(NamedTuple):
rejected: bool = False
def _positive_int(value: Any) -> Optional[int]:
if isinstance(value, int) and not isinstance(value, bool) and value > 0:
return value
return None
def _lmstudio_loaded_context(entry: Optional[dict]) -> Optional[int]:
"""First positive ``loaded_instances[*].config.context_length`` of a model entry."""
instances = entry.get("loaded_instances") if entry is not None else None
if not isinstance(instances, list):
return None
for instance in instances:
config = instance.get("config") if isinstance(instance, dict) else None
parsed = _positive_int(config.get("context_length") if isinstance(config, dict) else None)
if parsed is not None:
return parsed
return None
def ensure_lmstudio_model_loaded(
model: str,
base_url: Optional[str],
@@ -521,42 +500,13 @@ def ensure_lmstudio_model_loaded(
"""Ensure ``model`` is loaded and return verified runtime context.
Existing loaded-instance context is authoritative. Cold loads omit ``context_length`` unless the
caller supplied an explicit override; the returned context must come from LM Studio's echoed or
refreshed state.
"""
from hermes_cli.models import _lmstudio_fetch_raw_models, _urlopen_model_catalog_request
caller supplied an explicit override; the returned context comes from LM Studio's echoed or
refreshed state."""
from hermes_cli.models import _urlopen_model_catalog_request
def _result(
context_length: Optional[int],
*,
load_attempted: bool = False,
rejected: bool = False,
) -> Optional[int] | LMStudioLoadResult:
value = LMStudioLoadResult(context_length, load_attempted, rejected)
return value if return_load_result else context_length
def _positive_int(value: Any) -> Optional[int]:
if isinstance(value, int) and not isinstance(value, bool) and value > 0:
return value
return None
def _loaded_context(entry: dict) -> Optional[int]:
instances = entry.get("loaded_instances")
if not isinstance(instances, list):
return None
for instance in instances:
config = instance.get("config") if isinstance(instance, dict) else None
context = config.get("context_length") if isinstance(config, dict) else None
parsed = _positive_int(context)
if parsed is not None:
return parsed
return None
def _find_entry(raw_models: list[dict]) -> Optional[dict]:
for raw in raw_models:
if isinstance(raw, dict) and (raw.get("key") == model or raw.get("id") == model):
return raw
return None
def _result(context_length: Optional[int], *, load_attempted: bool = False, rejected: bool = False):
result = LMStudioLoadResult(context_length, load_attempted, rejected)
return result if return_load_result else context_length
server_root = _lmstudio_server_root(base_url)
if not server_root:
@@ -565,17 +515,7 @@ def ensure_lmstudio_model_loaded(
explicit_context = _positive_int(target_context_length)
if target_context_length is not None and explicit_context is None:
return _result(None)
headers = _lmstudio_request_headers(api_key)
try:
raw_models = _lmstudio_fetch_raw_models(api_key=api_key, base_url=base_url, timeout=10)
except Exception:
raw_models = None
if raw_models is None:
return _result(None)
target_entry = _find_entry(raw_models)
target_entry = _lmstudio_entry_for(_lmstudio_raw_models_or_none(api_key, base_url, 10) or [], model)
if target_entry is None:
return _result(None)
@@ -583,7 +523,7 @@ def ensure_lmstudio_model_loaded(
if explicit_context is not None and max_ctx is not None and explicit_context > max_ctx:
return _result(None, rejected=True)
current_context = _loaded_context(target_entry)
current_context = _lmstudio_loaded_context(target_entry)
if current_context is not None:
return _result(current_context)
@@ -594,14 +534,11 @@ def ensure_lmstudio_model_loaded(
load_payload: dict[str, Any] = {"model": model, "echo_load_config": True}
if explicit_context is not None:
load_payload["context_length"] = explicit_context
body = json.dumps(load_payload).encode()
load_headers = dict(headers)
load_headers["Content-Type"] = "application/json"
try:
load_request = urllib.request.Request(
server_root + "/api/v1/models/load",
data=body,
headers=load_headers,
data=json.dumps(load_payload).encode(),
headers={**_lmstudio_request_headers(api_key), "Content-Type": "application/json"},
method="POST",
)
with _urlopen_model_catalog_request(load_request, timeout=timeout) as resp:
@@ -614,23 +551,14 @@ def ensure_lmstudio_model_loaded(
except Exception:
response_payload = None
load_config = response_payload.get("load_config") if isinstance(response_payload, dict) else None
applied_context = (
_positive_int(load_config.get("context_length"))
if isinstance(load_config, dict)
else None
)
applied_context = _positive_int(load_config.get("context_length")) if isinstance(load_config, dict) else None
if applied_context is not None:
return _result(applied_context, load_attempted=True)
try:
refreshed_models = _lmstudio_fetch_raw_models(api_key=api_key, base_url=base_url, timeout=10)
except Exception:
refreshed_models = None
refreshed_models = _lmstudio_raw_models_or_none(api_key, base_url, 10)
if refreshed_models is None:
return _result(None, load_attempted=True)
refreshed_entry = _find_entry(refreshed_models)
refreshed_context = _loaded_context(refreshed_entry) if refreshed_entry is not None else None
return _result(refreshed_context, load_attempted=True)
return _result(_lmstudio_loaded_context(_lmstudio_entry_for(refreshed_models, model)), load_attempted=True)
def lmstudio_model_reasoning_options(
@@ -639,31 +567,16 @@ def lmstudio_model_reasoning_options(
api_key: Optional[str] = None,
timeout: float = 5.0,
) -> list[str]:
"""Return the reasoning ``allowed_options`` LM Studio publishes for ``model``.
Reads ``capabilities.reasoning.allowed_options`` from ``/api/v1/models``; returns ``[]``
when the model is unknown, the endpoint is unreachable, or no reasoning capability is
declared.
"""
from hermes_cli.models import _lmstudio_fetch_raw_models
try:
raw_models = _lmstudio_fetch_raw_models(api_key=api_key, base_url=base_url, timeout=timeout)
except Exception:
raw_models = None
if not raw_models:
return []
for raw in raw_models:
if not isinstance(raw, dict):
continue
if raw.get("key") != model and raw.get("id") != model:
continue
caps = raw.get("capabilities")
reasoning = caps.get("reasoning") if isinstance(caps, dict) else None
opts = reasoning.get("allowed_options") if isinstance(reasoning, dict) else None
if isinstance(opts, list):
return [str(o).strip().lower() for o in opts if isinstance(o, str)]
"""Reasoning ``allowed_options`` LM Studio publishes for ``model`` under
``capabilities.reasoning`` in ``/api/v1/models``; ``[]`` when unknown, unreachable, or absent."""
raw = _lmstudio_entry_for(_lmstudio_raw_models_or_none(api_key, base_url, timeout) or [], model)
if raw is None:
return []
caps = raw.get("capabilities")
reasoning = caps.get("reasoning") if isinstance(caps, dict) else None
opts = reasoning.get("allowed_options") if isinstance(reasoning, dict) else None
if isinstance(opts, list):
return [str(o).strip().lower() for o in opts if isinstance(o, str)]
return []
@@ -673,30 +586,21 @@ def ollama_model_supports_thinking(
api_key: Optional[str] = None,
timeout: float = 5.0,
) -> Optional[bool]:
"""Return True if an Ollama (Cloud or local) model advertises ``thinking``.
Probes native ``/api/show`` and checks ``capabilities`` — the authoritative source, since
the OpenAI-compat ``/v1/models`` endpoint omits it. Tri-state: True when ``thinking`` is
declared, False when the probe succeeded without it, None when the probe failed so the
caller picks the fallback (treated as "don't emit").
"""
"""Tri-state: True if an Ollama (Cloud or local) model advertises ``thinking`` in native
``/api/show`` ``capabilities`` (authoritative; OpenAI-compat ``/v1/models`` omits it), False
when the probe succeeded without it, None when it failed (caller treats as "don't emit")."""
import httpx
server_url = (base_url or "").strip().rstrip("/")
if server_url.endswith("/v1"):
server_url = server_url[:-3]
if not server_url:
return None
bare_model = _strip_ollama_cloud_suffix((model or "").strip())
if not bare_model:
if not server_url or not bare_model:
return None
token = str(api_key or "").strip()
headers = {"Authorization": f"Bearer {token}"} if token else {}
try:
with httpx.Client(timeout=timeout, headers=headers) as client:
with httpx.Client(timeout=timeout, headers={"Authorization": f"Bearer {token}"} if token else {}) as client:
resp = client.post(f"{server_url}/api/show", json={"name": bare_model})
if resp.status_code != 200:
return None
@@ -712,12 +616,8 @@ _OLLAMA_CLOUD_CACHE_TTL = 3600 # 1 hour
def _strip_ollama_cloud_suffix(model_id: str) -> str:
"""Strip :cloud / -cloud suffixes that models.dev appends to Ollama Cloud IDs.
The live API uses clean IDs (e.g. 'kimi-k2.6') while models.dev sometimes returns them as
'kimi-k2.6:cloud'. Normalising before the dedup merge prevents duplicate entries in the merged
model list.
"""
"""Strip the ``:cloud`` / ``-cloud`` suffix models.dev appends to Ollama Cloud IDs (the live
API uses bare ids), so the dedup merge does not produce duplicates."""
for suffix in (":cloud", "-cloud"):
if model_id.endswith(suffix):
return model_id[: -len(suffix)]
@@ -725,7 +625,6 @@ def _strip_ollama_cloud_suffix(model_id: str) -> str:
def _ollama_cloud_cache_path() -> Path:
"""Return the path for the Ollama Cloud model cache."""
from hermes_constants import get_hermes_home
return get_hermes_home() / "ollama_cloud_models_cache.json"
@@ -736,9 +635,7 @@ def _load_ollama_cloud_cache(*, ignore_ttl: bool = False) -> Optional[dict]:
try:
data = _read_json_cache(_ollama_cloud_cache_path())
if data is None:
return None
models = data.get("models")
models = data.get("models") if data is not None else None
if not (isinstance(models, list) and models):
return None
if not ignore_ttl and (time.time() - data.get("cached_at", 0)) > _OLLAMA_CLOUD_CACHE_TTL:
@@ -764,34 +661,18 @@ def fetch_ollama_cloud_models(
*,
force_refresh: bool = False,
) -> list[str]:
"""Fetch Ollama Cloud models by merging live API + models.dev, with disk cache.
Resolution order: 1. Disk cache (if fresh, < 1 hour, and not force_refresh) 2. Live
``/v1/models`` endpoint (primary — freshest source) 3. models.dev registry (secondary — fills
gaps for unlisted models) 4. Merge: live models first, then models.dev additions (deduped)
Returns a list of model IDs (never None — empty list on total failure).
"""
"""Ollama Cloud models: fresh disk cache (< 1h, unless force_refresh) → live ``/v1/models``
(freshest) merged with models.dev additions (deduped, live first) → stale cache → ``[]``.
Never None."""
from hermes_cli.models import fetch_api_models
# 1. Check disk cache
if not force_refresh:
cached = _load_ollama_cloud_cache()
if cached is not None:
return cached["models"]
# 2. Live API probe
if not api_key:
api_key = os.getenv("OLLAMA_API_KEY", "")
if not base_url:
base_url = os.getenv("OLLAMA_BASE_URL", "") or "https://ollama.com/v1"
live_models: list[str] = []
if api_key:
result = fetch_api_models(api_key, base_url, timeout=8.0)
if result:
live_models = result
# 3. models.dev registry
api_key = api_key or os.getenv("OLLAMA_API_KEY", "")
base_url = base_url or os.getenv("OLLAMA_BASE_URL", "") or "https://ollama.com/v1"
live_models = (fetch_api_models(api_key, base_url, timeout=8.0) or []) if api_key else []
mdev_models: list[str] = []
try:
from agent.models_dev import list_agentic_models
@@ -799,26 +680,13 @@ def fetch_ollama_cloud_models(
except Exception:
pass
# 4. Merge: live first, then models.dev additions (deduped, order-preserving)
if live_models or mdev_models:
seen: set[str] = set()
merged: list[str] = []
for m in live_models:
if m and m not in seen:
seen.add(m)
merged.append(m)
for m in mdev_models:
normalized = _strip_ollama_cloud_suffix(m)
if normalized and normalized not in seen:
seen.add(normalized)
merged.append(normalized)
if merged:
_save_ollama_cloud_cache(merged)
return merged
merged: list[str] = []
for m in [*live_models, *(_strip_ollama_cloud_suffix(m) for m in mdev_models)]:
if m and m not in merged:
merged.append(m)
if merged:
_save_ollama_cloud_cache(merged)
return merged
# Total failure — return stale cache if available (ignore TTL)
stale = _load_ollama_cloud_cache(ignore_ttl=True)
if stale is not None:
return stale["models"]
return []
return stale["models"] if stale is not None else []
hermes_cli/models_pricing.py
+177 -302
View File
@@ -2,10 +2,9 @@
OpenRouter-compatible ``/v1/models`` pricing fetch with a per-endpoint/per-credential cache,
Nous Portal sale chrome and org-policy filtering, and the Vercel AI Gateway / Novita / Fireworks /
DeepInfra pricing adapters.
Split out of ``hermes_cli.models``; every moved name is re-imported there, so
``hermes_cli.models.<name>`` keeps resolving (and monkeypatching) as before.
DeepInfra pricing adapters. Split out of ``hermes_cli.models``, which re-imports every name;
origin helpers and the cache dicts are looked up on ``hermes_cli.models`` at call time so
``patch("hermes_cli.models.<name>")`` mocks keep intercepting.
"""
from __future__ import annotations
@@ -21,17 +20,12 @@ from hermes_cli.models_reasoning_caps import _seed_reasoning_caps
# Cache: maps model_id → {"prompt": str, "completion": str} per endpoint
_pricing_cache: dict[str, dict[str, dict[str, str]]] = {}
# A failed fetch caches its empty result too, so an unreachable endpoint isn't
# re-dialed on every call — but only until this deadline. Cached forever, one
# bad moment (a blip during startup, a key that hadn't been written yet) turns
# into no live model discovery for the life of the process, and the processes
# that read this most are the ones that run for weeks: the gateway, the desktop
# backend. Every caller falls back to a curated list meanwhile, so the cost of
# the stale entry is silent and invisible.
# A failed fetch caches its empty result too, so an unreachable endpoint isn't re-dialed on every
# call — but only until this deadline. Cached forever, one blip at startup would mean no live model
# discovery for the life of a process that runs for weeks (gateway, desktop backend), silently:
# every caller falls back to a curated list meanwhile.
_FAILED_CATALOG_TTL_SECONDS = 120.0
_pricing_cache_retry_after: dict[str, float] = {}
@@ -54,23 +48,17 @@ def _cache_catalog(
result: dict[str, dict[str, Any]],
ttl_seconds: Optional[float] = None,
) -> dict[str, dict[str, Any]]:
"""Cache a catalog result, giving an empty one an expiry.
*ttl_seconds* expires a non-empty result too. Only a catalog whose contents depend on server-
side state the client cannot observe needs it — an org's model policy can change while a long-
lived process holds the entry.
"""
"""Cache a catalog result, giving an empty one an expiry. *ttl_seconds* expires a non-empty
result too — only for a catalog whose contents depend on server-side state the client cannot
observe (an org's model policy can change while a long-lived process holds the entry)."""
from hermes_cli.models import _pricing_cache, _pricing_cache_retry_after
_pricing_cache[cache_key] = result
if result:
if ttl_seconds:
_pricing_cache_retry_after[cache_key] = time.monotonic() + ttl_seconds
else:
_pricing_cache_retry_after.pop(cache_key, None)
if not result:
_pricing_cache_retry_after[cache_key] = time.monotonic() + _FAILED_CATALOG_TTL_SECONDS
elif ttl_seconds:
_pricing_cache_retry_after[cache_key] = time.monotonic() + ttl_seconds
else:
_pricing_cache_retry_after[cache_key] = (
time.monotonic() + _FAILED_CATALOG_TTL_SECONDS
)
_pricing_cache_retry_after.pop(cache_key, None)
return result
@@ -79,12 +67,9 @@ _PRICING_AUTH_KEY_PREFIX = "\x00auth:"
def _pricing_auth_fingerprint(api_key: str | None) -> str:
"""Key suffix identifying the credential a catalog was read with.
A governed endpoint answers each token with the catalog its org may reach, so two credentials
cannot share an entry. blake2b for cache-key fingerprinting only, same rationale as
:func:`_custom_endpoint_fingerprint`.
"""
"""Cache-key suffix identifying the credential a catalog was read with: a governed endpoint
answers each token with the catalog its org may reach, so two credentials cannot share an
entry. blake2b for fingerprinting only (same rationale as ``_custom_endpoint_fingerprint``)."""
if not api_key:
return ""
import hashlib
@@ -94,17 +79,11 @@ def _pricing_auth_fingerprint(api_key: str | None) -> str:
def peek_cached_pricing(base_url: str) -> dict[str, dict[str, Any]]:
"""Pricing already cached for *base_url*, or ``{}``. Never fetches.
Accepts a ``/v1``-suffixed URL as well as the pre-``/v1`` root the fetchers key on, and
prefers an authenticated catalog. Scans rather than rebuilding a key because callers hold no
credential — newest first, skipping expired entries, so a rotated credential does not keep
answering from the catalog its predecessor read.
"""
"""Pricing already cached for *base_url* (with or without ``/v1``), or ``{}``; never fetches.
Prefers an authenticated catalog, scanning newest first (callers hold no credential) and
skipping expired entries so a rotated credential does not answer from its predecessor's."""
from hermes_cli.models import _pricing_cache
root = (base_url or "").rstrip("/")
if root.endswith("/v1"):
root = root[:-3].rstrip("/")
root = _strip_v1((base_url or "").rstrip("/"))
authed_prefix = root + _PRICING_AUTH_KEY_PREFIX
for key in reversed(list(_pricing_cache)):
if key.startswith(authed_prefix):
@@ -114,16 +93,14 @@ def peek_cached_pricing(base_url: str) -> dict[str, dict[str, Any]]:
return _cached_catalog(root) or {}
def _strip_v1(url: str) -> str:
return url[:-3].rstrip("/") if url.endswith("/v1") else url
def _format_price_per_mtok(per_token_str: str) -> str:
"""Convert a per-token price string to a human-friendly $/Mtok string.
Always uses 2 decimal places so that prices align vertically when right-justified in a column
(the decimal point stays in the same position).
Sub-cent prices (e.g. deep-discount cache-hit promos) extend precision instead of collapsing to
"$0.00": the smallest decimal place that makes the value non-zero is found, then one extra digit
is kept and trailing zeros trimmed.
"""
"""Per-token price string → $/Mtok string. Always 2 decimals so right-justified prices align;
sub-cent prices (deep-discount cache-hit promos) widen precision until the value shows, keep
one extra digit and trim trailing zeros instead of collapsing to "$0.00"."""
try:
val = float(per_token_str)
except (TypeError, ValueError):
@@ -133,8 +110,6 @@ def _format_price_per_mtok(per_token_str: str) -> str:
per_m = val * 1_000_000
text = f"{per_m:.2f}"
if per_m < 0.01:
# Non-zero price below one cent per Mtok — widen precision until the
# value shows, keep one extra significant digit, trim trailing zeros.
prec = 3
while prec < 12 and round(per_m, prec) == 0:
prec += 1
@@ -142,80 +117,51 @@ def _format_price_per_mtok(per_token_str: str) -> str:
return f"${text}"
def compute_sale_discount(
prompt: str,
completion: str,
original: Any,
) -> tuple[int, str, str] | None:
"""Derive sale chrome from gateway ``pricing.original`` when cheaper.
def _price_float(raw: Any, *, positive: bool) -> float | None:
"""*raw* as a finite float (> 0, or >= 0 when not *positive*); None when unset/invalid/NaN."""
if raw in (None, ""):
return None
try:
n = float(raw)
except (TypeError, ValueError):
return None
if n != n or (n <= 0 if positive else n < 0):
return None
return n
Nous Portal-only feature: callers gate on the provider; this helper only sees ``original``
because the Nous fetch path opted in via ``include_sale_original=True``.
Returns ``(discount_percent, was_prompt_raw, was_completion_raw)`` only when ``original`` is a
dict and the current prompt (fallback: completion) rate is strictly below the corresponding
original.
"""
def _finite(raw: Any) -> float | None:
try:
n = float(raw)
except (TypeError, ValueError):
return None
return n if n > 0 and n == n else None # n == n rejects NaN
def _sale_pct(current: Any, original: Any) -> int | None:
"""Percent discount when *current* is strictly below *original* (both positive finite)."""
cur, orig = _price_float(current, positive=True), _price_float(original, positive=True)
if cur is None or orig is None or cur >= orig:
return None
return int(round((1.0 - (cur / orig)) * 100))
def _nonneg(raw: Any) -> float | None:
try:
n = float(raw)
except (TypeError, ValueError):
return None
return n if n >= 0 and n == n else None
def compute_sale_discount(prompt: str, completion: str, original: Any) -> tuple[int, str, str] | None:
"""Sale chrome from gateway ``pricing.original`` (Nous Portal only; callers gate on the provider
and opted in via ``include_sale_original=True``): ``(discount_percent, was_prompt_raw,
was_completion_raw)`` when ``original`` is a dict and the current prompt (fallback: completion)
rate is strictly below the original. Free / $0 models get a flat 100% off, with "was" prices
only when the gateway served an original (a natively-free stealth model gets bare "-100%")."""
orig_dict = original if isinstance(original, dict) else {}
was_prompt = orig_dict.get("prompt")
was_completion = orig_dict.get("completion")
was_prompt_str = str(was_prompt) if was_prompt not in (None, "") else ""
was_completion_str = str(was_completion) if was_completion not in (None, "") else ""
# Free / $0 models: flat 100% off, with "was" prices only when the
# gateway actually served an original (e.g. a :free sibling); a
# natively-free model (stealth/ox-alpha) gets bare "-100%" chrome.
cur_prompt_any = _nonneg(prompt) if prompt not in (None, "") else None
cur_comp_any = _nonneg(completion) if completion not in (None, "") else None
if cur_prompt_any == 0 and cur_comp_any in (0, None):
return (
100,
str(was_prompt) if was_prompt not in (None, "") else "",
str(was_completion) if was_completion not in (None, "") else "",
)
if _price_float(prompt, positive=False) == 0 and _price_float(completion, positive=False) in (0, None):
return (100, was_prompt_str, was_completion_str)
if not isinstance(original, dict):
if not isinstance(original, dict) or (not was_prompt_str and not was_completion_str):
return None
if was_prompt in (None, "") and was_completion in (None, ""):
return None
cur_prompt = _finite(prompt) if prompt not in (None, "") else None
orig_prompt = _finite(was_prompt) if was_prompt not in (None, "") else None
if cur_prompt is not None and orig_prompt is not None and cur_prompt < orig_prompt:
pct = int(round((1.0 - (cur_prompt / orig_prompt)) * 100))
if pct < 1:
return None
return (
pct,
str(was_prompt),
str(was_completion) if was_completion not in (None, "") else "",
)
cur_comp = _finite(completion) if completion not in (None, "") else None
orig_comp = _finite(was_completion) if was_completion not in (None, "") else None
if cur_comp is not None and orig_comp is not None and cur_comp < orig_comp:
pct = int(round((1.0 - (cur_comp / orig_comp)) * 100))
if pct < 1:
return None
return (
pct,
str(was_prompt) if was_prompt not in (None, "") else "",
str(was_completion),
)
pct = _sale_pct(prompt, was_prompt)
if pct is not None:
return (pct, was_prompt_str, was_completion_str) if pct >= 1 else None
pct = _sale_pct(completion, was_completion)
if pct is not None:
return (pct, was_prompt_str, was_completion_str) if pct >= 1 else None
return None
@@ -244,6 +190,15 @@ def _pricing_entry(pricing: dict, prompt_key: str = "prompt", completion_key: st
return entry
def _per_token(per_mtok: Any) -> str:
"""$/MTok → the per-token price string the picker expects."""
return str(float(per_mtok) / 1_000_000)
def _catalog_items(payload: dict) -> list[dict]:
return [item for item in payload.get("data", []) if isinstance(item, dict)]
def fetch_models_with_pricing(
api_key: str | None = None,
base_url: str = "https://openrouter.ai/api",
@@ -253,16 +208,10 @@ def fetch_models_with_pricing(
include_sale_original: bool = False,
cache_ttl_seconds: Optional[float] = None,
) -> dict[str, dict[str, Any]]:
"""Fetch ``/v1/models`` and return ``{model_id: {prompt, completion, ...}}``.
Results are cached per *base_url* and per credential, so repeated calls are free and one
caller's catalog never answers another's read. Works with any OpenRouter-compatible endpoint
(OpenRouter, Nous Portal).
When *include_sale_original* is true (Nous Portal only) and the gateway advertises a global
discount under ``pricing.original``, those pre-discount rates are copied through as a nested
``original`` dict so pickers can show sale chrome.
"""
"""Fetch ``/v1/models`` (any OpenRouter-compatible endpoint) → ``{model_id: {prompt, completion,
...}}``, cached per *base_url* and per credential so one caller's catalog never answers
another's read. *include_sale_original* (Nous Portal only) copies the gateway's pre-discount
``pricing.original`` rates through as a nested ``original`` dict for sale chrome."""
from hermes_cli.models import _HERMES_USER_AGENT
url_root = (base_url or "").rstrip("/")
cache_key = url_root + _pricing_auth_fingerprint(api_key)
@@ -279,25 +228,20 @@ def fetch_models_with_pricing(
if payload is None:
return _cache_catalog(cache_key, {})
# Same document the reasoning-capability fetch would pull, and every picker/pricing surface
# goes through here — mirror it so a later hot-path lookup (and the next process) has an
# answer without its own round-trip.
# Same document the reasoning-capability fetch would pull — mirror it so a later hot-path
# lookup (and the next process) has an answer without its own round-trip.
_seed_reasoning_caps(url, payload.get("data"))
result: dict[str, dict[str, Any]] = {}
for item in payload.get("data", []):
mid = item.get("id")
pricing = item.get("pricing")
mid, pricing = item.get("id"), item.get("pricing")
if mid and isinstance(pricing, dict):
entry = _pricing_entry(pricing)
# Sale chrome is Nous Portal-only; never copy pricing.original for other catalogs.
original = pricing.get("original") if include_sale_original else None
if isinstance(original, dict):
orig_entry = {
key: str(original[key])
for key in ("prompt", "completion", "input_cache_read", "input_cache_write")
if original.get(key) not in (None, "")
}
orig_entry = {key: str(original[key]) for key in ("prompt", "completion", "input_cache_read", "input_cache_write")
if original.get(key) not in (None, "")}
if orig_entry.get("prompt") or orig_entry.get("completion"):
entry["original"] = orig_entry
result[mid] = entry
@@ -305,16 +249,9 @@ def fetch_models_with_pricing(
return _cache_catalog(cache_key, result, cache_ttl_seconds)
def fetch_ai_gateway_pricing(
timeout: float = 8.0,
*,
force_refresh: bool = False,
) -> dict[str, dict[str, str]]:
"""Fetch Vercel AI Gateway /v1/models and return hermes-shaped pricing.
Vercel uses ``input`` / ``output`` field names; hermes's picker expects ``prompt`` /
``completion``. This translates. Cache read/write field names already match.
"""
def fetch_ai_gateway_pricing(timeout: float = 8.0, *, force_refresh: bool = False) -> dict[str, dict[str, str]]:
"""Vercel AI Gateway /v1/models pricing, translating its ``input`` / ``output`` field names to
the picker's ``prompt`` / ``completion`` (cache read/write names already match)."""
from hermes_constants import AI_GATEWAY_BASE_URL
cache_key = AI_GATEWAY_BASE_URL.rstrip("/")
@@ -328,11 +265,8 @@ def fetch_ai_gateway_pricing(
return _cache_catalog(cache_key, {})
result: dict[str, dict[str, str]] = {}
for item in payload.get("data", []):
if not isinstance(item, dict):
continue
mid = item.get("id")
pricing = item.get("pricing")
for item in _catalog_items(payload):
mid, pricing = item.get("id"), item.get("pricing")
if mid and isinstance(pricing, dict):
result[mid] = _pricing_entry(pricing, "input", "output")
return _cache_catalog(cache_key, result)
@@ -347,23 +281,17 @@ _DEFAULT_NOUS_INFERENCE_BASE = "https://inference-api.nousresearch.com"
def _resolve_nous_pricing_credentials() -> tuple[str, str]:
"""Return ``(api_key, base_url)`` for Nous Portal pricing.
Base URL precedence (mirrors runtime credential resolution): 1. ``NOUS_INFERENCE_BASE_URL`` env
override (staging / preview) 2. Resolved runtime credential ``base_url`` 3. Production default
Without (1), a staging profile's sale ``pricing.original`` never reaches the pickers — the
anonymous fallback would hit prod, which has no ``original`` field.
"""
"""``(api_key, base_url)`` for Nous Portal pricing; base_url is the bare origin (no ``/v1``).
Precedence mirrors runtime credential resolution: ``NOUS_INFERENCE_BASE_URL`` (staging /
preview) → resolved credential ``base_url`` → production default. Without the override a
staging profile's sale ``pricing.original`` would never reach the pickers."""
try:
from hermes_cli.auth import _nous_inference_env_override
env_base = _nous_inference_env_override()
except Exception:
env_base = None
api_key = ""
creds_base = ""
api_key = creds_base = ""
try:
from hermes_cli.auth import resolve_nous_runtime_credentials
@@ -373,27 +301,36 @@ def _resolve_nous_pricing_credentials() -> tuple[str, str]:
creds_base = (creds.get("base_url", "") or "").strip()
except Exception:
pass
base_url = (env_base or creds_base or _DEFAULT_NOUS_INFERENCE_BASE).rstrip("/")
# Credential bases arrive with or without the ``/v1`` suffix. Callers
# append their own path, so hand back the bare origin.
if base_url.endswith("/v1"):
base_url = base_url[:-3]
return (api_key, base_url)
# How long a Nous catalog stays trusted. Its contents depend on the org's policy, which an admin
# can change at any time and the client cannot observe, so a long-lived process must re-ask.
# Other providers' catalogs carry no such state and keep the default no-expiry caching.
_NOUS_CATALOG_TTL_SECONDS = 300.0
def _fetch_nous_pricing(api_key: str, base_url: str, *, force_refresh: bool) -> dict[str, dict[str, Any]]:
"""Shared by pricing and policy lookups so both read one cache entry."""
from hermes_cli.models import fetch_models_with_pricing
return fetch_models_with_pricing(
api_key=api_key,
base_url=base_url,
force_refresh=force_refresh,
include_sale_original=True, # Sale chrome (pricing.original) is Nous Portal-only.
cache_ttl_seconds=_NOUS_CATALOG_TTL_SECONDS,
)
def nous_policy_allowed_ids(*, force_refresh: bool = False) -> Optional[set[str]]:
"""The Nous model ids the caller's org may reach, or ``None`` to not filter.
The gateway omits policy-blocked rows from an authenticated ``GET /v1/models``, so that
response's keys are the reachable set.
``None`` means "leave the caller's list alone", for the three states that cannot support
narrowing one: no policy (or a token too old to say), an anonymous read whose catalog is
unfiltered, and an empty read, which is a fetch failure rather than an org that may reach
nothing.
"""
from hermes_cli.models import _resolve_nous_pricing_credentials, fetch_models_with_pricing
"""The Nous model ids the caller's org may reach (keys of an authenticated ``GET /v1/models``,
which omits policy-blocked rows), or ``None`` to not filter: no policy (or a token too old to
say), an anonymous read (unfiltered catalog), or an empty read (a fetch failure, not an org
that may reach nothing)."""
from hermes_cli.models import _resolve_nous_pricing_credentials
try:
from hermes_cli.nous_account import nous_policy_present
@@ -405,105 +342,62 @@ def nous_policy_allowed_ids(*, force_refresh: bool = False) -> Optional[set[str]
api_key, base_url = _resolve_nous_pricing_credentials()
if not api_key or not base_url:
return None
# Same arguments as get_pricing_for_provider's nous branch, so a caller
# asking for pricing too shares this entry instead of paying for a second
# request.
pricing = fetch_models_with_pricing(
api_key=api_key,
base_url=base_url,
force_refresh=force_refresh,
include_sale_original=True,
cache_ttl_seconds=_NOUS_CATALOG_TTL_SECONDS,
)
return set(pricing) or None
return set(_fetch_nous_pricing(api_key, base_url, force_refresh=force_refresh)) or None
# Past this size an allowed set reads as a whole catalog rather than an
# allowlist, and is not worth showing in place of an empty picker.
# Past this size an allowed set reads as a whole catalog rather than an allowlist, and is not
# worth showing in place of an empty picker.
_NOUS_POLICY_APPEND_MAX = 64
# How long a Nous catalog stays trusted. Its contents depend on the org's
# policy, which an admin can change at any time and the client cannot observe,
# so a long-lived process must re-ask instead of holding the first answer for
# its whole life. Other providers' catalogs carry no such state and keep the
# default no-expiry caching.
_NOUS_CATALOG_TTL_SECONDS = 300.0
def restrict_to_nous_policy(
model_ids: list[str],
allowed: Optional[set[str]],
*,
rescue_empty: bool = False,
) -> list[str]:
"""*model_ids* narrowed to *allowed*, preserving the caller's order.
A ``:free`` sibling is kept when its base model is reachable, mirroring the gateway, which
admits a row when any of its requestable ids passes. Prefer over-listing: that costs a 403 from
the authoritative gate, while hiding a row the gate would serve is unrecoverable from the
client.
"""
"""*model_ids* narrowed to *allowed*, preserving order. A ``:free`` sibling is kept when its
base model is reachable (the gateway admits a row when any requestable id passes); over-listing
costs a 403 from the authoritative gate, hiding a servable row is unrecoverable client-side.
*rescue_empty*: an allowlist naming only models the curated manifest lacks would leave an empty
picker — worse than no filter — so return the allowlist itself. Opt-in per list: an already-
empty list (a paid tier's gated models) means "nothing to gate", not "nothing survived"."""
if not allowed:
return list(model_ids)
kept = [
mid
for mid in model_ids
if mid in allowed or mid.split(":", 1)[0] in allowed
]
# An allowlist can name only models the curated manifest lacks, leaving an
# empty picker — worse than no filter, since the models the org may use are
# the ones dropped. Opt-in per list: an already-empty list (a paid tier's
# gated models) means "nothing to gate", not "nothing survived".
kept = [mid for mid in model_ids if mid in allowed or mid.split(":", 1)[0] in allowed]
if rescue_empty and not kept and len(allowed) <= _NOUS_POLICY_APPEND_MAX:
return sorted(allowed)
return kept
def _fetch_openrouter_pricing(*, force_refresh: bool = False) -> dict[str, dict[str, Any]]:
from hermes_cli.models import fetch_models_with_pricing
return fetch_models_with_pricing(
api_key=_resolve_openrouter_api_key(),
base_url="https://openrouter.ai/api",
force_refresh=force_refresh,
)
def _fetch_nous_pricing_for_provider(*, force_refresh: bool = False) -> dict[str, dict[str, Any]]:
from hermes_cli.models import _resolve_nous_pricing_credentials
api_key, base_url = _resolve_nous_pricing_credentials()
if not base_url:
return {}
return _fetch_nous_pricing(api_key, base_url, force_refresh=force_refresh)
def get_pricing_for_provider(provider: str, *, force_refresh: bool = False) -> dict[str, dict[str, str]]:
"""Return live pricing for providers that support it (openrouter, nous, ai-gateway, novita)."""
from hermes_cli.models import _resolve_nous_pricing_credentials, fetch_models_with_pricing, normalize_provider
normalized = normalize_provider(provider)
if normalized == "openrouter":
return fetch_models_with_pricing(
api_key=_resolve_openrouter_api_key(),
base_url="https://openrouter.ai/api",
force_refresh=force_refresh,
)
if normalized == "ai-gateway":
return fetch_ai_gateway_pricing(force_refresh=force_refresh)
if normalized == "novita":
return _fetch_novita_pricing(force_refresh=force_refresh)
if normalized == "deepinfra":
return _fetch_deepinfra_pricing(force_refresh=force_refresh)
if normalized == "fireworks":
return _fireworks_pricing_from_models_dev(force_refresh=force_refresh)
if normalized == "nous":
api_key, base_url = _resolve_nous_pricing_credentials()
if base_url:
return fetch_models_with_pricing(
api_key=api_key,
base_url=base_url,
force_refresh=force_refresh,
# Sale chrome (pricing.original) is Nous Portal-only.
include_sale_original=True,
cache_ttl_seconds=_NOUS_CATALOG_TTL_SECONDS,
)
return {}
"""Return live pricing for providers that support it (openrouter, nous, ai-gateway, novita,
deepinfra, fireworks); ``{}`` for everything else."""
from hermes_cli.models import normalize_provider
fetcher = _PRICING_FETCHERS.get(normalize_provider(provider))
return fetcher(force_refresh=force_refresh) if fetcher else {}
def _fireworks_pricing_from_models_dev(
*,
force_refresh: bool = False,
) -> dict[str, dict[str, str]]:
"""Derive Fireworks picker pricing from the models.dev registry cache.
No dedicated network fetch: ``fetch_models_dev()`` already maintains an in-memory + disk cache
(1h TTL) that every picker surface shares, so this is a pure dict transform on the picker path —
no added latency and no per-render network call.
"""
def _fireworks_pricing_from_models_dev(*, force_refresh: bool = False) -> dict[str, dict[str, str]]:
"""Fireworks picker pricing from the models.dev registry cache (``fetch_models_dev()`` keeps a
shared in-memory + disk cache, 1h TTL) — a pure dict transform, no per-render network call."""
cache_key = "models.dev/fireworks"
if not force_refresh:
cached = _cached_catalog(cache_key)
@@ -514,24 +408,16 @@ def _fireworks_pricing_from_models_dev(
try:
from agent.models_dev import _get_provider_models
models = _get_provider_models("fireworks") or {}
for mid, entry in models.items():
if not isinstance(entry, dict):
continue
cost = entry.get("cost")
for mid, entry in (_get_provider_models("fireworks") or {}).items():
cost = entry.get("cost") if isinstance(entry, dict) else None
if not isinstance(cost, dict):
continue
inp = cost.get("input")
out = cost.get("output")
inp, out = cost.get("input"), cost.get("output")
if inp is None and out is None:
continue
row: dict[str, str] = {
"prompt": str(float(inp or 0) / 1_000_000),
"completion": str(float(out or 0) / 1_000_000),
}
cache_read = cost.get("cache_read")
if cache_read:
row["input_cache_read"] = str(float(cache_read) / 1_000_000)
row = {"prompt": _per_token(inp or 0), "completion": _per_token(out or 0)}
if cost.get("cache_read"):
row["input_cache_read"] = _per_token(cost["cache_read"])
result[str(mid)] = row
except Exception:
result = {}
@@ -539,24 +425,15 @@ def _fireworks_pricing_from_models_dev(
return _cache_catalog(cache_key, result)
def _fetch_novita_pricing(
timeout: float = 8.0,
*,
force_refresh: bool = False,
) -> dict[str, dict[str, str]]:
"""Fetch pricing from NovitaAI /v1/models.
NovitaAI reports per-million-token prices in units of 0.0001 USD; they are converted to the
per-token strings the shared pricing formatter expects. Results are cached in
``_pricing_cache`` keyed on the resolved base URL so menu renders don't re-hit the network.
"""
def _fetch_novita_pricing(timeout: float = 8.0, *, force_refresh: bool = False) -> dict[str, dict[str, str]]:
"""NovitaAI /v1/models pricing (per-million prices in units of 0.0001 USD → per-token strings),
cached on the resolved base URL so menu renders don't re-hit the network."""
from hermes_cli.models import _HERMES_USER_AGENT
api_key = os.getenv("NOVITA_API_KEY", "").strip()
if not api_key:
return {}
base_url = os.getenv("NOVITA_BASE_URL", "").strip() or "https://api.novita.ai/openai/v1"
cache_key = base_url.rstrip("/")
cache_key = (os.getenv("NOVITA_BASE_URL", "").strip() or "https://api.novita.ai/openai/v1").rstrip("/")
if not force_refresh:
cached = _cached_catalog(cache_key)
if cached is not None:
@@ -568,15 +445,10 @@ def _fetch_novita_pricing(
return _cache_catalog(cache_key, {})
result: dict[str, dict[str, str]] = {}
for item in payload.get("data", []):
if not isinstance(item, dict):
continue
for item in _catalog_items(payload):
mid = item.get("id")
if not mid:
continue
inp = item.get("input_token_price_per_m")
out = item.get("output_token_price_per_m")
if inp is None and out is None:
inp, out = item.get("input_token_price_per_m"), item.get("output_token_price_per_m")
if not mid or (inp is None and out is None):
continue
result[str(mid)] = {
"prompt": str(float(inp or 0) / 10_000 / 1_000_000),
@@ -586,17 +458,10 @@ def _fetch_novita_pricing(
return _cache_catalog(cache_key, result)
def _fetch_deepinfra_pricing(
timeout: float = 5.0,
*,
force_refresh: bool = False,
) -> dict[str, dict[str, str]]:
"""Return picker-shape pricing for DeepInfra chat models.
DeepInfra publishes ``input_tokens``/``output_tokens``/``cache_read_tokens`` in $/MTok; the
picker expects per-token strings under ``prompt``/``completion``/``input_cache_read``
(OpenRouter shape). Cached via the catalog helper so repeated picker renders are free.
"""
def _fetch_deepinfra_pricing(timeout: float = 5.0, *, force_refresh: bool = False) -> dict[str, dict[str, str]]:
"""DeepInfra chat-model pricing: ``input_tokens`` / ``output_tokens`` / ``cache_read_tokens`` in
$/MTok → per-token ``prompt`` / ``completion`` / ``input_cache_read`` (cached by the by-tag
helper)."""
from hermes_cli.models import _fetch_deepinfra_models_by_tag
items = _fetch_deepinfra_models_by_tag("chat", timeout=timeout, force_refresh=force_refresh)
result: dict[str, dict[str, str]] = {}
@@ -606,10 +471,20 @@ def _fetch_deepinfra_pricing(
if not isinstance(pricing, dict):
continue
entry = {
ours: str(float(pricing[theirs]) / 1_000_000)
ours: _per_token(pricing[theirs])
for theirs, ours in (("input_tokens", "prompt"), ("output_tokens", "completion"), ("cache_read_tokens", "input_cache_read"))
if pricing.get(theirs) is not None
}
if entry:
result[item["id"]] = entry
return result
_PRICING_FETCHERS = {
"openrouter": _fetch_openrouter_pricing,
"ai-gateway": fetch_ai_gateway_pricing,
"novita": _fetch_novita_pricing,
"deepinfra": _fetch_deepinfra_pricing,
"fireworks": _fireworks_pricing_from_models_dev,
"nous": _fetch_nous_pricing_for_provider,
}
hermes_cli/models_reasoning_caps.py
+63 -101
View File
@@ -1,17 +1,15 @@
"""Per-model reasoning capabilities from OpenRouter-schema ``/v1/models`` catalogs.
Split out of ``hermes_cli.models``; every public/patched name is re-imported there. The
OpenRouter and Nous Portal catalogs share one implementation parametrized by
:class:`_CapsSource`; the per-source module globals (``_openrouter_reasoning_caps_cache``,
``_nous_caps_disk_checked``, ...) stay defined on ``hermes_cli.models`` — tests reset them there —
and are read/written by attribute name through the origin module.
Split out of ``hermes_cli.models``; every public/patched name is re-imported there. OpenRouter and
Nous Portal share one implementation parametrized by :class:`_CapsSource`; the per-source module
globals (``_openrouter_reasoning_caps_cache``, ``_nous_caps_disk_checked``, ...) stay defined on
``hermes_cli.models`` — tests reset them there — and are read/written by attribute name.
Tri-state contract for callers deciding whether to emit reasoning controls:
- dict with ``supports_reasoning: True`` (+ ``supported_efforts``, ``mandatory``) — the route
advertises reasoning controls;
- dict with ``supports_reasoning: False`` — the catalog knows the model and it does NOT accept
reasoning controls (definitive negative);
- ``None`` — unknown: catalog not loaded, model not listed (private/custom route), malformed.
Tri-state contract for callers deciding whether to emit reasoning controls: a dict with
``supports_reasoning: True`` (+ ``supported_efforts``, ``mandatory``) — the route advertises
reasoning controls; ``supports_reasoning: False`` — the catalog knows the model and it does NOT
accept them (definitive negative); ``None`` — unknown (catalog not loaded, model not listed,
malformed).
"""
from __future__ import annotations
@@ -34,7 +32,6 @@ Caps = dict[str, Optional[dict[str, Any]]]
def _origin():
from hermes_cli import models
return models
@@ -42,9 +39,8 @@ def parse_openrouter_reasoning_capabilities(item: Any) -> Optional[dict[str, Any
"""Normalize one OpenRouter catalog entry's reasoning metadata.
``supported_parameters`` contains ``"reasoning"`` when the route accepts reasoning controls at
all; a top-level ``reasoning`` object may add detail (``mandatory``, ``supported_efforts``).
A missing/malformed ``supported_parameters`` is "unknown" (None), mirroring the permissive
stance of ``_openrouter_model_supports_tools``.
all; a top-level ``reasoning`` object may add ``mandatory`` / ``supported_efforts``. A missing
or malformed ``supported_parameters`` is "unknown" (None), like ``_openrouter_model_supports_tools``.
"""
if not isinstance(item, dict):
return None
@@ -54,31 +50,22 @@ def parse_openrouter_reasoning_capabilities(item: Any) -> Optional[dict[str, Any
if "reasoning" not in params:
return {"supports_reasoning": False}
reasoning = item.get("reasoning")
mandatory = isinstance(reasoning, dict) and reasoning.get("mandatory") is True
if not isinstance(reasoning, dict):
reasoning = {}
raw_efforts = reasoning.get("supported_efforts")
efforts: Optional[list[str]] = None
if isinstance(reasoning, dict):
raw_efforts = reasoning.get("supported_efforts")
if isinstance(raw_efforts, list):
efforts = list(dict.fromkeys(
str(effort).strip().lower()
for effort in raw_efforts
if str(effort).strip()
))
return {
"supports_reasoning": True,
"supported_efforts": efforts,
"mandatory": mandatory,
}
if isinstance(raw_efforts, list):
efforts = list(dict.fromkeys(str(e).strip().lower() for e in raw_efforts if str(e).strip()))
return {"supports_reasoning": True, "supported_efforts": efforts, "mandatory": reasoning.get("mandatory") is True}
# ── Disk mirror ────────────────────────────────────────────────────────
#
# The in-process caches are always cold in a short-lived process, and every consumer is on a hot
# path that must never block on HTTP — so without a disk copy, `hermes -p`, a cron job, or a
# freshly booted gateway answers "capability unknown" for its whole first turn and falls back to
# the conservative wire shape. One file holds every catalog, keyed by the URL it came from:
# OpenRouter and the Nous Portal list different models, and a staging Portal must not answer for
# production.
# In-process caches are always cold in a short-lived process, and every consumer is on a hot path
# that must never block on HTTP — so without a disk copy, `hermes -p`, a cron job, or a freshly
# booted gateway answers "capability unknown" for its whole first turn. One file holds every
# catalog keyed by URL: OpenRouter and the Portal list different models, and a staging Portal must
# not answer for production.
_REASONING_CAPS_DISK_TTL_SECONDS = 24 * 3600
@@ -89,16 +76,13 @@ def _reasoning_caps_disk_path() -> Path:
def _read_reasoning_caps_disk() -> dict[str, Any]:
from hermes_cli.models import _read_json_cache
return _read_json_cache(_reasoning_caps_disk_path()) or {}
def _load_reasoning_caps_disk(url: str) -> tuple[Optional[Caps], float]:
"""Return ``(caps, age_seconds)`` for *url*, or ``(None, 0.0)``."""
entry = _origin()._read_reasoning_caps_disk().get(url)
if not isinstance(entry, dict):
return None, 0.0
caps = entry.get("caps")
caps = entry.get("caps") if isinstance(entry, dict) else None
if not isinstance(caps, dict) or not caps:
return None, 0.0
try:
@@ -111,7 +95,6 @@ def _load_reasoning_caps_disk(url: str) -> tuple[Optional[Caps], float]:
def _save_reasoning_caps_disk(url: str, caps: Caps) -> None:
"""Merge *url*'s catalog into the shared disk mirror, atomically."""
from hermes_cli.models import _write_json_cache
try:
data = _origin()._read_reasoning_caps_disk()
data[url] = {"ts": time.time(), "caps": caps}
@@ -121,27 +104,19 @@ def _save_reasoning_caps_disk(url: str, caps: Caps) -> None:
def _warm_reasoning_caps_async(refresh) -> None:
"""Run *refresh* in a background thread. Fire-and-forget.
Called from hot paths that found the cache cold or the disk copy stale, so the next call — or,
via the disk mirror, the next process — benefits without this turn ever blocking on HTTP.
Callers own the once-per-process guard; the fetch keeps its own failure TTL.
"""
"""Run *refresh* in a daemon thread (fire-and-forget) so a cold/stale cache is warm for the
next call or, via the disk mirror, the next process without this turn blocking on HTTP.
Callers own the once-per-process guard; the fetch keeps its own failure TTL."""
if os.environ.get("PYTEST_CURRENT_TEST"):
return
threading.Thread(target=refresh, name="reasoning-caps-warm", daemon=True).start()
def _hydrate_reasoning_caps_from_disk(url: str, refresh) -> Optional[Caps]:
"""The disk copy of *url*'s catalog, queueing *refresh* when it's stale.
A copy past its TTL is still returned — a stale verdict beats no verdict, and reasoning
capabilities change rarely — with a background refresh so the next run is current.
"""
"""The disk copy of *url*'s catalog, queueing *refresh* when it's stale. A copy past its TTL is
still returned — a stale verdict beats no verdict, and capabilities change rarely."""
caps, age = _load_reasoning_caps_disk(url)
if caps is None:
return None
if age >= _REASONING_CAPS_DISK_TTL_SECONDS:
if caps is not None and age >= _REASONING_CAPS_DISK_TTL_SECONDS:
_warm_reasoning_caps_async(refresh)
return caps
@@ -149,20 +124,17 @@ def _hydrate_reasoning_caps_from_disk(url: str, refresh) -> Optional[Caps]:
def _seed_reasoning_caps(url: str, items: Any) -> Optional[Caps]:
"""Parse a ``/v1/models`` ``data`` array and mirror it for *url*.
Takes the payload rather than fetching it, so picker and pricing fetches (which pull the same
document) leave the mirror warm at no network cost. Returns None when the array has no usable
entries, which callers remember as a failure rather than caching as empty.
Takes the payload rather than fetching it, so picker and pricing fetches (same document) leave
the mirror warm at no network cost. None when the array has no usable entries — callers
remember that as a failure rather than caching empty.
"""
if not isinstance(items, list):
return None
caps_by_id: Caps = {}
for item in items:
if not isinstance(item, dict):
continue
mid = str(item.get("id") or "").strip()
if not mid:
continue
caps_by_id[mid] = parse_openrouter_reasoning_capabilities(item)
mid = str(item.get("id") or "").strip() if isinstance(item, dict) else ""
if mid:
caps_by_id[mid] = parse_openrouter_reasoning_capabilities(item)
if not caps_by_id:
return None
_save_reasoning_caps_disk(url, caps_by_id)
@@ -170,16 +142,11 @@ def _seed_reasoning_caps(url: str, items: Any) -> Optional[Caps]:
def _fetch_reasoning_caps_catalog(url: str, timeout: float) -> Optional[Caps]:
"""Fetch one OpenRouter-shaped ``/v1/models`` catalog → per-model caps.
Returns None when the catalog is unreachable or has no usable entries, so callers remember the
failure and fall back rather than caching an empty result. Sends a User-Agent because the
Portal 403s anonymous catalog reads.
"""
"""Fetch one OpenRouter-shaped ``/v1/models`` catalog → per-model caps; None when unreachable or
empty so callers remember the failure. Sends a User-Agent: the Portal 403s anonymous reads."""
m = _origin()
headers = {"Accept": "application/json", "User-Agent": m._HERMES_USER_AGENT}
try:
req = urllib.request.Request(url, headers=headers)
req = urllib.request.Request(url, headers={"Accept": "application/json", "User-Agent": m._HERMES_USER_AGENT})
with m._urlopen_model_catalog_request(req, timeout=timeout) as resp:
payload = json.loads(resp.read().decode())
except Exception:
@@ -193,11 +160,10 @@ def _fetch_reasoning_caps_catalog(url: str, timeout: float) -> Optional[Caps]:
class _CapsSource:
"""One catalog's cache slots on ``hermes_cli.models`` plus how to name its URL.
``cache``: model id → parsed caps, populated by one full-catalog fetch and kept for the process
lifetime (capabilities don't change). ``failed_at``: monotonic timestamp of the last FAILED
fetch; suppresses re-fetch storms from per-turn callers while the catalog is unreachable (60s,
mirrors the LM Studio/Ollama capability-probe caching). ``disk_checked`` / ``warm_started``:
once-per-process guards for the disk hydrate and the background warm.
``cache``: model id → parsed caps from one full-catalog fetch, kept for the process lifetime.
``failed_at``: monotonic timestamp of the last FAILED fetch; suppresses re-fetch storms from
per-turn callers for 60s (mirrors the LM Studio/Ollama probe caching). ``disk_checked`` /
``warm_started``: once-per-process guards for the disk hydrate and the background warm.
"""
cache: str
failed_at: str
@@ -205,37 +171,40 @@ class _CapsSource:
warm_started: str
url: Callable[[], str]
def get(self, slot: str):
return getattr(_origin(), getattr(self, slot))
def set(self, slot: str, value) -> None:
setattr(_origin(), getattr(self, slot), value)
def _fetch_caps(src: _CapsSource, timeout: float = 6.0, *, force: bool = False) -> Optional[Caps]:
"""Fetch + cache the source's per-model caps. None (without poisoning the cache) when
unreachable, so callers retry later and fall back meanwhile."""
m = _origin()
cached = getattr(m, src.cache)
cached = src.get("cache")
if cached is not None and not force:
return cached
failed_at = getattr(m, src.failed_at)
failed_at = src.get("failed_at")
if failed_at is not None and (time.monotonic() - failed_at) < 60:
return None
caps_by_id = _fetch_reasoning_caps_catalog(src.url(), timeout)
if caps_by_id is None:
setattr(m, src.failed_at, time.monotonic())
src.set("failed_at", time.monotonic())
return None
setattr(m, src.cache, caps_by_id)
src.set("cache", caps_by_id)
return caps_by_id
def _caps_cached(src: _CapsSource) -> Optional[Caps]:
"""Cache-only caps: memory, else the disk mirror. Never HTTP.
Guarded to one disk attempt per process: for the Portal, naming the catalog means resolving
credentials, which can itself reach the network to refresh a token — far too expensive for a
caller that runs every turn.
One disk attempt per process: for the Portal, naming the catalog means resolving credentials,
which can itself reach the network to refresh a token — too expensive for a per-turn caller.
"""
m = _origin()
if getattr(m, src.cache) is None and not getattr(m, src.disk_checked):
setattr(m, src.disk_checked, True)
setattr(m, src.cache, _hydrate_reasoning_caps_from_disk(src.url(), lambda: _fetch_caps(src, force=True)))
return getattr(m, src.cache)
if src.get("cache") is None and not src.get("disk_checked"):
src.set("disk_checked", True)
src.set("cache", _hydrate_reasoning_caps_from_disk(src.url(), lambda: _fetch_caps(src, force=True)))
return src.get("cache")
def _model_caps(src: _CapsSource, model_id: Optional[str], *, timeout: float, allow_fetch: bool) -> Optional[dict[str, Any]]:
@@ -245,16 +214,13 @@ def _model_caps(src: _CapsSource, model_id: Optional[str], *, timeout: float, al
caps_by_id = _caps_cached(src)
if caps_by_id is None and allow_fetch:
caps_by_id = _fetch_caps(src, timeout=timeout)
if caps_by_id is None:
return None
return caps_by_id.get(model)
return caps_by_id.get(model) if caps_by_id is not None else None
def _warm_caps_async(src: _CapsSource) -> None:
m = _origin()
if getattr(m, src.warm_started) or _caps_cached(src) is not None:
if src.get("warm_started") or _caps_cached(src) is not None:
return
setattr(m, src.warm_started, True)
src.set("warm_started", True)
_warm_reasoning_caps_async(lambda: _fetch_caps(src, force=True))
@@ -275,11 +241,8 @@ _NOUS_CAPS = _CapsSource(
def nous_catalog_url() -> str:
"""The Portal ``/v1/models`` URL for the endpoint we actually talk to.
Resolved through the ladder ``NOUS_INFERENCE_BASE_URL`` → resolved credential base → prod
rather than pinned to production, so a staging profile reads staging's capabilities.
"""
"""The Portal ``/v1/models`` URL for the endpoint we actually talk to (``NOUS_INFERENCE_BASE_URL``
→ resolved credential base → prod), so a staging profile reads staging's capabilities."""
return f"{_origin()._resolve_nous_pricing_credentials()[1]}/v1/models"
@@ -287,7 +250,6 @@ def openrouter_model_reasoning_capabilities(
model_id: Optional[str], *, timeout: float = 6.0, allow_fetch: bool = False,
) -> Optional[dict[str, Any]]:
"""Live-catalog reasoning capabilities for an OpenRouter model (tri-state, see module doc).
CACHE-ONLY by default — safe on per-request hot paths (never blocks on HTTP)."""
return _model_caps(_OPENROUTER_CAPS, model_id, timeout=timeout, allow_fetch=allow_fetch)
hermes_cli/models_validate.py
+150 -210
View File
@@ -1,14 +1,11 @@
"""Validate a requested ``/model`` value against the active provider's catalog.
Split out of ``hermes_cli.models``; :func:`validate_requested_model` is re-imported there, so
``hermes_cli.models.validate_requested_model`` keeps resolving. Every catalog fetcher this module
calls is looked up on ``hermes_cli.models`` at call time (``_m.<name>``), so existing
``patch("hermes_cli.models.<name>")`` mocks keep intercepting.
Split out of ``hermes_cli.models``; :func:`validate_requested_model` is re-imported there. Every
catalog fetcher this module calls is looked up on ``hermes_cli.models`` at call time (``_m.<name>``)
so existing ``patch("hermes_cli.models.<name>")`` mocks keep intercepting.
Every provider branch returns one of four verdict shapes (see :func:`_verdict`) or ``None`` to
mean "not decided here — keep walking the ladder". The ladder ORDER is behavior: moa → whitespace
→ OpenRouter preset parse → LM Studio → Ollama native → custom → codex/xai static → MiniMax →
Anthropic native → Anthropic Messages → live listing → Bedrock → curated-catalog fallback.
Every provider branch returns a verdict dict (see :func:`_verdict`) or ``None`` for "not decided
here — keep walking the ladder". The ladder ORDER is behavior (see ``_LADDER``).
"""
from __future__ import annotations
@@ -16,7 +13,7 @@ from __future__ import annotations
import re
from dataclasses import dataclass
from difflib import get_close_matches
from typing import Any, Optional
from typing import Any, Callable, Optional
from utils import base_url_host_matches
@@ -25,8 +22,6 @@ from utils import base_url_host_matches
def _verdict(accepted: bool, persist: bool, recognized: bool, message: Optional[str],
corrected_model: Optional[str] = None) -> dict[str, Any]:
"""Build the verdict dict. ``corrected_model`` is only present when set (key order matters
to nobody, but keep it identical to the historical literals anyway)."""
out: dict[str, Any] = {"accepted": accepted, "persist": persist, "recognized": recognized}
if corrected_model is not None:
out["corrected_model"] = corrected_model
@@ -38,6 +33,10 @@ def _accept() -> dict[str, Any]:
return _verdict(True, True, True, None)
def _accept_with_note(message: str) -> dict[str, Any]:
return _verdict(True, True, True, message)
def _reject(message: str) -> dict[str, Any]:
return _verdict(False, False, False, message)
@@ -60,6 +59,16 @@ class _Match:
corrected: Optional[str] = None
suggestion_text: str = ""
def verdict(self, req: "_Request", *, keep_suffix: bool = False) -> Optional[dict[str, Any]]:
"""Accept on exact, auto-correct on a near-typo (re-attaching a preserved ``@preset/``
suffix when *keep_suffix*), else None so the branch composes its own message."""
if self.exact:
return _accept()
if self.corrected:
corrected = req.with_preset_suffix(self.corrected) if keep_suffix else self.corrected
return _corrected(req.requested, corrected)
return None
def _match_in_catalog(
query: str,
@@ -71,37 +80,31 @@ def _match_in_catalog(
suggest_cutoff: float = 0.5,
suggest_label: str = "Similar models",
) -> _Match:
"""The shared ladder: exact membership → typo auto-correct (cutoff .9) → suggestion text.
``case_insensitive`` matches on lower-cased ids and maps results back to the catalog's
spelling (MiniMax ships mixed-case ids). ``suggest_query`` overrides the string the
suggestion search uses (some branches search on the raw request, not the lookup form).
"""
candidates = list(candidates)
"""Shared ladder: exact membership → typo auto-correct (cutoff .9) → suggestion text.
``case_insensitive`` matches lower-cased ids and maps results back to the catalog's spelling
(MiniMax ships mixed-case ids). ``suggest_query`` overrides the string the suggestion search
uses (some branches search on the raw request, not the lookup form)."""
pool = list(candidates)
display = None
if suggest_query is None:
suggest_query = query
if case_insensitive:
display = {c.lower(): c for c in candidates}
display = {c.lower(): c for c in pool}
pool = list(display)
query = query.lower()
suggest_query = query if suggest_query is None else suggest_query.lower()
else:
display = None
pool = candidates
suggest_query = query if suggest_query is None else suggest_query
query, suggest_query = query.lower(), suggest_query.lower()
def _show(cid: str) -> str:
return display[cid] if display is not None else cid
if query in set(pool):
return _Match(exact=True)
if auto_correct:
auto = get_close_matches(query, pool, n=1, cutoff=0.9)
if auto:
return _Match(corrected=_show(auto[0]))
auto = get_close_matches(query, pool, n=1, cutoff=0.9) if auto_correct else []
if auto:
return _Match(corrected=_show(auto[0]))
suggestions = get_close_matches(suggest_query, pool, n=3, cutoff=suggest_cutoff)
text = ""
if suggestions:
text = f"\n {suggest_label}: " + ", ".join(f"`{_show(s)}`" for s in suggestions)
return _Match(suggestion_text=text)
if not suggestions:
return _Match()
return _Match(suggestion_text=f"\n {suggest_label}: " + ", ".join(f"`{_show(s)}`" for s in suggestions))
# ── Request context ──────────────────────────────────────────────────────
@@ -125,19 +128,25 @@ class _Request:
# ── Provider branches (None = not decided here) ─────────────────────────
def _validate_moa(requested: str) -> dict[str, Any]:
def _validate_moa(req: _Request) -> dict[str, Any]:
try:
from hermes_cli.config import load_config
from hermes_cli.moa_config import normalize_moa_config
cfg = normalize_moa_config(load_config().get("moa") or {})
if requested in cfg["presets"]:
if req.requested in cfg["presets"]:
return _accept()
return _reject(f"MoA preset `{requested}` was not found. Run `hermes moa list`.")
return _reject(f"MoA preset `{req.requested}` was not found. Run `hermes moa list`.")
except Exception as exc:
return _reject(f"Could not read MoA presets: {exc}")
def _reject_whitespace(req: _Request) -> Optional[dict[str, Any]]:
if any(ch.isspace() for ch in req.requested):
return _reject("Model names cannot contain spaces.")
return None
def _parse_openrouter_preset(req: _Request) -> Optional[dict[str, Any]]:
"""OpenRouter presets are account-scoped, so ``@preset/<slug>`` never appears in the public
/v1/models listing. A bare preset is accepted unverified; ``<model>@preset/<slug>`` validates
@@ -151,11 +160,8 @@ def _parse_openrouter_preset(req: _Request) -> Optional[dict[str, Any]]:
else:
preset_base, preset_slug = req.requested.split(marker, 1)
if re.fullmatch(r"[A-Za-z0-9._~-]+", preset_slug) is None:
return _reject(
"OpenRouter preset slugs must be non-empty URL-safe "
"identifiers using only letters, digits, '.', '_', "
"'~', or '-'."
)
return _reject("OpenRouter preset slugs must be non-empty URL-safe identifiers using only "
"letters, digits, '.', '_', '~', or '-'.")
req.preset_suffix = f"{marker}{preset_slug}"
if not preset_base:
return _soft_accept(None)
@@ -176,43 +182,30 @@ def _validate_lmstudio(req: _Request) -> dict[str, Any]:
if models is None:
return _reject(f"Could not reach LM Studio's `/api/v1/models` to validate `{req.requested}`.")
if not models:
return _reject(
f"LM Studio is reachable but no chat-capable models are loaded. "
f"Load `{req.requested}` in LM Studio (Developer tab → Load Model) and try again."
)
return _reject("LM Studio is reachable but no chat-capable models are loaded. "
f"Load `{req.requested}` in LM Studio (Developer tab → Load Model) and try again.")
if req.lookup in set(models):
return _accept()
return _reject(f"Model `{req.requested}` was not found in LM Studio's model listing.")
def _ollama_probe_headers(req: _Request) -> dict[str, str]:
"""Headers for the Ollama native probe.
Configured ``providers.ollama.extra_headers`` are only applied when the probed endpoint is
the configured one (never leak them to a different host). Caller headers win; a caller
``api_key`` becomes the Authorization header unless the caller already sent one.
"""
"""Headers for the Ollama native probe. Configured ``providers.ollama.extra_headers`` apply only
when the probed endpoint is the configured one (never leak them to a different host). Caller
headers win; a caller ``api_key`` becomes the Authorization header unless the caller sent one."""
from hermes_cli import models as _m
from hermes_cli.models_local import _configured_ollama_base_url
from hermes_cli.models_local import _configured_ollama_base_url, _drop_authorization
configured_base = _configured_ollama_base_url()
configured_allowed = not (
configured_base and not _m._same_ollama_native_root(req.base_url or "", configured_base)
)
configured_allowed = not configured_base or _m._same_ollama_native_root(req.base_url or "", configured_base)
configured = _m._get_ollama_native_headers(req.base_url, api_key=req.api_key) if configured_allowed else {}
if req.headers is None:
return _m._get_ollama_native_headers(req.base_url, api_key=req.api_key) if configured_allowed else {}
out: dict[str, str] = {}
if configured_allowed:
out.update(_m._get_ollama_native_headers(req.base_url, api_key=req.api_key))
for key in tuple(out):
if key.lower() == "authorization":
del out[key]
return configured
out = dict(configured)
_drop_authorization(out)
out.update(req.headers)
caller_has_authorization = any(key.lower() == "authorization" for key in req.headers)
if req.api_key and not caller_has_authorization:
for key in tuple(out):
if key.lower() == "authorization":
del out[key]
if req.api_key and not any(key.lower() == "authorization" for key in req.headers):
_drop_authorization(out)
out["Authorization"] = f"Bearer {req.api_key}"
return out
@@ -230,20 +223,16 @@ def _validate_ollama_native(req: _Request) -> Optional[dict[str, Any]]:
return None
models = _m.probe_ollama_local_models(req.base_url, headers=headers)
if models is None:
# A failed native probe is not authoritative; fall back to the OpenAI-compatible
# catalog before accepting blindly.
# A failed native probe is not authoritative; fall back to the OpenAI-compatible catalog.
models = _m.probe_api_models(
req.api_key,
_m._normalize_openai_base_url(req.base_url),
request_headers=headers,
req.api_key, _m._normalize_openai_base_url(req.base_url), request_headers=headers,
).get("models")
if models is None:
return _soft_accept(
f"Note: could not reach this Ollama endpoint's `/api/tags` model listing to validate `{req.requested}`. "
"Hermes will save the model name, but local Ollama model discovery could not verify it."
)
match = _match_in_catalog(req.lookup, models, auto_correct=False,
suggest_label="Similar local Ollama models")
match = _match_in_catalog(req.lookup, models, auto_correct=False, suggest_label="Similar local Ollama models")
if match.exact:
return _accept()
empty_hint = " No models are currently listed by `/api/tags`." if not models else ""
@@ -258,43 +247,36 @@ def _validate_custom(req: _Request) -> dict[str, Any]:
from hermes_cli import models as _m
# Probe with the auth shape the api_mode expects.
if req.api_mode == "anthropic_messages":
probe = _m.probe_api_models(req.api_key, req.base_url, api_mode=req.api_mode,
request_headers=req.headers)
else:
probe = _m.probe_api_models(req.api_key, req.base_url, request_headers=req.headers)
anthropic_style = req.api_mode == "anthropic_messages"
probe_kwargs = {"api_mode": req.api_mode} if anthropic_style else {}
probe = _m.probe_api_models(req.api_key, req.base_url, request_headers=req.headers, **probe_kwargs)
api_models = probe.get("models")
if api_models is not None:
match = _match_in_catalog(req.lookup, api_models, suggest_query=req.requested)
if match.exact:
return _accept()
if match.corrected:
return _corrected(req.requested, match.corrected)
verdict = match.verdict(req)
if verdict is not None:
return verdict
message = (
f"Note: `{req.requested}` was not found in this custom endpoint's model listing "
f"({probe.get('probed_url')}). It may still work if the server supports hidden or aliased models."
f"{match.suggestion_text}"
)
if probe.get("used_fallback"):
message += (
f"\n Endpoint verification succeeded after trying `{probe.get('resolved_base_url')}`. "
f"Consider saving that as your base URL."
)
message += (f"\n Endpoint verification succeeded after trying `{probe.get('resolved_base_url')}`. "
"Consider saving that as your base URL.")
return _soft_accept(message)
message = (
f"Note: could not reach this custom endpoint's model listing at `{probe.get('probed_url')}`. "
f"Hermes will still save `{req.requested}`, but the endpoint should expose `/models` for verification."
)
if req.api_mode == "anthropic_messages":
message += (
"\n Many Anthropic-compatible proxies do not implement the Models API "
"(GET /v1/models). The model name has been accepted without verification."
)
if anthropic_style:
message += ("\n Many Anthropic-compatible proxies do not implement the Models API (GET /v1/models). "
"The model name has been accepted without verification.")
if probe.get("suggested_base_url"):
message += f"\n If this server expects `/v1`, try base URL: `{probe.get('suggested_base_url')}`"
# Anthropic-style proxies routinely lack /v1/models, so only they are accepted unverified.
return _verdict(req.api_mode == "anthropic_messages", True, False, message)
return _verdict(anthropic_style, True, False, message)
def _static_catalog(normalized: str) -> list[str]:
@@ -330,19 +312,16 @@ def _validate_static_catalog(req: _Request) -> Optional[dict[str, Any]]:
return _accept()
base_guess = req.lookup[: -len(CODEX_CONTEXT_VARIANT_SUFFIX)]
return _reject(
f"`{req.requested}` is not a valid large-context variant — "
f"`{base_guess}` enforces the standard 272K window on "
f"Codex, so no `-900k` option exists for it. Pick the "
f"base model, or a verified variant from the `/model` "
f"picker (e.g. `gpt-5.6-sol-900k`)."
f"`{req.requested}` is not a valid large-context variant — `{base_guess}` enforces the "
"standard 272K window on Codex, so no `-900k` option exists for it. Pick the base model, "
"or a verified variant from the `/model` picker (e.g. `gpt-5.6-sol-900k`)."
)
if not catalog:
return None
match = _match_in_catalog(req.lookup, catalog)
if match.exact:
return _accept()
if match.corrected:
return _corrected(req.requested, match.corrected)
verdict = match.verdict(req)
if verdict is not None:
return verdict
label = _STATIC_LABELS[req.normalized]
# Plausibility gate: the soft-accept exists for entitlement-gated *hidden* slugs the curated
# listing hasn't caught up with — always the provider's own family (gpt-* / grok-*). An
@@ -352,12 +331,9 @@ def _validate_static_catalog(req: _Request) -> Optional[dict[str, Any]]:
lower = req.lookup.strip().lower()
if prefixes and not any(lower.startswith(p) for p in prefixes):
return _reject(
f"`{req.requested}` doesn't look like a {label} model "
f"and isn't in its listing, so it was not accepted. If it "
f"belongs to another configured provider, switch with "
f"`--provider <slug>` (or select it from the `/model` "
f"picker)."
f"{match.suggestion_text}"
f"`{req.requested}` doesn't look like a {label} model and isn't in its listing, so it was not "
"accepted. If it belongs to another configured provider, switch with `--provider <slug>` "
f"(or select it from the `/model` picker).{match.suggestion_text}"
)
return _soft_accept(
f"Note: `{req.requested}` was not found in the {label} model listing. "
@@ -373,11 +349,7 @@ def _validate_minimax(req: _Request) -> Optional[dict[str, Any]]:
if not catalog:
return None
match = _match_in_catalog(req.lookup, catalog, case_insensitive=True)
if match.exact:
return _accept()
if match.corrected:
return _corrected(req.requested, match.corrected)
return _soft_accept(
return match.verdict(req) or _soft_accept(
f"Note: `{req.requested}` was not found in the MiniMax catalog."
f"{match.suggestion_text}"
"\n MiniMax does not expose a /models endpoint, so Hermes cannot verify the model name."
@@ -387,21 +359,17 @@ def _validate_minimax(req: _Request) -> Optional[dict[str, Any]]:
def _validate_anthropic(req: _Request) -> Optional[dict[str, Any]]:
"""Native Anthropic: /v1/models needs x-api-key (or OAuth Bearer) + anthropic-version, so the
generic Bearer probe 401s — use the native fetcher. Returns None (fall through to the generic
ladder) when no token is resolvable or the network failed."""
generic Bearer probe 401s — use the native fetcher. None (fall through) when no token is
resolvable or the network failed."""
from hermes_cli import models as _m
models = _m._fetch_anthropic_models(base_url=req.base_url or None, api_key=req.api_key or None)
if models is None:
return None
match = _match_in_catalog(req.lookup, models, suggest_query=req.requested)
if match.exact:
return _accept()
if match.corrected:
return _corrected(req.requested, match.corrected)
# Accept anyway — Anthropic gates newer/preview models (snapshot IDs, early access) behind
# accounts even though they aren't listed on /v1/models.
return _soft_accept(
return match.verdict(req) or _soft_accept(
f"Note: `{req.requested}` was not found in Anthropic's /v1/models listing. "
f"It may still work if you have early-access or snapshot IDs."
f"{match.suggestion_text}"
@@ -414,17 +382,11 @@ def _validate_anthropic_messages(req: _Request) -> dict[str, Any]:
from hermes_cli import models as _m
models = _m.fetch_api_models(req.api_key, req.base_url, api_mode=req.api_mode)
if models is not None:
match = _match_in_catalog(req.lookup, models)
if match.exact:
return _accept()
if match.corrected:
return _corrected(req.requested, match.corrected)
return _soft_accept(
f"Note: could not verify `{req.requested}` against this endpoint's "
f"model listing. Many Anthropic-compatible proxies do not "
f"implement GET /v1/models. The model name has been accepted "
f"without verification."
verdict = _match_in_catalog(req.lookup, models).verdict(req) if models is not None else None
return verdict or _soft_accept(
f"Note: could not verify `{req.requested}` against this endpoint's model listing. Many "
"Anthropic-compatible proxies do not implement GET /v1/models. The model name has been accepted "
"without verification."
)
@@ -453,12 +415,9 @@ def _validate_live_listing(req: _Request) -> Optional[dict[str, Any]]:
if api_models is None:
return None
if req.normalized == "gemini":
# Gemini's OpenAI-compat listing prefixes ids with "models/"; curated list and user
# input use the bare id, so strip before comparing.
api_models = [
m[len("models/"):] if isinstance(m, str) and m.startswith("models/") else m
for m in api_models
]
# Gemini's OpenAI-compat listing prefixes ids with "models/"; curated list and user input
# use the bare id, so strip before comparing.
api_models = [m[len("models/"):] if isinstance(m, str) and m.startswith("models/") else m for m in api_models]
match = _match_in_catalog(req.lookup, api_models)
if match.exact:
return _accept()
@@ -471,9 +430,9 @@ def _validate_live_listing(req: _Request) -> Optional[dict[str, Any]]:
return _accept()
# Listed but not found: the account may reach models absent from the public listing
# (e.g. Z.AI Pro/Max plans use glm-5 on coding endpoints) — warn but allow where plausible.
if match.corrected:
corrected = req.with_preset_suffix(match.corrected)
return _corrected(req.requested, corrected)
verdict = match.verdict(req, keep_suffix=True)
if verdict is not None:
return verdict
# Curated-catalog soft-accept: providers omit valid models from live listings (stale cache,
# partial rollout, gated previews). EXCEPTION: official OpenAI hosts (canonical + data-
# residency regional) — their listing is access-scoped and authoritative, so an absent model
@@ -487,20 +446,15 @@ def _validate_live_listing(req: _Request) -> Optional[dict[str, Any]]:
if not listing_authoritative and _m._model_in_provider_catalog(
(variant_base or req.lookup).lower(), _m._provider_keys(req.normalized)
):
return _verdict(True, True, True,
f"Note: `{req.requested}` was not found in the live /v1/models listing "
f"but exists in the curated catalog — accepted.")
return _accept_with_note(f"Note: `{req.requested}` was not found in the live /v1/models listing "
"but exists in the curated catalog — accepted.")
# Nous: the Portal's recommended-models feed can list a model before the curated list or the
# docs-hosted manifest catches up; `hermes chat` already accepts those at model-list build
# time, so mirror that source of truth for per-message /model validation.
if req.normalized == "nous" and req.lookup.lower() in _nous_portal_recommended_names():
return _verdict(True, True, True,
f"Note: `{req.requested}` was not found in the live /v1/models "
f"listing but is a current Nous Portal recommendation — accepted.")
return _reject(
f"Model `{req.requested}` was not found in this provider's model listing."
f"{match.suggestion_text}"
)
return _accept_with_note(f"Note: `{req.requested}` was not found in the live /v1/models listing "
"but is a current Nous Portal recommendation — accepted.")
return _reject(f"Model `{req.requested}` was not found in this provider's model listing.{match.suggestion_text}")
def _validate_bedrock(req: _Request) -> Optional[dict[str, Any]]:
@@ -511,8 +465,7 @@ def _validate_bedrock(req: _Request) -> Optional[dict[str, Any]]:
region = resolve_bedrock_runtime_region()
discovered_ids = {m["id"] for m in discover_bedrock_models(region)}
match = _match_in_catalog(req.requested, list(discovered_ids), auto_correct=False,
suggest_cutoff=0.4)
match = _match_in_catalog(req.requested, list(discovered_ids), auto_correct=False, suggest_cutoff=0.4)
if match.exact:
return _accept()
# Still accept (custom inference profiles / cross-account access), but warn.
@@ -526,19 +479,16 @@ def _validate_bedrock(req: _Request) -> Optional[dict[str, Any]]:
def _validate_catalog_fallback(req: _Request) -> dict[str, Any]:
"""The /models probe was unreachable: validate against the curated ``provider_model_ids()``
list so gateway /model switches keep working while a provider's endpoint is down (otherwise
switch_model() would fail and the gateway never writes the session override). No catalog at
all → accept with a warning."""
"""/models unreachable: validate against the curated ``provider_model_ids()`` list so gateway
/model switches keep working while a provider's endpoint is down (otherwise switch_model() would
fail and the gateway never writes the session override). No catalog → accept with a warning."""
from hermes_cli import models as _m
label = _m._PROVIDER_LABELS.get(req.normalized, req.normalized)
catalog = _static_catalog(req.normalized)
if not catalog:
return _soft_accept(
f"Note: could not reach the {label} API to validate `{req.requested}`. "
f"If the service isn't down, this model may not be valid."
)
return _soft_accept(f"Note: could not reach the {label} API to validate `{req.requested}`. "
"If the service isn't down, this model may not be valid.")
match = _match_in_catalog(req.lookup, catalog, case_insensitive=True)
if match.exact:
return _accept()
@@ -547,10 +497,7 @@ def _validate_catalog_fallback(req: _Request) -> dict[str, Any]:
variant_base = _m._openrouter_variant_base(req.lookup)
if variant_base is not None and variant_base.lower() in {m.lower() for m in catalog}:
return _accept()
if match.corrected:
corrected = req.with_preset_suffix(match.corrected)
return _corrected(req.requested, corrected)
return _soft_accept(
return match.verdict(req, keep_suffix=True) or _soft_accept(
f"Note: `{req.requested}` was not found in the {label} curated catalog "
f"and the /models endpoint was unreachable.{match.suggestion_text}"
f"\n The model may still work if it exists on the provider."
@@ -559,6 +506,36 @@ def _validate_catalog_fallback(req: _Request) -> dict[str, Any]:
# ── Orchestrator ─────────────────────────────────────────────────────────
def _is_custom(req: _Request) -> bool:
return req.normalized == "custom" or req.normalized.startswith("custom:")
def _for(*providers: str) -> Callable[[_Request], bool]:
return lambda req: req.normalized in providers
# (gate, branch): the branch runs when the gate passes; the first non-None verdict wins. ORDER IS
# BEHAVIOR: moa → whitespace → OpenRouter preset parse → LM Studio → Ollama native → custom →
# codex/xai static → MiniMax → Anthropic native → Anthropic Messages → live listing → Bedrock →
# curated-catalog fallback (always decides).
_LADDER: tuple[tuple[Callable[[_Request], bool], Callable[[_Request], Optional[dict[str, Any]]]], ...] = (
(_for("moa"), _validate_moa),
(lambda req: True, _reject_whitespace),
(_for("openrouter"), _parse_openrouter_preset),
(_for("lmstudio"), _validate_lmstudio),
(lambda req: True, _validate_ollama_native),
(_is_custom, _validate_custom),
(_for("openai-codex", "xai-oauth"), _validate_static_catalog),
(_for("minimax", "minimax-cn"), _validate_minimax),
(_for("anthropic"), _validate_anthropic),
(lambda req: req.api_mode == "anthropic_messages", _validate_anthropic_messages),
(lambda req: True, _validate_live_listing),
# API unreachable — accept and persist, but warn so typos don't silently break things.
(_for("bedrock"), _validate_bedrock),
(lambda req: True, _validate_catalog_fallback),
)
def validate_requested_model(
model_name: str,
provider: Optional[str],
@@ -568,13 +545,9 @@ def validate_requested_model(
api_mode: Optional[str] = None,
headers: Optional[dict[str, str]] = None,
) -> dict[str, Any]:
"""Validate a ``/model`` value for the active provider.
Returns a dict with: - accepted: whether the CLI should switch to the requested model now -
persist: whether it is safe to save to config - recognized: whether it matched a known provider
catalog - message: optional warning / guidance for the user (- corrected_model: when a typo
was auto-corrected).
"""
"""Validate a ``/model`` value for the active provider → dict with ``accepted`` (switch now),
``persist`` (safe to save to config), ``recognized`` (matched a known provider catalog),
``message`` (optional warning / guidance) and ``corrected_model`` when a typo was fixed."""
from hermes_cli import models as _m
requested = (model_name or "").strip()
@@ -587,43 +560,10 @@ def validate_requested_model(
if not requested:
return _reject("Model name cannot be empty.")
if normalized == "moa":
return _validate_moa(requested)
if any(ch.isspace() for ch in requested):
return _reject("Model names cannot contain spaces.")
req = _Request(requested, lookup, provider, normalized, api_key, base_url, api_mode, headers)
if normalized == "openrouter":
verdict = _parse_openrouter_preset(req)
if verdict is not None:
return verdict
if normalized == "lmstudio":
return _validate_lmstudio(req)
verdict = _validate_ollama_native(req)
if verdict is not None:
return verdict
if normalized == "custom" or normalized.startswith("custom:"):
return _validate_custom(req)
if normalized in {"openai-codex", "xai-oauth"}:
verdict = _validate_static_catalog(req)
if verdict is not None:
return verdict
if normalized in {"minimax", "minimax-cn"}:
verdict = _validate_minimax(req)
if verdict is not None:
return verdict
if normalized == "anthropic":
verdict = _validate_anthropic(req)
if verdict is not None:
return verdict
if api_mode == "anthropic_messages":
return _validate_anthropic_messages(req)
verdict = _validate_live_listing(req)
if verdict is not None:
return verdict
# API unreachable — accept and persist, but warn so typos don't silently break things.
if normalized == "bedrock":
verdict = _validate_bedrock(req)
if verdict is not None:
return verdict
return _validate_catalog_fallback(req)
for gate, branch in _LADDER:
if gate(req):
verdict = branch(req)
if verdict is not None:
return verdict
raise AssertionError("unreachable: _validate_catalog_fallback always decides")