303 lines
13 KiB
Python
303 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""Normalize provider model metadata into safe Hermes routing candidates."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Iterable
|
|
|
|
|
|
EFFORTS = ("low", "medium", "high", "xhigh")
|
|
CAPABILITY_TIERS = ("economy", "balanced", "advanced")
|
|
EFFORT_TIERS = {
|
|
"low": "economy",
|
|
"medium": "balanced",
|
|
"high": "advanced",
|
|
"xhigh": "advanced",
|
|
}
|
|
SPECIALIST_MODEL_WORDS = frozenset(
|
|
{"audio", "embedding", "image", "moderation", "realtime", "speech", "transcrib", "tts"}
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Catalog:
|
|
"""Non-secret provider discovery result."""
|
|
|
|
provider: str
|
|
models: list[str]
|
|
live: bool
|
|
connected: bool
|
|
state: str
|
|
metadata: dict[str, dict[str, Any]] = field(default_factory=dict)
|
|
|
|
|
|
def unique_models(models: Iterable[str]) -> list[str]:
|
|
"""Return normalized model IDs without changing provider order."""
|
|
seen: set[str] = set()
|
|
result: list[str] = []
|
|
for model in models:
|
|
value = str(model or "").strip()
|
|
if value and value.lower() not in seen:
|
|
result.append(value)
|
|
seen.add(value.lower())
|
|
return result
|
|
|
|
|
|
def model_id(value: Any) -> str:
|
|
"""Extract a provider model ID from either a string or model-list record."""
|
|
if isinstance(value, str):
|
|
return value.strip()
|
|
if not isinstance(value, dict):
|
|
return ""
|
|
for key in ("id", "model", "model_id", "name", "value"):
|
|
candidate = value.get(key)
|
|
if isinstance(candidate, str) and candidate.strip():
|
|
return candidate.strip()
|
|
return ""
|
|
|
|
|
|
def public_metadata(value: Any) -> dict[str, Any]:
|
|
"""Keep provider routing fields while excluding unrelated response data."""
|
|
if not isinstance(value, dict):
|
|
return {}
|
|
allowed = {
|
|
"capabilities", "capability", "capability_tier", "cost", "cost_rank",
|
|
"created_at", "default_reasoning_effort", "description", "display_name", "displayName",
|
|
"disabled", "enabled", "hidden", "input_modalities", "inputModalities", "isDefault",
|
|
"is_default", "is_general_purpose", "isGeneralPurpose", "max_input_tokens", "max_tokens",
|
|
"output_modalities", "outputModalities", "pricing", "status",
|
|
"modelSpecialty", "resolvedModel", "supported_reasoning_efforts", "supportedReasoningEfforts", "supportedEffortLevels",
|
|
"supportsAdaptiveThinking", "supportsEffort", "tier", "upgrade",
|
|
"verified", "visibility",
|
|
}
|
|
return {key: value[key] for key in allowed if key in value}
|
|
|
|
|
|
def model_records(values: Iterable[Any]) -> tuple[list[str], dict[str, dict[str, Any]]]:
|
|
"""Normalize provider records into ordered IDs and safe per-model metadata."""
|
|
models: list[str] = []
|
|
metadata: dict[str, dict[str, Any]] = {}
|
|
for value in values:
|
|
model = model_id(value)
|
|
if not model:
|
|
continue
|
|
models.append(model)
|
|
if isinstance(value, dict):
|
|
metadata[model] = public_metadata(value)
|
|
unique = unique_models(models)
|
|
return unique, {model: metadata[model] for model in unique if model in metadata}
|
|
|
|
|
|
def merge_records(
|
|
*records: tuple[list[str], dict[str, dict[str, Any]]]
|
|
) -> tuple[list[str], dict[str, dict[str, Any]]]:
|
|
"""Merge ordered discovery sources, preferring metadata from later sources."""
|
|
models: list[str] = []
|
|
metadata: dict[str, dict[str, Any]] = {}
|
|
for source_models, source_metadata in records:
|
|
for model in source_models:
|
|
if model not in models:
|
|
models.append(model)
|
|
details = source_metadata.get(model)
|
|
if details:
|
|
metadata[model] = {**metadata.get(model, {}), **details}
|
|
return unique_models(models), metadata
|
|
|
|
|
|
def model_version(model: str) -> tuple[int, ...]:
|
|
"""Extract a sortable model version while ignoring dated aliases."""
|
|
value = re.sub(r"-\d{8}$", "", model.lower())
|
|
groups = re.findall(r"\d+(?:\.\d+)*", value)
|
|
parts: list[int] = []
|
|
for group in groups:
|
|
parts.extend(int(piece) for piece in group.split("."))
|
|
return tuple(parts) or (0,)
|
|
|
|
|
|
def _text_metadata(metadata: dict[str, Any]) -> str:
|
|
values: list[str] = []
|
|
for key in ("description", "display_name", "capability", "tier"):
|
|
value = metadata.get(key)
|
|
if isinstance(value, str):
|
|
values.append(value.lower())
|
|
capabilities = metadata.get("capabilities")
|
|
if isinstance(capabilities, list):
|
|
values.extend(str(value).lower() for value in capabilities)
|
|
return " ".join(values)
|
|
|
|
|
|
def metadata_tier(metadata: dict[str, Any]) -> tuple[str | None, str]:
|
|
"""Classify a model only from provider-declared capability information."""
|
|
for key in ("capability_tier", "tier", "capability"):
|
|
value = metadata.get(key)
|
|
if isinstance(value, str):
|
|
normalized = value.strip().lower()
|
|
if normalized in {"economy", "cheap", "fast", "low_cost"}:
|
|
return "economy", f"provider {key}={value!r}"
|
|
if normalized in {"balanced", "standard", "mid", "medium"}:
|
|
return "balanced", f"provider {key}={value!r}"
|
|
if normalized in {"advanced", "strongest", "premium", "high"}:
|
|
return "advanced", f"provider {key}={value!r}"
|
|
text = _text_metadata(metadata)
|
|
if any(term in text for term in ("strongest", "most capable", "most intelligent", "complex tasks")):
|
|
return "advanced", "provider description marks it strongest/complex"
|
|
if any(term in text for term in ("balanced", "general purpose", "everyday tasks", "routine tasks")):
|
|
return "balanced", "provider description marks it balanced/general-purpose"
|
|
if any(term in text for term in ("lowest cost", "low cost", "cheapest", "fastest", "fast and affordable")):
|
|
return "economy", "provider description marks it low-cost/fast"
|
|
return None, "provider metadata does not declare a capability tier"
|
|
|
|
|
|
def legacy_tier(provider: str, model: str) -> tuple[str | None, str]:
|
|
"""Migrate deployed pre-metadata families without predicting new IDs."""
|
|
known = {
|
|
"codex": {"luna": "economy", "terra": "balanced", "sol": "advanced"},
|
|
"claude": {
|
|
"haiku": "economy", "sonnet": "balanced", "opus": "advanced",
|
|
"fable": "advanced",
|
|
},
|
|
}
|
|
for marker, tier in known[provider].items():
|
|
if marker in model.lower():
|
|
return tier, "legacy deployed-family compatibility"
|
|
return None, "no provider capability metadata"
|
|
|
|
|
|
def _model_cost(metadata: dict[str, Any]) -> tuple[str, float] | None:
|
|
"""Return a cost only with its provider-declared comparable unit."""
|
|
for key in ("cost_rank", "cost"):
|
|
rank = metadata.get(key)
|
|
if not isinstance(rank, bool) and isinstance(rank, (int, float)) and rank >= 0:
|
|
return "ordinal", float(rank)
|
|
pricing = metadata.get("pricing")
|
|
sources = [pricing, metadata] if isinstance(pricing, dict) else [metadata]
|
|
for source in sources:
|
|
input_cost = source.get("input_cost_per_million")
|
|
output_cost = source.get("output_cost_per_million")
|
|
if (
|
|
not isinstance(input_cost, bool)
|
|
and not isinstance(output_cost, bool)
|
|
and isinstance(input_cost, (int, float))
|
|
and isinstance(output_cost, (int, float))
|
|
and input_cost >= 0
|
|
and output_cost >= 0
|
|
):
|
|
return "per-million", float(input_cost) + float(output_cost)
|
|
return None
|
|
|
|
|
|
def _comparable_costs(
|
|
pool: Iterable[tuple[str, dict[str, Any], str]]
|
|
) -> dict[str, float]:
|
|
"""Use cost only when every adequate candidate advertises one shared unit."""
|
|
records = [(model, _model_cost(details)) for model, details, _ in pool]
|
|
units = {cost[0] for _, cost in records if cost is not None}
|
|
if len(units) != 1 or any(cost is None for _, cost in records):
|
|
return {}
|
|
return {model: cost[1] for model, cost in records if cost is not None}
|
|
|
|
def _candidate_status(
|
|
provider: str, model: str, metadata: dict[str, Any], effort: str
|
|
) -> tuple[bool, str]:
|
|
if metadata.get("hidden") is True:
|
|
return False, "provider marks model hidden"
|
|
if str(metadata.get("visibility") or "").lower() in {"hidden", "internal", "private"}:
|
|
return False, "provider marks model non-public"
|
|
if metadata.get("enabled") is False or metadata.get("disabled") is True:
|
|
return False, "provider marks model disabled"
|
|
if str(metadata.get("status") or "").lower() in {"deprecated", "disabled", "retired"}:
|
|
return False, "provider marks model unavailable"
|
|
if metadata.get("is_general_purpose") is False or metadata.get("isGeneralPurpose") is False:
|
|
return False, "provider marks model specialist"
|
|
if metadata.get("modelSpecialty") not in (None, "", False):
|
|
return False, "provider marks model specialist"
|
|
modalities = metadata.get("input_modalities", metadata.get("inputModalities"))
|
|
if isinstance(modalities, list) and modalities and "text" not in {str(item).lower() for item in modalities}:
|
|
return False, "provider record is not text-capable"
|
|
if "specialist" in _text_metadata(metadata):
|
|
return False, "provider description identifies a specialist model"
|
|
if any(word in model.lower() for word in SPECIALIST_MODEL_WORDS):
|
|
return False, "model ID identifies a specialist model"
|
|
if metadata.get("supportsEffort") is False:
|
|
return False, "provider does not advertise effort support"
|
|
supported = metadata.get(
|
|
"supported_reasoning_efforts",
|
|
metadata.get("supportedReasoningEfforts", metadata.get("supportedEffortLevels")),
|
|
)
|
|
supported_efforts = {
|
|
str(item.get("reasoningEffort") or item.get("effort") or "").lower()
|
|
if isinstance(item, dict)
|
|
else str(item).lower()
|
|
for item in supported
|
|
} if isinstance(supported, list) else set()
|
|
if supported_efforts and effort not in supported_efforts:
|
|
return False, f"provider does not support {effort} effort"
|
|
if provider == "codex" and not model.lower().startswith("gpt-"):
|
|
return False, "model ID is outside the Codex general route namespace"
|
|
return True, "eligible"
|
|
|
|
|
|
def select_tier_model(
|
|
provider: str, models: Iterable[str], metadata: dict[str, dict[str, Any]],
|
|
desired: str, effort: str, current: str, *, allow_current_fallback: bool = True,
|
|
) -> tuple[str, dict[str, Any]]:
|
|
"""Select by declared tier and cost, retaining last-known-good only on outage."""
|
|
candidates: list[tuple[str, dict[str, Any], str]] = []
|
|
observations: dict[str, Any] = {}
|
|
for model in unique_models(models):
|
|
details = metadata.get(model, {})
|
|
eligible, reason = _candidate_status(provider, model, details, effort)
|
|
tier, tier_reason = metadata_tier(details)
|
|
if tier is None:
|
|
tier, legacy_reason = legacy_tier(provider, model)
|
|
tier_reason = legacy_reason if tier else tier_reason
|
|
cost = _model_cost(details)
|
|
observations[model] = {
|
|
"eligible": eligible,
|
|
"reason": reason if not eligible else tier_reason,
|
|
"tier": tier,
|
|
"cost": {"unit": cost[0], "value": cost[1]} if cost else None,
|
|
}
|
|
if eligible and tier:
|
|
candidates.append((model, details, tier))
|
|
exact = [item for item in candidates if item[2] == desired]
|
|
tier_order = {"economy": 0, "balanced": 1, "advanced": 2}
|
|
# When an economy model cannot honor the requested effort, use the nearest
|
|
# higher declared tier. A difficult request never falls to a lower tier.
|
|
higher = [
|
|
item for item in candidates
|
|
if tier_order[item[2]] > tier_order[desired]
|
|
]
|
|
nearest_higher = min(
|
|
(tier_order[item[2]] for item in higher), default=None
|
|
)
|
|
pool = exact or (
|
|
[item for item in higher if tier_order[item[2]] == nearest_higher]
|
|
if nearest_higher is not None else []
|
|
)
|
|
if not pool:
|
|
return (current if allow_current_fallback else ""), observations
|
|
comparable_costs = _comparable_costs(pool)
|
|
|
|
def key(item: tuple[str, dict[str, Any], str]) -> tuple[int, int, float, int, int, tuple[int, ...], str]:
|
|
model, details, tier = item
|
|
text = _text_metadata(details)
|
|
strongest = any(term in text for term in ("strongest", "most capable", "most intelligent"))
|
|
return (
|
|
0 if tier == desired else 1,
|
|
0 if strongest else 1,
|
|
comparable_costs.get(model, float("inf")),
|
|
# A provider moving its default is a current release signal. It
|
|
# supersedes an old advanced route, while current protects equal
|
|
# economy/balanced candidates where no comparable cost is known.
|
|
0 if desired == "advanced" and (
|
|
details.get("isDefault") is True or details.get("is_default") is True
|
|
) else 1,
|
|
0 if model == current else 1,
|
|
tuple(-part for part in model_version(model)),
|
|
model,
|
|
)
|
|
return min(pool, key=key)[0], observations
|