best_engine_ai_helper package

Subpackages

Submodules

Module contents

best_engine_ai_helper — pick the best local large language model (LLM) or vision-language model (VLM) for the current hardware.

Public API (importable without invoking the CLI):

from best_engine_ai_helper.detect import platform_name, chip_vendor, available_memory from best_engine_ai_helper.catalog import load_catalog, estimate_ram from best_engine_ai_helper.hardware import load_hardware, lookup_chip from best_engine_ai_helper.score import select, rank, effective_budget from best_engine_ai_helper import airllm_decide, airllm_plan # layer streaming from best_engine_ai_helper import airllm_compare # streamed vs native, timed from best_engine_ai_helper import imagine_plan, imagine_generate # text-to-image from best_engine_ai_helper import text_model, vision_model # cheap tag resolvers

An OpenAI-compatible HTTP surface sits on the same app as the GUI and the MCP endpoint (best-engine-ai-helper serve): point any OpenAI SDK at /v1 and ask for the model named best. See best_engine_ai_helper.openai_api.

The CLI entry point is best-engine-ai-helper (see cli.py). Importing this package does not trigger any CLI parsing or subprocess calls.

Downstream suite packages that just need “which model tag do I use?” should call text_model() / vision_model() — they resolve an env override, then the selection persisted by pull, then a safe default, without ever probing hardware, so they are cheap and deterministic (CI-safe).

Author

Warith Harchaoui <warithmetics@deraison.ai>

best_engine_ai_helper.available_memory()[source]

Detect available memory for model inference.

Reads three pools from os_helper’s hardware facts:

  1. Apple Silicon unified memory (macOS with Apple chip)

  2. NVIDIA/AMD VRAM (summed across all visible GPUs)

  3. System RAM (always populated)

Returns:

A dict with the following keys:

unified_gbfloat or None

Apple Silicon unified memory pool, in GB.

vram_gbfloat or None

Discrete GPU VRAM (sum of all visible GPUs), in GB.

ram_gbfloat

Total system RAM in GB. Always a positive float.

Return type:

dict[str, float | None]

Examples

>>> mem = available_memory()
>>> set(mem.keys()) == {'unified_gb', 'vram_gb', 'ram_gb'}
True
>>> mem['ram_gb'] > 0
True
best_engine_ai_helper.chip_name()[source]

Return the Apple Silicon chip name (e.g. "Apple M2 Max"), or None.

Delegates to os_helper.apple_chip_name(); None on non-macOS platforms or when the chip line is absent (old Intel Macs).

Return type:

str | None

best_engine_ai_helper.chip_vendor()[source]

Identify the primary compute vendor for model inference.

Returns:

One of: ‘apple’, ‘nvidia’, ‘amd’, ‘intel’, ‘cpu’. Delegates to os_helper.gpu_vendor().

Return type:

str

Examples

>>> chip_vendor() in ('apple', 'nvidia', 'amd', 'intel', 'cpu')
True
best_engine_ai_helper.compute_profile()[source]

Describe the machine’s inference accelerator and memory bandwidth.

Returns a dict with:

  • accelerator: "gpu-metal" (Apple Silicon), "gpu-cuda" (NVIDIA), "gpu-rocm" (AMD), or "cpu" (no discrete accelerator detected).

  • chip: the chip / GPU name when known, else None.

  • bandwidth_gbs: memory bandwidth in GB/s when the chip/GPU matches a known model in _APPLE_BANDWIDTH_GBS / _NVIDIA_BANDWIDTH_GBS / _AMD_BANDWIDTH_GBS, else None. This is the ceiling on decode throughput; token generation reads the whole active model from memory once per token, so tokens/s scales with it.

Bandwidth is tabulated for Apple Silicon (per-chip) and for discrete NVIDIA/AMD GPUs (per-board, matched on the model name os_helper reports). An unrecognised GPU model — a new SKU not yet in the table, or a multi-GPU box where the name string is ambiguous — degrades to bandwidth_gbs: None rather than a fabricated number; callers treat that as “throughput not estimated”, never as zero.

Return type:

dict[str, Any]

best_engine_ai_helper.default_backend()[source]

Return the backend for the current machine.

A three-branch decision tree, matching what each backend actually supports well rather than what it merely runs on:

  • macOS -> Ollama. vLLM ships no native macOS runtime.

  • CPU-only (Linux or macOS) -> Ollama. vLLM’s CPU-only path — notably on Linux — is its own reported weak spot; Ollama’s CPU path is the well-trodden one.

  • discrete GPU (NVIDIA/AMD) on Linux — the real production case -> vLLM, chiefly for the request-batching/concurrency handling it offers under real load, which Ollama does not match.

Windows falls back to Ollama too (vLLM has no native Windows wheel, only a WSL path), so in practice: vLLM only on Linux with a real discrete GPU; Ollama everywhere else. Endgame: when vLLM gains a first-class runtime on Mac, Windows, or CPU-only Linux, replace this whole body with return "vllm" and the suite is fully on vLLM.

Return type:

str

best_engine_ai_helper.disk_profile(measure=False, directory=None)[source]

Describe the storage this machine would stream model layers from.

The counterpart of compute_profile() for a disk-bound backend. Where memory bandwidth bounds an in-memory decode, disk bandwidth bounds a layer-streaming one (offload/lightweight), which reads every layer back per token.

Parameters:

  • measure (bool) – Run measure_disk_read_gbs() instead of reading the class table. Off by default: it writes a few hundred MB and takes a second or two, and a recommendation should stay a deterministic function of the hardware unless the caller asks for a measurement.

  • directory (str or None) – Volume to measure, when measure is True. None uses the temp directory.

Returns:

storagestr

Class from storage_class().

read_gbsfloat or None

Sequential read bandwidth in GB/s; None when the class is unknown and no measurement succeeded.

basis{‘measured’, ‘declared’, ‘unknown’}

Where read_gbs came from — the same honesty rule the throughput estimates follow: a measured number is never presented as a table lookup, nor the reverse.

free_gbfloat or None

Free space on the volume holding the home directory, where both the HuggingFace cache lands.

Return type:

dict[str, Any]

Examples

>>> p = disk_profile()
>>> p['basis'] in ('measured', 'declared', 'unknown')
True
>>> p['read_gbs'] is None or p['read_gbs'] > 0
True
best_engine_ai_helper.effective_budget(hw, headroom=0.5, load=None)[source]

Compute the memory budget (GB) a model may occupy at run time.

The budget is the accelerator’s usable memory pool, scaled by an extra headroom margin left for the operating system, your own application, and KV-cache growth as context fills. On Apple Silicon the usable pool is not the whole unified memory: Metal caps GPU allocations at about 66% of the pool at or below 36 GB and about 75% above it, beyond which inference spills to CPU. Compare a catalog entry’s ram_gb (already a peak-inference estimate, weights plus a moderate KV cache) against this budget.

Parameters:

  • hw (dict[str, float | None]) – Output of detect.available_memory(). Expected keys: unified_gb, vram_gb, ram_gb.

  • headroom (float) – Extra safety fraction applied on top of the accelerator cap, reserving room for the OS, the caller’s workload, and KV growth. Defaults to MAX_HEADROOM (0.5) and is clamped down to it — a larger value is never honoured, to keep picks realistic.

  • load (dict or None) – Live server state from detect.server_load() (available_ram_gb, cpu_percent, disk_free_gb, …). When given, the theoretical accelerator budget is additionally capped at what is ACTUALLY free right now (another process, or an already-running engine, holds memory the static hardware totals in hw know nothing about), and further derated when the CPU is already saturated or the disk is nearly full. None (the default) reproduces the pre-existing, load-blind behaviour exactly.

Returns:

Effective memory budget in GB.

Return type:

float

Examples

>>> effective_budget({'unified_gb': 96.0, 'vram_gb': None, 'ram_gb': 96.0})
36.0
>>> effective_budget({'unified_gb': None, 'vram_gb': 24.0, 'ram_gb': 64.0})
11.04
>>> # headroom above the 0.5 ceiling is clamped, not honoured
>>> effective_budget({'unified_gb': 96.0, 'vram_gb': None, 'ram_gb': 96.0}, headroom=0.85)
36.0
>>> # a busy machine gets a smaller budget than an idle one with the same hardware
>>> hw = {'unified_gb': 96.0, 'vram_gb': None, 'ram_gb': 96.0}
>>> effective_budget(hw, load={'available_ram_gb': 10.0})
10.0
best_engine_ai_helper.ensure(directory='.', *, brief='llm.brief.yaml', engine='llm.engine.yaml', backend='auto', endpoint=None, write=True)[source]

Return the engine descriptor for a repo, resolving it on first use.

Missing-file policy (the suite contract):

  • the engine file exists -> load and return it (fast path, no detection);

  • it is missing but the brief exists -> resolve from the brief, write the engine file (unless write=False), and return it;

  • both are missing -> raise. A committed brief is mandatory; its absence is a real bug, not a machine that has not run detection yet.

Parameters:

  • directory (str | Path) – Repo directory holding the two contract files.

  • brief (str) – Filenames within directory (default to the suite canonical names).

  • engine (str) – Filenames within directory (default to the suite canonical names).

  • backend (see resolve().)

  • endpoint (see resolve().)

  • write (bool) – Persist a freshly resolved engine file. False resolves in-memory only.

Return type:

dict[str, Any]

best_engine_ai_helper.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.estimated_tokens_per_second(entry, bandwidth_gbs, backend='ollama', calibrated=True, *, disk_read_gbs=None)[source]

Estimate local decode throughput (tokens/s) for a model on this machine.

Token generation is memory-bandwidth bound: each new token requires reading the model’s active weights from memory once, so the ceiling is bandwidth / model_bytes. The estimate derates that ceiling by a backend-specific decode efficiency (see _DECODE_EFFICIENCY_BY_BACKEND) to reflect KV-cache reads, kernel overhead, and sampling. It describes steady-state generation, not the compute-bound prefill of a long prompt.

Parameters:

  • entry (dict[str, Any]) – Catalog entry; its model_footprint_gb() is the active-model size.

  • bandwidth_gbs (float or None) – Memory bandwidth in GB/s from detect.compute_profile(). When None (unknown hardware, e.g. an unrecognised discrete-GPU model) the estimate is not computable and None is returned.

  • backend ({'ollama', 'vllm', 'lightweight'}) – Serving backend. 'lightweight' switches the whole model: throughput is bounded by disk read bandwidth rather than memory bandwidth, and disk_read_gbs is the input that matters (see offload.estimated_tokens_per_second()). For the other two it affects both the size proxy (heavier FP16 weights under vLLM decode more slowly than the Q4 Ollama figure) AND the decode efficiency (vLLM’s PagedAttention + CUDA-graph decode tracks closer to the bandwidth ceiling than llama.cpp’s — confirmed to differ on identical Ubuntu + discrete-GPU hardware, not just a cross-platform artifact).

  • calibrated (bool) – Apply this machine’s measured correction factor when one exists (see throughput.calibration_factor()). On by default: an estimate that has been corrected against a real measurement on this hardware is strictly better than one that has not. Pass False to see the raw bandwidth model, which is what the calibration itself is derived against – leaving it True there would feed the correction back into its own input. Ignored under backend='lightweight': the calibration was measured against a memory-bound decode and does not describe a disk-bound one.

  • disk_read_gbs (float or None) – Sequential disk read bandwidth in GB/s, from detect.disk_profile(). Only read under backend='lightweight'; None lets that branch look it up from the storage class.

Returns:

Estimated tokens/s, rounded to one decimal, or None when bandwidth or model size is unavailable. Under backend='lightweight' the figure is rounded to three decimals instead, because the honest answer there is usually a fraction of a token per second.

Return type:

float or None

Examples

>>> entry = {'id': 'a', 'ram_gb': 9.9, 'size_b': 14}
>>> estimated_tokens_per_second(entry, 400.0, calibrated=False)
26.3
>>> estimated_tokens_per_second(entry, None) is None
True
>>> # Same model, streamed off a 4 GB/s drive: a thousand times slower.
>>> estimated_tokens_per_second(entry, 400.0, 'lightweight', disk_read_gbs=4.0)
0.121
best_engine_ai_helper.family_brief(family_id)[source]

Return the representative brief for a family (ready for engine.resolve()).

Parameters:

family_id (str) – The family id.

Returns:

The brief block with mode: local.

Return type:

dict[str, Any]

Examples

>>> family_brief("F2")["kind"]
'llm'
best_engine_ai_helper.get_family(family_id)[source]

Return one family by id, with a helpful error when it is unknown.

Parameters:

family_id (str) – The family id ("F1", "F2" or "F3").

Returns:

The family dict.

Return type:

dict[str, Any]

Raises:

KeyError – If no family carries that id.

Examples

>>> get_family("F3")["name"]
'embeddings'
best_engine_ai_helper.get_usage(name)[source]

Return one profile by name, with a helpful error when it is unknown.

Parameters:

name (str) – The profile name (e.g. "text2sql").

Returns:

The profile dict.

Return type:

dict[str, Any]

Raises:

KeyError – If no profile carries that name; the message suggests close matches.

Examples

>>> get_usage("text2sql")["family"]
'F1'
best_engine_ai_helper.image_footprint_gb(entry, *, dtype=None, offload=False, width=None, height=None)

Resident memory (GB) a pipeline needs to produce one image.

weights x overhead + latents, where which weights depends on how the components take turns:

  • All resident (offload=False, the fast path): every component at once, so the weight term is the whole parameter count.

  • Component offloaded (offload=True): diffusers keeps only the component currently computing on the device, so the peak is the largest one — the denoiser, at _DENOISER_SHARE of the total unless the entry names its components. Slower by the handful of transfers per image that costs, and the difference between running FLUX on a 24 GB card and not running it.

The latent term is what makes resolution expensive: it scales with the output area, not with the model.

Parameters:

  • entry (dict[str, Any]) – Catalog entry; needs size_b.

  • dtype (str or None) – Compute dtype override.

  • offload (bool) – Size for a component-offloaded run rather than an all-resident one.

  • width (int or None) – Output size. None uses the model’s native_size(), which is the size it is good at anyway.

  • height (int or None) – Output size. None uses the model’s native_size(), which is the size it is good at anyway.

Returns:

Resident GB, or inf when the entry carries no parameter count — infinity rather than zero, so an unsized entry can never read as “fits everywhere” and win every comparison.

Return type:

float

Examples

>>> sdxl = {"id": "sdxl", "kind": "image", "size_b": 3.5, "native_size": "1024x1024"}
>>> footprint_gb(sdxl)
9.8
>>> footprint_gb(sdxl, offload=True)
7.2
>>> footprint_gb(sdxl, width=512, height=512)
9.0
>>> footprint_gb({"id": "sd15", "size_b": 1.07, "native_size": "512x512"})
2.9
>>> footprint_gb({"id": "unsized", "kind": "image"})
inf
best_engine_ai_helper.imagine_available()

Whether the diffusers runtime is importable in this interpreter.

False on a base install: it lives behind the imagine extra. Checked rather than assumed for the same reason offload.available() is — declared and importable are different claims, and a stripped container separates them even when the extra was asked for. A real import rather than find_spec for the same reason too: a package whose __init__ raises has a findable spec and is still unusable, which is how lightweight behaves on a Mac without mlx.

Returns:

True when import diffusers succeeds.

Return type:

bool

Examples

>>> available() in (True, False)
True
best_engine_ai_helper.imagine_generate(prompt, *, model, width=1024, height=1024, steps=28, guidance=None, negative_prompt=None, seed=None, dtype=None, offload=False, hf_token=None, loaded=None)

Generate one image and return it as PNG bytes.

Parameters:

  • prompt (str) – What to draw.

  • model (str) – HuggingFace repository id, as in load().

  • width (int) – Output size. Stay near the model’s native_size(): a diffusion model asked for an unfamiliar resolution does not produce more detail, it produces duplicated subjects.

  • height (int) – Output size. Stay near the model’s native_size(): a diffusion model asked for an unfamiliar resolution does not produce more detail, it produces duplicated subjects.

  • steps (int) – Denoising steps. Use the model’s own figure (steps_for()) — a distilled checkpoint at 28 steps is slower and worse.

  • guidance (float or None) – Classifier-free guidance scale. None uses the pipeline’s default, which is the right answer for a distilled model (they want 0.0) and for a normal one (7.5) alike, and those two defaults are why this is not a house constant.

  • negative_prompt (str or None) – What to avoid. Ignored by distilled pipelines that run without guidance; passed through when given.

  • seed (int or None) – Seed for reproducibility. None leaves the generator unseeded.

  • dtype (str | None) – Passed to load() when it has to load the pipeline.

  • offload (bool) – Passed to load() when it has to load the pipeline.

  • hf_token (str | None) – Passed to load() when it has to load the pipeline.

  • loaded (Any or None) – A pipeline from a previous load(), reused as-is. Loading one is seconds to minutes, so a caller making several images should pay it once.

Returns:

The image encoded as PNG.

Return type:

bytes

Raises:

ImportError – When diffusers cannot be imported.

Examples

>>> generate("a red bicycle", model="stabilityai/sdxl-turbo")
best_engine_ai_helper.imagine_load(model, *, dtype=None, offload=False, hf_token=None)

Load a diffusers pipeline.

The adapter, and the only function here that touches the upstream package.

Parameters:

  • model (str) – HuggingFace repository id, or a local path to a saved pipeline.

  • dtype (str or None) – Compute dtype (fp16, bf16, fp32). None lets torch pick float16 on an accelerator and float32 on CPU, which is the only pair that works on both without producing black images.

  • offload (bool) – Hand the pipeline to enable_model_cpu_offload(), so only the component currently computing stays on the device. Slower per image, and what makes a large pipeline fit at all.

  • hf_token (str or None) – HuggingFace token, for a gated repository (FLUX is gated).

Returns:

The diffusers pipeline, untouched.

Return type:

Any

Raises:

ImportError – When diffusers cannot be imported, with the command that fixes it.

Examples

>>> load("stabilityai/sdxl-turbo")
best_engine_ai_helper.imagine_pick(plans)

Choose one image model: the best that runs, else the fastest that fits.

Parameters:

plans (list of dict) – plan() dicts, typically from rank().

Returns:

The chosen plan, or None when nothing fits at all.

Return type:

dict or None

Examples

>>> rows = [
...     {"id": "sd15", "kind": "image", "size_b": 1.07, "steps": 28,
...      "native_size": "512x512", "hf_id": "x/sd15", "benchmarks": {"image": 60}},
...     {"id": "sdxl", "kind": "image", "size_b": 3.5, "steps": 28,
...      "native_size": "1024x1024", "hf_id": "x/sdxl", "benchmarks": {"image": 75}},
... ]
>>> pick(rank(rows, budget_gb=11.0, bandwidth_gbs=1000.0))["catalog_id"]
'sdxl'
>>> pick(rank(rows, budget_gb=4.0, bandwidth_gbs=1000.0))["catalog_id"]
'sd15'
>>> pick([]) is None
True
best_engine_ai_helper.imagine_plan(entry, *, budget_gb=None, bandwidth_gbs=None, dtype=None, width=None, height=None, steps=None)

Everything known about running ONE image model on this machine.

Tries the all-resident footprint first and falls back to the component-offloaded one when that does not fit — the same escalation offload.rank() makes with prefetching, for the same reason: the knob exists, the runtime exposes it, and trying it before rejecting a model is the difference between running FLUX on a 24 GB card and reporting that it cannot be done.

Parameters:

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

  • budget_gb (float or None) – Memory budget from score.effective_budget(). None skips the fit verdict.

  • bandwidth_gbs (float or None) – Memory bandwidth, for the time estimate.

  • dtype (str or None) – Compute dtype override.

  • width (int or None) – Output size; None uses the model’s native one.

  • height (int or None) – Output size; None uses the model’s native one.

  • steps (int or None) – Step override.

Returns:

model, catalog_id, resident_gb, offload (whether the fit needed it), size, steps, seconds_per_image, fits, usable and blockers.

Return type:

dict[str, Any]

Examples

>>> sdxl = {"id": "sdxl", "kind": "image", "size_b": 3.5, "denoiser_size_b": 2.6,
...         "steps": 28, "native_size": "1024x1024",
...         "hf_id": "stabilityai/stable-diffusion-xl-base-1.0"}
>>> p = plan(sdxl, budget_gb=11.0, bandwidth_gbs=1000.0)
>>> p["resident_gb"], p["offload"], p["seconds_per_image"], p["usable"]
(9.8, False, 10.3, True)
>>> # An 8 GB budget only fits it with the components taking turns.
>>> tight = plan(sdxl, budget_gb=8.0, bandwidth_gbs=1000.0)
>>> tight["resident_gb"], tight["offload"], tight["fits"]
(7.5, True, True)
>>> plan(sdxl, budget_gb=2.0)["blockers"][0][:38]
'needs 7.5 GB resident even with the co'
best_engine_ai_helper.imagine_rank(entries, *, budget_gb, bandwidth_gbs=None, dtype=None, width=None, height=None)

Rank the image models this machine could run, best first.

Parameters:

  • entries (list of dict) – Catalog entries; non-image kinds are dropped.

  • budget_gb (float) – Memory budget from score.effective_budget().

  • bandwidth_gbs (float or None) – Memory bandwidth, for the time estimates.

  • dtype (str or None) – Compute dtype override.

  • width (int or None) – Output size; None lets each model use its own native one, which is the fair comparison — a model is not worse for being a 512px model, it is cheaper.

  • height (int or None) – Output size; None lets each model use its own native one, which is the fair comparison — a model is not worse for being a 512px model, it is cheaper.

Returns:

plan() dicts, usable first, then by image-quality score, then by parameter count.

Return type:

list[dict[str, Any]]

Examples

>>> rows = [
...     {"id": "sd15", "kind": "image", "size_b": 1.07, "steps": 28,
...      "native_size": "512x512", "hf_id": "x/sd15",
...      "benchmarks": {"image": 60}},
...     {"id": "sdxl", "kind": "image", "size_b": 3.5, "steps": 28,
...      "native_size": "1024x1024", "hf_id": "x/sdxl",
...      "benchmarks": {"image": 75}},
...     {"id": "chat", "kind": "llm", "size_b": 8},
... ]
>>> [r["catalog_id"] for r in rank(rows, budget_gb=11.0, bandwidth_gbs=1000.0)]
['sdxl', 'sd15']
>>> # On a 4 GB card only the small one runs, and it says why about the other.
>>> tight = rank(rows, budget_gb=4.0, bandwidth_gbs=1000.0)
>>> [(r["catalog_id"], r["usable"]) for r in tight]
[('sd15', True), ('sdxl', False)]
best_engine_ai_helper.lightweight_available()

Whether the streaming runtime can be imported in this interpreter.

Normally True: it is this project’s own module over torch, transformers, accelerate and safetensors, all core dependencies. It is still checked rather than assumed, because a dependency being declared and a module being importable are different claims — a stripped container, a failed wheel build or a partially-installed environment separates them, and finding that out at resolution beats finding it out inside a job.

The check is a real import, and it is performed with stdout and stderr captured: importing torch can emit third-party chatter, and a capability probe has no business writing to the caller’s stdout when this package has a --json CLI and an OpenAI-compatible server. Cached, because importing torch is seconds, not milliseconds.

Returns:

True when the streaming runtime imports.

Return type:

bool

Examples

>>> available() in (True, False)
True
>>> available() is available()          # cached, so it is asked once
True
best_engine_ai_helper.lightweight_compare(entries, *, hw, disk, bandwidth_gbs, kind='llm', application=None, native_backend='ollama', new_tokens=300, prompt_tokens=0, batch=1, load=None)

Native model versus streamed model, on the same machine, side by side.

The decision this backend actually forces, which neither ranking answers alone: is the better model worth the wait? A machine that fits a 14B model comfortably can also stream a 70B one, and the two options differ by a handful of benchmark points in one direction and three orders of magnitude of latency in the other. Printing both, priced in the same unit — seconds for one answer of the same length — is the only way to make that call rather than guess at it.

Both sides are ranked by their own backend’s rules (see score.rank() for the native side, pick() for the streamed one, where bigger beats leaner), then compared on two axes only: benchmark score on the task’s axis, and wall-clock seconds for new_tokens.

One deliberate bias, stated so it can be discounted: the native timing counts decode only and ignores prompt processing, while the streamed timing is complete (its prompt rides free in the first pass — see seconds_per_pass()). That flatters the native side, which is the right direction for an honest comparison whose conclusion will usually be “stay native”.

Parameters:

  • entries (list of dict) – Catalog entries, unfiltered. Each side applies its own servability gate.

  • hw (dict[str, float | None]) – Memory pools from detect.available_memory().

  • disk (dict[str, Any]) – Storage profile from detect.disk_profile().

  • bandwidth_gbs (float or None) – Memory bandwidth from detect.compute_profile(), for the native side’s throughput estimate.

  • kind ({'llm', 'vlm'}) – Model kind to compare.

  • application (str or None) – Benchmark axis (code, math, vision, …), so both sides are scored on the task that matters.

  • native_backend ({'ollama', 'vllm'}) – The backend to compare against.

  • new_tokens (int) – Answer length both timings are computed for.

  • prompt_tokens (int) – Prompt length. Reported, and free on the streamed side.

  • batch (int) – Sequences at once. A real lever on the streamed side only, which is itself part of the answer.

  • load (dict or None) – Live load from detect.server_load(), applied to both budgets.

Returns:

native and streamed, each with budget_gb, best and rows; plus quality_gain (benchmark points streaming buys), slowdown (how many times longer the answer takes), verdict (one sentence for a human) and the inputs the comparison was made under.

Return type:

dict[str, Any]

Examples

>>> rows = [
...     {"id": "small:8b", "kind": "llm", "size_b": 8, "layers": 32,
...      "ram_gb": 5.6, "hf_id": "x/small", "benchmarks": {"general": 74}},
...     {"id": "big:70b", "kind": "llm", "size_b": 70, "layers": 80,
...      "ram_gb": 155, "hf_id": "x/big", "benchmarks": {"general": 86}},
... ]
>>> out = compare(
...     rows,
...     hw={"unified_gb": None, "vram_gb": 24.0, "ram_gb": 64.0},
...     disk={"read_gbs": 4.0, "free_gb": 4000.0, "storage": "nvme"},
...     bandwidth_gbs=1008.0,
... )
>>> out["native"]["best"]["catalog_id"], out["streamed"]["best"]["catalog_id"]
('small:8b', 'big:70b')
>>> out["quality_gain"]
12.0
>>> out["slowdown"] > 1000
True
>>> print(out["verdict"])
Streaming buys 12 benchmark points (86 vs 74) for 4751x the wait:
3.4 h instead of 3 s for a 300-token answer.
best_engine_ai_helper.lightweight_decide(*, request=None, conventional_resolved=True, compute=None, disk=None, allow_cpu=False, env=None)

Decide whether to stream layers on this machine, and say why.

The activation policy, in one place, returning a verdict rather than performing it — so the CLI, the HTTP API and engine.resolve() all ask the same question and get the same answer, and so the reasoning is inspectable (best-engine-ai-helper lightweight status) instead of implied.

The rule:

  • never → off, unconditionally. The escape hatch.

  • always → on, provided the machine has a runtime (runtime()) and the package is installed. A human asking for streaming is not second- guessed on whether it is wise, only told what it will cost.

  • auto → on only when the conventional backend resolved nothing for a kind the caller needs. Streaming is two to three orders of magnitude slower than loading a model that fits, so it is the answer to “no model at all”, never a preference.

Parameters:

  • request (object) – 'auto'/'always'/'never' or an alias, from a brief, a flag, or ENV_VAR. None resolves through requested().

  • conventional_resolved (bool) – Whether Ollama/vLLM already found a model that fits. True (the default) means the automatic trigger does not fire — there is nothing to rescue.

  • compute (dict or None) – detect.compute_profile() output. None probes it.

  • disk (dict or None) – detect.disk_profile() output. None probes it.

  • allow_cpu (bool) – Count a CPU-only machine as capable (the CPU path is experimental).

  • env (dict[str, str] or None) – Environment for ENV_VAR; None uses os.environ.

Returns:

activate : bool — the verdict. request : the normalised request that produced it. trigger : 'manual', 'fallback', 'disabled' or 'not-needed' — which branch of the rule fired. reason : one sentence, written for a human. installed / runtime / blockers : capability facts, kept even when activate is False so a caller can say what to fix.

Return type:

dict[str, Any]

Examples

>>> d = decide(request='never', conventional_resolved=False, env={})
>>> d['activate'], d['trigger']
(False, 'disabled')
>>> d = decide(request='auto', conventional_resolved=True,
...            compute={'accelerator': 'gpu-cuda'}, disk={'read_gbs': 2.5}, env={})
>>> d['activate'], d['trigger']
(False, 'not-needed')
>>> d = decide(request='auto', conventional_resolved=False,
...            compute={'accelerator': 'gpu-cuda'}, disk={'read_gbs': 2.5}, env={})
>>> d['trigger'], d['runtime']
('fallback', 'cuda')
>>> d = decide(request='always', compute={'accelerator': 'cpu'},
...            disk={'read_gbs': 2.5}, env={})
>>> d['activate'], d['blockers'][0][:37]
(False, 'this machine offers no accelerator to')
best_engine_ai_helper.lightweight_disk_required_gb(entry)

Free disk layer streaming needs: the checkpoint, and nothing else.

This used to be the constraint that actually bit. The previous runtime wrote per-layer shards beside the downloaded checkpoint before anything ran, so the cache briefly held both and a 70B model needed ~294 GB free rather than ~154 GB. That is what made nearly every row of lightweight plan read runs: NO on a normal laptop: the models fit in memory-per-layer terms and failed on scratch space.

best_engine_ai_helper.streaming reads one tensor at a time out of the original safetensors, which are memory-mapped and carry their own index. Nothing is copied and nothing is written, so peak and steady are both just the checkpoint. The two keys stay because callers and the engine file read them by name, and because “both figures are the same now” is worth being able to see.

Parameters:

entry (dict[str, Any]) – Catalog entry. Kept for call compatibility and ignored: this runtime does not requantise, so what is on disk is whatever the checkpoint ships.

Returns:

peak_gb and steady_gb, both the checkpoint’s own size.

Return type:

dict[str, float]

Examples

>>> disk_required_gb({"size_b": 70})
{'peak_gb': 140.0, 'steady_gb': 140.0}
>>> disk_required_gb({"size_b": 70, "dtype": "q4"})
{'peak_gb': 35.0, 'steady_gb': 35.0}
>>> disk_required_gb({"id": "unsized"})
{'peak_gb': inf, 'steady_gb': inf}
best_engine_ai_helper.lightweight_footprint_gb(entry, prefetching=True, context_k=4.0, batch=1)

Resident memory (GB) a model needs when its layers are streamed.

The whole point of the backend, and the one number that decides whether a machine can run a model it could never load. Derivation:

  1. A checkpoint of size_b parameters at bytes_per_param() bytes each, split across layer_count() shards, gives an average shard of checkpoint / layers.

  2. The biggest shard is the embedding/output matrix rather than a decoder layer, about _PEAK_SHARD_RATIO times the average on a modern large-vocabulary model, and peak memory follows the biggest.

  3. With prefetching, two shards are resident (the one computing and the one arriving); without it, one.

  4. Plus the KV cache, which streaming does not shrink: every layer’s keys and values stay resident for the whole generation (see kv_cache_gb()), so this term grows with context and with depth, and on a deep model it is the second-largest one.

  5. Plus _ACTIVATION_GB for activations and runtime buffers, which are bounded by one layer.

Checked against airllm’s published figures, and the check is worth stating precisely, because the headline “70B on a 4 GB GPU” is true at a context length nobody mentions. A Llama-3-70B comes to 3.0 GB serially at a 1k context — which is the demo, and it fits a 4 GB card — 3.9 GB serially at 4k, and 6.1 GB at 4k with prefetching on. The weights are the same in all three; what moves is the cache that is not streamed. So the useful way to read this function is backwards: a card of a given size buys a context length, and plan() is where that shows up as a verdict.

Parameters:

  • entry (dict[str, Any]) – Catalog entry; needs size_b, and layers when it is a MoE model.

  • prefetching (bool) – Whether prefetching is on (the default). Off halves the resident shard count at some cost in speed, and is the knob that gets a 70B model onto a 4 GB card.

  • context_k (float) – Context length in units of 1024 tokens, for the KV term. Long contexts are expensive under streaming in a way they are not elsewhere: the cache is the one thing that does not get streamed.

  • batch (int) – Sequences generated together. Multiplies the KV term, and only it — which is exactly the trade generation_time() describes from the other side: a batch of 8 answers arrives 8x sooner and costs 8 KV caches.

Returns:

Resident GB, or inf when the entry cannot be sized.

Return type:

float

Examples

>>> footprint_gb({"size_b": 70, "layers": 80})
6.1
>>> footprint_gb({"size_b": 70, "layers": 80}, prefetching=False)
3.9
>>> footprint_gb({"size_b": 70, "layers": 80, "dtype": "q4"})
2.8
>>> footprint_gb({"size_b": 32, "layers": 64})
4.0
>>> # A 32k context costs more than the weights do.
>>> footprint_gb({"size_b": 70, "layers": 80}, context_k=32)
14.9
>>> # Eight answers at once: eight KV caches, one set of shards.
>>> footprint_gb({"size_b": 70, "layers": 80}, batch=8)
14.9
>>> footprint_gb({"size_b": 671, "moe": True})     # depth unknown -> unsized
inf
best_engine_ai_helper.lightweight_generate(prompt, *, model, max_new_tokens=128, prefetching=True, hf_token=None, max_seq_len=4096, loaded=None)

Generate text with a streamed model, in this process.

There is no server to talk to: the runtime is a library, not a daemon, so this backend runs inside the caller’s interpreter. That is also why loaded exists — loading a 70B checkpoint is the expensive part, and a caller making several calls should pay it once.

Parameters:

  • prompt (str) – The text to continue.

  • model (str) – HuggingFace repository id, as in load().

  • max_new_tokens (int) – Tokens to generate. Keep it small: at streaming speeds every token is seconds of disk, so a 2000-token answer is an afternoon.

  • prefetching (bool) – Passed to load() when it has to load the model itself. Note that max_seq_len bounds prompt AND answer together, and that the tokenizer truncates to it — so a prompt near the limit leaves no room to answer.

  • hf_token (str | None) – Passed to load() when it has to load the model itself. Note that max_seq_len bounds prompt AND answer together, and that the tokenizer truncates to it — so a prompt near the limit leaves no room to answer.

  • max_seq_len (int) – Passed to load() when it has to load the model itself. Note that max_seq_len bounds prompt AND answer together, and that the tokenizer truncates to it — so a prompt near the limit leaves no room to answer.

  • loaded (Any or None) – An instance from a previous load(), reused as-is.

Returns:

The decoded continuation, with the prompt stripped from the front when the runtime echoes it back.

Return type:

str

Raises:

ImportError – When the runtime cannot be imported.

Examples

>>> generate("Hello", model="meta-llama/Llama-3.3-70B-Instruct")
best_engine_ai_helper.lightweight_generation_time(entry, read_gbs, *, new_tokens=128, prompt_tokens=0, batch=1, prefetching=True)

How long one streamed generation actually takes, end to end.

“What is this model’s speed when streamed” has three different answers, and quoting the wrong one is how a batch job gets planned as an interactive one:

  • decode_tps — the steady-state rate, one token per pass. The headline number, and the pessimistic one.

  • time_to_first_token_s — one pass, prompt included. Under streaming this is the same as the time for any other token, which is unusual: on every other backend a long prompt makes the first token the expensive one.

  • effective_tps — tokens actually delivered per second of wall clock, counting the whole batch. This is the one that decides whether a job finishes overnight, and the only one a caller can improve without changing the model.

And the lever that falls out of the pass model: because a pass costs the same whatever rides through it, a batch of B prompts costs one pass, not B passes. B independent questions answered together go B times faster per answer. A batch of 8 gives 8x. For a real batch job that is the larger of the two, and it costs nothing but the KV cache (which footprint_gb() sizes with the same argument).

Parameters:

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

  • read_gbs (float or None) – Sequential disk read bandwidth in GB/s.

  • new_tokens (int) – Tokens to generate per sequence. The first comes out of the prefill pass, so N tokens cost N passes.

  • prompt_tokens (int) – Prompt length. Costs no extra pass; reported so a caller can see that it does not, and warned about past _PREFILL_IO_BOUND_TOKENS, where per-layer compute starts to overtake the read.

  • batch (int) – Independent sequences generated together. Multiplies the tokens delivered per pass.

  • prefetching (bool) – Whether prefetching is on.

Returns:

seconds_per_pass, passes, time_to_first_token_s, total_seconds, decode_tps, effective_tps, tokens_delivered, prompt_tokens, prompt_is_free and batch. None when the drive or the model cannot be sized.

Return type:

dict[str, Any] or None

Examples

>>> t = generation_time({"size_b": 70}, 4.0, new_tokens=100)
>>> t["seconds_per_pass"], t["passes"], t["decode_tps"]
(41.176, 100, 0.024)
>>> round(t["total_seconds"] / 3600, 2)        # hours for 100 tokens
1.14
>>> t["time_to_first_token_s"] == t["seconds_per_pass"]
True
>>> # A 2000-token prompt adds nothing: the pass reads the same bytes.
>>> long_prompt = generation_time({"size_b": 70}, 4.0, new_tokens=100,
...                               prompt_tokens=2000)
>>> long_prompt["total_seconds"] == t["total_seconds"]
True
>>> # Eight questions at once cost the same passes and deliver eight times
>>> # the tokens.
>>> batched = generation_time({"size_b": 70}, 4.0, new_tokens=100, batch=8)
>>> batched["total_seconds"] == t["total_seconds"]
True
>>> round(batched["effective_tps"] / t["effective_tps"], 1)
8.1
>>> generation_time({"size_b": 70}, None) is None
True
best_engine_ai_helper.lightweight_load(model, *, prefetching=True, hf_token=None, max_seq_len=4096)

Load a model through this package’s layer-streaming runtime.

The seam between the arithmetic in this module and best_engine_ai_helper.streaming.StreamedModel, which owns the hooks that materialise one decoder layer at a time.

Parameters:

  • model (str) – HuggingFace repository id (meta-llama/Llama-3.3-70B-Instruct), or a local path to a downloaded checkpoint.

  • prefetching (bool) – Read the next layer’s tensors while the current layer computes.

  • hf_token (str or None) – HuggingFace token, for a gated repository.

  • max_seq_len (int) – Context length to configure, in tokens. Defaults to 4096, matching the context footprint_gb() prices. The tokenizer truncates to it rather than refusing a longer prompt, so the number matters. Raising this costs KV cache in exactly the proportion kv_cache_gb() describes, so size it against budget_gb() rather than maximising it.

Returns:

The lightweight.AutoModel instance, untouched, carrying its own .tokenizer, .device and .generate.

Return type:

Any

Raises:

ImportError – When the runtime cannot be imported, with the command that fixes it.

Examples

>>> load("meta-llama/Llama-3.3-70B-Instruct")
best_engine_ai_helper.lightweight_pick(plans)

Choose one streaming candidate, inverting the usual preference.

Every other backend in this package picks the LEANEST model within a few benchmark points of the best — the anti-greed rule, and the right one when a gigabyte of footprint is paid on every run and competes with the caller’s own workload. Streaming breaks that premise: resident memory is one layer whatever the model’s size, so leanness buys nothing, and the anti-greed rule lands on an 8B model — the one outcome that makes no sense at all, since an 8B model runs natively a hundred times faster.

So, among candidates that fit memory AND disk and clear COMFORT_TPS_STREAMING: the most capable one, by benchmark score, with parameter count as the tiebreak. Not simply the biggest — that was the first rule here and the side-by-side comparison made its error obvious: a 141B mixture-of-experts scoring 84 beat a dense 72B scoring 87 on size alone, while being twice as slow. A candidate that is bigger, slower AND weaker is dominated on every axis a caller cares about. Size decides only where the benchmark is silent, which is the convention the rest of this package follows: an absent score ranks low and is honest, a guessed one ranks wrongly and looks authoritative.

When nothing clears the floor, the fastest candidate that still fits, because “everything here is too slow” is better answered with the least-slow option than with the largest.

Parameters:

plans (list of dict) – plan() dicts, typically from rank().

Returns:

The chosen plan, or None when no candidate fits memory and disk.

Return type:

dict or None

Examples

>>> disk = {"read_gbs": 4.0, "free_gb": 4000.0}
>>> rows = [{"id": "s", "size_b": 8, "layers": 32, "hf_id": "x/s"},
...         {"id": "m", "size_b": 70, "layers": 80, "hf_id": "x/m"},
...         {"id": "xl", "size_b": 671, "layers": 61, "dtype": "fp8",
...          "hf_id": "x/xl"}]
>>> plans = rank(rows, budget_gb=40.0, disk=disk)
>>> pick(plans)["catalog_id"]          # unscored: size is all there is
'm'
>>> # With scores on the table, capability wins over sheer size — here the
>>> # smaller model is better AND faster, so bigger-but-weaker is dominated.
>>> scored = [{**rows[1], "benchmarks": {"general": 84}},
...           {**rows[0], "benchmarks": {"general": 87}}]
>>> pick(rank(scored, budget_gb=40.0, disk=disk))["catalog_id"]
's'
>>> # Nothing clears the floor on a slow drive: take the least-slow.
>>> slow = rank(rows, budget_gb=40.0, disk={"read_gbs": 0.12, "free_gb": 4000.0})
>>> pick(slow)["catalog_id"]
's'
>>> pick([]) is None
True
best_engine_ai_helper.lightweight_plan(entry, *, budget_gb=None, disk=None, prefetching=True, context_k=4.0, batch=1, new_tokens=300)

Everything known about streaming ONE model on this machine.

One call per candidate, so the caller never has to keep the footprint, the disk requirement and the throughput estimate in step by hand.

Parameters:

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

  • budget_gb (float or None) – The memory budget from score.effective_budget(). None skips the fit verdict.

  • disk (dict or None) – Output of detect.disk_profile(). None probes it (table lookup, no measurement).

  • prefetching (bool) – Whether prefetching is on.

  • context_k (float) – Context length in units of 1024 tokens, for the KV term.

  • batch (int) – Sequences generated together: multiplies the KV term and the tokens delivered per pass (see generation_time()).

  • new_tokens (int) – Answer length answer_seconds is reported for. 300 tokens is about a long paragraph — short enough to be a realistic streamed job, long enough that the wait is the real one.

Returns:

model (HuggingFace id or the raw tag), resident_gb, checkpoint_gb, layers/layers_estimated, disk (disk_required_gb()), disk_fits, est_tokens_per_s, seconds_per_token, seconds_per_pass, time_to_first_token_s, answer_seconds / answer_human for new_tokens, effective_tps, fits, usable (fits memory AND disk AND clears COMFORT_TPS_STREAMING), and blockers: the plain-text reasons it would not work here.

Return type:

dict[str, Any]

Examples

>>> p = plan(
...     {"id": "llama3.3:70b", "size_b": 70, "layers": 80,
...      "vllm_id": "meta-llama/Llama-3.3-70B-Instruct"},
...     budget_gb=11.0,
...     disk={"read_gbs": 4.0, "free_gb": 900.0, "basis": "declared"},
... )
>>> p["resident_gb"], p["fits"], p["est_tokens_per_s"]
(6.1, True, 0.024)
>>> p["answer_human"], p["time_to_first_token_s"]
('3.4 h', 41.176)
>>> p["seconds_per_token"], p["usable"], p["blockers"]
(41.7, True, [])
>>> slow = plan({"id": "y", "size_b": 671, "layers": 61, "dtype": "fp8",
...              "hf_id": "deepseek-ai/DeepSeek-V3"}, budget_gb=50.0,
...             disk={"read_gbs": 4.0, "free_gb": 4000.0})
>>> slow["usable"], slow["blockers"]
(False, ['estimated 0.005 tok/s (200.0 s/token) is below the 0.01 tok/s batch floor'])
>>> tight = plan({"id": "x", "size_b": 70, "layers": 80}, budget_gb=2.0,
...              disk={"read_gbs": 4.0, "free_gb": 20.0})
>>> tight["fits"], tight["disk_fits"]
(False, False)
best_engine_ai_helper.lightweight_rank(entries, *, budget_gb, disk=None, prefetching=True, kind=None, batch=1, new_tokens=300)

Rank catalog entries by what streaming unlocks on this machine.

Streaming inverts the usual ranking. With memory no longer the binding constraint, leanness stops being a virtue: what is wanted is the most capable model whose layers still fit and whose rate clears the batch floor, with parameter count deciding only where no benchmark does. The trade being made here is explicitly “slower for better”, so “better” is the sort key — see pick() for why that is not the same as “biggest”.

Parameters:

  • entries (list of dict) – Catalog entries.

  • budget_gb (float) – Memory budget from score.effective_budget().

  • disk (dict or None) – detect.disk_profile() output. None probes it.

  • prefetching (bool) – Whether prefetching is on. When it is and a candidate does not fit, that candidate is re-planned WITHOUT it before being rejected — see the comment in the body.

  • kind ({'llm', 'vlm', None}) – Restrict to one kind. A VLM satisfies an llm request (vision implies text), the same rule the rest of the catalog follows.

  • batch (int) – Sequences generated together, passed to plan().

  • new_tokens (int) – Answer length each plan’s answer_seconds is for.

Returns:

plan() dicts, usable ones first and most capable first within that, each carrying the entry’s size_b and benchmark score for display.

Return type:

list[dict[str, Any]]

Examples

>>> rows = [
...     {"id": "a", "kind": "llm", "size_b": 70, "layers": 80, "hf_id": "x/a"},
...     {"id": "b", "kind": "llm", "size_b": 7, "layers": 32, "hf_id": "x/b"},
... ]
>>> ranked = rank(rows, budget_gb=11.0,
...               disk={"read_gbs": 4.0, "free_gb": 900.0})
>>> [(r["catalog_id"], r["usable"]) for r in ranked]
[('a', True), ('b', True)]
>>> # Tighten the disk and the 70B stops being an option at all.
>>> [(r["catalog_id"], r["usable"]) for r in
...  rank(rows, budget_gb=11.0, disk={"read_gbs": 4.0, "free_gb": 100.0})]
[('b', True), ('a', False)]
best_engine_ai_helper.lightweight_runtime(compute=None, allow_cpu=False)

Which streaming runtime this machine offers, if any.

lightweight streams on CUDA (its primary path), on Apple Silicon through MLX, and on CPU experimentally. ROCm rides the same torch path as CUDA and is treated as capable; the figure that matters there is disk bandwidth anyway, which is vendor-independent.

Parameters:

  • compute (dict or None) – Output of detect.compute_profile(). None probes it.

  • allow_cpu (bool) – Whether to count a CPU-only machine as capable. False by default: the CPU path is slower by the same factor the disk is, but it works, so proposing it automatically would promise something that often does not load. A human can still ask for it.

Returns:

'cuda', 'rocm', 'mps', 'cpu', or None when this machine has no runtime streaming can use.

Return type:

str or None

Examples

>>> runtime({"accelerator": "gpu-cuda"})
'cuda'
>>> runtime({"accelerator": "gpu-metal"})
'mps'
>>> runtime({"accelerator": "cpu"}) is None
True
>>> runtime({"accelerator": "cpu"}, allow_cpu=True)
'cpu'
best_engine_ai_helper.lightweight_seconds_per_pass(entry, read_gbs, prefetching=True)

Seconds for ONE forward pass through a streamed model.

The atom of every timing here, and the reason those timings are not what intuition suggests. A streamed forward pass walks the layers in order, reading each layer, computing it over every token in the pass, then freeing it. So a pass costs what reading the checkpoint once costs — independent of how many tokens rode through it.

Every byte is charged to the drive, and that is correct rather than conservative. The tempting correction is to credit the operating system’s page cache with the layers it still holds, so a machine with free memory pays memory speed for part of the pass. It does not work, and the reason is the access pattern: a pass reads layers 1..N in order and the next pass starts again at layer 1, which the tail of the previous pass evicted. A cyclic scan over a working set larger than the cache is the textbook pathology for least-recently-used eviction, and Linux’s active/inactive approximation does not rescue it either – nothing is touched twice within a pass, so nothing is ever promoted to the active list. The expected hit rate is near zero, not free_memory / checkpoint.

And the regime where it would help is the regime where this backend is the wrong choice: memory that can cache the whole checkpoint is memory that can hold it, where a memory-mapped resident engine wins outright. So there is no mixing term here on purpose, and decide() is what steers a caller with that much memory back to a native backend.

Two consequences worth knowing before choosing this backend:

  • One generated token costs one pass. Decoding is autoregressive, so token N+1 needs its own walk through every layer. That is where the seconds-per-token figure comes from.

  • The prompt is nearly free. A 2000-token prompt is processed in the same single pass as a 10-token one, because its tokens ride through each layer together. No other backend has that property: under Ollama or vLLM a long prompt is real prefill work. It makes streaming comparatively good at large prompt, short answer jobs — classification, extraction, judging, a one-line verdict over a long document — and comparatively terrible at long answers.

Parameters:

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

  • read_gbs (float or None) – Sequential disk read bandwidth in GB/s. None returns None.

  • prefetching (bool) – Whether prefetching is on.

Returns:

Seconds per pass, or None when the drive or the model cannot be sized.

Return type:

float or None

Examples

>>> seconds_per_pass({"size_b": 70}, 4.0)
41.176
>>> seconds_per_pass({"size_b": 70, "dtype": "q4"}, 4.0)
10.294
>>> seconds_per_pass({"size_b": 70}, None) is None
True
best_engine_ai_helper.lightweight_tokens_per_second(entry, read_gbs, prefetching=True)

Estimate streaming decode throughput (tokens/s) from DISK bandwidth.

Under every other backend, decode is memory-bandwidth bound. Under streaming it is disk bound, and by a much harder bound: the weights do not stay anywhere, so each token costs one full forward pass over the checkpoint — see seconds_per_pass(), of which this is the reciprocal.

This is the steady-state decode rate for ONE sequence, which is the pessimistic reading of “how fast is it”. generation_time() gives the other two readings: the time to the first token (one pass, and a long prompt barely changes it) and the effective rate for a batch (B sequences ride through each pass together, so B answers arrive in the time of one).

This is also why the honest unit is often seconds per token — see seconds_per_token(). A 70B model at bf16 on a 4 GB/s NVMe drive is about 0.024 tok/s: forty seconds a token. The same model published at 4 bits reads a quarter of the bytes and so runs about four times faster, but that is a property of the checkpoint you point at rather than a conversion this package performs — and it is still a batch job.

Parameters:

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

  • read_gbs (float or None) – Sequential read bandwidth in GB/s, from detect.disk_profile(). None means “not estimated” and returns None rather than a fabricated rate.

  • prefetching (bool) – Whether reads overlap compute.

Returns:

Tokens per second, rounded to three decimals, or None when bandwidth or model size is unknown.

Return type:

float or None

Examples

>>> estimated_tokens_per_second({"size_b": 70}, 4.0)
0.024
>>> estimated_tokens_per_second({"size_b": 70, "dtype": "q4"}, 4.0)
0.097
>>> estimated_tokens_per_second({"size_b": 7}, 4.0)
0.243
>>> estimated_tokens_per_second({"size_b": 70}, None) is None
True
>>> estimated_tokens_per_second({"id": "unsized"}, 4.0) is None
True
best_engine_ai_helper.list_families()[source]

Enumerate families for discovery: id, name, members, summary.

Returns:

One compact row per family, in catalog order.

Return type:

list[dict[str, Any]]

Examples

>>> [r["id"] for r in list_families()]
['F1', 'F2', 'F3']
best_engine_ai_helper.list_usages()[source]

Enumerate profiles for discovery: name, family, status, summary.

Returns:

One compact row per profile, in catalog order.

Return type:

list[dict[str, Any]]

Examples

>>> rows = list_usages()
>>> {"name", "family", "status", "summary"} <= set(rows[0])
True
best_engine_ai_helper.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.load_config()[source]

Return the persisted selection dict, or an empty dict if absent/unreadable.

Reads ~/.best-engine-ai-helper/config.json (written by pull.write_env()). A missing file is the normal “never ran detection” case, so it maps to {} rather than an error; a corrupt file is treated the same way so a bad write can never break a downstream caller.

Returns:

The parsed config, or {} when the file does not exist or cannot be parsed as a JSON object.

Return type:

dict

best_engine_ai_helper.load_engine(path)[source]

Read an engine descriptor written by write_engine().

Parameters:

path (str | Path)

Return type:

dict[str, Any]

best_engine_ai_helper.load_families(usages_path=None)[source]

Load the bundled families merged with the user overlay.

Parameters:

usages_path (Path or None) – Path to the seed usages.yaml. Defaults to the bundled file.

Returns:

Family dicts, each with at least id, brief and members.

Return type:

list[dict[str, Any]]

Examples

>>> [f["id"] for f in load_families()]
['F1', 'F2', 'F3']
best_engine_ai_helper.load_hardware(hardware_path=None)[source]

Load the bundled hardware chip table merged with the user’s local cache.

Cache entries whose (chip, memory_gb) pair matches a seed entry overwrite it. New entries are appended. The seed file is never modified.

Parameters:

hardware_path (Path or None) – Path to the seed hardware.yaml. Defaults to the bundled file. Pass an explicit path in tests to use a fixture.

Returns:

Merged hardware entries. Each entry has at minimum: chip, vendor, memory_gb, ollama_usable_gb.

Return type:

list[dict[str, Any]]

Examples

>>> entries = load_hardware()
>>> len(entries) > 0
True
>>> all('chip' in e for e in entries)
True
best_engine_ai_helper.load_usages(usages_path=None)[source]

Load the bundled usage profiles merged with the user overlay.

Parameters:

usages_path (Path or None) – Path to the seed usages.yaml. Defaults to the bundled file; pass an explicit path in tests.

Returns:

Profile dicts, each with at least name, brief and family.

Return type:

list[dict[str, Any]]

Examples

>>> names = [p["name"] for p in load_usages()]
>>> "text2sql" in names and "embeddings" in names
True
best_engine_ai_helper.lookup_chip(chip_name, hardware)[source]

Find a hardware entry by a case-insensitive substring match on the chip name.

When multiple entries share the same chip name (for example, an Apple M2 Max at 32 GB and at 96 GB), this returns the first match in the list order. The caller should supply the most specific chip string available to avoid ambiguity.

Parameters:

  • chip_name (str) – Chip name or substring to search for, e.g. 'Apple M2 Max'.

  • hardware (list[dict[str, Any]]) – Hardware entries as returned by load_hardware().

Returns:

The first matching entry, or None if no entry contains chip_name as a case-insensitive substring.

Return type:

dict[str, Any] or None

Examples

>>> hw = load_hardware()
>>> entry = lookup_chip('Apple M2 Max', hw)
>>> entry is not None
True
>>> entry['vendor']
'apple'
best_engine_ai_helper.measure_disk_read_gbs(total_bytes=268435456, directory=None)[source]

Measure sequential read bandwidth (GB/s), defeating the page cache.

Writes a scratch file, flushes it to the device, then reads it back with caching disabled for that descriptor (F_NOCACHE on macOS, O_DIRECT on Linux). Without that step the read is served from RAM and reports the memory bus, not the drive — a number three times too high, which would make layer streaming look comfortable when it is not.

Returns None rather than a cached-read figure on any platform that offers no way to bypass the cache (Windows), so an unmeasurable machine falls back to the declared class figure instead of to a flattering one.

Parameters:

  • total_bytes (int) – Bytes to write and read back. Rounded down to a multiple of 4 MiB, the alignment O_DIRECT requires. Defaults to 256 MiB.

  • directory (str or None) – Where to put the scratch file. None uses the system temp directory. Point it at the volume holding the model cache when that differs.

Returns:

Measured GB/s (rounded to two decimals), or None when the measurement is unavailable or the platform cannot bypass its cache.

Return type:

float or None

Examples

>>> v = measure_disk_read_gbs(total_bytes=4 * 1024 * 1024)
>>> v is None or v > 0
True
best_engine_ai_helper.model_footprint_gb(entry, backend='ollama')[source]

Estimate a model’s peak inference memory (GB) on a given serving backend.

Three backends, three different physical situations, and sizing one against another’s figure is how a picker comes to recommend a model that will not load. The catalog’s ram_gb is an Ollama Q4 estimate. vLLM instead loads the full FP16/BF16 HuggingFace weights (~2 bytes/param) plus KV cache and runtime buffers, so the same model is markedly heavier there. Sizing a vLLM pick against ram_gb would over-promise and pick a model that will not actually fit — the “unrealistic recommendation” this guards against.

Parameters:

  • entry (dict[str, Any]) – Catalog entry; uses ram_gb (Ollama) or size_b (vLLM).

  • backend ({'ollama', 'vllm', 'lightweight'}) – Serving backend the footprint is estimated for. Unknown values are treated as 'ollama' (the conservative catalog figure). 'lightweight' streams layers off disk and so reports the resident size of ONE layer rather than of the model — see offload.footprint_gb().

Returns:

Estimated peak memory in GB.

Return type:

float

Examples

>>> model_footprint_gb({'id': 'a', 'ram_gb': 9.9, 'size_b': 14})
9.9
>>> model_footprint_gb({'id': 'a', 'ram_gb': 9.9, 'size_b': 14}, 'vllm')
32.2
>>> # Streaming sizes one layer, not the model: a 70B becomes laptop-sized.
>>> model_footprint_gb({'id': 'b', 'size_b': 70, 'layers': 80}, 'lightweight')
6.1
best_engine_ai_helper.model_for(engine, kind)[source]

Return (backend, base_url, model) for kind from an engine descriptor.

Raises KeyError if the descriptor has no entry for kind (e.g. asking for a VLM from an engine resolved for an llm-only brief).

Examples

>>> engine = {"backend": "ollama", "base_url": "http://localhost:11434",
...           "llm": {"model": "qwen3:14b"}}
>>> model_for(engine, "llm")
('ollama', 'http://localhost:11434', 'qwen3:14b')
>>> # A kind that fell back to streaming carries its own backend.
>>> engine["vlm"] = {"model": "Qwen/Qwen2.5-VL-72B-Instruct", "backend": "lightweight"}
>>> model_for(engine, "vlm")
('lightweight', '', 'Qwen/Qwen2.5-VL-72B-Instruct')
Parameters:
Return type:

tuple[str, str, str]

best_engine_ai_helper.parse_task(task)[source]

Turn a vague task phrase into the model kinds and benchmark axis it implies.

Returns a dict with kinds (subset of ["llm", "vlm"] in pull order), application (the benchmark axis for the text model), matched (the keywords that fired, for the report’s justification), and language (best-effort ISO 639-1 code from _detect_language(), or None). A task that mentions nothing visual still gets an LLM on the generalist axis; any vision keyword adds a VLM.

Parameters:

task (str or None) – Free-text task description. None, a blank/whitespace-only string, or text with no detectable language (symbols/digits only) falls back to a generic text-assistant profile, but logs a loud warning first: a recommendation with no clean task description carries no useful label for activity/cost monitoring or for the report’s justification, so a caller skipping it should see that reflected back.

Return type:

dict[str, Any]

Examples

>>> parse_task("write product descriptions and check photo quality")["kinds"]
['llm', 'vlm']
>>> parse_task(None)["application"]  # logs a WARNING, still resolves
'generalist'
best_engine_ai_helper.platform_name()[source]

Return the current OS as a short lowercase string.

Returns:

One of: ‘darwin’, ‘linux’, ‘windows’. Delegates to os_helper.platform_name().

Return type:

str

Examples

>>> platform_name() in ('darwin', 'linux', 'windows')
True
best_engine_ai_helper.rank(hw, catalog, kind, headroom=0.5, application=None, backend='ollama', load=None, bandwidth_gbs=None, min_tps=None, installed=None, require_structured=True, disk_read_gbs=None)[source]

Return all candidates sorted by benchmark score, annotated with fit status.

Each entry in the result gets a _fits key (bool) indicating whether it fits within the effective budget. The top entry is identical to what select() would return.

Parameters:

  • hw (dict[str, float | None]) – Output of detect.available_memory().

  • catalog (list[dict[str, Any]]) – Merged model entries from catalog.load_catalog().

  • kind ({'llm', 'vlm'}) – The inference kind to filter and rank by.

  • headroom (float) – Extra safety on top of the accelerator cap. Clamped to MAX_HEADROOM (0.5); default 0.5.

  • application (str or None) – Optional use-case keyword ("code", "math", "ocr", "vision", "chat", "generalist"). Drives which benchmark column is used for ranking.

  • backend ({'ollama', 'vllm'}) – Serving backend, so _fits reflects the footprint that actually loads (FP16 for vLLM, Q4 ram_gb for Ollama).

  • load (dict or None) – Live server state from detect.server_load(), forwarded to effective_budget(). None reproduces the load-blind behaviour.

  • bandwidth_gbs (float or None) – Memory bandwidth from detect.compute_profile(), used to estimate throughput for candidates with no measurement. None leaves the throughput annotations unpopulated unless a measurement exists.

  • disk_read_gbs (float or None) – Sequential disk read bandwidth from detect.disk_profile(), used only under backend='lightweight', where throughput is bounded by the drive rather than the memory bus.

  • min_tps (float or None) – Throughput floor in tokens/s. A candidate measured below it on this machine is demoted out of auto-selection, in the same way and for the same reason as one that does not fit memory. A candidate merely estimated below it is annotated and left in place: the estimator was measured here to run optimistic by a wide margin, so letting it disqualify anything would reject models that run perfectly well. None disables the check entirely.

  • require_structured (bool) – Whether schema-constrained output is a requirement of the job. True (the default) demotes a model that cannot honour Ollama structured output below one that can, however well it scores. False drops that key entirely, for a caller whose brief says it does not need schemas.

  • installed (set of str or None) – Model tags already present on the machine. Used only to break ties between candidates of comparable quality (within QUALITY_TIE_BAND), never to promote a worse model over a better one: at equal quality a model already on disk beats a 20 GB download, and at unequal quality the better model still wins. Passed in rather than probed here so that ranking stays a pure function and no network call hides inside a sort.

Returns:

Candidates sorted descending by benchmark score, each annotated with _fits, and with _tps / _tps_basis / _comfy when a throughput figure could be obtained.

Return type:

list[dict[str, Any]]

Examples

>>> catalog = [{'id': 'v', 'kind': 'vlm', 'ram_gb': 9.0,
...             'benchmarks': {'vision': 80}}]
>>> rank({'unified_gb': 96.0, 'vram_gb': None, 'ram_gb': 96.0}, catalog, 'vlm')[0]['_fits']
True
best_engine_ai_helper.recommend_engines(hw, catalog, task=None, *, headroom=0.5, compute=None, min_tps=15.0, backend='ollama', kinds=None, load=None, installed=None, require_structured=True, disk=None)

Recommend the best engine per needed kind for this hardware and task.

Parameters:

  • hw (dict) – Memory description from detect.available_memory().

  • catalog (list of dict) – Benchmark catalog from catalog.load_catalog().

  • task (str or None) – Free-text or keyword task. None means a generalist text assistant.

  • headroom (float) – Memory safety margin passed to score.effective_budget() (clamped to score.MAX_HEADROOM).

  • compute (dict or None) – Compute profile from detect.compute_profile() (accelerator + bandwidth). None disables the throughput estimate.

  • min_tps (float) – Comfort throughput floor; a fitting model below it is only picked when no comfortable one exists (and the choice is warned about).

  • backend ({'ollama', 'vllm', 'lightweight'}) – Serving backend, so memory fit and throughput reflect what actually loads. Threaded to score.rank() and _candidate_row(). 'lightweight' ranks layer-streaming candidates: ram_gb becomes one resident layer and the rate becomes disk-bound, which is why disk matters there and nowhere else.

  • disk (dict or None) – Storage profile from detect.disk_profile(). Read only under backend='lightweight', where it bounds throughput; also echoed into the report so a streaming recommendation states the drive it assumed.

  • require_structured (bool) – Whether the job needs schema-constrained output. True (the default) keeps a model that cannot honour it out of auto-selection; False lets such a model win on score alone, which is what a brief saying structured_output: false is asking for.

  • installed (set of str or None) – Tags the local daemon already serves. Breaks ties between candidates of comparable quality and comparable footprint, so a model already on disk is preferred over an equivalent 20 GB download. Never overrides quality or leanness.

  • kinds (list of str or None) – Explicit kinds to resolve (["llm"], ["vlm"] or both), overriding the kinds inferred from task. The task text still selects the benchmark axis. Used when the caller already knows what it needs (e.g. a brief that declares kind: both).

  • load (dict or None) – Live server state from detect.server_load() (current free RAM, CPU/GPU/disk usage, already-running engines). Forwarded to score.effective_budget()/score.rank() so the recommendation reflects what else is happening on this machine right now, not only its theoretical capacity. None (the default) reproduces the load-blind behaviour exactly; also included as server_load in the returned report when given, for activity monitoring.

Returns:

A JSON-ready report: the parsed task, hardware, memory budget, and for each needed kind the chosen model plus the full ranked candidate table, with per-model fit and estimated throughput.

Return type:

dict

best_engine_ai_helper.resolve(brief, *, backend='auto', endpoint=None, catalog=None, hw=None, compute=None, cloud_catalog=None, disk=None)[source]

Resolve a usage brief into a concrete engine descriptor.

The brief’s mode selects local vs cloud (default local):

  • local (default) -> a hardware-specific descriptor: the backend chosen for this machine (Ollama/vLLM) plus the model per kind.

  • cloud -> a provider descriptor (provider, model, optional base_url/api_key_env) plus a local fallback resolved from the SAME brief, so a failed paid call degrades to the always-available local model (paid -> local, the safe direction). model is OPTIONAL: when omitted, the best paid model is auto-picked from the bundled pricing.yaml catalog on a quality-vs-price trade-off (see _resolve_cloud() and best_engine_ai_helper.cloud_catalog).

In local mode there is one escalation: when the conventional backend resolves nothing that FITS for a kind the brief needs, resolution consults offload.decide() and, if this machine can stream, rescues that kind alone with a layer-streamed model (backend: lightweight inside the kind’s own section — see model_for()). The descriptor records the decision and its reason under lightweight, including when the answer was no. Set lightweight: never in the brief, or BEST_ENGINE_LIGHTWEIGHT=never, to switch the escalation off.

Parameters:

  • brief (str | Path | dict) – The input brief (path to llm.brief.yaml or an already-loaded dict). Keys: mode (local/cloud, default local), kind (llm/vlm/both), headroom, min_tps, structured_output, task (free text); cloud-only: provider, model (both optional -> auto-pick), vlm_model, base_url, api_key_env, quality_vs_cost (0..1, default cloud_catalog.DEFAULT_QUALITY_VS_COST, only used when auto-picking); streaming-only: lightweight (auto/always/never, default auto) and lightweight (auto/always/never).

  • backend ({'auto', 'ollama', 'vllm', 'lightweight'}) – auto picks per default_backend() — which never returns lightweight: layer streaming is reached automatically only as a rescue (see below), and otherwise by naming it here. Local mode only.

  • endpoint (str or None) – Override the server base URL (defaults to the local endpoint for the backend, or the provider’s default API root for cloud mode).

  • catalog (optional) – Injectable for tests; default to the live local catalog and detected hardware.

  • hw (optional) – Injectable for tests; default to the live local catalog and detected hardware.

  • compute (optional) – Injectable for tests; default to the live local catalog and detected hardware.

  • cloud_catalog (optional) – Injectable for tests; defaults to the live pricing.yaml catalog. Cloud mode only, and only consulted when the brief omits model.

  • disk (optional) – Injectable storage profile (detect.disk_profile()). Read only on the layer-streaming path, where the drive bounds throughput.

Returns:

The engine descriptor (see write_engine() for the on-disk shape).

Return type:

dict

best_engine_ai_helper.resolve_family(family_id, *, backend='auto', endpoint=None, catalog=None, hw=None, compute=None)[source]

Resolve a whole family into one machine-specific engine descriptor.

Resolving a family yields a single model for the group (the shared pick), whereas resolve_usage() yields the possibly-specialised model for one profile. The result is machine-specific — persist it gitignored, never commit.

Parameters:
Returns:

The engine descriptor, annotated with the family’s metadata.

Return type:

dict[str, Any]

Examples

>>> hw = {"unified_gb": 96.0, "vram_gb": None, "ram_gb": 96.0}
>>> cat = [{"id": "emb", "kind": "embed", "ram_gb": 1.2,
...         "benchmarks": {"mteb": 66}}]
>>> eng = resolve_family("F3", catalog=cat, hw=hw,
...                      compute={"chip": "M2", "accelerator": "apple"})
>>> eng["embed"]["model"], eng["family"]
('emb', 'F3')
best_engine_ai_helper.resolve_usage(name, *, backend='auto', endpoint=None, catalog=None, hw=None, compute=None)[source]

Resolve a usage profile into a machine-specific engine descriptor.

This is the “give me the model for profile name” entry point. It reads only the profile’s needs, then lets best-engine choose the concrete model for this machine. The returned descriptor is machine-specific: persist it to a gitignored file (see engine.write_engine()), never commit it.

Parameters:

  • name (str) – The profile name ("text2sql", "rag-answer", …).

  • backend ({'auto', 'ollama', 'vllm'}) – Serving backend; auto picks per engine.default_backend().

  • endpoint (str or None) – Override the server base URL.

  • catalog (optional) – Injectable for tests; default to the live catalog and detected hardware.

  • hw (optional) – Injectable for tests; default to the live catalog and detected hardware.

  • compute (optional) – Injectable for tests; default to the live catalog and detected hardware.

Returns:

The engine descriptor, annotated with the profile’s metadata.

Return type:

dict[str, Any]

Examples

>>> hw = {"unified_gb": 96.0, "vram_gb": None, "ram_gb": 96.0}
>>> compute = {"accelerator": "apple", "chip": "M2", "bandwidth_gbs": None}
>>> cat = [{"id": "coder", "kind": "llm", "size_b": 7, "ram_gb": 5.0,
...         "benchmarks": {"general": 66, "code": 85},
...         "structured_output": True, "vllm_id": "org/Coder"}]
>>> eng = resolve_usage("text2sql", backend="ollama", catalog=cat,
...                     hw=hw, compute=compute)
>>> eng["llm"]["model"], eng["usage"], eng["status"]
('coder', 'text2sql', 'stable')
best_engine_ai_helper.resolved_models()[source]

Return both resolved tags in one call, reading the config file at most once.

Returns:

{"text": <tag>, "vision": <tag>} — the same values text_model() and vision_model() would return.

Return type:

dict

best_engine_ai_helper.seconds_per_image(entry, bandwidth_gbs, *, steps=None, width=None, height=None)[source]

Estimate how long one image takes on this machine.

steps x denoiser passes, each pass proportional to the denoiser’s size and to the output area. The constant behind it (_STEP_SECONDS_PER_B_PER_MP) is anchored on one measured point and scaled by memory bandwidth, which is the only compute proxy this package has — so treat the result as “seconds or minutes?”, never as a benchmark. It is still the figure that decides usability: a model needing four minutes an image is a different tool from one needing four seconds.

Parameters:

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

  • bandwidth_gbs (float or None) – Memory bandwidth from detect.compute_profile(). None returns None: no estimate rather than a fabricated one.

  • steps (int or None) – Override the catalog’s step count.

  • width (int or None) – Output size; None uses the native one.

  • height (int or None) – Output size; None uses the native one.

Returns:

Seconds per image, rounded to one decimal, or None when bandwidth or model size is unknown.

Return type:

float or None

Examples

>>> sdxl = {"id": "sdxl", "size_b": 3.5, "denoiser_size_b": 2.6,
...         "steps": 28, "native_size": "1024x1024"}
>>> seconds_per_image(sdxl, 1000.0)
10.3
>>> # The same model distilled to 4 steps is the same arithmetic, 7x less of it.
>>> seconds_per_image({**sdxl, "steps": 4}, 1000.0)
1.5
>>> # Half the bandwidth, twice the wait.
>>> seconds_per_image(sdxl, 500.0)
20.6
>>> seconds_per_image(sdxl, None) is None
True
>>> seconds_per_image({"id": "unsized"}, 1000.0) is None
True
best_engine_ai_helper.select(hw, catalog, kind, headroom=0.5, application=None, backend='ollama', load=None, require_structured=True)[source]

Pick the best-scoring model that fits in available memory.

Candidates for a ‘vlm’ selection include both VLMs and LLMs with vision capability (kind == ‘vlm’). Candidates for a ‘llm’ selection include text-only LLMs and VLMs (since a VLM handles text-only prompts equally).

Selection order:

  1. Filter: keep entries whose ram_gb fits within the effective budget.

  2. Rank: structured-output capability first (a model that cannot honour Ollama structured output is never chosen over one that can), then benchmark score (application-specific if given, else vision for VLM or general for LLM). This matches rank(), so rank(...)[0] and select(...) agree.

  3. Last resort: if nothing fits, return the smallest model in the catalog rather than raising; the caller decides whether to warn the user.

Parameters:

  • hw (dict[str, float | None]) – Output of detect.available_memory().

  • catalog (list[dict[str, Any]]) – Merged model entries from catalog.load_catalog().

  • kind ({'llm', 'vlm'}) – The type of model to select.

  • headroom (float) – Extra safety on top of the accelerator cap. Clamped to MAX_HEADROOM (0.5); default 0.5.

  • application (str or None) – Optional use-case keyword ("code", "math", "ocr", "vision", "chat", "generalist"). Selects the benchmark axis used for scoring. None uses the default kind-based rule.

  • backend ({'ollama', 'vllm'}) – Serving backend, so memory fit is checked against the footprint that actually loads (FP16 for vLLM, Q4 ram_gb for Ollama).

  • load (dict or None) – Live server state from detect.server_load(), forwarded to effective_budget(). None reproduces the load-blind behaviour.

  • require_structured (bool)

Returns:

The selected catalog entry.

Return type:

dict[str, Any]

Raises:

ValueError – If the catalog is empty.

Examples

>>> hw = {'unified_gb': 96.0, 'vram_gb': None, 'ram_gb': 96.0}
>>> catalog = [{'id': 'v', 'kind': 'vlm', 'ram_gb': 9.0,
...             'benchmarks': {'vision': 80}}]
>>> select(hw, catalog, kind='vlm')['id']
'v'
best_engine_ai_helper.storage_class()[source]

Classify the drive that holds the model cache.

Layer streaming reads the weights off this drive once per token, so its class — not the memory bus — is what bounds throughput. Classification is deliberately coarse (four classes, see _DISK_READ_GBS): the goal is the right order of magnitude for an estimate, not a device database.

Returns:

One of 'apple-internal', 'nvme', 'ssd-sata', 'hdd' or 'unknown'. 'unknown' is returned rather than a guess when the platform exposes no rotational/transport hint, so callers can say “not estimated” instead of inventing a speed.

Return type:

str

Examples

>>> storage_class() in ('apple-internal', 'nvme', 'ssd-sata', 'hdd', 'unknown')
True
best_engine_ai_helper.streaming_budget_gb(hw=None, load=None)

Memory budget (GB) a streamed model may occupy on this machine.

Deliberately the accelerator’s whole usable pool (score.usable_pool_gb()) rather than the headroom-reduced score.effective_budget() every other backend sizes against. Two reasons, and they are the same reason twice: what the headroom reserves is already accounted for here. It exists to leave room for KV-cache growth — which footprint_gb() now sizes explicitly, per layer, at a named context length — and for the OS and the caller’s own workload, which the Metal cap (66-75%) and the VRAM fraction (92%) inside usable_pool_gb already hold back. Applying it a second time reserves the same gigabytes twice and rejects exactly the models this backend exists to run: a 70B needs about 4 GB resident, which clears a 4 GB card’s real budget and fails a doubly-discounted one.

Parameters:

  • hw (dict or None) – Output of detect.available_memory(). None probes it.

  • load (dict or None) – Live load from detect.server_load(). When given, the budget is additionally capped at the RAM actually free right now — a shard still has to fit beside whatever else is running.

Returns:

Budget in GB.

Return type:

float

Examples

>>> budget_gb({'unified_gb': 8.0, 'vram_gb': None, 'ram_gb': 8.0})
5.28
>>> budget_gb({'unified_gb': None, 'vram_gb': 4.0, 'ram_gb': 32.0})
3.68
>>> budget_gb({'unified_gb': 96.0, 'vram_gb': None, 'ram_gb': 96.0},
...           load={'available_ram_gb': 12.0})
12.0
best_engine_ai_helper.text_model()[source]

Return the model tag to use for text-only prompts (lint, summaries, …).

Precedence: BEST_LLM_TEXT env (or legacy SPREZZATURE_LLM_TEXT) -> persisted config.json -> DEFAULT_TEXT_MODEL. Never probes hardware, never raises.

Return type:

str

best_engine_ai_helper.to_markdown(report)[source]

Render a recommend() report as a Markdown document.

Parameters:

report (dict[str, Any])

Return type:

str

best_engine_ai_helper.usable_pool_gb(hw)[source]

The accelerator’s usable memory pool (GB), before any safety headroom.

The first half of effective_budget(), split out because one backend needs it on its own. On Apple Silicon the usable pool is NOT the whole unified memory — Metal caps GPU allocations at about 66% of a pool up to 36 GB and 75% above it, beyond which inference spills to CPU. A discrete GPU’s VRAM is nearly all usable; a CPU-only machine is held to half its RAM.

Every caller sizing a model that STAYS in memory wants effective_budget() instead: the extra headroom is what leaves room for the OS, the caller’s own workload and KV-cache growth. The exception is layer streaming, where what occupies memory is one transient layer that the runtime frees per layer, and where the KV cache is sized explicitly rather than absorbed into a margin (see offload.footprint_gb()) — applying the headroom there would reserve the same memory twice and reject the models the backend exists to run.

Parameters:

hw (dict[str, float | None]) – Output of detect.available_memory().

Returns:

Usable pool in GB.

Return type:

float

Examples

>>> usable_pool_gb({'unified_gb': 96.0, 'vram_gb': None, 'ram_gb': 96.0})
72.0
>>> usable_pool_gb({'unified_gb': 8.0, 'vram_gb': None, 'ram_gb': 8.0})
5.28
>>> usable_pool_gb({'unified_gb': None, 'vram_gb': 24.0, 'ram_gb': 64.0})
22.08
>>> usable_pool_gb({'unified_gb': None, 'vram_gb': None, 'ram_gb': 32.0})
16.0
best_engine_ai_helper.usage_brief(name)[source]

Return the resolvable brief for a profile (ready for engine.resolve()).

Parameters:

name (str) – The profile name.

Returns:

The brief block with mode: local.

Return type:

dict[str, Any]

Examples

>>> usage_brief("classification")["kind"]
'llm'
best_engine_ai_helper.vision_model()[source]

Return the model tag to use for prompts that include images (alt-text, OCR).

Precedence: BEST_LLM_VISION env (or legacy SPREZZATURE_LLM_VISION) -> persisted config.json -> DEFAULT_VISION_MODEL. Never probes hardware, never raises.

Return type:

str

best_engine_ai_helper.write_engine(engine, path)[source]

Write an engine descriptor to path as YAML with a do-not-commit header.

The file is machine-specific (it encodes the chosen backend and models for this hardware), so it belongs in .gitignore, not in version control.

Parameters:
Return type:

Path

best_engine_ai_helper.write_report(report, stem)[source]

Write both a JSON and a Markdown rendering; return (md_path, json_path).

Parameters:
Return type:

tuple[Path, Path]