99 lines
3.2 KiB
Python
99 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Resolve stable Switchyard selectors through the stewarded model catalog."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
CATALOG_PATH = Path(
|
|
os.environ.get("HERMES_ROUTING_CATALOG_PATH", "/routing-catalog/catalog.json")
|
|
)
|
|
DEFAULTS = {
|
|
"codex": {
|
|
"low": "gpt-5.6-luna",
|
|
"medium": "gpt-5.6-terra",
|
|
"high": "gpt-5.6-sol",
|
|
"xhigh": "gpt-5.6-sol",
|
|
},
|
|
"claude": {
|
|
"low": "claude-haiku-4-5-20251001",
|
|
"medium": "claude-sonnet-5",
|
|
"high": "claude-opus-5",
|
|
"xhigh": "claude-opus-5",
|
|
},
|
|
}
|
|
PREFIXES = {"codex": "gpt-", "claude": "claude-"}
|
|
TIER_DEFAULTS = {
|
|
"codex": {
|
|
"luna": "gpt-5.6-luna",
|
|
"terra": "gpt-5.6-terra",
|
|
"sol": "gpt-5.6-sol",
|
|
},
|
|
"claude": {
|
|
"haiku": "claude-haiku-4-5-20251001",
|
|
"fable": "claude-fable-5",
|
|
"sonnet": "claude-sonnet-5",
|
|
"opus": "claude-opus-5",
|
|
},
|
|
}
|
|
|
|
|
|
def load_catalog(path: Path = CATALOG_PATH) -> dict[str, Any]:
|
|
"""Load a valid routing catalog, falling back to an empty mapping."""
|
|
try:
|
|
value = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError, json.JSONDecodeError):
|
|
return {}
|
|
return value if isinstance(value, dict) else {}
|
|
|
|
|
|
def resolve_model(
|
|
provider: str,
|
|
selector: str,
|
|
effort: str,
|
|
catalog: dict[str, Any] | None = None,
|
|
) -> str:
|
|
"""Resolve an automatic or capability-tier selector to an exact model ID."""
|
|
if provider not in DEFAULTS or effort not in DEFAULTS[provider]:
|
|
raise ValueError("unsupported provider or effort")
|
|
prefix = PREFIXES[provider]
|
|
if selector.startswith(prefix):
|
|
return selector
|
|
|
|
document = catalog if catalog is not None else load_catalog()
|
|
providers = document.get("providers", {}) if isinstance(document, dict) else {}
|
|
record = providers.get(provider, {}) if isinstance(providers, dict) else {}
|
|
if not isinstance(record, dict):
|
|
record = {}
|
|
mapping_name = "resolved" if selector == "auto" else "tiers"
|
|
mapping = record.get(mapping_name, {})
|
|
candidate = mapping.get(effort if selector == "auto" else selector, "") if isinstance(mapping, dict) else ""
|
|
if not isinstance(candidate, str) or not candidate.startswith(prefix):
|
|
candidate = (
|
|
DEFAULTS[provider][effort]
|
|
if selector == "auto"
|
|
else TIER_DEFAULTS[provider].get(selector, DEFAULTS[provider][effort])
|
|
)
|
|
return candidate
|
|
|
|
|
|
def resolve_route(route: str, catalog: dict[str, Any] | None = None) -> str:
|
|
"""Resolve a route/<provider>/<selector>/<effort> target to a model ID."""
|
|
parts = route.split("/")
|
|
if len(parts) != 4 or parts[0] != "route":
|
|
return route
|
|
return resolve_model(parts[1], parts[2], parts[3], catalog)
|
|
|
|
|
|
def resolve_worker_route(route: str, catalog: dict[str, Any] | None = None) -> str:
|
|
"""Resolve a worker selector while preserving the worker route envelope."""
|
|
parts = route.split("/")
|
|
if len(parts) != 4 or parts[0] != "worker":
|
|
raise ValueError("unsupported worker route")
|
|
model = resolve_model(parts[1], parts[2], parts[3], catalog)
|
|
return f"worker/{parts[1]}/{model}/{parts[3]}"
|