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

refactor(hermes_cli): dispatch-table pricing fetchers, shared price parsing, slot accessors on _CapsSource

Browse Source
This commit is contained in:
Teknium
2026-09-02 21:12:47 -07:00
parent 1cf91f6e13
commit 07904249a3
2 changed files with 233 additions and 331 deletions
Download Patch File Download Diff File Expand all files Collapse all files

408
hermes_cli/models_pricing.py
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] = {}
@@ -56,21 +50,18 @@ def _cache_catalog(
) -> 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.
*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 +70,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
@@ -97,14 +85,11 @@ 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.
prefers an authenticated catalog. Scans (callers hold no credential) newest first, skipping
expired entries, so a rotated credential does not keep answering from its predecessor's catalog.
"""
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,15 +99,16 @@ 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.
"""Per-token price string → 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.
Always 2 decimals so right-justified prices align on the decimal point. 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)
@@ -133,8 +119,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 +126,58 @@ def _format_price_per_mtok(per_token_str: str) -> str:
return f"${text}"
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
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 compute_sale_discount(
prompt: str,
completion: str,
original: Any,
) -> tuple[int, str, str] | None:
"""Derive sale chrome from gateway ``pricing.original`` when cheaper.
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``.
"""Derive sale chrome from gateway ``pricing.original`` when cheaper (Nous Portal only; callers
gate on the provider and 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.
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%" chrome).
"""
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 _nonneg(raw: Any) -> float | None:
try:
n = float(raw)
except (TypeError, ValueError):
return None
return n if n >= 0 and n == n else None
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 +206,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",
@@ -255,13 +226,10 @@ def fetch_models_with_pricing(
) -> 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.
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). *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("/")
@@ -310,11 +278,8 @@ def fetch_ai_gateway_pricing(
*,
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.
"""
"""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,9 +293,7 @@ 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
for item in _catalog_items(payload):
mid = item.get("id")
pricing = item.get("pricing")
if mid and isinstance(pricing, dict):
@@ -347,13 +310,12 @@ _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.
"""``(api_key, base_url)`` for Nous Portal pricing; base_url is the bare origin (no ``/v1``).
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.
Base URL precedence mirrors runtime credential resolution: ``NOUS_INFERENCE_BASE_URL`` env
override (staging / preview) → resolved runtime credential ``base_url`` → production default.
Without the override a staging profile's sale ``pricing.original`` would never reach the
pickers (prod has no ``original`` field).
"""
try:
from hermes_cli.auth import _nous_inference_env_override
@@ -375,25 +337,39 @@ def _resolve_nous_pricing_credentials() -> tuple[str, str]:
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.
response's keys are the reachable set. ``None`` (leave the caller's list alone) covers the
three states that cannot narrow it: no policy (or a token too old to say), an anonymous read
whose catalog is unfiltered, and an empty read (a fetch failure, not an org that may reach
nothing).
"""
from hermes_cli.models import _resolve_nous_pricing_credentials, fetch_models_with_pricing
from hermes_cli.models import _resolve_nous_pricing_credentials
try:
from hermes_cli.nous_account import nous_policy_present
@@ -405,33 +381,14 @@ 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]],
@@ -442,68 +399,51 @@ def restrict_to_nous_policy(
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.
the authoritative gate, while hiding a row the gate would serve 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.
"""
"""Fireworks picker pricing from the models.dev registry cache — no dedicated network fetch:
``fetch_models_dev()`` already keeps a shared in-memory + disk cache (1h TTL), so this is a
pure dict transform with no per-render network call."""
cache_key = "models.dev/fireworks"
if not force_refresh:
cached = _cached_catalog(cache_key)
@@ -514,24 +454,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 = {}
@@ -544,19 +476,15 @@ def _fetch_novita_pricing(
*,
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.
"""
"""NovitaAI /v1/models pricing. Novita reports per-million-token prices in units of 0.0001 USD;
converted to the per-token strings the shared formatter expects. 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 +496,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),
@@ -591,12 +514,9 @@ def _fetch_deepinfra_pricing(
*,
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.
"""
"""Picker-shape pricing for DeepInfra chat models: ``input_tokens`` / ``output_tokens`` /
``cache_read_tokens`` in $/MTok → per-token ``prompt`` / ``completion`` / ``input_cache_read``.
Cached via the models-by-tag helper so repeated picker renders are free."""
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 +526,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,
}

156
hermes_cli/models_reasoning_caps.py
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
@@ -42,9 +40,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 +51,28 @@ 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()
))
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,
"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
@@ -96,9 +90,7 @@ def _read_reasoning_caps_disk() -> dict[str, Any]:
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:
@@ -121,27 +113,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 +133,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,12 +151,8 @@ 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:
@@ -193,11 +170,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 +181,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 +224,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 +251,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 +260,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)