A Coding Guide to Cactus Whistle: Does a 16.9 MB Speech Model Hold Up Against Whisper Base? Word Error Rate, Speed, Streaming, Timestamps, Keywords, and Speech-to-Tool Calls on a CPU

In this tutorial, we use Whistle, Cactus Compute’s speech-to-text model, a 16.9 MB file that runs on a CPU, transcribes seven languages, and returns word timestamps and probabilities. The model card claims it beats Whisper base on LibriSpeech; we test it on seeded test-clean and test-other samples, comparing Whistle with Whisper tiny and base on identical clips, judged by paired speaker-cluster intervals with Holm correction. We find a lead over greedy Whisper base on test-other, no significant gap against beam search, and a speed advantage that narrows as clips grow. Beyond the headline, we probe a 30-second window where the pinned engine silently drops the middle of longer audio, plus streaming latency, timestamps, keyword biasing, other languages, noise, embeddings, and speech-to-tool calls. Our figures are one-run samples from an Apple M4, and published full-split numbers are quoted, not measured. Copy CodeCopiedUse a different Browser

import os
import io
import re
import sys
import json
import math
import time
import random
import hashlib
import shutil
import zipfile
import platform
import traceback
import subprocess
import collections
import urllib.error
import urllib.request
 
RESULTS = {}
VERDICTS = []
 
 
def banner(title):
    print("\n" + "=" * 78)
    print(title)
    print("=" * 78)
 
 
def section(name):
    def wrap(fn):
        def run(*a, **kw):
            banner(name)
            t0 = time.time()
            try:
                out = fn(*a, **kw)
                RESULTS[name] = out if isinstance(out, str) else "ok"
                return out
            except Exception as e:
                RESULTS[name] = f"SKIPPED / FAILED -> {type(e).__name__}: {e}"
                print(f"\n[!] {name} did not complete: {type(e).__name__}: {e}")
                traceback.print_exc(limit=3)
                return None
            finally:
                print(f"\n[{time.time() - t0:.0f} s]")
                json.dump({"results": RESULTS, "verdicts": VERDICTS}, open("results.json", "w"), indent=1)
        return run
    return wrap
 
 
banner("1. Install Whistle, pin its weights and engine by Hub revision, transcribe one clip")
subprocess.run([sys.executable, "-m", "pip", "install", "-q", "cactus-needle==3.2.0", "openai-whisper==20250625",
                "jiwer==4.0.0", "soundfile==0.14.0", "soxr==1.1.0"], check=True)
os.environ["NEEDLE_TELEMETRY"] = "0"            # the package sends anonymous usage counts unless told not to
os.environ["HF_HUB_DISABLE_PROGRESS_BARS"] = "1"
 
import numpy as np
import soundfile as sf
import soxr
from huggingface_hub import hf_hub_download
 
# cactus-needle fetches three files on first use, from Hugging Face repos whose files are replaced in place: the 16.9 MB
# whistle.cact weights, the ~1 MB engine library (version 3.3.0 in this package; the needle3 repo received new binaries
# on 2026-10-10) and the 35 MB needle3.cact used in step 12. Pinning the Hub revision of each, and printing the hashes,
# is what makes this run reproducible. The engine version is read from the package, not typed in.
WHISTLE_REV = "ca5287601bef25af26dcf2e1b2bdc0843a7c19e5"   # Cactus-Compute/whistle, 2026-10-10
NEEDLE3_REV = "93e88ede867e4311c39151285e74df513e4b835b"   # Cactus-Compute/needle3 (engine + Needle 3 weights), 2026-10-10
from needle.agent import fetch as needle_fetch
TAG = needle_fetch._platform_tag()
ENGINE_VERSION = needle_fetch.engine_version(3)
weights = hf_hub_download("Cactus-Compute/whistle", "whistle.cact", revision=WHISTLE_REV)
wheel = hf_hub_download("Cactus-Compute/needle3", needle_fetch.engine_wheel(ENGINE_VERSION, TAG, 3), revision=NEEDLE3_REV)
with zipfile.ZipFile(wheel) as z:
    member = next(n for n in z.namelist() if n.startswith("needle/libneedle"))
    os.makedirs("engine", exist_ok=True)
    engine = os.path.abspath(os.path.join("engine", os.path.basename(member)))
    open(engine, "wb").write(z.read(member))
needle3_weights = hf_hub_download("Cactus-Compute/needle3", "needle3.cact", revision=NEEDLE3_REV)
os.makedirs(needle_fetch.cache_dir(3), exist_ok=True)           # Needle(tools=...) reads its weights from this cache path
shutil.copyfile(needle3_weights, os.path.join(needle_fetch.cache_dir(3), needle_fetch.base_weights(3)))
os.environ["NEEDLE_WHISTLE_WEIGHTS"] = weights   # both variables are honoured by the package
os.environ["NEEDLE3_LIB_PATH"] = engine
 
import needle
import importlib.metadata as md
sha = lambda p: hashlib.sha256(open(p, "rb").read()).hexdigest()[:12]
CPU = platform.processor() or platform.machine()
if sys.platform == "linux" and os.path.exists("/proc/cpuinfo"):
    CPU = next((l.split(":", 1)[1].strip() for l in open("/proc/cpuinfo") if l.startswith("model name")), CPU)
elif sys.platform == "darwin":
    CPU = subprocess.run(["sysctl", "-n", "machdep.cpu.brand_string"], capture_output=True, text=True).stdout.strip() or CPU
print(f"cactus-needle {needle.__version__} (engine {ENGINE_VERSION}, wheel tag {TAG})   python {platform.python_version()}   "
      f"soundfile {md.version('soundfile')}  soxr {md.version('soxr')}  numpy {np.__version__}")
print(f"hardware: {CPU}, {os.cpu_count()} cpus visible")
print(f"whistle.cact  {os.path.getsize(weights)/1e6:.1f} MB  sha256 {sha(weights)}   @ {WHISTLE_REV[:12]}")
print(f"engine        {os.path.getsize(engine)/1e6:.1f} MB  sha256 {sha(engine)}   @ {NEEDLE3_REV[:12]}")
print(f"needle3.cact  {os.path.getsize(needle3_weights)/1e6:.1f} MB  sha256 {sha(needle3_weights)}   @ {NEEDLE3_REV[:12]}")
 
SR = 16000
W = needle.Whistle()      # one model per process, not thread-safe; every later step reuses this object
 
 
def fetch(url, tries=7):
    """GET with retries. The datasets-server rate-limits anonymous callers (HTTP 429): honour Retry-After when it is
    sent, otherwise back off 5, 10, 20, 40 ... seconds, and never give up on the first refusal."""
    for k in range(tries):
        try:
            with urllib.request.urlopen(url, timeout=60) as r:
                return r.read()
        except urllib.error.HTTPError as e:
            if k == tries - 1:
                raise
            wait = int(e.headers.get("Retry-After", 0) or 0) if e.code == 429 else 0
            time.sleep(max(wait, 5 * 2 ** k) if e.code in (429, 500, 502, 503, 504) else 2)
        except Exception:
            if k == tries - 1:
                raise
            time.sleep(3 * (k + 1))
 
 
def page(dataset, config, split, offset, length=100):
    """One page of a Hub dataset through the datasets-server rows API: metadata plus signed audio URLs."""
    url = (f"https://datasets-server.huggingface.co/rows?dataset={dataset}&config={config}"
           f"&split={split}&offset={offset}&length={length}")
    time.sleep(0.25)                                  # stay under the anonymous rate limit when paging a whole split
    return json.loads(fetch(url))
 
 
def load_audio(row):
    """Download one row's clip; return 16 kHz mono float32 samples and the sha256 of the original file."""
    data = fetch(row["audio"][0]["src"])
    audio, sr = sf.read(io.BytesIO(data), dtype="float32")
    if audio.ndim > 1:
        audio = audio.mean(axis=1)
    if sr != SR:
        audio = soxr.resample(audio, sr, SR, quality="HQ")
    return audio, hashlib.sha256(data).hexdigest()[:12]
 
 
first = page("openslr/librispeech_asr", "clean", "test", 0, 1)["rows"][0]["row"]
clip, _ = load_audio(first)
t0 = time.perf_counter()
out = W.transcribe(clip, word_timestamps=True)
wall = (time.perf_counter() - t0) * 1000
print(f"\nclip {first['id']}  {len(clip)/SR:.1f} s   reference: {first['text']}")
print(f"-> {out['text']!r}")
print(f"   language {out['language']}   engine time to first token {out['ttft_ms']:.1f} ms   decode {out['decode_tps']:.0f} tok/s"
      f"   wall {wall:.0f} ms")
print("   first words:", [(w["word"], w["start"], w["end"], w["probability"]) for w in out["words"][:3]])
RESULTS["1. Install Whistle, pin its weights and engine by Hub revision, transcribe one clip"] = \
    f"{os.path.getsize(weights)/1e6:.1f} MB weights, {len(clip)/SR:.1f} s clip in {wall:.0f} ms"

We install cactus-needle 3.2.0 with its dependencies pinned and turn off its anonymous telemetry. The package downloads Whistle’s weights, its engine and Needle 3’s weights from Hugging Face repositories whose files are replaced in place, so we pin both repositories to fixed revisions and print SHA-256 hashes. The page() helper reads the datasets-server rows API with rate-limit backoff, and load_audio() resamples clips to 16 kHz. The first test-clean clip, 3.5 s of speech, transcribes in 33 ms of wall time, each word carrying its start, end and probability. Copy CodeCopiedUse a different Browser

import jiwer
import whisper
import torch
from whisper.normalizers import EnglishTextNormalizer, BasicTextNormalizer
 
NORM_EN = EnglishTextNormalizer()      # the normalisers Whistle's own benchmarks use; applied to both sides, always
NORM_BASIC = BasicTextNormalizer()
SEED = 2026
N_WHISTLE = 300                        # clips per LibriSpeech split that Whistle transcribes (it is cheap)
N_PAIRED = {"other": 100, "clean": 60}  # the first N of them also go through Whisper (it is not); test-other is primary
MAX_CLIP_S = 30.0                      # Whistle's single-pass window; longer clips are declared out of scope here
WHISPER_SECONDS = collections.Counter()
 
 
def all_rows(dataset, config, split):
    out, offset = [], 0
    while True:
        rows = page(dataset, config, split, offset)["rows"]
        out += [r["row"] for r in rows]
        offset += 100
        if len(rows) = 0).mean())
    return {"a": label_a, "b": label_b, "diff": corpus_wer(errs_a, words) - corpus_wer(errs_b, words),
            "lo": lo, "hi": hi, "p": max(p, 1 / B), "clusters": len(keys)}
 
 
def holm(ps, alpha=0.05):
    order, keep = np.argsort(ps), [False] * len(ps)
    for rank, i in enumerate(order):
        if ps[i]  0 and Holm-significant -> Whistle ahead;
    entirely  Whistle behind; otherwise no significant difference at this sample size."""
    if c["lo"] > 0 and significant:
        return "Whistle ahead"
    if c["hi"]  0 and Holm-adjusted p ")
    print("    Whistle ahead; entirely  behind; otherwise no significant difference at this n")
    print("    family A (published claims): test-other and test-clean vs base greedy and vs base beam 5, MLS pooled vs base greedy;")
    print("    family B: the keyword contrasts. Holm at alpha = 0.05 within each family. Whisper tiny rows are descriptive.\n")
    global MANIFEST
    EN, manifest = {}, {"dataset_sha": dataset_sha("openslr/librispeech_asr"), "seed": SEED, "splits": {}}
    MANIFEST = manifest
    for split in ("other", "clean"):
        t0 = time.time()
        rows = all_rows("openslr/librispeech_asr", split, "test")
        idx = random.Random(SEED).sample(range(len(rows)), N_WHISTLE)
        kept, too_long, empty = [], 0, 0
        for i in idx:
            r = rows[i]
            audio, h = load_audio(r)
            ref = NORM_EN(r["text"])
            if len(audio) / SR > MAX_CLIP_S:
                too_long += 1
                continue
            if not ref:
                empty += 1
                continue
            kept.append({"id": r["id"], "spk": str(r["speaker_id"]), "text": r["text"], "ref": ref,
                         "audio": audio, "dur": len(audio) / SR, "sha": h})
        EN[split] = kept
        manifest["splits"][split] = {"rows": len(rows), "sampled": N_WHISTLE, "too_long": too_long, "empty_ref": empty,
                                     "ids": [c["id"] for c in kept], "sha": [c["sha"] for c in kept]}
        durs = [c["dur"] for c in kept]
        pk = kept[:N_PAIRED[split]]
        print(f"test-{split}: {len(rows)} rows -> {len(kept)} clips kept ({too_long} over {MAX_CLIP_S:.0f} s, {empty} empty), "
              f"{len(set(c['spk'] for c in kept))} speakers, {sum(durs)/60:.1f} min, clip {np.mean(durs):.1f} s mean / {max(durs):.1f} s max"
              f"   [