best_engine_ai_helper.cloud_catalog module

cloud_catalog — pick the best PAID model for a task, the cloud counterpart of best_engine_ai_helper.score.

Local mode (score, recommend) auto-picks the best model from a catalog by weighing benchmark quality against memory fit — free, since the hardware is already paid for. Cloud mode auto-picks the best model from a different catalog by weighing benchmark quality against price: every candidate fits on any machine (the provider owns the hardware), so cost is the resource being budgeted instead of memory.

The catalog is pricing.yaml at the package root: the same table best_engine_ai_helper.observe already reads to price a completed call, extended with provider, kind (llm/vlm), structured_output, and benchmarks per entry. One file, two consumers — pricing after the call, ranking before it — so cost and quality can never drift apart into two different numbers for the same model.

Author

Warith Harchaoui <warithmetics@deraison.ai>

best_engine_ai_helper.cloud_catalog.CLOUD_CACHE_PATH = PosixPath('~/.best-engine-ai-helper/pricing_cache.yaml'): Public alias, so the CLI and tests can name the cache without reaching into a private attribute (mirrors catalog.CACHE_PATH).

best_engine_ai_helper.cloud_catalog.REPRICEABLE: frozenset[str] = frozenset({'context_length', 'fetched_at', 'input_per_1m', 'kind', 'output_per_1m', 'source', 'structured_output'}): Fields a refresh is allowed to overwrite. Everything else on a seed entry — benchmarks above all — survives, because the price source carries no quality scores and blanking them would silently disarm the ranker.

best_engine_ai_helper.cloud_catalog.STALE_AFTER_DAYS: int = 180: How old the effective price table may get before every cloud surface says so. Providers reprice and retire models on a scale of months — the seed shipped gemini-1.5-pro long after Google retired the 1.5 line — so half a year is already generous. It is a warning, never a refusal: an old price is still better than no price, and the user may have no network.

best_engine_ai_helper.cloud_catalog.adopt_rated(fetched, ratings, known)[source]

Build catalog entries for priced models that a rating source also rates.

This is the rule that was missing, and the reason the paid catalog sat at ten models from 2024 while the price source knew three hundred. The seed could be repriced from a source, never extended from one, because an entry needs a quality score to be rankable and the price table publishes none. So the catalog could learn what a model cost and never that it existed.

An entry is adopted only where both sources speak about the same model: a list price from the price table, and a rating from the leaderboard. That is two citations, neither of them invented, and it is the whole admission test – a priced model nobody rated is still not adopted, and a rated model nobody prices still cannot be billed.

What is deliberately not filled in: general, code, math, ocr and vision stay null. The leaderboard measures conversational preference; synthesising an MMLU-style number from it would put a measurement-shaped claim in a file that bills people. score reads the Arena axes directly (see score._APP_BENCH_PRIORITY), so a null there costs nothing.

Parameters:

fetched (list[dict[str, Any]]) – Normalised price rows (sources.litellm_prices.fetch_cloud_prices()).
ratings (dict[str, Any]) – A sources.arena.fetch_arena_ratings() result.
known (set[str]) – Ids the catalog already carries; those are repriced, not re-adopted.

Returns:

Adoptable entries, cheapest first. Empty when the two sources share no model, which is a normal outcome and not an error.

Return type:

list[dict[str, Any]]

Examples

>>> priced = [
...     {"id": "claude-opus-5-5", "provider": "anthropic", "kind": "vlm",
...      "input_per_1m": 4.0, "output_per_1m": 20.0, "source": "litellm",
...      "fetched_at": "2026-10-28"},
...     {"id": "nobody-rates-this", "provider": "openai", "kind": "llm",
...      "input_per_1m": 1.0, "output_per_1m": 2.0, "source": "litellm",
...      "fetched_at": "2026-10-28"},
... ]
>>> rated = {"snapshot": "2026-10-07",
...          "scores": {"arena": {"claude-opus-5.5-high": 85.2}}}
>>> adopted = adopt_rated(priced, rated, known=set())
>>> [e["id"] for e in adopted]           # priced AND rated, only
['claude-opus-5-5']
>>> adopted[0]["benchmarks"]["arena"], adopted[0]["benchmarks"]["general"]
(85.2, None)
>>> adopted[0]["source"]
'litellm+arena'

best_engine_ai_helper.cloud_catalog.blended_price_per_1m(entry)[source]

A single $/1M-token comparability figure for one catalog entry.

A real workload’s input/output ratio varies by task (a long-document summary is input-heavy; an open-ended generation is output-heavy), so this is a simple average of the two list prices, not a workload-specific estimate — the same “rough estimate, not a bill” honesty pricing.yaml already documents for the ledger side.

Returns None when the entry carries no price at all, which is not the same as a price of zero. This used to collapse to 0.0, and the consequence was quiet and bad: the ranker inverts price into a cost score, so an entry nobody had priced came out looking free and won the cost axis outright. A model whose cost is unknown has not earned that; it has simply not been measured, and the two deserve different answers.

Parameters:: entry (dict[str, Any]) – A catalog entry with input_per_1m/output_per_1m (USD).
Returns:: (input_per_1m + output_per_1m) / 2, or None when neither price is present. A single missing side is treated as zero for that side, since a provider that bills only for output is a real arrangement.
Return type:: float or None

Examples

>>> blended_price_per_1m({"input_per_1m": 2.0, "output_per_1m": 10.0})
6.0
>>> blended_price_per_1m({"input_per_1m": 0.0, "output_per_1m": 0.0})
0.0
>>> blended_price_per_1m({"id": "nobody-priced-this"}) is None
True

best_engine_ai_helper.cloud_catalog.catalog_as_of(path=None)[source]

The date the effective price table was last known good, as ISO YYYY-MM-DD.

Reads the cache’s newest fetched_at when a refresh has happened, and otherwise the seed’s own as_of stamp. Returns None when neither exists, which is itself worth reporting: a table of prices with no date is one nobody can judge.

Parameters:: path (Path or None) – Override the seed file (tests use a fixture).
Returns:: ISO date, or None when the table carries no stamp at all.
Return type:: str or None

best_engine_ai_helper.cloud_catalog.load_cloud_catalog(path=None)[source]

Load the bundled paid-model catalog (pricing.yaml).

Parameters:: path (Path or None) – Override the catalog file (tests use a fixture). Defaults to the bundled pricing.yaml next to pyproject.toml.
Returns:: One dict per model, each carrying its YAML key as id plus provider, kind, structured_output, benchmarks, input_per_1m, output_per_1m.
Return type:: list[dict[str, Any]]
Raises:: FileNotFoundError – If path is given explicitly and does not exist.

Examples

>>> entries = load_cloud_catalog()
>>> all({"id", "provider", "kind"} <= e.keys() for e in entries)
True

best_engine_ai_helper.cloud_catalog.pick_cloud(catalog, kind, application=None, quality_vs_cost=0.7, provider=None)[source]

Return the single best paid candidate, or None if nothing matches.

Thin wrapper over rank_cloud() — rank_cloud(...)[0], or None on an empty ranking, so callers do not need an IndexError guard.

Parameters:

catalog (see rank_cloud().)
kind (see rank_cloud().)
application (see rank_cloud().)
quality_vs_cost (see rank_cloud().)
provider (see rank_cloud().)

Returns:

The top-ranked entry, or None when no candidate matches kind (and provider, if given).

Return type:

dict[str, Any] or None

Examples

>>> catalog = [{"id": "m", "provider": "p", "kind": "llm",
...             "structured_output": True, "benchmarks": {"general": 80},
...             "input_per_1m": 1.0, "output_per_1m": 2.0}]
>>> pick_cloud(catalog, "llm")["id"]
'm'
>>> pick_cloud(catalog, "vlm") is None
True

best_engine_ai_helper.cloud_catalog.provider_status(catalog=None)[source]

Which cloud providers are usable right now, and for the unusable, why.

This package claims to route on the keys you actually hold, and that claim is only worth something if a caller can see it being applied. A provider whose key is missing is not a candidate, and saying so plainly — with the name of the variable to set — turns a silently shorter list of options into an actionable one.

Never returns, logs, or compares a key value: only whether one resolved, and the name of the variable it would come from. The whole point of storing api_key_env rather than the key is that the secret stays out of every descriptor, report and screen, and that holds here too.

Parameters:: catalog (list of dict or None) – Cloud entries to take providers from. Defaults to the bundled pricing.yaml.
Returns:: One row per provider, sorted by name: {"provider", "key_env", "available", "reason", "models"}.
Return type:: list of dict

best_engine_ai_helper.cloud_catalog.rank_cloud(catalog, kind, application=None, quality_vs_cost=0.7, provider=None)[source]

Rank paid catalog candidates by a quality/price trade-off.

Mirrors score.rank()’s shape (structured-output capability first, then a combined score), so the local and cloud pickers read the same way. Each candidate gets score (raw benchmark), price_per_1m (blended_price_per_1m()), and combined — a 0..1 blend of min-max-normalized quality and (inverted) price across the candidate pool.

Parameters:

catalog (list[dict[str, Any]]) – Entries from load_cloud_catalog().
kind ({'llm', 'vlm'}) – vlm keeps only vision-capable entries; llm keeps both llm and vlm entries (a VLM answers text-only prompts too — same rule as score.rank()).
application (str or None) – Benchmark axis keyword ("code", "math", "ocr", "vision", "chat", "generalist"), forwarded to score._benchmark_score(). None uses the default kind-based rule (vision axis for a VLM, general for an LLM).
quality_vs_cost (float) – 0..1 weight on quality vs. price; clamped into range. 1.0 picks purely on benchmark score (price ignored); 0.0 picks purely on price (quality ignored, subject to the structured-output gate). Defaults to DEFAULT_QUALITY_VS_COST.
provider (str or None) – Restrict candidates to one provider (e.g. "openai"), case- insensitive. None ranks across every provider in the catalog.

Returns:

Candidates sorted best-first. Empty when nothing matches kind (and provider, if given).

Return type:

list[dict[str, Any]]

Examples

>>> catalog = [
...     {"id": "cheap", "provider": "p", "kind": "llm", "structured_output": True,
...      "benchmarks": {"general": 70}, "input_per_1m": 0.1, "output_per_1m": 0.1},
...     {"id": "strong", "provider": "p", "kind": "llm", "structured_output": True,
...      "benchmarks": {"general": 90}, "input_per_1m": 5.0, "output_per_1m": 15.0},
... ]
>>> rank_cloud(catalog, "llm", quality_vs_cost=1.0)[0]["id"]  # quality only
'strong'
>>> rank_cloud(catalog, "llm", quality_vs_cost=0.0)[0]["id"]  # price only
'cheap'

best_engine_ai_helper.cloud_catalog.reprice(seed, fetched)[source]

Overlay fetched prices onto the catalog we carry, and name what was missed.

The merge is deliberately narrow. Only REPRICEABLE fields are taken from the source; benchmarks and everything else stay as the seed has them, because the price table carries no quality scores and a refresh that blanked them would leave rank_cloud ranking every model at zero.

Nothing new is added either. A model the source knows and the seed does not has a price but no score, so adding it would either flood the ranking with unscorable candidates or require inventing a number. Those ids come back as the second return value for a human to triage — see catalog update --cloud, which prints them.

Parameters:

seed (list[dict[str, Any]]) – The catalog as loaded today.
fetched (list[dict[str, Any]]) – Normalised source entries (sources.litellm_prices.fetch_cloud_prices()).

Returns:

The repriced entries, and the ids of seed models the source did not know. An id in that second list is a model the source has never heard of: usually retired, occasionally renamed, and in both cases worth a look rather than an automatic deletion.

Return type:

tuple[list[dict[str, Any]], list[str]]

Examples

>>> seed = [{"id": "m", "provider": "p", "kind": "llm",
...          "benchmarks": {"general": 80}, "input_per_1m": 9.0,
...          "output_per_1m": 9.0}]
>>> live = [{"id": "m", "provider": "p", "kind": "vlm",
...          "input_per_1m": 1.0, "output_per_1m": 2.0,
...          "context_length": 128000}]
>>> merged, missing = reprice(seed, live)
>>> merged[0]["input_per_1m"], merged[0]["context_length"], missing
(1.0, 128000, [])
>>> merged[0]["benchmarks"]          # never touched by a price source
{'general': 80}
>>> reprice(seed, [])[1]
['m']

best_engine_ai_helper.cloud_catalog.staleness_note(today=None, path=None)[source]

One sentence about how old the prices are, or None when they are fresh.

Every cloud surface prints this rather than deciding for the user. An old price is still better than no price — and the machine may have no network — so this never refuses a resolution, it only stops a stale number from passing itself off as current.

Parameters:

today (str or None) – ISO date to measure against; defaults to today (UTC).
path (Path or None) – Override the seed file (tests use a fixture).

Returns:

The note, or None when the table is within STALE_AFTER_DAYS.

Return type:

str or None

Notes

This has deliberately no doctest. The answer depends on machine-local state outside the repository – catalog_as_of reads the user’s price cache first, and that file exists only where somebody has run catalog update –cloud. A doctest asserting None therefore passed on the author’s machine, where a cache supplies a stamp, and failed in CI, where the shipped seed carries none on purpose (see the test below for why). Both branches are pinned in tests/test_cloud_prices.py, which monkeypatches _CLOUD_CACHE_PATH away and writes its own seed, so the assertion reads the same on every machine.

best_engine_ai_helper.cloud_catalog.write_cloud_cache(entries, cache_path=None)[source]

Write repriced entries to the user cache, leaving the bundled seed alone.

Parameters:

entries (list[dict[str, Any]]) – Entries to persist, each carrying an id.
cache_path (Path or None) – Destination. Defaults to CLOUD_CACHE_PATH; override in tests.

Returns:

The path written.

Return type:

Path