best_engine_ai_helper.catalog module

catalog — load and merge the bundled model catalog with the user’s cache.

The catalog lives in two layers:

  1. The bundled seed (models.yaml in the package root). Hand-maintained, always present, never deleted by auto-refresh.

  2. The user cache (~/.best-engine-ai-helper/catalog_cache.yaml). Written by catalog update; entries keyed by id overwrite matching seed entries. Absent on first run; load_catalog silently skips it in that case.

Keeping the seed immutable and the cache additive means offline machines always have a usable catalog and updates never lose hand-curated data.

Author

Warith Harchaoui <warithmetics@deraison.ai>

best_engine_ai_helper.catalog.estimate_ram(disk_gb, quant)[source]

Estimate peak inference RAM from on-disk model size.

The estimate covers the model weights plus KV cache at the default context length (4K tokens). For models with very large context windows (256K+), actual RAM may exceed this estimate significantly; treat it as a lower bound.

Parameters:

  • disk_gb (float) – On-disk footprint of the model in GB.

  • quant (str) – Quantization identifier, e.g. ‘Q4_K_M’, ‘Q8_0’, ‘FP16’, ‘Q2_K’.

Returns:

Estimated peak RAM in GB.

Return type:

float

Examples

>>> estimate_ram(6.1, 'Q4_K_M')
6.832
>>> estimate_ram(10.0, 'FP16')
10.5
best_engine_ai_helper.catalog.fetch_arena_safely(timeout=30.0)[source]

Fetch Arena ratings without ever raising, and say what happened.

The “a dead mirror must not sink a good refresh” contract lives here, in one place, because two callers need it: merge_arena_axes(), which scores entries, and the cloud refresh, which also adopts models from the same snapshot. Threading one fetch through both beats fetching twice, and beats a second copy of the contract more.

Parameters:

timeout (float) – Per-request network timeout, in seconds.

Returns:

result is a sources.arena.fetch_arena_ratings() payload, or None when no mirror answered. message is always worth showing.

Return type:

(result, message)

best_engine_ai_helper.catalog.load_catalog(catalog_path=None)[source]

Load the bundled seed catalog merged with the user’s local cache.

Cache entries whose id matches a seed entry overwrite the seed entry. New cache entries (no matching seed id) are appended. The seed is never modified on disk.

Parameters:

catalog_path (Path or None) – Path to the seed models.yaml. Defaults to the bundled file next to pyproject.toml. Pass an explicit path in tests to use a fixture.

Returns:

Merged model entries. Each entry is guaranteed to have at minimum: id, kind, ram_gb, benchmarks.

Return type:

list[dict[str, Any]]

Raises:

FileNotFoundError – If catalog_path is given explicitly and does not exist.

Examples

>>> entries = load_catalog()
>>> len(entries) > 0
True
>>> all('id' in e for e in entries)
True
best_engine_ai_helper.catalog.merge_arena_axes(entries, timeout=30.0, ratings=None)[source]

Add LM Arena conversational axes to entries, and report what happened.

Lives here rather than in the CLI because the command line and the HTTP ends mirror each other 1:1; a copy in each would be two places to drift. The callers do the printing, this does the deciding.

Never raises on a dead mirror. The refresh that precedes this call has already succeeded, and throwing away a good catalog because a third-party scrape rotted would punish the user for someone else’s outage. What it refuses to do is return quietly: merged is False and message says why whenever no rating was applied, because an axis the user believes is populated is worse than one they know is empty.

Coverage skews to hosted models. Arena rates what its voters can chat with, which is frontier endpoints rather than small quantised local builds, so this axis serves the cloud catalog far more than the local one.

Parameters:

  • entries (list[dict[str, Any]]) – Catalog entries to enrich. Not mutated; arena.merge_into copies.

  • timeout (float) – Per-request network timeout, in seconds. Ignored when ratings is supplied.

  • ratings (dict or None) – A snapshot already fetched by fetch_arena_safely(), so a caller that needs the same ratings twice fetches once. None fetches here.

Returns:

entries enriched when merged is True, returned unchanged otherwise. message is always worth showing the user.

Return type:

(entries, message, merged)

best_engine_ai_helper.catalog.normalize_apxml_spec(spec, fetched_at)[source]

Map one ApXML spec dict onto a catalog entry.

The ApXML adapter (best_engine_ai_helper.sources.apxml) yields spec and memory-fit metadata but no numeric benchmarks, so the benchmarks block is left null. disk_gb is estimated from the Q4 VRAM figure by dividing out the quant overhead — the inverse of estimate_ram().

Parameters:

  • spec (dict[str, Any]) – A normalized spec as returned by apxml.parse_model_page.

  • fetched_at (str) – ISO 8601 date recorded on the entry as its refresh timestamp.

Returns:

A catalog entry, or None when the spec lacks the slug needed to key it (such an entry could never be merged or pulled).

Return type:

dict[str, Any] or None

best_engine_ai_helper.catalog.normalize_apxml_specs(specs, fetched_at=None)[source]

Normalize a batch of ApXML specs into catalog entries, dropping unusable ones.

Parameters:

  • specs (list[dict[str, Any]]) – Specs as returned by apxml.fetch_open_weight_models.

  • fetched_at (str or None) – Refresh timestamp for every entry; defaults to today (UTC).

Returns:

Catalog entries, in input order, minus specs with no slug.

Return type:

list[dict[str, Any]]

best_engine_ai_helper.catalog.normalize_ollama_spec(spec, fetched_at)[source]

Map one Ollama-library spec onto a catalog entry.

Unlike the ApXML path, id here is already a real ollama pull tag, so the entry carries no backends restriction and servable() lets it through for Ollama. That is the whole point of this source: it widens the pool of models that can actually be served, where ApXML widens a pool that only vLLM can load.

Benchmarks are left null. The library publishes descriptions and sizes, not leaderboard results, and an invented score would be worse than an absent one — a null ranks the entry low, which is honest, whereas a guess would rank it wrongly and look authoritative.

Parameters:

  • spec (dict[str, Any]) – A spec from sources.ollama_library.to_specs.

  • fetched_at (str) – ISO 8601 date recorded as the entry’s refresh timestamp.

Returns:

A catalog entry, or None when the spec has no id or no usable size — an entry whose footprint is unknown cannot be sized against memory, and one that silently became zero would “fit” every machine.

Return type:

dict[str, Any] or None

best_engine_ai_helper.catalog.normalize_ollama_specs(specs, fetched_at=None)[source]

Normalize a batch of Ollama-library specs, dropping unusable ones.

Parameters:

  • specs (list[dict[str, Any]]) – Specs as returned by ollama_library.fetch_pullable_models.

  • fetched_at (str or None) – Refresh timestamp for every entry; defaults to today (UTC).

Returns:

Catalog entries, in input order, minus specs that could not be sized.

Return type:

list[dict[str, Any]]

best_engine_ai_helper.catalog.servable(entry, backend)[source]

Whether a backend can actually serve this catalog entry.

Parameters:

  • entry (dict[str, Any]) – A catalog entry.

  • backend (str) – ollama, vllm, lightweight, or anything else (unknown backends are trusted: this is a gate against a known mismatch, not a whitelist).

Returns:

  • bool – True when the entry names something the backend can load.

  • An entry may say so itself with backends: [...]; otherwise the rule is

  • the source. A bare name is not treated as suspect — gemma3 is a

  • perfectly legal pull tag that Ollama resolves to gemma3:latest — so the

  • discriminator is where the entry came from, not how it is spelled.

Return type:

bool

Examples

>>> servable({"id": "qwen3-vl:8b"}, "ollama")
True
>>> servable({"id": "gemma3"}, "ollama")          # bare tags are legal
True
>>> servable({"id": "qwen36-35b-a3b", "source": "apxml"}, "ollama")
False
>>> servable({"id": "x", "source": "apxml"}, "vllm")   # vLLM is not gated
True
>>> servable({"id": "x", "backends": ["vllm"]}, "ollama")
False
>>> servable({"id": "whatever"}, "some-future-backend")
True
>>> servable({"id": "llama3.3:70b", "vllm_id": "meta-llama/x"}, "lightweight")
True
>>> servable({"id": "gemma3:12b"}, "lightweight")      # no Hub repo to stream
False
best_engine_ai_helper.catalog.servable_only(rows, backend)[source]

Keep the entries a backend can serve, and say what was dropped.

Parameters:

  • rows (list[dict[str, Any]]) – Catalog entries.

  • backend (str) – The backend about to be resolved for.

Returns:

The servable subset, in the original order.

Return type:

list[dict[str, Any]]

Examples

>>> rows = [{"id": "gemma3:12b"}, {"id": "llama-3-70b", "source": "apxml"}]
>>> [r["id"] for r in servable_only(rows, "ollama")]
['gemma3:12b']
>>> len(servable_only(rows, "vllm"))        # vLLM keeps both
2
best_engine_ai_helper.catalog.write_cache(entries, cache_path=None)[source]

Merge entries into the user catalog cache by id and write it to disk.

Existing cache entries are preserved; an incoming entry whose id matches one already cached overwrites it, so a refresh is idempotent and never loses previously cached models. The bundled seed is untouched.

Parameters:

  • entries (list[dict[str, Any]]) – Catalog entries to add or update, e.g. from normalize_apxml_specs().

  • cache_path (Path or None) – Destination cache file. Defaults to CACHE_PATH; override in tests.

Returns:

The path written.

Return type:

Path