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

Merge simp/r3-21-A (models/catalog/normalize/guards)

Browse Source
This commit is contained in:
Teknium
2026-09-02 22:18:17 -07:00
parent 113f04616b 2a9481274c
commit 8f24dd4c70
7 changed files with 799 additions and 1910 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/model_catalog.py
+72 -202
View File
@@ -1,13 +1,9 @@
"""Remote model catalog fetcher.
Pipeline -------- 1. ``get_catalog()`` — returns a parsed manifest dict. - Checks in-process cache
(invalidated by TTL). - Reads disk cache at ``~/.hermes/cache/model_catalog.json``. - Fetches the
master URL if disk cache is stale or missing. - On any fetch failure, keeps using the stale cache
(or empty dict).
2. ``get_curated_openrouter_models()`` / ``get_curated_nous_models()`` — thin accessors returning
the shapes existing callers expect. Each falls back to the in-repo hardcoded list on any lookup
failure.
``get_catalog()`` returns the parsed manifest: in-process cache (TTL) → disk cache at
``~/.hermes/cache/model_catalog.json`` → master URL fetch; any fetch failure keeps the stale copy
(or ``{}``). ``get_curated_openrouter_models()`` / ``get_curated_nous_models()`` are thin accessors
whose callers fall back to the in-repo lists on ``None``.
"""
from __future__ import annotations
@@ -26,43 +22,26 @@ from utils import atomic_replace
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
DEFAULT_CATALOG_URL = (
"https://hermes-agent.nousresearch.com/docs/api/model-catalog.json"
)
# Fallback fetch chain. The Docusaurus site is served through Vercel, which
# occasionally returns HTTP 403 + x-vercel-mitigated: challenge for non-
# browser clients (urllib, curl). When that happens the disk cache goes
# stale and new model releases never reach the picker. The raw GitHub URL
# is the same manifest published from the same repo and is not bot-gated,
# so we fall through to it whenever the primary URL fails.
"https://hermes-agent.nousresearch.com/docs/api/model-catalog.json")
# The Docusaurus site sits behind Vercel, which occasionally 403s non-browser clients (bot
# challenge); the raw GitHub copy is the same manifest and is not bot-gated.
DEFAULT_CATALOG_FALLBACK_URLS: tuple[str, ...] = (
"https://raw.githubusercontent.com/NousResearch/hermes-agent/main/website/static/api/model-catalog.json",
)
DEFAULT_TTL_MINUTES = 20
# Legacy key. ``ttl_hours`` is honoured only when the user set it explicitly;
# the shipped default is ``ttl_minutes`` above.
# Legacy key, honoured only when the user set it explicitly; ``ttl_minutes`` is the shipped default.
DEFAULT_TTL_HOURS = DEFAULT_TTL_MINUTES / 60.0
DEFAULT_FETCH_TIMEOUT = 8.0
SUPPORTED_SCHEMA_VERSION = 1
_HERMES_USER_AGENT = f"hermes-cli/{_HERMES_VERSION}"
# In-process cache to avoid repeated disk + parse work across multiple
# calls within the same session. Invalidated by TTL against the disk file's
# mtime, so calling code never has to think about this.
# In-process cache, invalidated against the disk file's mtime and TTL.
_catalog_cache: dict[str, Any] | None = None
_catalog_cache_source_mtime: float = 0.0
# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------
def _load_catalog_config() -> dict[str, Any]:
"""Load the ``model_catalog`` config block with defaults filled in."""
try:
@@ -70,16 +49,12 @@ def _load_catalog_config() -> dict[str, Any]:
cfg = load_config() or {}
except Exception:
cfg = {}
raw = cfg.get("model_catalog")
if not isinstance(raw, dict):
raw = {}
# ``ttl_minutes`` is the shipped default (20). ``ttl_hours`` is the legacy
# key: honoured when a user set it explicitly and ``ttl_minutes`` is still
# at its default (load_config() deep-merges the default in, so "present"
# alone doesn't mean "user-set"), so old customized configs keep their
# chosen window.
# ``ttl_hours`` (legacy) is honoured only when ``ttl_minutes`` is still at its default —
# load_config() deep-merges the default in, so "present" alone doesn't mean "user-set".
ttl_minutes = raw.get("ttl_minutes")
try:
ttl_minutes = float(ttl_minutes) if ttl_minutes not in (None, "") else DEFAULT_TTL_MINUTES
@@ -97,31 +72,19 @@ def _load_catalog_config() -> dict[str, Any]:
"enabled": bool(raw.get("enabled", True)),
"url": str(raw.get("url") or DEFAULT_CATALOG_URL),
"ttl_hours": ttl_minutes / 60.0,
"providers": raw.get("providers") if isinstance(raw.get("providers"), dict) else {},
}
"providers": raw.get("providers") if isinstance(raw.get("providers"), dict) else {}}
def _cache_path() -> Path:
"""Return the disk cache path. Import lazily so tests can monkeypatch home."""
"""Disk cache path; imported lazily so tests can monkeypatch home."""
from hermes_constants import get_hermes_home
return get_hermes_home() / "cache" / "model_catalog.json"
# ---------------------------------------------------------------------------
# Fetch + validate + cache
# ---------------------------------------------------------------------------
def _fetch_manifest(url: str, timeout: float) -> dict[str, Any] | None:
"""HTTP GET the manifest URL and return a parsed dict, or None on failure."""
"""HTTP GET the manifest URL and return a validated dict, or None on failure."""
try:
req = urllib.request.Request(
url,
headers={
"Accept": "application/json",
"User-Agent": _HERMES_USER_AGENT,
},
)
req = urllib.request.Request(url, headers={"Accept": "application/json", "User-Agent": _HERMES_USER_AGENT})
with urllib.request.urlopen(req, timeout=timeout) as resp:
data = json.loads(resp.read().decode())
except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, OSError) as exc:
@@ -130,25 +93,17 @@ def _fetch_manifest(url: str, timeout: float) -> dict[str, Any] | None:
except Exception as exc: # pragma: no cover — defensive
logger.info("model catalog fetch errored (%s): %s", url, exc)
return None
if not _validate_manifest(data):
logger.info("model catalog at %s failed schema validation", url)
return None
return data
def _fetch_manifest_with_fallback(
primary_url: str,
timeout: float,
fallback_urls: tuple[str, ...] = DEFAULT_CATALOG_FALLBACK_URLS,
primary_url: str, timeout: float, fallback_urls: tuple[str, ...] = DEFAULT_CATALOG_FALLBACK_URLS
) -> dict[str, Any] | None:
"""Try ``primary_url`` first, then walk ``fallback_urls``.
Returns the first manifest that fetches and validates, or None when every URL fails. Skips
fallback URLs identical to the primary so an operator who configured the catalog URL to point at
the raw GitHub copy doesn't double-fetch.
"""
"""First manifest that fetches and validates from ``primary_url`` then ``fallback_urls`` (skipping
any equal to the primary so a raw-GitHub-configured operator doesn't double-fetch), or None."""
data = _fetch_manifest(primary_url, timeout)
if data is not None:
return data
@@ -168,9 +123,7 @@ def _validate_manifest(data: Any) -> bool:
return False
version = data.get("version")
if not isinstance(version, int) or version > SUPPORTED_SCHEMA_VERSION:
# Future schema version we don't understand — refuse rather than
# guess. Older schemas (version < 1) aren't supported either.
return False
return False # future schema we don't understand — refuse rather than guess
providers = data.get("providers")
if not isinstance(providers, dict):
return False
@@ -180,11 +133,8 @@ def _validate_manifest(data: Any) -> bool:
models = pblock.get("models")
if not isinstance(models, list):
return False
for m in models:
if not isinstance(m, dict):
return False
if not isinstance(m.get("id"), str) or not m["id"].strip():
return False
if not all(isinstance(m, dict) and isinstance(m.get("id"), str) and m["id"].strip() for m in models):
return False
return True
@@ -193,16 +143,11 @@ def _read_disk_cache() -> tuple[dict[str, Any] | None, float]:
path = _cache_path()
try:
mtime = path.stat().st_mtime
except (OSError, FileNotFoundError):
return (None, 0.0)
try:
with open(path, encoding="utf-8") as fh:
data = json.load(fh)
except (OSError, json.JSONDecodeError):
return (None, 0.0)
if not _validate_manifest(data):
return (None, 0.0)
return (data, mtime)
return (data, mtime) if _validate_manifest(data) else (None, 0.0)
def _write_disk_cache(data: dict[str, Any]) -> None:
@@ -218,9 +163,8 @@ def _write_disk_cache(data: dict[str, Any]) -> None:
logger.info("model catalog cache write failed: %s", exc)
# Stale-while-revalidate machinery: at most one background manifest refresh
# in flight per process. The refreshed manifest lands on disk; the NEXT
# get_catalog() call picks it up via the mtime check.
# Stale-while-revalidate: at most one background manifest refresh in flight per process. The
# refreshed manifest lands on disk; the NEXT get_catalog() call picks it up via the mtime check.
_catalog_swr_lock = threading.Lock()
_catalog_swr_inflight = False
@@ -248,73 +192,42 @@ def _spawn_catalog_swr_refresh(url: str) -> None:
threading.Thread(target=_refresh, daemon=True, name="model-catalog-swr").start()
# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------
def _remember(data: dict[str, Any], mtime: float) -> dict[str, Any]:
global _catalog_cache, _catalog_cache_source_mtime
_catalog_cache, _catalog_cache_source_mtime = data, mtime
return data
def get_catalog(*, force_refresh: bool = False) -> dict[str, Any]:
"""Return the parsed model catalog manifest, or an empty dict on failure.
Callers should treat a missing provider/model as "use the in-repo fallback" — never raise from
this function so the CLI keeps working offline.
"""
global _catalog_cache, _catalog_cache_source_mtime
"""Parsed model catalog manifest, or ``{}`` on failure — never raises, so the CLI works offline
(callers treat a missing provider/model as "use the in-repo fallback")."""
cfg = _load_catalog_config()
if not cfg["enabled"]:
return {}
ttl_seconds = max(0.0, cfg["ttl_hours"] * 3600.0)
disk_data, disk_mtime = _read_disk_cache()
now = time.time()
disk_fresh = disk_data is not None and (now - disk_mtime) < ttl_seconds
# In-process cache hit: disk hasn't changed since we loaded it and still fresh.
if (
not force_refresh
and _catalog_cache is not None
and disk_data is not None
and disk_mtime == _catalog_cache_source_mtime
and disk_fresh
):
return _catalog_cache
# Disk is fresh enough — use it without a network hit.
if not force_refresh and disk_fresh and disk_data is not None:
_catalog_cache = disk_data
_catalog_cache_source_mtime = disk_mtime
return disk_data
# Stale-while-revalidate: an expired disk copy is served immediately and
# refreshed off-thread, so interactive surfaces (the /model picker calls
# this via get_curated_nous_model_ids on every open) never block on the
# manifest fetch. Only a cold cache (no disk copy at all) still blocks.
if not force_refresh and disk_data is not None:
_catalog_cache = disk_data
_catalog_cache_source_mtime = disk_mtime
_spawn_catalog_swr_refresh(cfg["url"])
return disk_data
if disk_fresh and _catalog_cache is not None and disk_mtime == _catalog_cache_source_mtime:
return _catalog_cache
if not disk_fresh:
# Stale-while-revalidate: serve the expired disk copy now and refresh off-thread so the
# /model picker (which calls this on every open) never blocks on the manifest fetch.
# Only a cold cache (no disk copy at all) still blocks.
_spawn_catalog_swr_refresh(cfg["url"])
return _remember(disk_data, disk_mtime)
# Need to (re)fetch. If it fails, fall back to any stale disk copy.
fetched = _fetch_manifest_with_fallback(cfg["url"], DEFAULT_FETCH_TIMEOUT)
if fetched is not None:
_write_disk_cache(fetched)
new_disk_data, new_mtime = _read_disk_cache()
if new_disk_data is not None:
_catalog_cache = new_disk_data
_catalog_cache_source_mtime = new_mtime
return new_disk_data
_catalog_cache = fetched
_catalog_cache_source_mtime = now
return fetched
return _remember(new_disk_data, new_mtime)
return _remember(fetched, now)
if disk_data is not None:
_catalog_cache = disk_data
_catalog_cache_source_mtime = disk_mtime
return disk_data
return _remember(disk_data, disk_mtime)
return {}
@@ -324,12 +237,9 @@ def refresh_interval_seconds() -> float:
def refresh_catalogs() -> bool:
"""Force-refresh every remote model catalog the picker reads from.
Fetches the curated manifest, the OpenRouter live list (tool-support / free-pricing filter) and
the Nous Portal recommendations, writing each to its disk cache so the next ``/model`` open in
ANY process on this machine sees the new lists. Blocking; run it off the event loop.
"""
"""Force-refresh every remote catalog the picker reads (manifest, OpenRouter live list, Nous Portal
recommendations), writing each disk cache so the next ``/model`` open in ANY process sees them.
Blocking; run it off the event loop."""
if not _load_catalog_config()["enabled"]:
return False
catalog = get_catalog(force_refresh=True)
@@ -354,96 +264,56 @@ def _fetch_provider_override(provider: str) -> dict[str, Any] | None:
override_url = provider_cfg.get("url")
if not isinstance(override_url, str) or not override_url.strip():
return None
# Override fetches skip the disk cache because they're usually
# third-party self-hosted. Re-request on every call but with a short
# timeout so they don't block the picker.
# Overrides are usually third-party self-hosted: skip the disk cache, re-request every call.
return _fetch_manifest(override_url.strip(), DEFAULT_FETCH_TIMEOUT)
def _block_of(manifest: dict[str, Any] | None, provider: str) -> dict[str, Any] | None:
block = (manifest or {}).get("providers", {}).get(provider)
return block if isinstance(block, dict) else None
def _get_provider_block(provider: str) -> dict[str, Any] | None:
"""Return the provider's manifest block, respecting per-provider overrides."""
override = _fetch_provider_override(provider)
if override is not None:
block = override.get("providers", {}).get(provider)
if isinstance(block, dict):
return block
return _block_of(_fetch_provider_override(provider), provider) or _block_of(get_catalog(), provider)
catalog = get_catalog()
if not catalog:
return None
block = catalog.get("providers", {}).get(provider)
return block if isinstance(block, dict) else None
def _block_ids(block: dict[str, Any] | None) -> list[tuple[str, dict[str, Any]]]:
"""``(id, entry)`` for every model entry of ``block`` with a non-empty id."""
models = (block or {}).get("models", [])
return [(mid, m) for m in models if isinstance(m, dict) and (mid := str(m.get("id") or "").strip())]
def get_curated_openrouter_models() -> list[tuple[str, str]] | None:
"""Return OpenRouter's curated ``[(id, description), ...]`` from the manifest."""
block = _get_provider_block("openrouter")
if not block:
return None
out: list[tuple[str, str]] = []
for m in block.get("models", []):
mid = str(m.get("id") or "").strip()
if not mid:
continue
desc = str(m.get("description") or "")
out.append((mid, desc))
return out or None
"""OpenRouter's curated ``[(id, description), ...]`` from the manifest."""
rows = _block_ids(_get_provider_block("openrouter"))
return [(mid, str(m.get("description") or "")) for mid, m in rows] or None
def get_curated_nous_models() -> list[str] | None:
"""Return Nous Portal's curated list of model ids from the manifest."""
block = _get_provider_block("nous")
if not block:
return None
out: list[str] = []
for m in block.get("models", []):
mid = str(m.get("id") or "").strip()
if mid:
out.append(mid)
return out or None
"""Nous Portal's curated model ids from the manifest."""
return [mid for mid, _ in _block_ids(_get_provider_block("nous"))] or None
def _default_model_from_block(block: dict[str, Any] | None) -> str | None:
"""Return the id of the model entry labeled ``"default": true``, or None."""
if not isinstance(block, dict):
return None
for m in block.get("models", []):
if isinstance(m, dict) and m.get("default"):
mid = str(m.get("id") or "").strip()
if mid:
return mid
return None
"""Id of the model entry labeled ``"default": true``, or None."""
return next((mid for mid, m in _block_ids(block) if m.get("default")), None)
def get_default_model_from_cache(provider: str) -> str | None:
"""Return the catalog's labeled default model for ``provider`` — cache only.
The manifest marks exactly one model entry per provider with ``"default": true``; that entry is
the model Hermes silently lands on when the user never picked one.
"""
if _catalog_cache is not None:
block = _catalog_cache.get("providers", {}).get(provider)
found = _default_model_from_block(block)
if found:
return found
"""The manifest's labeled default for ``provider`` (the model Hermes silently lands on when the
user never picked one) — in-process then disk cache only, never a fetch."""
found = _default_model_from_block(_block_of(_catalog_cache, provider)) if _catalog_cache is not None else None
if found:
return found
disk_data, _mtime = _read_disk_cache()
if disk_data is not None:
block = disk_data.get("providers", {}).get(provider)
return _default_model_from_block(block)
return None
return _default_model_from_block(_block_of(disk_data, provider)) if disk_data is not None else None
def seed_cache_from_checkout(project_root: "Path | str") -> bool:
"""Overwrite the disk cache with the catalog shipped in a local checkout.
``hermes update`` pulls the latest repo, so the freshly-pulled ``website/static/api/model-
catalog.json`` IS the newest catalog — no network round-trip needed. Copying it straight over
the disk cache keeps the model picker current even when the remote manifest fetch is bot-gated
or the Portal hiccups.
Reads the shipped manifest, validates it against the schema, and writes it to
``~/.hermes/cache/model_catalog.json`` via the same atomic writer the network path uses.
"""
"""Overwrite the disk cache with the checkout's ``website/static/api/model-catalog.json``.
After ``hermes update`` that file IS the newest catalog, so the picker stays current even when
the remote fetch is bot-gated. Validated, then written via the same atomic writer."""
src = Path(project_root) / "website" / "static" / "api" / "model-catalog.json"
try:
with open(src, encoding="utf-8") as fh:
hermes_cli/model_cost_guard.py
+14 -50
View File
@@ -28,10 +28,8 @@ class ExpensiveModelWarning:
def _to_decimal(value: object) -> Optional[Decimal]:
if value is None:
return None
try:
return Decimal(str(value))
return None if value is None else Decimal(str(value))
except (InvalidOperation, ValueError):
return None
@@ -41,15 +39,10 @@ def _format_money(value: Optional[Decimal]) -> str:
def _pricing_from_model_info(
model_info: Optional[ModelInfo],
) -> tuple[Optional[Decimal], Optional[Decimal], str]:
model_info: Optional[ModelInfo]) -> tuple[Optional[Decimal], Optional[Decimal], str]:
if model_info is None or not model_info.has_cost_data():
return None, None, ""
return (
_to_decimal(model_info.cost_input),
_to_decimal(model_info.cost_output),
"models.dev",
)
return _to_decimal(model_info.cost_input), _to_decimal(model_info.cost_output), "models.dev"
def _known_models_dev_provider(provider: Optional[str]) -> Optional[str]:
@@ -58,23 +51,16 @@ def _known_models_dev_provider(provider: Optional[str]) -> Optional[str]:
def _can_trust_model_info_pricing(
provider: Optional[str],
model_info: Optional[ModelInfo],
) -> bool:
provider: Optional[str], model_info: Optional[ModelInfo]) -> bool:
expected_provider = _known_models_dev_provider(provider)
if not expected_provider or model_info is None:
return False
actual_provider = str(getattr(model_info, "provider_id", "") or "").strip().lower()
return not actual_provider or actual_provider == expected_provider
def _can_trust_pricing_lookup(
model_name: str,
*,
provider: Optional[str],
base_url: Optional[str],
) -> bool:
model_name: str, *, provider: Optional[str], base_url: Optional[str]) -> bool:
try:
from agent.usage_pricing import resolve_billing_route
@@ -85,18 +71,11 @@ def _can_trust_pricing_lookup(
def expensive_model_warning(
model_name: str,
*,
provider: Optional[str] = None,
base_url: Optional[str] = None,
api_key: Optional[str] = None,
model_info: Optional[ModelInfo] = None,
model_name: str, *, provider: Optional[str] = None, base_url: Optional[str] = None,
api_key: Optional[str] = None, model_info: Optional[ModelInfo] = None,
) -> Optional[ExpensiveModelWarning]:
"""Return a warning payload when known pricing exceeds safety thresholds.
The guard only triggers when pricing is known. Callers should use this after model resolution so
aliases and provider-specific model IDs have settled.
"""
"""Warning payload when KNOWN pricing exceeds the safety thresholds (never fires on unknown
pricing). Call after model resolution so aliases / provider-specific ids have settled."""
model = (model_name or "").strip()
if not model:
return None
@@ -104,7 +83,6 @@ def expensive_model_warning(
input_cost: Optional[Decimal] = None
output_cost: Optional[Decimal] = None
source = ""
if _can_trust_model_info_pricing(provider, model_info):
input_cost, output_cost, source = _pricing_from_model_info(model_info)
@@ -115,9 +93,7 @@ def expensive_model_warning(
try:
from agent.models_dev import get_model_info
input_cost, output_cost, source = _pricing_from_model_info(
get_model_info(provider, model)
)
input_cost, output_cost, source = _pricing_from_model_info(get_model_info(provider, model))
except Exception:
pass
@@ -125,12 +101,7 @@ def expensive_model_warning(
try:
from agent.usage_pricing import get_pricing_entry
entry = get_pricing_entry(
model,
provider=provider,
base_url=base_url,
api_key=api_key,
)
entry = get_pricing_entry(model, provider=provider, base_url=base_url, api_key=api_key)
except Exception:
entry = None
if entry is not None:
@@ -139,7 +110,6 @@ def expensive_model_warning(
source = entry.source
is_known_gpt55_pro_confusion = model.lower() == GPT55_PRO_OPENROUTER_ID
over_input = input_cost is not None and input_cost > INPUT_COST_WARNING_THRESHOLD
over_output = output_cost is not None and output_cost > OUTPUT_COST_WARNING_THRESHOLD
if not over_input and not over_output and not is_known_gpt55_pro_confusion:
@@ -151,8 +121,7 @@ def expensive_model_warning(
f"{model} has known pricing above Hermes' safety threshold.",
f"Input tokens: {_format_money(input_cost)}",
f"Output tokens: {_format_money(output_cost)}",
"Threshold: more than $20/M input tokens or more than $100/M output tokens.",
]
"Threshold: more than $20/M input tokens or more than $100/M output tokens."]
if source:
lines.append(f"Pricing source: {source}.")
if is_known_gpt55_pro_confusion:
@@ -160,10 +129,5 @@ def expensive_model_warning(
lines.append("Confirm only if you intend to use this model.")
return ExpensiveModelWarning(
model=model,
provider=(provider or "").strip(),
input_cost_per_million=input_cost,
output_cost_per_million=output_cost,
source=source or "unknown",
message="\n".join(lines),
)
model=model, provider=(provider or "").strip(), input_cost_per_million=input_cost,
output_cost_per_million=output_cost, source=source or "unknown", message="\n".join(lines))
hermes_cli/model_data_policy_guard.py
+12 -32
View File
@@ -1,11 +1,8 @@
"""Data-policy confirmation helpers for model selection surfaces.
Some inference tiers are cheap *because* the vendor trains future models on your prompts and
completions. Selecting one for the low price without realising the data trade-off is a real footgun.
Why a static table (not a ProviderProfile hook): the guard runs inside core selection code
(``auth.py`` / ``web_server.py``), which never calls into the active provider profile for a
selection-time warning.
Some inference tiers are cheap *because* the vendor trains on your prompts. A static rule table (not
a ProviderProfile hook) because the guard runs inside core selection code (``auth.py`` /
``web_server.py``), which never calls into the active provider profile.
"""
from __future__ import annotations
@@ -23,18 +20,12 @@ class DataTrainingWarning:
message: str
# ── Rule table ────────────────────────────────────────────────────────────
# Each rule: (human label, predicate over (model_lower, provider_lower), message).
# Extensible — new data-collection tiers from other vendors slot in here without
# touching the call sites. Predicates are intentionally conservative: match an
# explicit, vendor-documented id rather than guessing from price alone (price is
# only a corroborating signal and can change).
# Rule predicates are deliberately conservative: match an explicit, vendor-documented id rather
# than guessing from price (a corroborating signal that can change).
def _is_meta_contributor(model_lower: str, provider_lower: str) -> bool:
# Meta Model API "contributor" tier (muse-spark-1.2-contributor and any
# future -contributor checkpoints). Match on the id suffix; do not require a
# specific provider id so it fires whether selected via the meta-ai plugin,
# a gateway, or a custom endpoint that serves the same model id.
# Meta "contributor" tier, matched on the id alone (no provider check) so it fires whether
# selected via the meta-ai plugin, a gateway, or a custom endpoint serving the same id.
return model_lower.endswith("-contributor") or "contributor" in model_lower.split("-")
@@ -71,27 +62,16 @@ def data_training_warning(
provider: Optional[str] = None,
base_url: Optional[str] = None, # noqa: ARG001 — reserved for host-scoped rules
) -> Optional[DataTrainingWarning]:
"""Return a warning payload when *model_name* selects a data-training tier.
Returns ``None`` when no rule matches (the common case). Callers should run this after model
resolution so aliases / provider-specific ids have settled, and surface ``.message`` as a
confirm prompt.
"""
"""Warning payload when *model_name* selects a data-training tier, else ``None``. Call after model
resolution; surface ``.message`` as a confirm prompt."""
model = (model_name or "").strip()
if not model:
return None
model_lower = model.lower()
provider_lower = (provider or "").strip().lower()
model_lower, provider_lower = model.lower(), (provider or "").strip().lower()
for predicate, message in _RULES:
try:
if predicate(model_lower, provider_lower):
return DataTrainingWarning(
model=model,
provider=(provider or "").strip(),
message=message,
)
return DataTrainingWarning(model=model, provider=(provider or "").strip(), message=message)
except Exception:
# A misbehaving predicate must never break model selection.
continue
continue # a misbehaving predicate must never break model selection
return None
hermes_cli/model_normalize.py
+59 -223
View File
@@ -5,15 +5,8 @@ from __future__ import annotations
import re
from typing import Optional
# ---------------------------------------------------------------------------
# Vendor prefix mapping
# ---------------------------------------------------------------------------
# Maps the first hyphen-delimited token of a bare model name to the vendor
# slug used by aggregator APIs (OpenRouter, Nous, etc.).
#
# Example: "claude-sonnet-4.6" -> first token "claude" -> vendor "anthropic"
# -> aggregator slug: "anthropic/claude-sonnet-4.6"
# First hyphen-delimited token of a bare model name → vendor slug used by aggregator APIs
# ("claude-sonnet-4.6" → "anthropic/claude-sonnet-4.6").
_VENDOR_PREFIXES: dict[str, str] = {
"claude": "anthropic",
"gpt": "openai",
@@ -32,34 +25,23 @@ _VENDOR_PREFIXES: dict[str, str] = {
"trinity": "arcee-ai",
"nemotron": "nvidia",
"llama": "meta-llama",
"step": "stepfun",
"trinity": "arcee-ai",
}
"step": "stepfun"}
# Providers whose APIs consume vendor/model slugs.
_AGGREGATOR_PROVIDERS: frozenset[str] = frozenset({
"openrouter",
"nous",
"ai-gateway",
"kilocode",
})
"openrouter", "nous", "ai-gateway", "kilocode"})
# Providers that want bare names with dots replaced by hyphens.
_DOT_TO_HYPHEN_PROVIDERS: frozenset[str] = frozenset({
"anthropic",
})
"anthropic"})
# Providers that want bare names with dots preserved.
_STRIP_VENDOR_ONLY_PROVIDERS: frozenset[str] = frozenset({
"copilot",
"copilot-acp",
"openai-codex",
})
"copilot", "copilot-acp", "openai-codex"})
# Providers whose native naming is authoritative -- pass through unchanged.
_AUTHORITATIVE_NATIVE_PROVIDERS: frozenset[str] = frozenset({
"huggingface",
})
"huggingface"})
# Direct providers that accept bare native names but should repair a matching
# provider/ prefix when users copy the aggregator form into config.yaml.
@@ -78,8 +60,7 @@ _MATCHING_PREFIX_STRIP_PROVIDERS: frozenset[str] = frozenset({
"nebius-token-factory",
"custom",
"gemini",
"xai",
})
"xai"})
# Providers whose API serves ``vendor/model`` ids but whose endpoint can also
# front arbitrary self-hosted models, so a bare name cannot be prefixed
@@ -91,97 +72,43 @@ _MATCHING_PREFIX_STRIP_PROVIDERS: frozenset[str] = frozenset({
# Without this repair a bare ``nemotron-3-ultra-550b-a55b`` reaches the API
# and returns a bare ``404 page not found`` that never names the model (#78796).
_CATALOGUE_PREFIX_REPAIR_PROVIDERS: frozenset[str] = frozenset({
"nvidia",
})
"nvidia"})
# Providers whose APIs require lowercase model IDs. Xiaomi's
# ``api.xiaomimimo.com`` rejects mixed-case names like ``MiMo-V2.5-Pro``
# that users might copy from marketing docs — it only accepts
# ``mimo-v2.5-pro``. After stripping a matching provider prefix, these
# providers also get ``.lower()`` applied.
# Providers whose APIs require lowercase model IDs (Xiaomi rejects ``MiMo-V2.5-Pro`` copied from
# marketing docs; only ``mimo-v2.5-pro`` works). Applied after matching-prefix stripping.
_LOWERCASE_MODEL_PROVIDERS: frozenset[str] = frozenset({
"xiaomi",
})
"xiaomi"})
# ---------------------------------------------------------------------------
# DeepSeek special handling
# ---------------------------------------------------------------------------
# DeepSeek's direct API only accepts first-class V-series IDs after the
# 2026-07-24 cut-off. Legacy aliases and fuzzy names are remapped here so
# saved configs / picker leftovers cannot keep sending retired IDs.
_DEEPSEEK_REASONER_KEYWORDS: frozenset[str] = frozenset({
"reasoner",
"r1",
"think",
"reasoning",
"cot",
})
# Retired on 2026-07-24 15:59 UTC. Official docs: both aliases mapped to
# deepseek-v4-flash (chat = non-thinking, reasoner = thinking). Thinking
# mode itself is controlled by extra_body.thinking on the DeepSeek profile.
# DeepSeek's direct API only accepts first-class V-series IDs after the 2026-07-24 cut-off (HTTP 400
# otherwise). Both retired aliases map to deepseek-v4-flash per the official docs (thinking mode is
# controlled by extra_body.thinking on the profile), so saved configs can't keep sending them.
_DEEPSEEK_RETIRED_ALIASES: frozenset[str] = frozenset({
"deepseek-chat",
"deepseek-reasoner",
})
"deepseek-chat", "deepseek-reasoner"})
_DEEPSEEK_CANONICAL_MODELS: frozenset[str] = frozenset({
"deepseek-v4-pro", # V4 Pro — first-class model ID
"deepseek-v4-flash", # V4 Flash — first-class model ID
})
"deepseek-v4-pro", "deepseek-v4-flash"})
# First-class V-series IDs (``deepseek-v4-pro``, ``deepseek-v4-flash``,
# future ``deepseek-v5-*``, dated variants like ``deepseek-v4-flash-20260423``).
# Verified empirically 2026-04-24: DeepSeek's Chat Completions API returns
# ``provider: DeepSeek`` / ``model: deepseek-v4-flash-20260423`` when called
# with ``model=deepseek/deepseek-v4-flash``, so these names are not aliases
# of ``deepseek-chat`` and must not be folded into it.
# First-class V-series IDs incl. future ``deepseek-v5-*`` and dated variants
# (``deepseek-v4-flash-20260423``): verified real model ids, NOT aliases of ``deepseek-chat``.
_DEEPSEEK_V_SERIES_RE = re.compile(r"^deepseek-v\d+([-.].+)?$")
def _normalize_for_deepseek(model_name: str) -> str:
"""Map a model input to a DeepSeek-accepted identifier.
Retired aliases ``deepseek-chat``/``deepseek-reasoner`` and known canonicals map as expected;
anything matching ``deepseek-v<digit>...`` passes through so future V-series ids work without a
release; reasoner keywords and everything else fall back to ``deepseek-v4-flash``.
"""
"""Map a model input to a DeepSeek-accepted id: canonicals and ``deepseek-v<digit>…`` pass
through (future V-series work without a release); retired aliases and everything else become
``deepseek-v4-flash``."""
bare = _strip_vendor_prefix(model_name).lower()
# Retired aliases must rewrite — DeepSeek returns HTTP 400 after the
# 2026-07-24 cut-off if these IDs are sent on the wire.
if bare in _DEEPSEEK_RETIRED_ALIASES:
return "deepseek-v4-flash"
if bare in _DEEPSEEK_CANONICAL_MODELS:
if bare in _DEEPSEEK_CANONICAL_MODELS or _DEEPSEEK_V_SERIES_RE.match(bare):
return bare
# V-series first-class IDs (v4-pro, v4-flash, future v5-*, dated variants)
if _DEEPSEEK_V_SERIES_RE.match(bare):
return bare
# Check for reasoner-like keywords anywhere in the name
for keyword in _DEEPSEEK_REASONER_KEYWORDS:
if keyword in bare:
return "deepseek-v4-flash"
return "deepseek-v4-flash"
# ---------------------------------------------------------------------------
# Helper utilities
# ---------------------------------------------------------------------------
def _strip_vendor_prefix(model_name: str) -> str:
"""Remove a ``vendor/`` prefix if present."""
if "/" in model_name:
return model_name.split("/", 1)[1]
return model_name
return model_name.split("/", 1)[1] if "/" in model_name else model_name
def _dots_to_hyphens(model_name: str) -> str:
"""Replace dots with hyphens in a model name."""
return model_name.replace(".", "-")
@@ -199,108 +126,66 @@ def _normalize_provider_alias(provider_name: str) -> str:
def _strip_matching_provider_prefix(model_name: str, target_provider: str) -> str:
"""Strip ``provider/`` only when the prefix matches the target provider.
Prevents arbitrary slash-bearing ids from being mangled on native providers while still
repairing config values like ``zai/glm-5.1`` for ``zai``. ``custom`` is a bucket, not a vendor:
an alias that merely resolves to it (e.g. ``ollama``) may be a real routing prefix required by a
proxy such as LiteLLM, so only a literal ``custom/`` prefix is treated as redundant.
"""
"""Strip ``provider/`` only when the prefix matches the target provider, so arbitrary slash-bearing
ids aren't mangled while ``zai/glm-5.1`` is repaired for ``zai``. ``custom`` is a bucket, not a
vendor: an alias resolving to it (``ollama``) may be a real LiteLLM-style routing prefix, so only a
literal ``custom/`` prefix is redundant there."""
if "/" not in model_name:
return model_name
prefix, remainder = model_name.split("/", 1)
if not prefix.strip() or not remainder.strip():
return model_name
normalized_target = _normalize_provider_alias(target_provider)
if normalized_target == "custom":
if prefix.strip().lower() == "custom":
return remainder.strip()
return model_name
return remainder.strip() if prefix.strip().lower() == "custom" else model_name
normalized_prefix = _normalize_provider_alias(prefix)
if normalized_prefix and normalized_prefix == normalized_target:
return remainder.strip()
return model_name
return remainder.strip() if normalized_prefix and normalized_prefix == normalized_target else model_name
def detect_vendor(model_name: str) -> Optional[str]:
"""Detect the vendor slug from a bare model name."""
"""Vendor slug from a bare model name: an existing ``vendor/`` prefix, the first hyphen token,
or a ``_VENDOR_PREFIXES`` key the name starts with (``qwen3.5-plus`` → ``qwen``)."""
name = model_name.strip()
if not name:
return None
# If there's already a vendor/ prefix, extract it
if "/" in name:
return name.split("/", 1)[0].lower() or None
name_lower = name.lower()
# Try first hyphen-delimited token (exact match)
first_token = name_lower.split("-")[0]
if first_token in _VENDOR_PREFIXES:
return _VENDOR_PREFIXES[first_token]
# Handle patterns where the first token includes version digits,
# e.g. "qwen3.5-plus" -> first token "qwen3.5", but prefix is "qwen"
for prefix, vendor in _VENDOR_PREFIXES.items():
if name_lower.startswith(prefix):
return vendor
return None
return next((vendor for prefix, vendor in _VENDOR_PREFIXES.items() if name_lower.startswith(prefix)), None)
def _prepend_vendor(model_name: str) -> str:
"""Prepend the detected ``vendor/`` prefix if missing.
For aggregators that require ``vendor/model``. Names already containing ``/`` or with no
detectable vendor are returned unchanged (the aggregator may still accept them).
"""
"""Prepend the detected ``vendor/`` for aggregators; names with ``/`` or no detectable vendor
pass through (the aggregator may still accept them)."""
if "/" in model_name:
return model_name
vendor = detect_vendor(model_name)
if vendor:
return f"{vendor}/{model_name}"
return model_name
return f"{vendor}/{model_name}" if vendor else model_name
def _repair_prefix_from_catalogue(model_name: str, provider: str) -> str:
"""Restore a dropped ``vendor/`` prefix using the provider's catalogue.
Unlike :func:`_prepend_vendor`, this never guesses from the model's name shape — it only repairs
a bare id that matches **exactly one** curated entry for this provider modulo the prefix.
"""
"""Restore a dropped ``vendor/`` prefix only when the bare id matches **exactly one** curated
entry for this provider modulo the prefix — a lookup, never a guess from name shape."""
if "/" in model_name:
return model_name
try:
from hermes_cli.models import _PROVIDER_MODELS
except Exception:
return model_name
catalogue = _PROVIDER_MODELS.get(provider) or []
# Compare against the catalogue's own suffix, tag included: a bare
# ``…:free`` id must resolve to the ``:free`` entry, not its paid sibling.
# Compare against the catalogue's own suffix, tag included: a bare ``…:free`` id must resolve to
# the ``:free`` entry, not its paid sibling.
needle = model_name.strip().lower()
matches = {
entry
for entry in catalogue
if "/" in entry and entry.split("/", 1)[1].strip().lower() == needle
}
if len(matches) == 1:
return matches.pop()
return model_name
catalogue = _PROVIDER_MODELS.get(provider) or []
matches = {e for e in catalogue if "/" in e and e.split("/", 1)[1].strip().lower() == needle}
return matches.pop() if len(matches) == 1 else model_name
def suggest_prefixed_model_id(provider: str, model_name: str) -> Optional[str]:
"""Return the prefixed catalogue id for a bare *model_name*, if unambiguous.
Diagnostic counterpart to :func:`_repair_prefix_from_catalogue`, used to explain a provider's
content-free 404 when the configured id lost its ``vendor/`` prefix. Returns ``None`` when the
name already has a prefix, the provider has no catalogue, or nothing matches — so callers stay
silent rather than guess.
"""
"""Prefixed catalogue id for a bare *model_name* if unambiguous, else ``None`` — the diagnostic
counterpart to :func:`_repair_prefix_from_catalogue` for explaining a content-free 404."""
name = (model_name or "").strip()
if not name or "/" in name:
return None
@@ -312,59 +197,35 @@ def suggest_prefixed_model_id(provider: str, model_name: str) -> Optional[str]:
return repaired if repaired != name else None
# ---------------------------------------------------------------------------
# Main normalisation entry point
# ---------------------------------------------------------------------------
def normalize_model_for_provider(model_input: str, target_provider: str) -> str:
"""Translate a model name into the format the target provider's API expects.
Primary entry point for model-name normalisation. Accepts bare, vendor-prefixed or native ids;
``target_provider`` should already be normalised via ``normalize_provider()``. Never raises —
always returns a best-effort string.
"""
"""Translate a model name (bare, vendor-prefixed or native) into what the target provider's API
expects. ``target_provider`` should already be canonical. Never raises."""
name = (model_input or "").strip()
if not name:
return name
provider = _normalize_provider_alias(target_provider)
# --- Aggregators: need vendor/model format ---
if provider in _AGGREGATOR_PROVIDERS:
return _prepend_vendor(name)
# --- OpenCode Zen / OpenCode Go: flat-namespace resellers.
# Their /v1/models API returns bare IDs only (no vendor prefix), and
# the inference endpoint rejects vendor-prefixed names with HTTP 401
# "Model not supported". Strip ANY leading ``vendor/`` so config
# entries like ``minimax/minimax-m2.7`` or ``deepseek/deepseek-v4-flash``
# — commonly copied from aggregator slugs into fallback_model lists —
# resolve to bare ``minimax-m2.7`` / ``deepseek-v4-flash`` the API
# actually serves. See PR reviewing opencode-go fallback 401s. ---
# OpenCode Zen / Go are flat-namespace resellers: /v1/models returns bare IDs and inference 401s
# vendor-prefixed names, so strip ANY leading ``vendor/`` (commonly copied from aggregator slugs).
from hermes_cli.models import opencode_provider_family
_oc_family = opencode_provider_family(provider)
if _oc_family is not None:
if "/" in name:
_, bare_after_slash = name.split("/", 1)
name = bare_after_slash.strip() or name
name = name.split("/", 1)[1].strip() or name
if _oc_family == "opencode-zen" and name.lower().startswith("claude-"):
return _dots_to_hyphens(name)
return name
# --- Anthropic: strip matching provider prefix, dots -> hyphens ---
if provider in _DOT_TO_HYPHEN_PROVIDERS:
bare = _strip_matching_provider_prefix(name, provider)
if "/" in bare:
return bare
return _dots_to_hyphens(bare)
return bare if "/" in bare else _dots_to_hyphens(bare)
# --- Copilot / Copilot ACP: delegate to the Copilot-specific
# normalizer. It knows about the alias table (vendor-prefix
# stripping for Anthropic/OpenAI, dash-to-dot repair for Claude)
# and live-catalog lookups. Without this, vendor-prefixed or
# dash-notation Claude IDs survive to the Copilot API and hit
# HTTP 400 "model_not_supported". See issue #6879.
# Copilot's own normalizer knows the alias table (vendor stripping, dash-to-dot repair for Claude)
# and live-catalog lookups; without it dash-notation Claude ids hit HTTP 400 model_not_supported.
if provider in {"copilot", "copilot-acp"}:
try:
from hermes_cli.models import normalize_copilot_model_id
@@ -373,50 +234,25 @@ def normalize_model_for_provider(model_input: str, target_provider: str) -> str:
if normalized:
return normalized
except Exception:
# Fall through to the generic strip-vendor behaviour below
# if the Copilot-specific path is unavailable for any reason.
pass
pass # fall through to the generic strip-vendor behaviour
# --- Copilot / Copilot ACP / openai-codex fallback:
# strip matching provider prefix, keep dots ---
if provider in _STRIP_VENDOR_ONLY_PROVIDERS:
stripped = _strip_matching_provider_prefix(name, provider)
if stripped == name and name.startswith("openai/"):
# openai-codex maps openai/gpt-5.4 -> gpt-5.4
return name.split("/", 1)[1]
return name.split("/", 1)[1] # openai-codex maps openai/gpt-5.4 -> gpt-5.4
return stripped
# --- DeepSeek: map to one of two canonical names ---
if provider == "deepseek":
bare = _strip_matching_provider_prefix(name, provider)
if "/" in bare:
return bare
return _normalize_for_deepseek(bare)
return bare if "/" in bare else _normalize_for_deepseek(bare)
# --- Direct providers: repair matching provider prefixes only ---
if provider in _MATCHING_PREFIX_STRIP_PROVIDERS:
result = _strip_matching_provider_prefix(name, provider)
# Some providers require lowercase model IDs (e.g. Xiaomi's API
# rejects "MiMo-V2.5-Pro" but accepts "mimo-v2.5-pro").
if provider in _LOWERCASE_MODEL_PROVIDERS:
result = result.lower()
return result
return result.lower() if provider in _LOWERCASE_MODEL_PROVIDERS else result
# --- Catalogue-backed prefix repair: restore a dropped ``vendor/`` on a
# bare id that matches exactly one curated entry. Unknown names (a
# local NIM container, a proxied model) pass through untouched. ---
# Unknown names (a local NIM container, a proxied model) pass through untouched.
if provider in _CATALOGUE_PREFIX_REPAIR_PROVIDERS:
return _repair_prefix_from_catalogue(name, provider)
# --- Authoritative native providers: preserve user-facing slugs as-is ---
if provider in _AUTHORITATIVE_NATIVE_PROVIDERS:
return name
# --- Custom & all others: pass through as-is ---
# Authoritative native providers, custom and all others: pass through as-is.
return name
# ---------------------------------------------------------------------------
# Batch / convenience helpers
# ---------------------------------------------------------------------------
hermes_cli/model_search.py
+6 -17
View File
@@ -7,18 +7,12 @@ _MODEL_SEARCH_ALIASES: dict[str, tuple[str, ...]] = {
"k3": ("kimi-k3", "kimi"),
# OpenCode Zen serves the "Ox Alpha" stealth model under an opaque
# preview slug; let users find it by its public codename.
"x-preview-f-free": ("ox-alpha", "ox"),
}
"x-preview-f-free": ("ox-alpha", "ox")}
# Lowercased wire id → canonical public slug it aliases. Used by picker
# dedup so a live bare id and its curated public slug (``k3`` / ``kimi-k3``)
# don't render as two rows for the same model. Derived from the FIRST alias
# entry, which by convention is the full public slug.
# Lowercased wire id → canonical public slug (the FIRST alias by convention), so picker dedup doesn't
# render a live bare id and its curated slug (``k3`` / ``kimi-k3``) as two rows.
_MODEL_ALIAS_CANONICAL: dict[str, str] = {
wire_id: aliases[0].lower()
for wire_id, aliases in _MODEL_SEARCH_ALIASES.items()
if aliases
}
wire_id: aliases[0].lower() for wire_id, aliases in _MODEL_SEARCH_ALIASES.items() if aliases}
def model_alias_canonical(model: str) -> str:
@@ -28,14 +22,9 @@ def model_alias_canonical(model: str) -> str:
def model_search_text(model: str) -> str:
"""Return the haystack used for fuzzy/substring model search.
Never changes the wire id passed to the provider.
"""
"""Haystack for fuzzy/substring model search; never changes the wire id sent to the provider."""
mid = (model or "").strip()
if not mid:
return model or ""
aliases = _MODEL_SEARCH_ALIASES.get(mid.lower())
if not aliases:
return mid
return f"{mid} {' '.join(aliases)}"
return f"{mid} {' '.join(aliases)}" if aliases else mid
hermes_cli/model_selection_guards.py
+22 -66
View File
@@ -24,44 +24,28 @@ class SelectionWarning:
def _wrap(kind: str, title: str, warning, model_name: str, provider: Optional[str]):
"""Lift a raw guard payload into a :class:`SelectionWarning` (None passes through).
Duck-typed access: tests (and future guard payloads) may supply objects carrying only
``.message``.
"""
"""Lift a raw guard payload into a :class:`SelectionWarning` (None passes through). Duck-typed:
payloads may carry only ``.message``."""
if warning is None:
return None
return SelectionWarning(
kind=kind,
title=title,
model=getattr(warning, "model", model_name),
provider=getattr(warning, "provider", provider or ""),
message=warning.message,
)
kind=kind, title=title, model=getattr(warning, "model", model_name),
provider=getattr(warning, "provider", provider or ""), message=warning.message)
def _cost_guard(
model_name: str,
provider: Optional[str],
base_url: Optional[str],
api_key: Optional[str],
model_info: Optional[ModelInfo],
) -> Optional[SelectionWarning]:
model_name: str, provider: Optional[str], base_url: Optional[str], api_key: Optional[str],
model_info: Optional[ModelInfo]) -> Optional[SelectionWarning]:
from hermes_cli.model_cost_guard import expensive_model_warning
warning = expensive_model_warning(
model_name, provider=provider, base_url=base_url, api_key=api_key, model_info=model_info
)
model_name, provider=provider, base_url=base_url, api_key=api_key, model_info=model_info)
return _wrap("cost", "Expensive Model Warning", warning, model_name, provider)
def _data_policy_guard(
model_name: str,
provider: Optional[str],
base_url: Optional[str],
api_key: Optional[str],
model_info: Optional[ModelInfo],
) -> Optional[SelectionWarning]:
model_name: str, provider: Optional[str], base_url: Optional[str], api_key: Optional[str],
model_info: Optional[ModelInfo]) -> Optional[SelectionWarning]:
from hermes_cli.model_data_policy_guard import data_training_warning
warning = data_training_warning(model_name, provider=provider, base_url=base_url)
@@ -74,22 +58,11 @@ _GUARDS = (_cost_guard, _data_policy_guard)
def selection_warnings(
model_name: str,
*,
provider: Optional[str] = None,
base_url: Optional[str] = None,
api_key: Optional[str] = None,
model_info: Optional[ModelInfo] = None,
include_kinds: Optional[Iterable[str]] = None,
) -> List[SelectionWarning]:
"""Run every registered selection guard and return the warnings that fired.
Returns an empty list in the common case (no guard fired). Callers should run this after model
resolution so aliases / provider-specific ids have settled, then surface the messages as a
confirm step. ``include_kinds`` optionally restricts which guard kinds run (e.g.
A misbehaving guard must never break model selection: individual guard exceptions are swallowed.
"""
model_name: str, *, provider: Optional[str] = None, base_url: Optional[str] = None,
api_key: Optional[str] = None, model_info: Optional[ModelInfo] = None,
include_kinds: Optional[Iterable[str]] = None) -> List[SelectionWarning]:
"""Warnings from every registered guard (empty in the common case). ``include_kinds`` restricts
which kinds are returned. Guard exceptions are swallowed — never break model selection."""
wanted = set(include_kinds) if include_kinds is not None else None
results: List[SelectionWarning] = []
for guard in _GUARDS:
@@ -103,39 +76,22 @@ def selection_warnings(
def combined_message(warnings: List[SelectionWarning]) -> str:
"""Join multiple warnings into one confirm-prompt body.
Used by surfaces with a single confirm dialog when more than one guard fires (rare) — one
prompt showing both blocks beats two sequential prompts.
"""
"""One confirm-prompt body for several warnings (one prompt beats two sequential ones)."""
return "\n\n".join(w.message for w in warnings)
def combined_selection_warning(
model_name: str,
*,
provider: Optional[str] = None,
base_url: Optional[str] = None,
api_key: Optional[str] = None,
model_info: Optional[ModelInfo] = None,
model_name: str, *, provider: Optional[str] = None, base_url: Optional[str] = None,
api_key: Optional[str] = None, model_info: Optional[ModelInfo] = None,
) -> Optional[SelectionWarning]:
"""Drop-in replacement for ``expensive_model_warning`` call sites.
Returns ``None`` when no guard fired, the single :class:`SelectionWarning` when one fired,
or a merged ``kind="multiple"`` warning stacking every message — so surfaces rendering one
confirm dialog from ``warning.message`` can switch without reshaping control flow.
"""
"""Drop-in for ``expensive_model_warning`` call sites: ``None``, the single warning, or a merged
``kind="multiple"`` warning stacking every message."""
warnings = selection_warnings(
model_name, provider=provider, base_url=base_url, api_key=api_key, model_info=model_info
)
model_name, provider=provider, base_url=base_url, api_key=api_key, model_info=model_info)
if not warnings:
return None
if len(warnings) == 1:
return warnings[0]
return SelectionWarning(
kind="multiple",
title="Model Selection Warning",
model=warnings[0].model,
provider=warnings[0].provider,
message=combined_message(warnings),
)
kind="multiple", title="Model Selection Warning", model=warnings[0].model,
provider=warnings[0].provider, message=combined_message(warnings))
hermes_cli/models.py
+614 -1320
View File
File diff suppressed because it is too large Load Diff