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

refactor(hermes_cli): runtime_provider_custom/backends — shared _custom_runtime builder, azure key helper, compact docs

Browse Source
This commit is contained in:
Teknium
2026-09-02 21:21:10 -07:00
parent 6ae394025f
commit b8c7add3a9
2 changed files with 112 additions and 175 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/runtime_provider_custom.py
+74 -120
View File
@@ -1,11 +1,11 @@
"""Custom-provider resolution: ``providers:`` / ``custom_providers:`` lookup, identity
recovery, custom credential pools, and the named-custom runtime builder.
Extracted from :mod:`hermes_cli.runtime_provider`; every public/private name here is
re-exported there. Origin-internal collaborators (``load_config``, ``_get_model_config``,
``load_pool``, ``has_usable_secret``, ``custom_provider_pool_key_candidates``, …) are looked up
on the origin module AT CALL TIME via :func:`_rp` so ``monkeypatch.setattr(runtime_provider,
name, …)`` in tests keeps working for moved bodies.
Extracted from :mod:`hermes_cli.runtime_provider`; every name here is re-exported there.
Origin-internal collaborators (``load_config``, ``_get_model_config``, ``load_pool``,
``has_usable_secret``, ``custom_provider_pool_key_candidates``, …) are looked up on the origin
module AT CALL TIME via :func:`_rp` so ``monkeypatch.setattr(runtime_provider, name, …)`` keeps
working for moved bodies.
"""
from __future__ import annotations
@@ -19,6 +19,8 @@ from utils import base_url_hostname
logger = logging.getLogger("hermes_cli.runtime_provider")
_LLAMACPP_ALIASES = ("llamacpp", "llama.cpp", "llama-cpp")
def _rp():
"""Origin module, late-bound so test patches on ``hermes_cli.runtime_provider.*`` apply."""
@@ -65,11 +67,9 @@ def _lift_model_capabilities(entry: Dict[str, Any], model: Optional[str], result
def _lift_max_output_tokens(entry: Dict[str, Any], result: Dict[str, Any]) -> None:
"""``max_output_tokens`` or ``max_tokens`` on a provider entry pins its own output limit.
Gateway/CLI map it onto ``AIAgent.max_tokens`` only when top-level ``model.max_tokens`` is
unset, so the documented global key still wins.
"""
"""``max_output_tokens`` or ``max_tokens`` on a provider entry pins its own output limit;
gateway/CLI map it onto ``AIAgent.max_tokens`` only when top-level ``model.max_tokens`` is
unset, so the documented global key still wins."""
for key in ("max_output_tokens", "max_tokens"):
value = entry.get(key)
if isinstance(value, int) and value > 0:
@@ -85,12 +85,7 @@ def _lift_extra_headers(entry: Dict[str, Any], result: Dict[str, Any]) -> None:
def _lift_common_custom_fields(
entry: Dict[str, Any],
result: Dict[str, Any],
*,
provider_key: str,
key_env: str,
api_mode: Optional[str],
entry: Dict[str, Any], result: Dict[str, Any], *, provider_key: str, key_env: str, api_mode: Optional[str]
) -> None:
"""Copy the optional fields shared by ``providers:`` and legacy ``custom_providers:`` entries."""
if key_env:
@@ -115,11 +110,11 @@ def _lift_common_custom_fields(
def _shadowed_by_builtin(requested_norm: str) -> bool:
"""Raw names map to custom providers only when they are not canonical built-ins.
Explicit ``custom:<name>`` keys always target the saved entry, and bare ``custom`` is
exempt: a user may literally name a ``providers:`` entry "custom" (returning None before
the config scan made such cron jobs fail with ``auth_unavailable``). Defer to the built-in
only when the raw name IS the canonical provider (``nous``); an entry matching merely an
alias (``kimi`` → ``kimi-coding``) is the user's target.
Explicit ``custom:<name>`` keys always target the saved entry, and bare ``custom`` is exempt: a
user may literally name a ``providers:`` entry "custom" (returning None before the config scan
made such cron jobs fail with ``auth_unavailable``). Defer to the built-in only when the raw
name IS the canonical provider (``nous``); an entry matching merely an alias (``kimi`` →
``kimi-coding``) is the user's target.
"""
if requested_norm == "custom" or requested_norm.startswith("custom:"):
return False
@@ -162,9 +157,7 @@ def _match_new_style_provider(requested_norm: str, providers: Dict[str, Any]) ->
# v12 migration writes ``transport``; hand-edited configs may still use ``api_mode``.
# Accept both or migrated configs silently downgrade to chat_completions.
_lift_common_custom_fields(
entry, result,
provider_key=_clean(ep_name),
key_env=key_env,
entry, result, provider_key=_clean(ep_name), key_env=key_env,
api_mode=rp._parse_api_mode(entry.get("api_mode") or entry.get("transport")),
)
return result
@@ -187,9 +180,7 @@ def _match_legacy_custom_provider(requested_norm: str, custom_providers) -> Opti
if model_name:
result["model"] = model_name
_lift_common_custom_fields(
entry, result,
provider_key=provider_key,
key_env=_clean(entry.get("key_env", "")),
entry, result, provider_key=provider_key, key_env=_clean(entry.get("key_env", "")),
api_mode=_rp()._parse_api_mode(entry.get("api_mode")),
)
return result
@@ -200,7 +191,6 @@ def _get_named_custom_provider(requested_provider: str) -> Optional[Dict[str, An
requested_norm = _normalize_custom_provider_name(requested_provider or "")
if not requested_norm or requested_norm == "auto" or _shadowed_by_builtin(requested_norm):
return None
rp = _rp()
config = rp.load_config()
providers = config.get("providers")
@@ -208,7 +198,6 @@ def _get_named_custom_provider(requested_provider: str) -> Optional[Dict[str, An
found = _match_new_style_provider(requested_norm, providers)
if found:
return found
if isinstance(config.get("custom_providers"), dict):
logger.warning(
"custom_providers in config.yaml is a dict, not a list. "
@@ -223,10 +212,8 @@ def _get_named_custom_provider(requested_provider: str) -> Optional[Dict[str, An
def has_named_custom_provider(requested_provider: str) -> bool:
"""True when config defines a ``providers:`` / ``custom_providers:`` entry matching the request.
Public wrapper so other modules (e.g. the cronjob tool) need not reach into a private helper.
"""
"""True when config defines a ``providers:`` / ``custom_providers:`` entry matching the request
(public wrapper so e.g. the cronjob tool need not reach into a private helper)."""
try:
return _rp()._get_named_custom_provider(requested_provider) is not None
except Exception:
@@ -244,13 +231,11 @@ def _find_custom_identity(matches: Callable[[Dict[str, Any]], bool]) -> Optional
config = rp.load_config()
except Exception:
return None
providers = config.get("providers")
if isinstance(providers, dict):
for ep_name, entry in providers.items():
if isinstance(entry, dict) and matches(entry):
return custom_provider_slug(str(ep_name), str(ep_name))
try:
custom_providers = rp.get_compatible_custom_providers(config)
except Exception:
@@ -265,64 +250,54 @@ def _find_custom_identity(matches: Callable[[Dict[str, Any]], bool]) -> Optional
def find_custom_provider_identity(base_url: str) -> Optional[str]:
"""Map an endpoint URL back to its canonical ``custom:<name>`` menu key.
Session persistence stores the agent's *resolved* provider, and for every named custom
endpoint that is the literal string ``"custom"`` — the entry name is lost, and the api_key is
deliberately never persisted.
"""
"""Map an endpoint URL back to its canonical ``custom:<name>`` menu key. Session persistence
stores the agent's *resolved* provider, which for every named custom endpoint is the literal
string ``"custom"`` — the entry name is lost, and the api_key is deliberately never persisted."""
target = _normalize_base_url_for_match(base_url)
if not target:
return None
return _find_custom_identity(lambda entry: _normalize_base_url_for_match(_entry_url(entry)) == target)
def find_custom_provider_identity_by_model(model: str) -> Optional[str]:
"""Map a model id back to the ``custom:<name>`` entry that serves it.
def _model_id_matches(value: Any, target: str) -> bool:
return isinstance(value, str) and value.strip().lower() == target
Companion to :func:`find_custom_provider_identity` for persistence paths where no base_url
survived the round-trip: the session row always stores the model name.
"""
def find_custom_provider_identity_by_model(model: str) -> Optional[str]:
"""Map a model id back to the ``custom:<name>`` entry that serves it — companion to
:func:`find_custom_provider_identity` for persistence paths where no base_url survived the
round-trip (the session row always stores the model name)."""
target = str(model or "").strip().lower()
if not target:
return None
def _entry_serves_model(entry: Dict[str, Any]) -> bool:
for key in ("model", "default_model"):
value = entry.get(key)
if isinstance(value, str) and value.strip().lower() == target:
return True
if any(_model_id_matches(entry.get(key), target) for key in ("model", "default_model")):
return True
models = entry.get("models")
if isinstance(models, dict):
return any(str(mid).strip().lower() == target for mid in models)
if isinstance(models, list):
for item in models:
if isinstance(item, str) and item.strip().lower() == target:
return True
if isinstance(item, dict):
mid = item.get("id") or item.get("name")
if isinstance(mid, str) and mid.strip().lower() == target:
return True
return any(
_model_id_matches(item.get("id") or item.get("name") if isinstance(item, dict) else item, target)
for item in models
)
return False
return _find_custom_identity(_entry_serves_model)
def canonical_custom_identity(
*,
base_url: Optional[str] = None,
config_provider: Optional[str] = None,
model: Optional[str] = None,
*, base_url: Optional[str] = None, config_provider: Optional[str] = None, model: Optional[str] = None
) -> Optional[str]:
"""Recover a routable ``custom:<name>`` identity for a bare custom provider.
Every path that persists or restores a session's provider override must run the resolved
provider through this so a bare ``"custom"`` is upgraded back to its durable
``custom:<name>`` menu key. Recovery sources, in priority order: (1) ``base_url`` reverse
lookup — the one fact that always survives the round-trip when a URL was recorded; (2)
``model`` reverse lookup (``model``/``default_model``/``models`` catalog); (3) the configured
provider (arg, then ``model.provider``, then ``HERMES_INFERENCE_PROVIDER``) when it names a
real entry.
provider through this so a bare ``"custom"`` is upgraded back to its durable ``custom:<name>``
menu key. Sources in priority order: (1) ``base_url`` reverse lookup — the one fact that always
survives the round-trip when a URL was recorded; (2) ``model`` reverse lookup
(``model``/``default_model``/``models`` catalog); (3) the configured provider (arg, then
``model.provider``, then ``HERMES_INFERENCE_PROVIDER``) when it names a real entry.
"""
rp = _rp()
if base_url:
@@ -333,7 +308,6 @@ def canonical_custom_identity(
identity = find_custom_provider_identity_by_model(model)
if identity:
return identity
candidate = str(config_provider or "").strip()
if not candidate:
try:
@@ -342,7 +316,6 @@ def canonical_custom_identity(
candidate = ""
if not candidate:
candidate = os.environ.get("HERMES_INFERENCE_PROVIDER", "").strip()
candidate_norm = _normalize_custom_provider_name(candidate)
# A bare/non-routable candidate cannot heal a bare custom override.
if not candidate_norm or candidate_norm in {"custom", "auto", "openrouter"}:
@@ -367,11 +340,11 @@ def canonical_custom_identity(
def is_routable_provider(provider: Optional[str]) -> bool:
"""Whether a provider name currently resolves to a routable route.
Empty/None/``auto`` is vacuously routable (agent build falls back to the configured
default). Bare ``custom`` is the resolved billing class shared by every named entry — not a
routable identity; restore paths must heal it (:func:`canonical_custom_identity`) or fall
back. Anything else is routable iff the full chain (built-in -> ``providers:`` ->
``custom_providers:`` -> models.dev) resolves it.
Empty/None/``auto`` is vacuously routable (agent build falls back to the configured default).
Bare ``custom`` is the resolved billing class shared by every named entry — not a routable
identity; restore paths must heal it (:func:`canonical_custom_identity`) or fall back. Anything
else is routable iff the full chain (built-in -> ``providers:`` -> ``custom_providers:`` ->
models.dev) resolves it.
"""
name = str(provider or "").strip()
if not name or name.lower() == "auto":
@@ -383,9 +356,7 @@ def is_routable_provider(provider: Optional[str]) -> bool:
rp = _rp()
config = rp.load_config()
return resolve_provider_full(
name, config.get("providers"), rp.get_compatible_custom_providers(config)
) is not None
return resolve_provider_full(name, config.get("providers"), rp.get_compatible_custom_providers(config)) is not None
except Exception:
return False
@@ -394,10 +365,7 @@ def is_routable_provider(provider: Optional[str]) -> bool:
def _try_resolve_from_custom_pool(
base_url: str,
provider_label: str,
api_mode_override: Optional[str] = None,
provider_name: Optional[str] = None,
base_url: str, provider_label: str, api_mode_override: Optional[str] = None, provider_name: Optional[str] = None
) -> Optional[Dict[str, Any]]:
"""Runtime dict from the first credential pool that owns this custom endpoint, else None."""
rp = _rp()
@@ -424,12 +392,8 @@ def _try_resolve_from_custom_pool(
# substitutes "no-key-required" for a loopback endpoint — this was the one gap.
pool_api_key = "no-key-required"
return rp._runtime(
provider_label,
api_mode_override or rp._detect_api_mode_for_url(base_url) or "chat_completions",
base_url,
pool_api_key,
source=f"pool:{pool_key}",
credential_pool=pool,
provider_label, api_mode_override or rp._detect_api_mode_for_url(base_url) or "chat_completions",
base_url, pool_api_key, source=f"pool:{pool_key}", credential_pool=pool,
)
except Exception:
continue
@@ -447,12 +411,9 @@ def _apply_custom_provider_extras(
custom_provider: Dict[str, Any], target_model: Optional[str], result: Dict[str, Any]
) -> None:
"""Copy model / capabilities / max_output_tokens / extra_headers / request_overrides onto a
resolved custom runtime.
An explicit ``target_model`` wins over the provider's configured default (auxiliary slots /
background-review resolve a concrete model and must not fall back to ``default_model``).
``extra_headers`` may carry credentials — NEVER log them.
"""
resolved custom runtime. An explicit ``target_model`` wins over the provider's configured
default (auxiliary slots / background-review resolve a concrete model and must not fall back to
``default_model``). ``extra_headers`` may carry credentials — NEVER log them."""
model_name = target_model or custom_provider.get("model")
if model_name:
result["model"] = model_name
@@ -482,12 +443,9 @@ def _resolve_llamacpp_runtime(requested_provider: str, explicit_api_key: Optiona
endpoint = None
if endpoint:
return rp._runtime(
"custom",
"chat_completions",
endpoint["base_url"],
"custom", "chat_completions", endpoint["base_url"],
(explicit_api_key or "").strip() or endpoint["api_key"] or "no-key-required",
source="local-runtime",
requested_provider=requested_provider,
source="local-runtime", requested_provider=requested_provider,
)
try:
enabled = bool((rp.load_config().get("local_runtime") or {}).get("enabled"))
@@ -506,6 +464,14 @@ def _resolve_llamacpp_runtime(requested_provider: str, explicit_api_key: Optiona
)
def _custom_runtime(rp, base_url: str, api_key: Any, api_mode: Optional[str], **extra: Any) -> Dict[str, Any]:
"""``custom`` runtime dict with URL-detected api_mode fallback and the no-auth placeholder."""
return rp._runtime(
"custom", api_mode or rp._detect_api_mode_for_url(base_url) or "chat_completions", base_url,
api_key or "no-key-required", **extra,
)
def _resolve_direct_alias_runtime(
requested_provider: str, explicit_api_key: Optional[str], explicit_base_url: str
) -> Dict[str, Any]:
@@ -521,15 +487,8 @@ def _resolve_direct_alias_runtime(
# OLLAMA_API_KEY gets its own gate here: without it a `model_aliases:` entry pointing at
# Ollama Cloud resolved no key at all.
candidates = [(explicit_api_key or "").strip(), *rp._host_gated_env_key_candidates(base_url, ollama=True)]
api_key = next((c for c in candidates if rp.has_usable_secret(c)), "") or "no-key-required"
return rp._runtime(
"custom",
rp._detect_api_mode_for_url(base_url) or "chat_completions",
base_url,
api_key,
source="direct-alias",
requested_provider=requested_provider,
)
api_key = next((c for c in candidates if rp.has_usable_secret(c)), "")
return _custom_runtime(rp, base_url, api_key, None, source="direct-alias", requested_provider=requested_provider)
def _opencode_family_for_custom(requested_provider: str, base_url: str) -> Optional[str]:
@@ -562,7 +521,7 @@ def _resolve_named_custom_runtime(
"""
rp = _rp()
requested_norm = (requested_provider or "").strip().lower()
if requested_norm in ("llamacpp", "llama.cpp", "llama-cpp") and not explicit_base_url:
if requested_norm in _LLAMACPP_ALIASES and not explicit_base_url:
return _resolve_llamacpp_runtime(requested_provider, explicit_api_key)
if requested_norm and requested_norm != "custom" and rp._resolves_to_custom(requested_norm):
requested_norm = "custom"
@@ -577,9 +536,7 @@ def _resolve_named_custom_runtime(
return None
pool_result = rp._try_resolve_from_custom_pool(
base_url,
"custom",
custom_provider.get("api_mode"),
base_url, "custom", custom_provider.get("api_mode"),
provider_name=custom_provider.get("provider_key") or custom_provider.get("name"),
)
if pool_result:
@@ -587,8 +544,9 @@ def _resolve_named_custom_runtime(
_apply_custom_provider_extras(custom_provider, target_model, pool_result)
return pool_result
explicit_key = (explicit_api_key or "").strip()
candidates = [
(explicit_api_key or "").strip(),
explicit_key,
_clean(custom_provider.get("api_key", "")),
rp._getenv(_clean(custom_provider.get("key_env", "")), "").strip(),
*rp._host_gated_env_key_candidates(base_url, ollama=False),
@@ -599,7 +557,7 @@ def _resolve_named_custom_runtime(
# mid-session); both wire clients accept a callable api_key (the Entra ID contract). An
# explicit --api-key still wins as the one-off recovery escape hatch.
key_cmd = _clean(custom_provider.get("key_cmd", ""))
if key_cmd and not rp.has_usable_secret((explicit_api_key or "").strip()):
if key_cmd and not rp.has_usable_secret(explicit_key):
from agent.command_token_source import build_command_token_provider
token_provider = build_command_token_provider(
@@ -608,13 +566,9 @@ def _resolve_named_custom_runtime(
if token_provider is not None:
api_key = token_provider
result = rp._runtime(
"custom",
custom_provider.get("api_mode") or rp._detect_api_mode_for_url(base_url) or "chat_completions",
base_url,
api_key or "no-key-required",
source=f"custom_provider:{custom_provider.get('name', requested_provider)}",
requested_provider=requested_provider,
result = _custom_runtime(
rp, base_url, api_key, custom_provider.get("api_mode"),
source=f"custom_provider:{custom_provider.get('name', requested_provider)}", requested_provider=requested_provider,
)
_apply_custom_provider_extras(custom_provider, target_model, result)