在这个教程中，我们使用Whistle,Cactus Compute的语音转文本模型，一个运行在CPU上的16.9 MB文件，转录七种语言，并返回单词时间戳和概率。 它的型号卡声称击败了LibriSpeech上的Whisper基数；我们用种子的测试干净和测试其他样品来测试它，将Whistle与Whisper微小和基数在相同的剪辑上进行比较，通过配对的扬声器-集群间隔与霍尔姆修正来判断。 我们发现一个领先者超越贪婪的Whisper基在测试 -其他，没有显著的缺口对光束搜索，和速度优势缩小随着剪辑增长。 在标题之外，我们探测到一个30秒窗口，被封住的引擎静悄悄地放下了较长的音频，加上流畅的空档，时间戳，关键词偏差，其他语言，噪音，嵌入，和语音对工具的调用。 我们的数字是Apple M4的单运行样本，公布的全块数字被引用，没有测量。
导入 Os
导入 io
导入重置
导入 sys
导入 Json
导入数学
导入时间
导入随机
导入大麻脂
导入 hipil
导入 zip 文件
导入平台
导入追溯
导入子进程
导入收藏
导入urlib.error
导入urlib。请求
结果 =
VERDICTS = []
def 横幅( 标题):
打印("\n" + "=" * 78)
打印(标题)
打印 ("=" * 78)
def section(name):
fn) 包装：
f 运行( *a, ** kw):
横幅( 名称)
t0 = 时间。 time ()
尝试：
输出 = fn(*a,**kw)
RESULTS[名] = 如果出现( 出，str) 其他"OK"。
返回出来
例外作为e:
RESULTS[name] = f"SKIPPED / FAILED -> {type(e).__name__}: {e}"
打印 (f)\n [!] {name} 未完成：{type (e). name: {e}>
回溯.print exc(限制=3)
返回无
终于：
打印(f)\n [{time.time() - t0: 0f} s] )
json.dump({"结果"):ResULTS,"核查":VERDICTS}，开("结果.json","w")，缩进=1
返回运行
返回包装
横幅 ("1. 安装口哨，通过 Hub 修订版标注其重量和引擎，转录一个剪辑" )
subprocess.run([sys.executable, "-m", "pip", "install", "-q", "cactus-needle==3.2.0", "openai-whisper==20250625",
(原始内容存档于2017-07-02). "jiwer = 4.0.0", "Soundfile = 0.14.0", "soxr = 1.1.0"], check=True).
os.environ ["NEEDLE TELEMTRY"] = "0" # 包裹发送匿名使用量计数，除非被告知不要
os.environ ["HF HUB DISBAL PROGRESS BARS"] = "1" (中文(简体) ).
导入数字为 np
导入声音文件为 sf
导入 soxr
从 huggingface hub 导入 hf hub 下载
# 仙人掌 - Needle 从已替换文件的 Hugging Face Repos 中获取三个关于首次使用的文件：16. 9 MB
# great.cact 权重，~1MB 引擎库(本包第3.3.0版；针3 repo 收到了新的二进制)
#在2026-10-10上，以及第12步中使用的35MB针3.cact. 固定每个节点的修改，并打印散列
# 是什么使得这个运行可以复制。 引擎版本从软件包读取，不输入。
WHISTLE REV = "ca5287601 bef25af26dcf2e1b2bdc0843a7c19e5"#仙人掌-计算机/微信，2026-10-10
NEALE3 REV = "93e88ede867e4311c39151285e74df513e4b835b" 仙人掌-计算/需要3(引擎+针头3重),2026-10-10
从针头中获取。 agent 导入为针脚
TAG = 针头 fetch. platform tag ()
ENGINE VERSION = 针头牵引。 engine version(3)
重量 = hf hub download("Cactus-Compute/whistle"),"whistle.cact"，修订版=WHISTLE REV).
车轮 = hf hub 下載("Cactus-Compute/nectle3"，针形文字.engine 輪(ENGINE VERSION, TAG, 3)，修订=NEEDLE3 REV) 车轮车轮车轮车轮车轮车轮车轮车轮车轮车轮车轮车轮车轮车轮
带拉链文件。 ZipFile( 轮) 作为 z:
成員 = 下一个(n for n in z.namelist () if n. starts with ("Needle/libnedle")).
os. makedirs (“ 引擎 ”，存在 ok= True)
引擎 = os.path.abspath(os.path.join) (“引擎”、os.path.basename(成员))
打开(引擎，“ wb ”). write(z.read(成员))
针头3 重量 = hf hub download("Cactus-Compute/Needle3","Needle3.cact"，修订=NEEDLE3 REV) 针头3 重量针头3 重量 = hf hub download("Cactus-Compute/Needle3","Needle3.cact"，修订=NEEDLE3 REV) 针头针头针头针头针头针头针头针头针头针头针头重量重量 = hf hub 下载量("Cactus-Compute/ Needle3","Needle3.cact"，修订=NEEDLE3 REV)
os.makedirs(需要 fetch.cache dir(3)，存在 ok=True)# Needle(工具=...)从此缓存路径读取其重量
hopil.copyfile(需要3 重量，os.path.join(需要 fetch.cache dir(3)，针 fetch.base 重量(3))).
os.environ ["NEEDLE WHISTLE WEIGHTS"] = 重量 # 两个变量都被包所遵守
os.environ ["NEEDLE3 LIB PATH"]=引擎互联网电影数据库(IMDb)上"NEEDLE3"的资料(英文)
导入针头
导入 importlib.metadata 作为 md
sha = lambda p: hashlib.sha256(open(p,"rb").read()). hexdigiest () [:12] (中文(简体) ).
CPU = 平台。 处理器()或平台。 机器()
如果 sys.plateform = "linux" 和 os.path.sexits ("/proc/cpuinfo") 则：
CPU = next((l.split (":"), 1) [1].strip () for l in open ("/proc/cpuinfo")，如果 l. starts with ("模型名称"), CPU).
elif sys.plateform=中的日期值 (帮助) "Darwin":
CPU = sub process.run(["sysctl","-n","machdep.cpu.brand string"])，抓取 output=True，文本=True.stdout.strip()或CPU.
print(f)"cactus - Needle {nedyle. version }(引擎 {ENGINE VERSION}，轮标 {TAG}) python {plateform.python version()}"
f" soundfile { md.version ('soundfile')} soxr { md.version ('soxr')} numby {np. version } ) (中文(简体) ).
打印(f) 硬件：{CPU},{os.cpu count()} cpus 可见”
print(f)"whistle.cact {os.path.getsize(重量)/1e6:1f}MB sha256 {sha(重量)} @ {WHISTLE REV[:12]}"
打印(f) “ engine {os. path. getsize( engine)/ 1e6: 1f} MB sha256 {sha( engine)} @ {NEEEDLE3 REV[: 12]} )
print(f"needle3.cact {os.path.getsize(needle3_weights)/1e6:.1f} MB sha256 {sha(needle3_weights)} @ {NEEDLE3_REV[:12]}")
SR = 16000
W=针。 Whistle( ) # 每个过程一个模型，不保证线程安全；每一步后都会重用此对象
def 获取( url, trice=7):
""GET带复刻。 数据集 - 服务器费率限制 - 匿名呼叫者(HTTP 429)：荣誉重试 - 当是时
发送，否则退 5,10,20,40... 秒，永远不要放弃第一次拒绝。"
k 在范围( 三次):
尝试：
使用urlib.request.urlopen(url，超时=60)作为r:
返回 r. read ()
除了urlib.error.HTTPError作为e:
如果K ==尝试 -1:
提高
等待 = int(e.headers.get("Retry-After, 0)或 0) 如果 e.code == 429 其它 0
时间.sleep(max(等待，5 * 2 ** k) 如果e.code in (429, 500, 502, 503, 504 else 2) 则使用
除外情形：
如果K ==尝试 -1:
提高
时间。 sleep( 3 * (k + 1))
def 页面( 数据集、 配置、 分割、 偏移、 长度=100):
""一个Hub数据集的一页通过数据集-服务器行 API：元数据加上签名的音频URL".
url = (f"https://datasets-server.huggingface.co/rows?dataset={dataset}&config={config}"
f"和split =split==========================================================================================
tim.sleep( 0. 25) # 拨打整张分页时，保持在匿名率限制下
返回 json.loads(牵引(url))
def 负载 audio( row ):
"""卸下一行的夹子；还原16千赫单浮32样本和原文件的sha256".
数据=获取(row [“adio”][0][“src”])
音频，sr = sf.read(io.BytesIO(data), dtype = "float32")
如果音频.ndim > 1:
音频 = audio.mean(轴=1)
如果有的话！
音频=soxr.resample(音频，sr,SR，质量="HQ").
返回音频，hashlib.sha256(数据). hexdigiest() [:12]
首页=("openslr/librispeech asr","干净","测试",0,1) ["rows"][0]["row"]
剪贴板，= 负载 audio( 第一个)
t0 = 时间.perf counter ()
输出 = W.trancis (clip, word timestamps=True)
壁=(时间.perf counter() - t0) * 1000
print(f)\nclip {first ['id']}{len(clip)/SR: 1f} s 引用：{first ['text'}} )
打印( f) - > {out ['text'!r}!}
print(f) 语言 {out ['language'] 引擎时间 {out [ttft ms']: 1f} ms 解码 {out ['decode tps']:.0f} tok/s]
f" 壁 {wall:.0f} ms] (中文(简体) ).
print("第一字："),[[("词"),w["开始"],w["结束",w["概率"]]，用于w的输出["词"][:3]]].
ResULTS["1. 安装口哨，通过Hub修订版固定其重量和引擎，转录一个片段"]==\
f"{os.path.getsize(重量)/1e6:1f} MB重量，{len( clip)/SR: 1f} {wall: 0f} ms 中的剪辑
我们安装Cactus -neightle 3.2.0 它的依赖性被固定并关闭它的匿名遥测。 软件包下载了Whistle的权重、引擎和Needle 3的权重。 Hugging Face的存储器已经更换了文件，所以我们把两个存储器都固定在SHA-256散列上。 页面()帮助读取数据集-服务器行 API 带速率限制回放，加载 audio() 复标剪辑到 16 kHz. 第一个测试干净的片段，3.5 s的语音，在33ms的墙壁时间中转录，每个词都带有它的起始，结束和概率。
导入 jiwer
导入 whisper
导入火炬
从 whisper. 常态器导入 EnglishTextormalizer, BasicTextormormalizer
NORM EN = EnglishTextNormalizer () # 正常人 Whistle 自己的基准测试用途；适用于两侧，总是
NORM BASIC = BasicTextormalizer( ) () 化学反应器
SEED = 2026年
N WHISTLE = 300 # 剪辑/ LibriSpeech 拆分 Whistle 转录(很便宜)
N PAIRED={其他>:100,"干净":60}# 其中第一个N也经过Whisper(不是)；测试-其他是初级
MAQQCLIP S = 30:00 # Whistle的单行道窗口；较长的片段在此宣布退出范围。
WHISPER SECONDS = 藏书。 计数器( )
def all rows(数据集，配置，拆分):
页：1
而 True 时：
行=页面(数据集，配置，分割，偏移) ["rows"]
[r ['row'] 对于r排
页：1
如果len( rows) < 100:
返回出来
def 数据集 sha( 数据集):
返回 json.loads (f"https://huggingface.co/api/datasets/{dataset}") ["sha"][:12]
数据( ref, hyp):
""错误(S+D+I)和一个正常对的参考词；空假说都是删除的".
n = len(ref.split ()) (中文(简体) ).
如果不是hyp.strip (): (中文(简体) ).
返回n,n
o = jiwer.process words(ref, hyp) (中文(简体) ).
返回 o. 替换 + o. 删除 + o. 插入，n
sdi( ref, hyp) 类型：
如果不是hyp.strip (): (中文(简体) ).
返回 0, len( ref.split ()), 0
o = jiwer.process words(ref, hyp) (中文(简体) ).
替换、删除、插入
defecur wer(errs, words) (英语).
返回 1000.0 * 和( errs) / 最大(和( 单词) 1)
def by 组( 组，*阵列):
tot = 集合.defaultdict(lambda: [0.0] * len(阵列))
i, c 在列表(组):
用于 j, a in listene(阵列):
tot[c][j] += a[i]
返回吨数
def wer ci(集群，错误，词，B=2000，种子=0):
""百分之九十五的间距水体WER，重印集群( 扬声器) 与替换"。"
tot = by cluster(集群、错误、词)
键，rng = 排序(tot)，随机。 随机(种子)
靴子 = []
在范围(B)中：
d = rng.choices(钥匙，k=len(钥匙))
boots.append(100 * sum(tot[c][0] for c in d) / max(sum(tot[c][1] for c in d), 1))
返回 np.percentile(boots, [2.5, 97.5])。
def配对 contrast(集群，错误 a，词，错误 b，标签 a，标签 b,B=2000，种子=0):
""WER(a) - WER(b) 带有对齐的扬声器集群靴子：两个系统相同的重印扬声器". "
tot = by cluster(clusters，错误 a，错误 b，词)
键，rng = 排序(tot)，随机。 随机(种子)
靴子 = []
在范围(B)中：
d = rng.choices(钥匙，k=len(钥匙))
w = 最大值(和(tot[c][2], c in d), 1)
boogle. 附录(100 * (d中c的和(tot[c][0)) -- -- 和(d中c的tot[c][1])/w)
靴子 = np.array(boots)
lo, hi = np.percentile(boots, [2.5, 97.5])
p = 2 * 分钟(( boots Q 0. 0 ), ( boots Q 0. ) )
返回 {'a"：标签 a,"b"：标签 b,"diff":abus wer(errs a，单词) -abus wer(errs b，单词)
"lo":lo,"hi":hi,"p"：最大(p, 1/B),"组":len(键)}
def holm(ps, alpha=0.05):
顺序，保持=np.argsort(ps),[法 * len(ps)
对于军衔，我以列表(顺序)表示：
如果 ps[i] QQ alpha / (len(ps) - 级别):
保持[i]=对
其他情况：
换行
返回保留
(c，重大)
"''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''' CI 完全 > 0 和 Holm- 重大 - > 先吹哨声；
完全 < 0 且重要 - > 后面吹口哨，否则这个样本大小没有显著差别".
如果 c [“ lo] > 0 且重要：
返回"哨声前进"
如果 c [“ hi] < 0 且重要：
返回"哨声后面"
返回“此 n 没有显著差异”
ff fmt contrast(c): (中文(简体) ).
返回 f"{c['diff']:+.2f} pts (95% CI {c['lo']:+.2f}. {c['hi']:+.2f},{c['组']}扬声器)".
@section("2. 固定，种子样本 LibriSpeech 测试-清洁和测试-其他，以及判决规则").
daf 样本英语( ):
全球环境
打印("以下判决的规则，在最后运行前写在本笔记本上"(探索探针在"))
print(“ 相同的分裂是先出现的；流派称是 ” )
打印(f) * 简单随机样本 {N WHISTLE} 每行(种子 {SEED} )；第一个 {N PAIRED ['other'} (试验-其他)和 "
f"{N PAIRED ['clean']] (test-clean)是配对的Whisper子集").
打印(f) * 剪辑长于 {MAQQCLIP S:
本文由 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 Atlas. All rights belong to the original author.