이 튜토리얼에서는 Whistle을 사용합니다. Cactus Compute의 음성-텍스트 모델로, 16.9 MB 크기의 파일로 CPU에서 실행되며 7개 언어를 전사하고 단어 단위 타임스탬프와 확률을 반환합니다. 모델 카드는 LibriSpeech에서 Whisper base를 능가한다고 주장하며, 우리는 시드 고정된 test-clean 및 test-other 샘플에서 이를 테스트하여 동일한 클립을 대상으로 Whistle을 Whisper tiny 및 base와 비교하고, Holm 보정을 적용한 쌍별 화자-클러스터 구간으로 판정합니다. 그 결과 test-other에서 greedy Whisper base에 대한 우위를 확인했고, beam search와는 유의미한 격차가 없었으며, 클립이 길어질수록 줄어드는 속도 이점을 발견했습니다. 헤드라인 너머로, 고정된 엔진이 30초 창을 넘는 긴 오디오의 중간을 조용히 삭제하는 현상을 조사하고, 스트리밍 지연 시간, 타임스탬프, 키워드 바이어싱, 기타 언어, 노이즈, 임베딩, 음성-툴 호출도 살펴봅니다. 우리의 수치는 Apple M4에서 얻은 1회 실행 샘플이며, 공개된 전체 분할 수치는 측정이 아닌 인용입니다.
코드 복사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"
우리는 cactus-needle 3.2.0을 의존성을 고정하여 설치하고 익명 텔레메트리를 끕니다. 이 패키지는 파일이 제자리에서 교체되는 Hugging Face 저장소에서 Whistle의 가중치, 엔진, Needle 3의 가중치를 다운로드하므로, 두 저장소 모두를 고정 리비전에 고정하고 SHA-256 해시를 출력합니다. page() 헬퍼는 레이트 리밋 백오프와 함께 datasets-server rows API를 읽고, load_audio()는 클립을 16 kHz로 리샘플링합니다. 3.5 s 분량의 음성으로 이루어진 첫 번째 test-clean 클립은 33 ms의 실제 시간에 전사되며, 각 단어는 시작, 끝, 확률을 함께 담고 있습니다.
코드 복사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:.0f} s are excluded and counted (Whistle's single-pass window), as are empty references")
print(" * the same normaliser on both sides of each comparison (Whisper's EnglishTextNormalizer for LibriSpeech, its")
print(" BasicTextNormalizer for MLS); corpus WER = summed errors / summed reference words")
print(" * interval: 95% speaker-cluster paired bootstrap, 2,000 draws, the same resampled speakers for every system")
print(" * verdict per contrast: unadjusted CI of WER(other) - WER(Whistle) entirely > 0 and Holm-adjusted p < 0.05 ->")
print(" Whistle ahead; entirely < 0 and significant -> 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" [{time.time()-t0:.0f} s]")
print(f" paired subset: {len(pk)} clips, {len(set(c['spk'] for c in pk))} speakers, {sum(c['dur'] for c in pk):.0f} s of audio, "
f"{sum(len(c['ref'].split()) for c in pk)} reference words")
json.dump(manifest, open("manifest.json", "w"))
print(f"\nmanifest.json: LibriSpeech revision {manifest['dataset_sha']}, clip ids and file hashes (the MLS and MInDS-14 clips are added to it in")
print("steps 9 and 12), so a rerun can prove it used the same audio")
return f"{len(EN['other'])} + {len(EN['clean'])} clips, manifest written"
sample_english()
최종 수치가 나온 후에 판정 기준을 정한다면 그 판정은 큰 의미가 없으므로, 우리는 동일한 클립 일부를 건드린 탐색적 조사 이후, 최종 실행 전에 판정 기준을 노트북에 기록해 두었습니다. 우리는 분할당 300행씩의 시드 고정 샘플을 뽑고 Whistle의 단일 패스 창보다 긴 클립을 제거하여 각 299개 클립을 남겼습니다. 쌍별 부분 집합인 첫 100개의 test-other 클립과 60개의 test-clean 클립은 Whisper에 통과시켰고, 적은 수의 화자로 인해 구간이 근사치가 됩니다. 차이는 95% 쌍별 화자-클러스터 부트스트랩 구간이 0을 제외하고 Holm 보정된 p가 0.05 미만일 때만 유효한 것으로 간주합니다.
코드 복사WHISPER = {}
def whisper_model(size):
if size not in WHISPER:
WHISPER[size] = whisper.load_model(size, device="cpu")
return WHISPER[size]
def run_whisper(size, audio, language="en", **kw):
"""openai-whisper with its defaults (greedy with temperature fallback) unless kw says otherwise. Seeded per call,
because the fallback samples. Returns text, whether any segment needed the fallback, and the wall time."""
torch.manual_seed(0)
WHISPER_SECONDS[size] += len(audio) / SR
t0 = time.perf_counter()
r = whisper_model(size).transcribe(audio, language=language, fp16=False, **kw)
fell_back = any(s.get("temperature", 0) > 0 for s in r["segments"])
return r["text"], fell_back, time.perf_counter() - t0
SYSTEMS = {"other": [("whisper tiny, greedy", "tiny", {}), ("whisper base, greedy", "base", {}), ("whisper base, beam 5", "base", {"beam_size": 5})],
"clean": [("whisper tiny, greedy", "tiny", {}), ("whisper base, greedy", "base", {}), ("whisper base, beam 5", "base", {"beam_size": 5})]}
FAMILY_A = {"test-clean: Whistle vs whisper base, greedy", "test-clean: Whistle vs whisper base, beam 5",
"test-other: Whistle vs whisper base, greedy", "test-other: Whistle vs whisper base, beam 5",
"MLS pooled: Whistle vs whisper base, greedy"}
PUBLISHED = {"other": "Whistle 10.49 / Whisper base 11.0 (paper table 9, beam) / 12.4 (table 8, greedy)",
"clean": "Whistle 4.31 / Whisper base 4.9 (paper table 9, beam) / 5.0 (table 8, greedy)"}
CONTRASTS = []
@section("3. Word error rate against Whisper tiny and base, paired by clip, with a verdict per contrast")
def english_wer():
global HYP
HYP = {}
for split in ("other", "clean"):
clips = EN[split]
t0 = time.perf_counter()
hw = [W.transcribe(c["audio"], language="en")["text"] for c in clips]
tw = time.perf_counter() - t0
HYP[(split, "Whistle")] = hw
e_w, n_w = zip(*[wer_counts(c["ref"], NORM_EN(h)) for c, h in zip(clips, hw)])
lo, hi = wer_ci([c["spk"] for c in clips], e_w, n_w)
print(f"test-{split}: Whistle on all {len(clips)} clips: WER {corpus_wer(e_w, n_w):.2f}% (95% speaker-cluster CI {lo:.2f}..{hi:.2f}) "
f"in {tw:.1f} s ({1000*tw/len(clips):.0f} ms per clip)")
print(f" published full-split figures, different sample: {PUBLISHED[split]}")
pk = clips[:N_PAIRED[split]]
spk = [c["spk"] for c in pk]
e_wp, n_p = list(e_w[:len(pk)]), list(n_w[:len(pk)])
s = np.sum([sdi(c["ref"], NORM_EN(h)) for c, h in zip(pk, hw[:len(pk)])], axis=0)
print(f" paired subset, {len(pk)} clips / {len(set(spk))} speakers / {sum(n_p)} words:")
print(f" {'Whistle, 5-beam':22s} {corpus_wer(e_wp, n_p):6.2f}% S/D/I {s[0]:3d}/{s[1]:3d}/{s[2]:3d}")
for label, size, kw in SYSTEMS[split]:
res = [run_whisper(size, c["audio"], **kw) for c in pk]
hyps = [r[0] for r in res]
HYP[(split, label)] = hyps
e_o = [wer_counts(c["ref"], NORM_EN(h))[0] for c, h in zip(pk, hyps)]
s = np.sum([sdi(c["ref"], NORM_EN(h)) for c, h in zip(pk, hyps)], axis=0)
con = paired_contrast(spk, e_o, n_p, e_wp, label, "Whistle")
con["name"] = f"test-{split}: Whistle vs {label}"
CONTRASTS.append(con)
better, worse = sum(a < b for a, b in zip(e_wp, e_o)), sum(a > b for a, b in zip(e_wp, e_o))
print(f" {label:22s} {corpus_wer(e_o, n_p):6.2f}% S/D/I {s[0]:3d}/{s[1]:3d}/{s[2]:3d} fallback on {sum(r[1] for r in res):2d} clips "
f"{np.median([r[2] for r in res])*1000:5.0f} ms/clip | {fmt_contrast(con)} clips Whistle better/worse {better}/{worse}")
fam = [c for c in CONTRASTS if c["name"] in FAMILY_A]
print(f"\nContrasts so far (family A: {len(fam)} of 5; the Holm correction and the verdicts come once the MLS contrast is in):")
for c in CONTRASTS:
tag = "family A" if c["name"] in FAMILY_A else "descriptive"
print(f" {c['name']:48s} {fmt_contrast(c)} p~{c['p']:.3f} [{tag}]")
print("\n Whistle decodes with a 5-beam length-normalised search. openai-whisper's default is greedy with temperature")
print(" fallback; the Whisper paper's LibriSpeech numbers come from beam search, so the beam-5 rows are the like-for-like ones.")
print(" Every gap here is a system gap (model plus decoder), not a model-only one.")
return " | ".join(f"test-{s}: Whistle {corpus_wer(*zip(*[wer_counts(c['ref'], NORM_EN(h)) for c, h in zip(EN[s][:N_PAIRED[s]], HYP[(s, 'Whistle')][:N_PAIRED[s]])])):.2f}% "
f"vs base greedy {corpus_wer(*zip(*[wer_counts(c['ref'], NORM_EN(h)) for c, h in zip(EN[s][:N_PAIRED[s]], HYP[(s, 'whisper base, greedy')])])):.2f}%" for s in ("other", "clean"))
english_wer()
Whisper의 기본 디코더는 temperature fallback을 사용하는 greedy 방식입니다. 동시에 논문의 LibriSpeech 수치는 beam search에서 나온 것이고 Whistle은 5빔 검색으로 디코딩하므로, 모든 비교는 기준선을 명시해야 합니다. 하나의 정규화기로 채점했을 때 Whistle의 WER은 test-other 샘플에서 10.47%로, 인용만 했을 뿐 측정하지는 않은 공개된 전체 분할 수치 10.49에 근접합니다. 쌍별 부분 집합에서는 greedy Whisper base의 12.46%, beam search의 10.83%에 대비해 10.27%를 기록했습니다. greedy 대비 2.19포인트 우위는 Holm 보정을 통과했지만 beam과의 격차는 통과하지 못했으므로, beam 기준선이 승리합니다.
코드 복사def audio_seconds(clips, seconds):
"""Exactly `seconds` of speech: clips concatenated in sample order, then trimmed."""
out, total = [], 0
for c in clips:
out.append(c["audio"])
total += c["dur"]
if total >= seconds:
break
return np.concatenate(out)[:int(seconds * SR)]
PROBE_SCRIPT = r"""
import sys, json, time, resource, numpy as np
kind, paths = sys.argv[1], sys.argv[2:]
scale = 1 << 20 if sys.platform == "darwin" else 1 << 10 # ru_maxrss is bytes on macOS, kilobytes on Linux
rss = lambda: resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / scale
out = {"kind": kind}
if kind.startswith("whistle"):
import needle
out["after_import"] = rss()
W = needle.Whistle()
run = lambda a: W.transcribe(a, language="en")
else:
import whisper, torch
out["after_import"] = rss()
size, kw = kind.split("-")[1], ({"beam_size": 5} if kind.endswith("beam5") else {})
m = whisper.load_model(size, device="cpu")
def run(a):
torch.manual_seed(0)
return m.transcribe(a, language="en", fp16=False, **kw)
clips = {p: np.load(p) for p in paths}
run(clips[paths[0]]) # warm-up
reps = 5 if kind.startswith("whistle") else 3
for p, a in clips.items():
ts = []
for _ in range(reps):
t0 = time.perf_counter(); r = run(a); ts.append(time.perf_counter() - t0)
out[p] = {"wall_ms": 1000 * float(np.median(ts)), "spread_ms": 1000 * float(np.ptp(ts))}
if kind.startswith("whistle"):
out[p].update(ttft_ms=r["ttft_ms"], tps=r["decode_tps"])
out["peak"] = rss()
print(json.dumps(out))
"""
@section("4. Speed and footprint: identical arrays, one fresh process per engine, total wall time and peak memory")
def speed():
pool = EN["other"]
print(f"hardware: {CPU}, {os.cpu_count()} cpus visible. Each engine runs in its own fresh process so torch's threads cannot slow")
print("Whistle down (or the reverse); Whistle's engine picks its own thread count. Primary measure: total wall time on the same")
print("float32 array, median of 5 (Whistle) or 3 (Whisper) after a warm-up. ttft_ms is the engine's own clock, for reference.\n")
for secs in (5, 10, 30):
np.save(f"clip{secs}.npy", audio_seconds(pool, secs))
open("probe.py", "w").write(PROBE_SCRIPT)
res = {}
for kind in ("whistle", "whisper-tiny", "whisper-base", "whisper-base-beam5"):
out = subprocess.run([sys.executable, "probe.py", kind, "clip5.npy", "clip10.npy", "clip30.npy"], capture_output=True, text=True, env=os.environ)
if out.returncode != 0:
raise RuntimeError(f"{kind} probe failed: {out.stderr[-300:]}")
res[kind] = json.loads(out.stdout.strip().splitlines()[-1])
print(f"{'clip':>6s} | {'Whistle wall':>12s} {'spread':>7s} {'ttft':>6s} {'tok/s':>6s} | {'tiny greedy':>11s} | {'base greedy':>11s} | {'base beam 5':>11s}")
for secs in (5, 10, 30):
p = f"clip{secs}.npy"
w = res["whistle"][p]
print(f"{secs:5d} s | {w['wall_ms']:9.0f} ms {w['spread_ms']:5.0f} ms {w['ttft_ms']:6.1f} {w['tps']:6.0f} | "
f"{res['whisper-tiny'][p]['wall_ms']:8.0f} ms | {res['whisper-base'][p]['wall_ms']:8.0f} ms | {res['whisper-base-beam5'][p]['wall_ms']:8.0f} ms")
w5, w10, w30 = (res["whistle"][p]["wall_ms"] for p in ("clip5.npy", "clip10.npy", "clip30.npy"))
b5, b10, b30 = (res["whisper-base"][p]["wall_ms"] for p in ("clip5.npy", "clip10.npy", "clip30.npy"))
bb5, bb10, bb30 = (res["whisper-base-beam5"][p]["wall_ms"] for p in ("clip5.npy", "clip10.npy", "clip30.npy"))
print(f"\nreal-time factor on the 10 s clip: Whistle {w10/10000:.4f}, base greedy {b10/10000:.3f}, base beam 5 {bb10/10000:.3f}")
print(f"base greedy / Whistle wall time: {b5/w5:.1f}x at 5 s, {b10/w10:.1f}x at 10 s, {b30/w30:.1f}x at 30 s; beam 5 / Whistle: "
f"{bb5/w5:.1f}x, {bb10/w10:.1f}x, {bb30/w30:.1f}x")
print(f"from 5 to 10 s the wall time grows {w10/w5:.2f}x for Whistle, {b10/b5:.2f}x for base greedy, {bb10/bb5:.2f}x for base beam 5;")
print(f"from 10 to 30 s: {w30/w10:.2f}x, {b30/b10:.2f}x, {bb30/bb10:.2f}x. Whisper's encoder always sees a padded 30 s window; its decoder")
print("still grows with the words it emits, so the padding fixes only part of its cost. The encoder is not timed separately here.")
print("\npeak RSS of each process (model loaded, three clips transcribed) and the RSS right after its imports:")
for kind in ("whistle", "whisper-tiny", "whisper-base"):
r = res[kind]
print(f" {kind:13s} peak {r['peak']:6.0f} MB after imports {r['after_import']:4.0f} MB -> {r['peak']-r['after_import']:4.0f} MB for the model and the run")
sizes = {"whistle.cact + engine": (os.path.getsize(weights) + os.path.getsize(engine)) / 1e6}
for size in ("tiny", "base"):
p = os.path.expanduser(f"~/.cache/whisper/{size}.pt")
sizes[f"whisper {size}.pt"] = os.path.getsize(p) / 1e6 if os.path.exists(p) else float("nan")
print("files on disk: " + ", ".join(f"{k} {v:.1f} MB" for k, v in sizes.items()) + " (file sizes, not deployed footprint)")
off = subprocess.run([sys.executable, "-c", "import needle, numpy as np; W = needle.Whistle(); print(W.transcribe(np.zeros(16000, np.float32))['text'] == '')"],
capture_output=True, text=True, env={**os.environ, "HF_HUB_OFFLINE": "1"})
print(f"fresh process with HF_HUB_OFFLINE=1 and the pinned paths: {'loads and runs' if 'True' in off.stdout else 'FAILED: ' + off.stderr[-200:]}")
return f"10 s clip: Whistle {w10:.0f} ms vs base greedy {b10:.0f} ms ({b10/w10:.1f}x, {b5/w5:.1f}x at 5 s, {b30/w30:.1f}x at 30 s); peak RSS {res['whistle']['peak']:.0f} vs {res['whisper-base']['peak']:.0f} MB"
speed()
엔진들이 하나의 프로세스를 공유하면 torch 스레드가 한쪽을 느리게 할 수 있어 속도 비교가 실패합니다. 따라서 각 엔진은 동일한 float32 배열을 대상으로 별도의 새 프로세스에서 실행되며, 워밍업 후 중앙값으로 측정됩니다. 격차는 클립이 길어질수록 줄어들어, 5 s에서 Whistle 시간의 7.7배에서 30 s에서 2.8배가 되는데, 이는 부분적으로 Whisper의 고정 패딩 창 때문입니다. 최대 메모리는 Whistle이 124 MB, Whisper base가 823 MB였습니다. Whistle의 클립당 시간은 두 분할 간에 거의 3배 차이가 났으며 이는 클립 길이로 설명되지 않으므로, 비율은 근사치로 취급하십시오.
코드 복사def quiet_windows(a, limit=30.0, search=5.0, hop=0.2):
"""Cut before `limit` seconds at the quietest `hop`-second patch inside the last `search` seconds of each window."""
out, i, n, s, h = [], 0, int(limit * SR), int(search * SR), int(hop * SR)
while len(a) - i > n:
seg = a[i + n - s:i + n]
energy = [np.mean(seg[j:j + h] ** 2) for j in range(0, len(seg) - h, h // 4)]
cut = i + n - s + int(np.argmin(energy)) * (h // 4) + h // 2
out.append(a[i:cut])
i = cut
out.append(a[i:])
return out
def speaker_sessions(min_s=40, max_s=100, k=8):
"""Same-speaker concatenations (0.4 s gaps) from the sampled clips of both splits, longest speakers first."""
by = collections.defaultdict(list)
for split in ("other", "clean"):
for c in EN[split]:
by[c["spk"]].append(c)
out, gap = [], np.zeros(int(0.4 * SR), np.float32)
for spk, cs in sorted(by.items(), key=lambda kv: -sum(c["dur"] for c in kv[1])):
parts, used, total = [], [], 0
for c in cs:
if total + c["dur"] + 0.4 > max_s:
break
parts += [c["audio"], gap]
used.append(c)
total += c["dur"] + 0.4
if total >= min_s:
out.append({"spk": spk, "clips": used, "audio": np.concatenate(parts), "ref": " ".join(c["ref"] for c in used), "dur": total})
if len(out) == k:
break
return out
OTHER_ENGINE_SCRIPT = r"""
import sys, os, numpy as np
os.environ["NEEDLE3_LIB_PATH"] = sys.argv[1]
import needle
W = needle.Whistle()
a = np.load(sys.argv[2])
for secs in (30.0, 30.01, 45.0):
try:
r = W.transcribe(a[:int(secs * 16000)], language="en", word_timestamps=True)
print(f" {secs:6.2f} s -> {len(r['words'])} words, last word ends at {r['words'][-1]['end']:.2f} s")
except RuntimeError as e:
print(f" {secs:6.2f} s -> RuntimeError: {e}")
"""
@section("5. The 30-second window: what transcribe() does with longer audio, and two ways around it")
def long_audio():
global SESSIONS
SESSIONS = speaker_sessions()
s = SESSIONS[0]
gap = np.zeros(int(0.4 * SR), np.float32)
print(f"speaker {s['spk']}: {len(s['clips'])} utterances joined with 0.4 s gaps, {s['dur']:.0f} s, {len(s['ref'].split())} reference words")
print("the docs say transcribe() takes 'audio of any length, one pass up to 30 seconds and windowed beyond it'.")
print(f"the pinned engine {ENGINE_VERSION}, on whole utterances cut from the session so the reference is exact:\n")
print(f"{'audio':>7s} | {'ref words':>9s} {'hyp words':>9s} {'WER':>7s} | gaps longer than 3 s in the word timeline")
for target in (20, 29, 35, 45, 60, 90):
parts, refs, total = [], [], 0
for c in s["clips"]:
if total + c["dur"] + 0.4 > target:
break
parts += [c["audio"], gap]
refs.append(c["ref"])
total += c["dur"] + 0.4
if total < min(target, s["dur"]) - 8:
continue
a, ref = np.concatenate(parts), " ".join(refs)
r = W.transcribe(a, language="en", word_timestamps=True)
hyp = NORM_EN(r["text"])
gaps = [f"{w1['end']:.0f}-{w2['start']:.0f} s" for w1, w2 in zip(r["words"], r["words"][1:]) if w2["start"] - w1["end"] > 3]
print(f"{total:5.0f} s | {len(ref.split()):9d} {len(hyp.split()):9d} {100*jiwer.wer(ref, hyp or '<empty>'):6.1f}% | {', '.join(gaps) or 'none'}")
full = W.transcribe(s["audio"], language="en", word_timestamps=True)
hyp = NORM_EN(full["text"])
print(f"\nthe whole {s['dur']:.0f} s in one call: {len(hyp.split())} of {len(s['ref'].split())} words, WER {100*jiwer.wer(s['ref'], hyp or '<empty>'):.1f}%, "
f"nothing raised, engine ttft {full['ttft_ms']:.0f} ms")
# the same repo revision also holds a 3.2.0 engine wheel; a child process shows what that build does with the same audio
other_wheel = hf_hub_download("Cactus-Compute/needle3", needle_fetch.engine_wheel("3.2.0", TAG, 3), revision=NEEDLE3_REV)
with zipfile.ZipFile(other_wheel) as z:
member = next(n for n in z.namelist() if n.startswith("needle/libneedle"))
other = os.path.abspath(os.path.join("engine", "v3.2.0-" + os.path.basename(member)))
open(other, "wb").write(z.read(member))
np.save("session0.npy", s["audio"])
open("other_engine.py", "w").write(OTHER_ENGINE_SCRIPT)
out = subprocess.run([sys.executable, "other_engine.py", other, "session0.npy"], capture_output=True, text=True, env=os.environ)
print(f"\nthe 3.2.0 engine wheel from the same repo revision (sha256 {sha(other)}), in a child process, on the same audio:")
print(out.stdout.rstrip() or out.stderr[-300:])
print(" two builds, two behaviours: one windows and loses the middle, the other refuses. Either way the split is yours to make.\n")
print("workarounds on the whole clip, scored against the exact reference (one clip; the joins are 0.4 s of silence):")
results = {}
for name, wins in (("fixed 30 s windows", [s["audio"][i:i + 30 * SR] for i in range(0, len(s["audio"]), 30 * SR)]),
("fixed 25 s windows", [s["audio"][i:i + 25 * SR] for i in range(0, len(s["audio"]), 25 * SR)]),
("quiet-point cuts <= 30 s", quiet_windows(s["audio"], 30)),
("quiet-point cuts <= 20 s", quiet_windows(s["audio"], 20))):
t0 = time.perf_counter()
hyp = NORM_EN(" ".join(W.transcribe(w, language="en")["text"] for w in wins))
results[name] = 100 * jiwer.wer(s["ref"], hyp or "<empty>")
print(f" {name:26s} {len(wins):2d} calls WER {results[name]:5.2f}% {1000*(time.perf_counter()-t0):4.0f} ms "
f"windows of {', '.join(f'{len(w)/SR:.1f}' for w in wins)} s")
t0 = time.perf_counter()
steps = list(W.stream(iter([s["audio"][i:i + SR] for i in range(0, len(s["audio"]), SR)]), language="en"))
hyp = NORM_EN(" ".join(x["text"] for x in steps if x["text"]))
results["stream(), 1 s chunks"] = 100 * jiwer.wer(s["ref"], hyp or "<empty>")
print(f" {'stream(), 1 s chunks':26s} {len(steps):2d} passes WER {results['stream(), 1 s chunks']:5.2f}% {1000*(time.perf_counter()-t0):4.0f} ms")
best = min(results, key=results.get)
return (f"one call on {s['dur']:.0f} s keeps {len(NORM_EN(full['text']).split())}/{len(s['ref'].split())} words; {best}: {results[best]:.2f}% WER "
f"(fixed 30 s windows {results['fixed 30 s windows']:.2f}%)")
long_audio()
한 화자의 발화를 이어 붙이면 정확한 참조 전사가 있는 긴 클립이 되고, README는 30 s 제한을 넘는 창 처리를 약속합니다. 더 짧게 자른 것은 거의 단어 그대로 돌아오지만, 고정된 3.3.0 엔진은 전체 클립의 중간을 조용히 삭제하여 오류 없이 절반 이하의 단어만 반환합니다. 3.2.0 엔진은 대신 RuntimeError로 그런 오디오를 거부하므로, 두 빌드가 서로 다르며 어느 쪽을 택할지는 우리의 몫입니다. 조용한 지점에서 분할하면 고정 창의 6.74%에 대비해 3.75%의 WER을 기록하지만, 하나의 클립으로 이를 확정할 수는 없습니다.
코드 복사@section("6. Streaming: cost per pass, how long a word waits to be committed, and what chunk size changes")
def streaming():
durs = [s["dur"] for s in SESSIONS]
print(f"{len(SESSIONS)} same-speaker sessions of {', '.join(f'{x:.0f}' for x in durs)} s, "
f"{sum(len(s['ref'].split()) for s in SESSIONS)} reference words; the joins between utterances are artificial")
off = {c["id"]: h for split in ("other", "clean") for c, h in zip(EN[split], HYP[(split, "Whistle")])}
off_err = [sum(wer_counts(c["ref"], NORM_EN(off[c["id"]]))[0] for c in s["clips"]) for s in SESSIONS]
words = [len(s["ref"].split()) for s in SESSIONS]
print(f"the same utterances one at a time through transcribe() (the step 3 outputs): WER {100*sum(off_err)/sum(words):.2f}%\n")
print(f"{'chunk':>6s} {'sessions':>8s} | {'WER':>6s} | {'pass ms mean / max':>18s} | {'passes > 1 s':>12s} | {'commit latency median / p90':>27s} | {'words only in tail':>18s} | {'pass vs time r':>14s}")
summary, stream_err = {}, None
for chunk_s, sessions in ((1.0, SESSIONS), (0.5, SESSIONS[:2]), (1.0, SESSIONS[:2]), (2.0, SESSIONS[:2])):
n = int(chunk_s * SR)
errs, tail, allw = [], 0, 0
pass_ms, lat, rs = [], [], []
for s in sessions:
steps = list(W.stream(iter([s["audio"][i:i + n] for i in range(0, len(s["audio"]), n)]), language="en"))
errs.append(wer_counts(s["ref"], NORM_EN(" ".join(x["text"] for x in steps if x["text"])))[0])
body = steps[:-1]
pass_ms += [x["pass_ms"] for x in body]
lat += [x["received"] - wd["end"] for x in steps for wd in x["words"]]
tail += len(steps[-1]["words"])
allw += sum(len(x["words"]) for x in steps)
if len(body) > 2:
rs.append(np.corrcoef([x["received"] for x in body], [x["pass_ms"] for x in body])[0, 1])
w = words[:len(sessions)]
summary[(chunk_s, len(sessions))] = (100 * sum(errs) / sum(w), np.median(lat), np.mean(pass_ms))
if chunk_s == 1.0 and len(sessions) == len(SESSIONS):
stream_err = errs
print(f"{chunk_s:5.1f} s {len(sessions):8d} | {100*sum(errs)/sum(w):5.2f}% | {np.mean(pass_ms):9.0f} / {max(pass_ms):4.0f} | {sum(p > 1000 for p in pass_ms):12d} | "
f"{np.median(lat):12.2f} / {np.percentile(lat, 90):4.2f} s | {100*tail/allw:17.1f}% | {np.mean(rs):+.2f}")
c = paired_contrast(list(range(len(SESSIONS))), stream_err, words, off_err, "stream, 1 s chunks", "one utterance at a time")
print(f"\nstream (1 s chunks, 8 sessions) minus utterance-by-utterance: {c['diff']:+.2f} pts (95% session-cluster CI {c['lo']:+.2f}..{c['hi']:+.2f}, "
f"{c['clusters']} sessions) -> {'no significant difference' if c['lo'] <= 0 <= c['hi'] else 'a real gap'} at this n")
print(" pass_ms is the engine time per chunk; below the chunk length means real time. Commit latency is the stream time")
print(" at which a word was committed minus the word's end: each chunk re-transcribes the recent audio and commits what two")
print(" passes agree on, so the wait scales with the chunk. r is the correlation of pass time with stream time (buffer growth).")
print(" The three chunk sizes are compared on the same two sessions; the eight-session row is the headline for 1 s chunks.")
a8, a1, a05, a2 = summary[(1.0, len(SESSIONS))], summary[(1.0, 2)], summary[(0.5, 2)], summary[(2.0, 2)]
return (f"1 s chunks on 8 sessions: WER {a8[0]:.2f}% vs {100*sum(off_err)/sum(words):.2f}% utterance by utterance, median commit {a8[1]:.2f} s; "
f"on 2 sessions the median commit is {a05[1]:.2f} / {a1[1]:.2f} / {a2[1]:.2f} s for 0.5 / 1 / 2 s chunks")
streaming()
stream()을 통한 스트리밍은 최근 오디오를 다시 전사하며, 연속 두 번의 패스가 일치하면 단어를 확정(커밋)하므로 대기 시간이 청크 크기에 비례해 늘어납니다. 발화들을 이어 붙여 만든 동일 화자 8개 세션을 1초 청크로 스트리밍하면 WER이 10.05%인 반면, 동일한 발화들을 하나씩 전사할 때는 10.41%로 유의미하지 않은 차이입니다. 단어는 끝난 후 중앙값 1.68초 뒤에 확정되며, 두 세션에서는 중앙값이 0.5초 청크에서 0.78초에서 2초 청크에서 3.44초로 상승합니다. 이어 붙인 부분은 인위적이므로 이는 대화가 아니라 메커니즘을 시험하는 것입니다.
Copy Codedef auroc(scores, labels):
"""Probability that a random positive scores above a random negative (Mann-Whitney)."""
pos = [s for s, l in zip(scores, labels) if l]
neg = [s for s, l in zip(scores, labels) if not l]
if not pos or not neg:
return float("nan")
order = sorted((s, i) for i, s in enumerate(scores))
ranks, i = {}, 0
while i < len(order):
j = i
while j + 1 < len(order) and order[j + 1][0] == order[i][0]:
j += 1
for k in range(i, j + 1):
ranks[order[k][1]] = (i + j) / 2 + 1
i = j + 1
pos_idx = [i for i, l in enumerate(labels) if l]
return (sum(ranks[i] for i in pos_idx) - len(pos) * (len(pos) + 1) / 2) / (len(pos) * len(neg))
@section("7. Word timestamps and word probabilities: consistency, a known shift, agreement with Whisper, and what p < 0.7 means")
def timestamps():
import difflib
pk = EN["other"][:40]
key = lambda w: re.sub(r"[^a-z']", "", w.lower())
diffs, same, mono, shifts = [], 0, 0, []
probs, correct, durs = [], [], []
replay_hits = replay_n = 0
for i, c in enumerate(pk):
ws = W.transcribe(c["audio"], language="en", word_timestamps=True)["words"]
mono += all(a["start"] <= a["end"] <= b["start"] + 1e-6 for a, b in zip(ws, ws[1:]))
if i < 20: # a known shift: 2.000 s of zeros in front should move every word by 2.000 s
ws2 = W.transcribe(np.concatenate([np.zeros(2 * SR, np.float32), c["audio"]]), language="en", word_timestamps=True)["words"]
if ws2 and ws:
shifts.append(ws2[0]["start"] - ws[0]["start"])
if i < 20 and len(ws) >= 6: # cut the audio where word k ends and transcribe only the cut
k = len(ws) // 2
got = NORM_EN(W.transcribe(c["audio"][:int(ws[k]["end"] * SR)], language="en")["text"]).split()
exp = NORM_EN(" ".join(w["word"] for w in ws[:k + 1])).split()
replay_n += 1
replay_hits += got == exp
torch.manual_seed(0)
WHISPER_SECONDS["base"] += c["dur"]
ww = [w for s in whisper_model("base").transcribe(c["audio"], language="en", fp16=False, word_timestamps=True)["segments"] for w in s["words"]]
if [key(w["word"]) for w in ws] == [key(w["word"]) for w in ww]:
same += 1
diffs += [x["start"] - y["start"] for x, y in zip(ws, ww)] + [x["end"] - y["end"] for x, y in zip(ws, ww)]
H, tp, td = [], [], []
for w in ws:
for t in NORM_EN(w["word"]).split(): # "hmm" normalises to nothing, "didn't" to two tokens, as in the reference
H.append(t)
tp.append(w["probability"])
td.append(w["end"] - w["start"])
ok = [False] * len(H)
for tag, i1, i2, j1, j2 in difflib.SequenceMatcher(a=c["ref"].split(), b=H).get_opcodes():
if tag == "equal":
for j in range(j1, j2):
ok[j] = True
probs += tp
correct += ok
durs += td
diffs, shifts = np.abs(diffs), np.array(shifts)
print(f"{len(pk)} test-other clips: word times monotonic on {mono}/{len(pk)}")
print(f"2.000 s of silence prepended on 20 clips: first word moved by {np.median(shifts):.3f} s (median), within 50 ms of 2.000 on "
f"{100*np.mean(np.abs(shifts-2) <= 0.05):.0f}% of clips")
print(f"cut-and-replay: cutting the audio at a word's 'end' and transcribing the cut reproduced the exact word prefix on {replay_hits}/{replay_n} clips")
print(f"agreement with Whisper base's DTW word times, on the {same}/{len(pk)} clips where both emit the same words ({len(diffs)//2} words; a selected, easier subset):")
print(f" |boundary difference| median {np.median(diffs):.3f} s, within 100 ms {100*np.mean(diffs <= 0.1):.0f}%, within 200 ms "
f"{100*np.mean(diffs <= 0.2):.0f}% (agreement between two systems, not accuracy: there is no gold alignment)")
probs, correct, durs = np.array(probs), np.array(correct), np.array(durs)
print(f"\nword probability over {len(probs)} normalised tokens from {len(pk)} clips: median {np.median(probs):.3f}, 10th percentile {np.percentile(probs, 10):.3f}")
for lo, hi in ((0, 0.5), (0.5, 0.7), (0.7, 0.9), (0.9, 1.01)):
m = (probs >= lo) & (probs < hi)
if m.sum():
print(f" p in [{lo:.1f}, {min(hi, 1):.1f}): {m.sum():4d} words, {100*(~correct[m]).mean():5.1f}% wrong")
print(f" AUROC for 'this word is wrong': 1 - probability {auroc(-probs, ~correct):.3f}; word duration as a baseline (shorter = riskier) {auroc(-durs, ~correct):.3f}")
low, high = probs < 0.7, probs >= 0.7
return (f"shift {np.median(shifts):.2f} s; within 200 ms of Whisper on {100*np.mean(diffs <= 0.2):.0f}% of {len(diffs)//2} matched words; "
f"p<0.7 words wrong {100*(~correct[low]).mean():.0f}% vs {100*(~correct[high]).mean():.1f}%")
timestamps()
정답(gold) 정렬이 없으므로, 타이밍이 반드시 충족해야 하는 조건과 비교하여 검증합니다. 확인한 모든 클립에서 단어 시간은 단조 증가하며, 20개 클립 앞에 2초 무음을 추가하면 첫 단어가 중앙값 2.000초만큼 이동합니다. 단어의 끝에서 잘라내면 그 접두부가 재현되어야 하는데, 이는 20개 클립 중 5분의 1에서만 성립합니다. 이는 종료 시각의 정확성이 아니라 안정성을 시험하는 것입니다. 확률이 더 강한 신호입니다. 0.5 미만으로 평가된 단어는 69.2%의 경우 틀렸던 반면, 0.9 이상에서는 2.5%였고, 틀린 단어에 대한 AUROC는 0.916인 반면 길이에 대해서는 0.509였습니다.
Copy Code@section("8. Keyword biasing: an oracle ceiling, a matched wrong-keyword control and a distractor list")
def keywords():
import difflib
clips = EN["other"]
base = HYP[("other", "Whistle")]
freq = collections.Counter(w for c in clips for w in c["ref"].split())
def error_words(ref, hyp):
out = []
for tag, i1, i2, j1, j2 in difflib.SequenceMatcher(a=ref.split(), b=NORM_EN(hyp).split()).get_opcodes():
if tag in ("replace", "delete"):
out += ref.split()[i1:i2]
return sorted(set(out))
oracle = [error_words(c["ref"], h) for c, h in zip(clips, base)]
idx = [i for i, k in enumerate(oracle) if k]
print(f"{len(idx)}/{len(clips)} test-other clips have at least one Whistle error ({sum(len(oracle[i]) for i in idx)} distinct wrong words)")
print(" oracle keywords = the reference words Whistle got wrong on that clip (a CEILING: it reads the answer)")
print(" matched wrong = the next error clip's oracle list, minus every word the target clip actually says")
print(" distractor (8) = the 8 most frequent reference words of the whole sample that the target clip does not say")
print(" distractor (2) = the same list cut to 2 words, so its size matches the oracle and matched arms\n")
def matched_wrong(i):
donor = oracle[idx[(idx.index(i) + 1) % len(idx)]]
return [w for w in donor if w not in clips[i]["ref"].split()] or None
arms = {}
distractor = lambda i, k: [w for w, _ in freq.most_common(60) if w not in clips[i]["ref"].split()][:k]
for name, keylist in (("no keywords", lambda i: None), ("oracle (ceiling)", lambda i: oracle[i]), ("matched wrong", matched_wrong),
("distractor (8)", lambda i: distractor(i, 8)), ("distractor (2)", lambda i: distractor(i, 2))):
errs, words, ins, recovered, total_kw, bad_kw, bad_kw_base, n_kw = [], [], 0, 0, 0, 0, 0, 0
for i in idx:
c = clips[i]
kws = keylist(i)
n_kw += len(kws or [])
h = base[i] if name == "no keywords" else W.transcribe(c["audio"], language="en", keywords=kws)["text"]
s, d, ins_i = sdi(c["ref"], NORM_EN(h))
errs.append(s + d + ins_i)
words.append(len(c["ref"].split()))
ins += ins_i
still_wrong = set(error_words(c["ref"], h))
recovered += sum(w not in still_wrong for w in oracle[i])
total_kw += len(oracle[i])
if kws and name != "oracle (ceiling)":
bad_kw += sum(w in NORM_EN(h).split() for w in kws)
bad_kw_base += sum(w in NORM_EN(base[i]).split() for w in kws)
arms[name] = (errs, words)
extra = f" wrong keywords in the output {bad_kw:3d} (already {bad_kw_base} without biasing)" if name not in ("no keywords", "oracle (ceiling)") else ""
print(f" {name:18s} {n_kw/len(idx):4.1f} keywords/clip WER {corpus_wer(errs, words):6.2f}% insertions {ins:3d} oracle words now right {recovered:3d}/{total_kw}{extra}")
spk = [clips[i]["spk"] for i in idx]
w = arms["no keywords"][1]
fam = []
for a, b in (("no keywords", "oracle (ceiling)"), ("matched wrong", "no keywords"), ("distractor (8)", "no keywords"),
("distractor (2)", "no keywords"), ("matched wrong", "oracle (ceiling)")):
c = paired_contrast(spk, arms[a][0], w, arms[b][0], a, b)
fam.append(c)
sig = holm([c["p"] for c in fam])
print(f"\n family B, {len(fam)} contrasts, Holm at 0.05:")
for c, s in zip(fam, sig):
print(f" WER({c['a']}) - WER({c['b']}) = {fmt_contrast(c)} p~{c['p']:.3f} {'significant' if s else 'not significant'}")
print(" the decisive contrast is matched wrong minus oracle: how much of the gain needs the right words, not just any words")
return (f"oracle ceiling {corpus_wer(*arms['oracle (ceiling)']):.2f}% vs {corpus_wer(*arms['no keywords']):.2f}% baseline; "
f"wrong keywords {corpus_wer(*arms['matched wrong']):.2f}%")
keywords()
키워드 편향(biasing)에는 대조군이 필요한데, 정답을 읽어내는 오라클은 성공처럼 보이기 때문입니다. 오라클은 Whistle이 틀린 단어들을 나열하고, 매칭된 대조군은 다른 클립의 틀린 단어를 사용하되 이 클립에서 말하는 단어는 제외합니다. 오류가 있는 185개 클립에서 오라클은 WER을 14.74%에서 7.62%로 낮추는 반면, 매칭된 틀린 단어는 14.82%로 유의미하지 않은 변화를 보입니다. 클립에 없는 자주 쓰이는 단어 8개를 추가하면 WER이 2.05포인트 상승합니다. 오라클에 비해 매칭된 틀린 단어는 7.20포인트 더 나쁘며, Holm 보정 후 유의미하므로 올바른 단어가 중요합니다.
Copy CodeMLS = {"de": "german", "fr": "french", "es": "spanish", "it": "italian", "nl": "dutch", "pl": "polish"}
WHISTLE_LANGS = ("en", "de", "fr", "es", "it", "nl", "pl")
N_PER_LANG = 12
PUBLISHED_MLS_BASE = {"nl": 28.4, "fr": 26.6, "de": 17.7, "it": 31.1, "pl": 22.8, "es": 12.8} # Whisper paper, table 10 (full splits)
def clopper_pearson_lower(k, n, alpha=0.05):
"""Exact one-sided 95% lower bound for a binomial proportion, by bisection on the binomial tail."""
lo, hi = 0.0, 1.0
for _ in range(60):
p = (lo + hi) / 2
tail = sum(math.comb(n, j) * p ** j * (1 - p) ** (n - j) for j in range(k, n + 1))
lo, hi = (p, hi) if tail < alpha else (lo, p)
return lo
@section("9. Six more languages on Multilingual LibriSpeech, pooled, with a symmetric language-ID test")
def multilingual():
rng = random.Random(SEED)
print(f"sampling rule, fixed in advance: {N_PER_LANG} clips per language from 3 random pages of 100 rows of each test split (seed {SEED});")
print("MLS pages are speaker-sorted, so the pages are the first stage of the sample. Normaliser: Whisper's BasicTextNormalizer on both sides.\n")
pooled = {"Whistle (forced)": [], "Whistle (detected)": [], "whisper base, greedy": []}
words, clusters, lid_w, lid_b, rows_out = [], [], 0, 0, []
wb = whisper_model("base")
for code, cfg in MLS.items():
total = page("facebook/multilingual_librispeech", cfg, "test", 0, 1)["num_rows_total"]
offsets = rng.sample(range(0, total, 100), 3)
rows = [r["row"] for off in offsets for r in page("facebook/multilingual_librispeech", cfg, "test", off)["rows"]]
sample = rng.sample(rows, N_PER_LANG)
e, n_words, spk = collections.defaultdict(list), [], []
for r in sample:
audio, h = load_audio(r)
MANIFEST.setdefault("mls", {}).setdefault(code, []).append([r["id"], h])
ref = NORM_BASIC(r["transcript"])
forced = W.transcribe(audio, language=code)["text"]
det = W.transcribe(audio)
lid_w += det["language"] == code
hb, _, _ = run_whisper("base", audio, language=code)
mel = whisper.log_mel_spectrogram(whisper.pad_or_trim(audio), n_mels=wb.dims.n_mels)
_, probs = wb.detect_language(mel)
lid_b += max(WHISTLE_LANGS, key=lambda l: probs.get(l, 0)) == code
for name, h in (("Whistle (forced)", forced), ("Whistle (detected)", det["text"]), ("whisper base, greedy", hb)):
e[name].append(wer_counts(ref, NORM_BASIC(h))[0])
n_words.append(len(ref.split()))
spk.append(f"{code}-{r['speaker_id']}")
for name in pooled:
pooled[name] += e[name]
words += n_words
clusters += spk
rows_out.append((code, len(sample), len(set(spk)), sum(n_words), corpus_wer(e["Whistle (forced)"], n_words), corpus_wer(e["whisper base, greedy"], n_words)))
print(f"{'lang':>4s} {'clips':>5s} {'spk':>3s} {'words':>5s} | {'Whistle':>8s} {'base':>8s} | Whisper base published on the full split (paper table 10)")
for code, n, s, nw, ww, wbase in rows_out:
print(f"{code:>4s} {n:5d} {s:3d} {nw:5d} | {ww:7.1f}% {wbase:7.1f}% | {PUBLISHED_MLS_BASE[code]}")
print(" per-language rows are descriptive: 12 clips from a handful of speakers cannot rank the systems within a language\n")
c = paired_contrast(clusters, pooled["whisper base, greedy"], words, pooled["Whistle (forced)"], "whisper base, greedy", "Whistle")
c["name"] = "MLS pooled: Whistle vs whisper base, greedy"
CONTRASTS.append(c)
print(f"pooled over {len(words)} clips / {c['clusters']} speakers / {sum(words)} words: Whistle (forced) {corpus_wer(pooled['Whistle (forced)'], words):.2f}%, "
f"whisper base greedy {corpus_wer(pooled['whisper base, greedy'], words):.2f}%; {fmt_contrast(c)} [family A]")
print(f"Whistle with the language detected instead of forced: {corpus_wer(pooled['Whistle (detected)'], words):.2f}%")
n = len(words)
print(f"language identification over the same seven labels: Whistle {lid_w}/{n} (one-sided 95% lower bound {100*clopper_pearson_lower(lid_w, n):.0f}%, "
f"treating clips as independent), whisper base {lid_b}/{n} (lower bound {100*clopper_pearson_lower(lid_b, n):.0f}%)")
json.dump(MANIFEST, open("manifest.json", "w"))
fam = [x for x in CONTRASTS if x["name"] in FAMILY_A]
sig = dict(zip([x["name"] for x in fam], holm([x["p"] for x in fam])))
print(f"\nfamily A complete ({len(fam)} contrasts), Holm at 0.05 (the printed CIs are unadjusted; the verdict uses the adjusted p):")
for x in fam:
v = verdict(x, sig[x["name"]])
VERDICTS.append((x["name"], fmt_contrast(x), v))
print(f" {x['name']:48s} {fmt_contrast(x)} p~{x['p']:.3f} -> {v}")
return f"pooled WER Whistle {corpus_wer(pooled['Whistle (forced)'], words):.1f}% vs base {corpus_wer(pooled['whisper base, greedy'], words):.1f}%; language ID {lid_w}/{n} vs {lid_b}/{n}"
multilingual()
실제 언어를 강제로 지정하면 언어 감지 실패가 숨겨지므로, 각 시스템은 언어도 감지해야 합니다. 사전에 고정된 세 개의 무작위 페이지에서 언어당 12개 클립을 추출하고, 양쪽 모두 BasicTextNormalizer로 채점합니다. MLS 개정판(revision)을 고정하지 않으므로 대신 각 클립을 해시합니다. 22명 화자의 72개 클립을 합산하면, 언어를 강제 지정한 Whistle은 26.40%를 기록한 반면 greedy Whisper base는 24.06%로, Whisper에 유리한 이 격차는 유의미하지 않습니다. 두 시스템 모두 72개 클립 전부에서 언어를 감지했습니다. 언어별 행은 설명적일 뿐입니다.
Copy Code@section("10. Non-speech, quiet speech and white noise at three SNRs, Whistle against Whisper base")
def noise():
rng = np.random.default_rng(SEED)
quiet = EN["other"][0]["audio"] * (10 ** (-30 / 20) / np.sqrt(np.mean(EN["other"][0]["audio"] ** 2)))
probes = {"5 s digital silence": np.zeros(5 * SR, np.float32),
"white noise, -40 dBFS": (rng.standard_normal(5 * SR) * 0.01).astype(np.float32),
"white noise, -20 dBFS": (rng.standard_normal(5 * SR) * 0.1).astype(np.float32),
"440 Hz tone, -20 dBFS": (0.1 * np.sqrt(2) * np.sin(2 * np.pi * 440 * np.arange(5 * SR) / SR)).astype(np.float32),
"speech scaled to -30 dBFS": quiet.astype(np.float32)}
print(f"{'input':>26s} | Whistle text / language | Whisper base text")
for name, x in probes.items():
w = W.transcribe(x, language="en")
b = run_whisper("base", x)[0].strip()
print(f"{name:>26s} | {w['text']!r} {w['language']!r} | {b!r}")
print(f" reference for the quiet clip: {EN['other'][0]['ref']}")
print(" an empty Whisper output is its no-speech gate, not evidence of quality; an empty Whistle output on real speech would be a miss\n")
clips = EN["other"][:20]
spk = [c["spk"] for c in clips]
words = [len(c["ref"].split()) for c in clips]
print(f"white noise on {len(clips)} test-other clips, {sum(words)} words, {len(set(spk))} speakers (directional at this n):")
print(f"{'condition':>14s} | {'Whistle':>8s} {'base greedy':>11s} | WER(base) - WER(Whistle), 95% speaker-cluster CI")
out = []
for snr in (None, 20, 10, 5):
xs = []
for c in clips:
if snr is None:
xs.append(c["audio"])
else:
n = rng.standard_normal(len(c["audio"])).astype(np.float32)
n *= np.sqrt((c["audio"] ** 2).mean() / ((n ** 2).mean() * 10 ** (snr / 10)))
y = c["audio"] + n
xs.append((y * min(1, 0.99 / np.abs(y).max())).astype(np.float32))
e_w = [wer_counts(c["ref"], NORM_EN(W.transcribe(x, language="en")["text"]))[0] for c, x in zip(clips, xs)]
e_b = [wer_counts(c["ref"], NORM_EN(run_whisper("base", x)[0]))[0] for c, x in zip(clips, xs)]
c = paired_contrast(spk, e_b, words, e_w, "base", "Whistle")
label = "clean" if snr is None else f"{snr} dB SNR"
out.append((label, corpus_wer(e_w, words), corpus_wer(e_b, words)))
print(f"{label:>14s} | {corpus_wer(e_w, words):7.1f}% {corpus_wer(e_b, words):10.1f}% | {c['diff']:+6.1f} ({c['lo']:+.1f}..{c['hi']:+.1f})")
print("\n Whistle's front end keeps roughly 250-3500 Hz while Whisper's covers 0-8 kHz, so the two systems do not see the same")
print(" noise; this compares them as shipped, and at n = 20 none of the differences is distinguishable from zero.")
return "; ".join(f"{l}: {w:.0f}% vs {b:.0f}%" for l, w, b in out[1:])
noise()
디지털 무음, 두 가지 소음 수준, 순수 톤 모두 빈 텍스트를 반환합니다. Whisper의 경우 이는 음성 없음(no-speech) 게이트이지 품질의 증거가 아니며, 음성 입력에서 Whistle이 빈 출력을 내는 것은 실패가 될 것입니다. -30 dBFS의 실제 문장은 Whistle에서 단어 하나까지 정확히 반환되지만, Whisper base는 'traced'를 'crazed'로 씁니다. 테스트한 가장 가혹한 SNR에서 백색 소음은 20개 test-other 클립의 WER을 Whistle의 경우 12.4%에서 37.1%로, Whisper base의 경우 15.2%에서 39.9%로 올리고, 어떤 격차도 0과 구별되지 않습니다.
Copy Code@section("11. Speech embeddings: the measured frame rate, and finding a spoken word inside another clip")
def embeddings():
frames = {}
for secs in (1, 10, 30):
v = np.asarray(W.embed(audio_seconds(EN["other"], secs)), np.float32)
frames[secs] = len(v) // 256
print(f"{secs:2d} s of audio -> {len(v):7d} floats = {len(v)//256:4d} frames x 256 ({len(v)/256/secs:.1f} frames/s, {1000*secs/(len(v)/256):.1f} ms per frame)")
fps = frames[10] / 10
try:
W.embed(np.zeros(31 * SR, np.float32))
except RuntimeError as e:
print(f"31 s -> RuntimeError: {e}")
print("\nspoken-term search: embed clip A whole, keep the frames under one word's own timestamps as the query (the encoder is")
print("non-causal, so those frames carry context), slide the query over clip B's frames with a cosine score. Pool: the first 120")
print("test-other clips; words of 5+ letters and 0.25 s+, first two occurrences in different clips; a hit is a peak within 0.3 s.")
clips = [c for c in EN["other"][:120]]
key = lambda w: re.sub(r"[^a-z']", "", w.lower())
occurrences = collections.defaultdict(list)
for i, c in enumerate(clips):
for w in W.transcribe(c["audio"], language="en", word_timestamps=True)["words"]:
if len(key(w["word"])) >= 5 and w["end"] - w["start"] >= 0.25:
occurrences[key(w["word"])].append((i, w["start"], w["end"]))
pairs = [(k, o[0], o[1]) for k, o in occurrences.items() if len(o) >= 2 and o[0][0] != o[1][0]]
cache = {}
def frames_of(i):
if i not in cache:
v = np.asarray(W.embed(clips[i]["audio"]), np.float32).reshape(-1, 256)
cache[i] = v / (np.linalg.norm(v, axis=1, keepdims=True) + 1e-8)
return cache[i]
hits = n = 0
chance, tol = [], 0.3
for k, (ia, sa, ea), (ib, sb, eb) in pairs[:60]:
q = frames_of(ia)[int(sa * fps):max(int(sa * fps) + 2, int(ea * fps))]
T = frames_of(ib)
if len(T) <= len(q):
continue
scores = np.array([(q * T[j:j + len(q)]).sum(1).mean() for j in range(len(T) - len(q) + 1)])
peak = (scores.argmax() + len(q) / 2) / fps
hit = sb - tol <= peak <= eb + tol
hits += hit
n += 1
chance.append((eb - sb + 2 * tol) / clips[ib]["dur"])
if n <= 3:
print(f" {k!r:14s} query {ea-sa:.2f} s from clip {ia:3d} -> peak at {peak:5.2f} s in clip {ib:3d}; the word is spoken at {sb:.2f}-{eb:.2f} s: {'hit' if hit else 'miss'}")
print(f" {hits}/{n} peaks inside the target word (+-{tol} s); a random position would land there about {100*np.mean(chance):.0f}% of the time")
print(" the encoder output is a content representation at frame level; it is not a speaker or sentence embedding")
return f"{fps:.1f} frames/s x 256 floats; spoken-term search {hits}/{n} hits (chance ~{100*np.mean(chance):.0f}%)"
embeddings()
검색이 프레임을 초 단위로 매핑하므로 프레임 레이트를 측정합니다. 10초 클립은 초당 25개의 프레임을 산출하며 각 프레임은 256-부동소수점 벡터이고, 임베딩 한도를 넘는 클립은 RuntimeError를 발생시킵니다. 클립의 프레임은 양쪽을 모두 보기 때문에 클립 전체를 한 번에 임베딩하고, 한 단어의 프레임들을 다른 클립 위에서 코사인 점수로 슬라이드합니다. 5글자 이상인 34쌍의 단어 중 32쌍에서 피크가 단어의 0.3초 이내에 떨어지며, 우연이라면 약 17%입니다. 출력은 프레임 수준의 내용이지 문장 임베딩이 아닙니다.
Copy CodeINTENTS = {"abroad": "Using the card or account while travelling or abroad",
"address": "Change, update or confirm the postal address on the account",
"app_error": "The banking app or website shows an error or does not work",
"atm_limit": "The daily ATM cash withdrawal limit",
"balance": "Check the current account balance",
"business_loan": "Apply for or ask about a business loan",
"card_issues": "A card is lost, stolen, blocked, declined or not working",
"cash_deposit": "Deposit or transfer cash into the account",
"direct_debit": "Set up, change or cancel a direct debit or standing order",
"freeze": "Freeze, block or unfreeze the account or card",
"high_value_payment": "Make or authorize a large or high-value payment or transfer",
"joint_account": "Open or ask about a joint account with another person",
"latest_transactions": "See the latest or recent transactions and payments",
"pay_bill": "Pay a bill or make a payment to a company"}
N_MINDS = 200
@section("12. From speech to a tool call: Whistle transcripts into Needle 3 on MInDS-14 banking requests")
def speech_to_tools():
tools = [{"name": n, "description": d, "parameters": {"type": "object", "properties": {}}} for n, d in INTENTS.items()]
print(f"14 tools, one per MInDS-14 intent, descriptions fixed before the first call (sha256 {hashlib.sha256(json.dumps(tools).encode()).hexdigest()[:12]})")
first = page("PolyAI/minds14", "en-US", "train", 0, 1)
names = next(f for f in first["features"] if f["name"] == "intent_class")["type"]["names"]
assert names == list(INTENTS), names
rows = all_rows("PolyAI/minds14", "en-US", "train")
sample = random.Random(SEED).sample(rows, N_MINDS)
clips, too_long = [], 0
for r in sample:
audio, h = load_audio(r) # 8 kHz telephone wav, resampled to 16 kHz by load_audio
MANIFEST.setdefault("minds14", []).append([r["path"], h])
if len(audio) / SR > MAX_CLIP_S: # the same pre-declared filter as step 2: the engine refuses longer clips
too_long += 1
continue
clips.append({"audio": audio, "ref": r["transcription"], "intent": names[r["intent_class"]], "dur": len(audio) / SR})
t0 = time.perf_counter()
hyps = [W.transcribe(c["audio"], language="en")["text"] for c in clips]
tw = time.perf_counter() - t0
e, n = zip(*[wer_counts(NORM_EN(c["ref"]), NORM_EN(h)) for c, h in zip(clips, hyps)])
json.dump(MANIFEST, open("manifest.json", "w"))
print(f"{len(clips)} of the {N_MINDS} sampled clips ({len(rows)} rows in the split; {too_long} dropped for exceeding {MAX_CLIP_S:.0f} s; "
f"{sum(c['dur'] for c in clips)/60:.1f} min of 8 kHz telephone speech): Whistle WER {corpus_wer(e, n):.1f}% "
f"against the casual MInDS-14 transcriptions, {1000*tw/len(clips):.0f} ms per clip")
for c, h in list(zip(clips, hyps))[:2]:
print(f" [{c['intent']}] ref: {c['ref'][:72]}\n {' ' * (len(c['intent']) + 2)} hyp: {h[:72]}")
t0 = time.perf_counter()
agent = needle.Needle(tools=tools, auto_date=False) # no date fact in the prompt, so the result does not depend on the clock
print(f"\nNeedle 3 loaded in {time.perf_counter()-t0:.1f} s into the same engine; Whistle stays loaded beside it")
def route(texts):
out = []
for t in texts:
r = agent.complete(t)
agent.reset()
calls = [c["name"] for c in r.get("function_calls", [])]
out.append((calls[0] if calls else None, r.get("confidence") or 0.0, r.get("peak_ram_mb")))
return out
truth = [c["intent"] for c in clips]
res = {}
for label, texts in (("reference transcripts", [c["ref"] for c in clips]), ("Whistle transcripts", hyps)):
t0 = time.perf_counter()
res[label] = route(texts)
right = np.array([p == y for (p, _, _), y in zip(res[label], truth)])
conf = np.array([c for _, c, _ in res[label]])
print(f" routing from {label:21s}: {100*right.mean():.1f}% correct, {sum(p is None for p, _, _ in res[label])} refusals, "
f"{1000*(time.perf_counter()-t0)/len(texts):.0f} ms/call, confidence when right {conf[right].mean():.2f} vs wrong {conf[~right].mean():.2f}")
a = [p == y for (p, _, _), y in zip(res["reference transcripts"], truth)]
b = [p == y for (p, _, _), y in zip(res["Whistle transcripts"], truth)]
only_ref, only_wh = sum(x and not y for x, y in zip(a, b)), sum(y and not x for x, y in zip(a, b))
m = only_ref + only_wh
p_mc = min(1.0, 2 * sum(math.comb(m, j) for j in range(min(only_ref, only_wh) + 1)) / 2 ** m) if m else 1.0
print(f" discordant clips: right only from the reference text {only_ref}, right only from Whistle's text {only_wh} (exact McNemar p = {p_mc:.2f})")
conf = collections.Counter((y, p) for (p, _, _), y in zip(res["Whistle transcripts"], truth) if p != y)
print(" most common confusions (truth -> call): " + ", ".join(f"{y} -> {p} x{k}" for (y, p), k in conf.most_common(3)))
off = route([NORM_EN(h) for h in HYP[("other", "Whistle")][:40]])
print(f" 40 LibriSpeech sentences (off-topic): {sum(p is None for p, _, _ in off)} refused, {sum(p is not None for p, _, _ in off)} given a banking tool call")
print(f" peak_ram_mb reported by Needle: {res['Whistle transcripts'][-1][2]} MB (this whole process, Whisper models included). MInDS-14 is public,")
print(" so training-set overlap cannot be ruled out for either model, and the tool descriptions are ours: the number moves with them.")
return f"routing {100*np.mean(b):.0f}% right from Whistle's text vs {100*np.mean(a):.0f}% from the reference text; WER {corpus_wer(e, n):.0f}%"
speech_to_tools()
Whistle은 표본으로 추출한 MInDS-14 뱅킹 요청을 비격식 참조 전사 기준 31.9% WER로 전사합니다. Needle 3는 각 전사를, 뱅킹 인텐트당 하나씩이며 설명을 우리가 작성한, 14개 도구 사이에서 라우팅합니다. 반환된 함수 호출이 라우팅이 되고, 호출 없이 답변하면 거절로 간주합니다. 라우팅은 Whistle의 텍스트에서 50.3%, 참조 텍스트에서 51.8% 정확했으며 유의미하지 않은 격차이므로, 이 설명들로는 라우팅이 한계이지 전사가 아닙니다. 주제에서 벗어난 LibriSpeech 문장 중 17개는 거절되었고 나머지는 뱅킹 도구 호출을 받았습니다.
Copy Codebanner("SUMMARY")
for name, res in RESULTS.items():
print(f" {name:<104s} {res}")
print("\nVerdicts on the published-claim contrasts (family A, 95% speaker-cluster paired bootstrap, unadjusted CIs, Holm over 5):")
for claim, measured, v in VERDICTS:
print(f" {claim:<48s} {measured:<58s} {v}")
print(f"\nWhisper audio transcribed in this run: " + ", ".join(f"{k} {v/60:.1f} min" for k, v in WHISPER_SECONDS.items())
+ f"; hardware {CPU}")
print("""
What to carry over
- Pin the Hub revisions of whistle.cact, the engine wheel and needle3.cact and print their hashes; both repos replace files in
place, and the engine version the package wants is in needle.agent.fetch.ENGINE_VERSIONS, not in the package version.
- Score with the same normalizer on both sides of each comparison, sample with a fixed seed, cluster the bootstrap by speaker.
- Say which Whisper you compared against: greedy (the package default) and beam search (the paper's numbers) differ.
- Keep a single transcribe() call under 30 s; cut longer audio yourself (quiet points worked best on this clip) or use stream().
- Word probability is a usable error signal, keyword biasing has a ceiling, embeddings are content frames.
Where to go next
- pip install "cactus-needle[mic,compare]" then needle whistle compare clip.wav runs one clip through Whistle, Whisper and Moonshine.
- needle download <platform> fetches the engine for macOS, Linux, Android, iOS, RISC-V, MIPS, the browser or WASI; the native
runner takes --model needle3.cact --model whistle.cact --tools tools.json --audio clip.wav for audio-to-tool-calls in one call.
- The weights also ship as safetensors, and fine-tuning for new languages is being discussed in the repository's issues.
- Docs: cactuscompute.com/whistle Code: github.com/cactus-compute/needle Weights: huggingface.co/Cactus-Compute/whistle
""")
요약은 각 섹션의 한 줄 결과를 출력한 뒤, 다섯 개의 공개된 주장에 대한 판정을 보여주는데, greedy Whisper base에 대한 test-other만이 Whistle에 유리합니다. 또한 Whisper가 전사한 오디오, 즉 tiny의 경우 18.4분, base의 경우 68.7분을 보고하며, 13개 셀 모두 실패 없이 실행되었습니다. 이식할 습관은 Hub 개정판 고정, 시드 고정, 하나의 정규화기로 양쪽 채점, Whisper 디코더 명명입니다.
결론적으로, Apple M4 CPU에서 Whistle은 test-other 샘플에서 10.47%를 기록했는데, 이는 발표된 전체 분할 10.49(인용된 수치이지 측정된 값이 아님)에 근접한 결과입니다. 짝지어진 하위 집합에서는 greedy Whisper base보다 2.19포인트 앞섰지만, beam search, test-clean, 통합 MLS에서는 유의미한 차이가 없었습니다. 10초 클립에서 greedy base보다 5.7배 빠르게 실행되었습니다. 1초 청크로 스트리밍할 경우 단어가 끝난 후 중앙값 1.68초 뒤에 확정되었으며, 유의미한 손실은 없었습니다. 30초 제한이 가장 중요합니다: 고정된 엔진은 더 긴 오디오의 중간 부분을 조용히 삭제하므로, 분할하거나 스트리밍해야 합니다. 이것들은 단일 실행 샘플이며, 어떤 단일 클립이나 작은 집합으로도 주장을 확정할 수 없습니다.
다음을 확인해 보세요 전체 코드는 여기에 있습니다. 이 프로젝트의 연구자에게 모든 공이 돌아갑니다. 또한 언제든지 저희를 팔로우해 주세요. 트위터 그리고 저희 ML 서브레딧 150k+ 그리고 구독하세요 저희 뉴스레터. 잠깐! 텔레그램을 쓰고 계신가요? 이제 텔레그램에서도 저희와 함께하실 수 있습니다.
GitHub 저장소, Hugging Face 페이지, 제품 출시, 웨비나 등을 홍보하기 위해 저희와 파트너십이 필요하신가요? 저희와 소통하세요
해당 게시물 Cactus Whistle 코딩 가이드: 16.9 MB 음성 모델은 Whisper Base와 겨룰 만한가? CPU에서의 단어 오류율, 속도, 스트리밍, 타임스탬프, 키워드, 음성-툴 호출 최초 게재: MarkTechPost.