Cactus Whistle的编码指南：一个16.9 MB的语音模型对起Whisper基站吗？单词出错率、 速度、 流线、 时间戳、 关键字和对工具的调用

中文版正在随流水线自动整理，稍后刷新即可；可先阅读下方英文原文。

本文由 AI宝盒自动抓取并翻译，版权归原作者所有；如需查看原文请点下方链接。

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.

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.

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) < 100:

return out

def dataset_sha(dataset):

return json.loads(fetch(f"https://huggingface.co/api/datasets/{dataset}"))["sha"][:12]

def wer_counts(ref, hyp):

"""Errors (S+D+I) and reference words for one normalised pair; an empty hypothesis is all deletions."""

n = len(ref.split())

if not hyp.strip():

return n, n

o = jiwer.process_words(ref, hyp)

return o.substitutions + o.deletions + o.insertions, n

def sdi(ref, hyp):

if not hyp.strip():

return 0, len(ref.split()), 0

o = jiwer.process_words(ref, hyp)

return o.substitutions, o.deletions, o.insertions

def corpus_wer(errs, words):

return 100.0 * sum(errs) / max(sum(words), 1)

def _by_cluster(clusters, *arrays):

tot = collections.defaultdict(lambda: [0.0] * len(arrays))

for i, c in enumerate(clusters):

for j, a in enumerate(arrays):

tot[c][j] += a[i]

return tot

def wer_ci(clusters, errs, words, B=2000, seed=0):

"""95% percentile interval of a corpus WER, resampling clusters (speakers) with replacement."""

tot = _by_cluster(clusters, errs, words)

keys, rng = sorted(tot), random.Random(seed)

boots = []

for _ in range(B):

d = rng.choices(keys, k=len(keys))

boots.append(100 * sum(tot[c][0] for c in d) / max(sum(tot[c][1] for c in d), 1))

return np.percentile(boots, [2.5, 97.5])

def paired_contrast(clusters, errs_a, words, errs_b, label_a, label_b, B=2000, seed=0):

"""WER(a) - WER(b) with a paired speaker-cluster bootstrap: the same resampled speakers for both systems."""

tot = _by_cluster(clusters, errs_a, errs_b, words)

keys, rng = sorted(tot), random.Random(seed)

boots = []

for _ in range(B):

d = rng.choices(keys, k=len(keys))

w = max(sum(tot[c][2] for c in d), 1)

boots.append(100 * (sum(tot[c][0] for c in d) - sum(tot[c][1] for c in d)) / w)

boots = np.array(boots)

lo, hi = np.percentile(boots, [2.5, 97.5])

p = 2 * min((boots <= 0).mean(), (boots >= 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] <= alpha / (len(ps) - rank):

keep[i] = True

else:

break

return keep

def verdict(c, significant):

"""Verdict rule: diff = WER(other system) - WER(Whistle). CI entirely > 0 and Holm-significant -> Whistle ahead;

entirely < 0 and significant -> Whistle behind; otherwise no significant difference at this sample size."""

if c["lo"] > 0 and significant:

return "Whistle ahead"

if c["hi"] < 0 and significant:

return "Whistle behind"

return "no significant difference at this n"

def fmt_contrast(c):

return f"{c['diff']:+.2f} pts (95% CI {c['lo']:+.2f}..{c['hi']:+.2f}, {c['clusters']} speakers)"

@section("2. Fixed, seeded samples of LibriSpeech test-clean and test-other, and the rules for the verdicts")

def sample_english():

global EN

print("Rules for the verdicts below, written into this notebook before its final run (exploratory probes on the")

print("same splits came first; the prose says which):")

print(f" * simple random sample of {N_WHISTLE} rows per split (seed {SEED}); the first {N_PAIRED['other']} (test-other) and "

f"{N_PAIRED['clean']} (test-clean) are the paired Whisper subsets")

print(f" * clips longer than {MAX_CLIP_S:

Auto-fetched and translated by AI Gem Box. All rights belong to the original author.