best_engine_ai_helper.offload module

offload — run a model that does NOT fit in memory, by streaming its layers.

Every other backend in this package answers the same question: which model fits? This one answers the question that follows when the honest answer is “none of the good ones”: a transformer never needs all of its layers resident at once. Decoding reads layer 1, then layer 2, then layer 3; only the layer being computed has to be in memory. Keep one layer resident, stream the rest off the disk, and the memory a model needs stops scaling with its parameter count and starts scaling with its widest single layer — a 70B model becomes a ~5 GB resident footprint instead of a 141 GB one.

The idea is not new: airllm (https://github.com/lyogavin/airllm, Apache-2.0) demonstrated it, and the arithmetic below was calibrated against its published figures. The runtime here is this package’s own (best_engine_ai_helper.streaming), for three reasons that were measured rather than argued:

1. airllm did not work on Apple Silicon at all. Its AutoModel dispatches on architecture everywhere except macOS, where it hard-returns a hand-written Llama transformer in MLX with bias=False throughout. Qwen carries attention biases, Llama 3.2 ties its embeddings so there is no lm_head tensor, and older exports persist rotary_emb.inv_freq — three different crashes on the platform this suite is developed on.

2. It wrote a second copy of every checkpoint to disk as per-layer shards before running, so a 70B needed ~294 GB free rather than ~140. That, not memory, is what made nearly every row of lightweight plan read runs: NO on a normal laptop.

3. It pulled MLX, bitsandbytes and a per-family subclass for each architecture. All of that left with it.

What this module owns, and what it does not:

• The arithmetic is here. Which model a machine can stream, what it costs in resident memory, how many seconds per token the disk imposes, and whether streaming is worth activating at all are selection questions, which is this package’s entire job. See footprint_gb() and estimated_tokens_per_second().

• The forward pass is not. best_engine_ai_helper.streaming builds the model on the meta device and hangs a pre-hook on each decoder layer that materialises its weights out of the checkpoint’s own memory-mapped safetensors, plus a post-hook that frees exactly those again. transformers owns the pass itself, which is why Qwen, Gemma and GPT-NeoX work with no per-family code here.

Activation

Streaming is never the default: it trades two to three orders of magnitude of speed for capability, and a machine that can comfortably run a 14B model should run it. decide() is the policy, and it fires on exactly two triggers:

• Automatic — the conventional backend found nothing that fits this machine’s memory budget for a kind the caller needs. Streaming is then the difference between a model and no model, so it is proposed with its cost stated in seconds per token.

• Manual — a human asks for it, by --backend lightweight, by lightweight: always in the brief, or by BEST_ENGINE_LIGHTWEIGHT=always in the environment. A human who wants a 235B model on a laptop overnight is making a legitimate choice that no automatic rule should have to guess.

BEST_ENGINE_LIGHTWEIGHT=never switches the whole path off, including the automatic trigger.

Author

Warith Harchaoui <warithmetics@deraison.ai>

best_engine_ai_helper.offload.COMFORT_TPS_STREAMING: float = 0.01

Floor below which streaming stops being worth proposing even as a batch job. 0.01 tok/s is one token every hundred seconds: a paragraph in three hours, an overnight answer, which IS the trade lightweight offers and plenty of people take. Under it, a single sentence costs most of a day and a caller is better served by a smaller model or a cloud call — so the floor reports a blocker rather than offering a technically-correct eternity.

Deliberately NOT score.COMFORT_TPS (15 tok/s), and deliberately not a tenth of it either. Streaming never clears an interactive floor; judging it by one rejects every candidate and hides the capability behind a rule written for a different backend. A floor an order of magnitude too strict is the same mistake in miniature: at 0.2 tok/s this gate admitted nothing above an 8B checkpoint, which is to say nothing worth streaming, since an 8B model runs natively at a hundred times the speed.

best_engine_ai_helper.offload.ENV_VAR = 'BEST_ENGINE_LIGHTWEIGHT'

Environment variable that forces the policy decide() applies. auto (the default) activates streaming only as a last resort; always activates it whenever the runtime can, never disables it outright.

best_engine_ai_helper.offload.INSTALL_HINT = "pip install --upgrade 'best-engine-ai-helper[lightweight]'"

What to tell a caller whose interpreter cannot import the runtime. These live behind the lightweight extra, so reaching this usually means the feature was never opted into — and sometimes means a stripped environment or a failed wheel build. One command covers both.

best_engine_ai_helper.offload.PACKAGE = 'torch/transformers/accelerate/safetensors'

What the runtime is built out of. Not a third-party streaming package any more: best_engine_ai_helper.streaming is this project’s own, and these are the libraries it stands on — all of them already core dependencies for other reasons.

best_engine_ai_helper.offload.STREAMING_CAVEATS: tuple[str, ...] = ('the checkpoint has to publish safetensors; a repository that ships only a pickled pytorch_model.bin is refused, because reading one layer out of a pickle means reading the whole file once per layer', 'a sparse mixture-of-experts layer is materialised whole, so its resident figure is the full expert set rather than the few a token routes to')

What this runtime does not do, stated so a plan is not read as a promise. These are properties of best_engine_ai_helper.streaming, not of a platform: the per-platform caveats that used to live here were artefacts of routing Apple Silicon through MLX, and went away with the dependency.

best_engine_ai_helper.offload.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.offload.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.offload.bytes_per_param(entry)

Bytes each parameter occupies on disk for a streaming load.

Parameters:

entry (dict[str, Any]) – Catalog entry; reads dtype (default bf16).

Returns:

Bytes per parameter.

Return type:

float

Examples

>>> bytes_per_param({})
2.0
>>> bytes_per_param({"dtype": "fp8"})
1.0
>>> bytes_per_param({"dtype": "q4"})
0.5
>>> bytes_per_param({"dtype": "nonsense"})
2.0

best_engine_ai_helper.offload.caveats(runtime_name)

Known limitations of streaming, as sentences for a human.

Separate from blockers: a blocker says streaming cannot start, a caveat says it starts and then depends on the model. Both belong in the engine file, because the consumer reading it is the one that finds out otherwise.

Parameters:

runtime_name (str or None) – Output of runtime(). None means nothing will run, so nothing is worth warning about.

Returns:

Empty when there is no runtime at all.

Return type:

tuple[str, …]

Examples

>>> caveats("cuda") == caveats("mps")      # not a per-platform list
True
>>> len(caveats("mps"))
2
>>> caveats(None)
()

best_engine_ai_helper.offload.checkpoint_gb(entry)

Size of the whole checkpoint a streaming backend reads per token (GB).

Parameters:

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

Returns:

Checkpoint size in GB, or inf when the entry carries no parameter count. Infinity, not zero: an unsized entry must never read as “free” and win every comparison, the defect this package hunts by name.

Return type:

float

Examples

>>> checkpoint_gb({"size_b": 70})
140.0
>>> checkpoint_gb({"size_b": 70, "dtype": "q4"})
35.0
>>> checkpoint_gb({"id": "unsized"})
inf

best_engine_ai_helper.offload.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.offload.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.offload.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.offload.estimated_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.offload.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.offload.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.offload.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.offload.hf_id(entry)

The HuggingFace repository a streaming backend would load.

lightweight reads safetensors from the Hub, not GGUF from Ollama’s registry, so a catalog entry is only streamable when it names a Hub repo. hf_id is preferred; vllm_id is the same thing under the name this catalog has always used for it, and is accepted so no entry has to be edited twice.

Parameters:

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

Returns:

The repository id, or None when the entry names none.

Return type:

str or None

Examples

>>> hf_id({"id": "llama3:70b", "vllm_id": "meta-llama/Llama-3.3-70B-Instruct"})
'meta-llama/Llama-3.3-70B-Instruct'
>>> hf_id({"id": "llama3:70b", "hf_id": "a/b", "vllm_id": "c/d"})
'a/b'
>>> hf_id({"id": "gemma3:12b"}) is None
True

best_engine_ai_helper.offload.human_duration(seconds)

Render a duration the way a person would say it.

Streaming timings span six orders of magnitude — a second, a minute, an afternoon, a week — and “20361.6 s” communicates none of them.

Parameters:

seconds (float or None) – Duration in seconds.

Returns:

A short human reading, or '?' when the duration is unknown.

Return type:

str

Examples

>>> human_duration(0.4), human_duration(42)
('0.4 s', '42 s')
>>> human_duration(185), human_duration(7200)
('3 min', '2.0 h')
>>> human_duration(200000), human_duration(None)
('2.3 days', '?')

best_engine_ai_helper.offload.kv_cache_gb(layers, context_k=4.0)

KV-cache size (GB) for a streamed decode at a given context length.

Parameters:

• layers (int or None) – Decoder depth. None yields 0.0 — an unknown depth is reported as an unsized footprint upstream, so adding a guessed cache here would only obscure that.

• context_k (float) – Context length in units of 1024 tokens.

Returns:

GB of KV cache.

Return type:

float

Examples

>>> kv_cache_gb(80)              # a 70B at 4k context
1.25
>>> kv_cache_gb(32)
0.5
>>> kv_cache_gb(80, context_k=32)
10.0
>>> kv_cache_gb(None)
0.0

best_engine_ai_helper.offload.layer_count(entry)

Decoder depth of a catalog entry, and whether it had to be estimated.

Depth is the divisor of the whole footprint model, so where it comes from matters as much as its value. A declared layers is used as given; a dense model without one is interpolated from _DEPTH_BY_SIZE_B; a mixture-of-experts model without one is refused, because MoE breaks the depth-versus-size relation outright (DeepSeek-V3 is 671B across 61 layers, where the dense trend would predict well over 120) and an interpolated guess there would be wrong by a factor of two in the direction that over-promises.

Parameters:

entry (dict[str, Any]) – Catalog entry; reads layers, size_b and moe.

Returns:

(layers, estimated). layers is None when depth is unknowable from the entry; estimated is True when the figure was interpolated rather than declared.

Return type:

tuple[int | None, bool]

Examples

>>> layer_count({"layers": 80})
(80, False)
>>> layer_count({"size_b": 70})
(80, True)
>>> layer_count({"size_b": 20})          # between the 14B and 27B points
(56, True)
>>> layer_count({"size_b": 671, "moe": True})
(None, False)
>>> layer_count({"id": "mystery"})
(None, False)

best_engine_ai_helper.offload.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.offload.normalize_request(value)

Normalise a human’s streaming request into auto/always/never.

Accepts what people actually write in a brief or an environment variable — yes, 1, off, a real bool, None — and refuses to guess at anything else.

Parameters:

value (object) – Raw value from a brief key, a CLI flag, or ENV_VAR. None means “did not say” and yields auto.

Returns:

One of 'auto', 'always', 'never'.

Return type:

str

Raises:

ValueError – When the value is neither empty nor a recognised alias. Silently reading an unknown word as auto would turn a typo (lightweight: alwyas) into a quietly disabled feature.

Examples

>>> normalize_request(None)
'auto'
>>> normalize_request(True), normalize_request(False)
('always', 'never')
>>> normalize_request(' Always ')
'always'
>>> normalize_request("maybe")
Traceback (most recent call last):
ValueError: lightweight request must be one of auto/always/never, got 'maybe'

best_engine_ai_helper.offload.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.offload.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.offload.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.offload.requested(brief_value=None, env=None)

Resolve the effective streaming request, brief first, environment second.

Precedence mirrors every other setting in this package: an explicit value in the brief (or on the command line) wins, the environment is the fallback, and auto is the default.

Parameters:

• brief_value (object) – The brief’s lightweight: key, or a CLI flag’s value. None defers to the environment.

• env (dict[str, str] or None) – Environment mapping to read ENV_VAR from. None uses os.environ; tests pass a dict.

Returns:

One of 'auto', 'always', 'never'.

Return type:

str

Examples

>>> requested(None, env={})
'auto'
>>> requested(None, env={'BEST_ENGINE_LIGHTWEIGHT': 'never'})
'never'
>>> requested('always', env={'BEST_ENGINE_LIGHTWEIGHT': 'never'})
'always'

best_engine_ai_helper.offload.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.offload.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.offload.seconds_per_token(tps)

Invert a streaming rate into the unit humans can picture.

“0.024 tokens per second” reads as a small number; “41 seconds per token” reads as what it is. Every user-facing surface shows both.

Parameters:

tps (float or None) – Tokens per second, typically from estimated_tokens_per_second().

Returns:

Seconds per token rounded to one decimal, or None when the rate is unknown or non-positive.

Return type:

float or None

Examples

>>> seconds_per_token(0.024)
41.7
>>> seconds_per_token(2.0)
0.5
>>> seconds_per_token(None) is None
True
>>> seconds_per_token(0.0) is None
True

best_engine_ai_helper.offload.stream_efficiency(prefetching=True)

Fraction of raw disk bandwidth a streaming pass actually achieves.

Two independent deratings, multiplied: the I/O one (prefetching overlaps the next layer’s read with the current layer’s compute, or it does not.

Parameters:

prefetching (bool) – Whether prefetching is on.

Returns:

Multiplier on read_gbs, between 0 and 1.

Return type:

float

Examples

>>> stream_efficiency()
0.85
>>> stream_efficiency(prefetching=False)
0.7