在本教程中，我们使用 MSEB，即来自 Google Research 的 Massive Sound Embedding Benchmark，并从排行榜数字实际意味着什么的角度来切入：评估器表面。我们安装该包并梳理其三个层次，然后针对框架自有的抽象基类编写两个刻意不同的编码器：一个测量随时间变化的响度，一个测量音色，并对我们在 notebook 中生成的小型合成语料库进行编码，这样无需下载任何内容。我们在这些嵌入上驱动分类、聚类、检索和分割评估器，直接调用度量函数以查看每个度量奖励什么，最后组装真实提交所携带的 TaskMetadata。结果是一个对比，其中两个编码器会根据所询问的评估器而互换位置，这正是用数字而非文字为多任务基准测试所做的论证。
复制代码已复制使用其他浏览器
import os
import sys
import json
import math
import traceback
import subprocess
import numpy as np
RESULTS = {}
BENCH = {}
def banner(title):
print("\n" + "=" * 78)
print(title)
print("=" * 78)
def section(name):
def wrap(fn):
def run(*a, **kw):
banner(name)
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
return run
return wrap
banner("0. Install MSEB and map the three layers we will use")
subprocess.run([sys.executable, "-m", "pip", "install", "-q", "mseb==0.1.0"], check=True)
import mseb
from mseb import types, encoder as encoder_lib, evaluator as evaluator_lib, metrics
from mseb.evaluators import (
classification_evaluator,
clustering_evaluator,
retrieval_evaluator,
segmentation_evaluator,
)
print(f" mseb {mseb.__version__} | Python {sys.version.split()[0]} | numpy {np.__version__}")
print("\n MSEB is three layers, and a benchmark run walks down them:")
print(" types -> Sound, SoundEmbedding, Score, TaskMetadata: the shapes every task speaks")
print(" encoder -> MultiModalEncoder: the contract YOUR model implements")
print(" evaluators -> classification, clustering, retrieval, reranking, transcription, segmentation, ...")
print("\n evaluator entry points we will drive:")
for module, cls in [(classification_evaluator, "ClassificationEvaluator"),
(clustering_evaluator, "ClusteringEvaluator"),
(retrieval_evaluator, "RetrievalEvaluator"),
(segmentation_evaluator, "SegmentationEvaluator")]:
print(f" {module.__name__.split('.')[-1]:28s} {cls}")
print("\n Everything below runs on CPU with no dataset download: we synthesise the audio.")
我们安装 mseb，并导入一次基准测试运行会逐层深入的三个层。types 模块保存每个任务都使用的形状，即 Sound、SoundEmbedding、Score 和 TaskMetadata；encoder 模块保存 MultiModalEncoder，即我们自己的模型所实现的契约；evaluators 包则为每个任务族保存一个模块。我们只导入本 notebook 驱动的四个评估器，因为 classification、clustering、retrieval 和 segmentation 模块所依赖的不过是 NumPy 和 scikit-learn。相比之下，reranking 和 transcription 评估器会引入 Whisper，而任务运行器会引入 TensorFlow 和 apache-beam。因此，下面的一切都在免费的 CPU 运行时上运行，无需下载数据集，也无需加速器。
Copy CodeCopiedUse a different Browser
SR = 16000
@section("1. The type contract: Sound, SoundEmbedding, Score")
def type_contract():
t = np.arange(SR) / SR
waveform = (0.5 * np.sin(2 * np.pi * 440 * t)).astype(np.float32)
sound = types.Sound(
waveform=waveform,
context=types.SoundContextParams(id="demo_000", sample_rate=SR, length=len(waveform),
language="en_us", text="a 440 Hz tone"),
)
print(f" Sound id={sound.context.id!r} {sound.waveform.shape} @ {sound.context.sample_rate} Hz"
f" -> {sound.size_bytes:,} bytes")
embedding = types.SoundEmbedding(
embedding=np.zeros((1, 16), dtype=np.float32), # (N, D): one utterance-level vector
timestamps=np.array([[0.0, 1.0]], dtype=np.float32), # (M, 2): [start, end] in seconds
context=sound.context,
encoding_stats=types.EncodingStats(input_size_bytes=sound.size_bytes, embedding_size_bytes=16 * 4),
)
print(f" SoundEmbedding embedding{embedding.embedding.shape} timestamps{embedding.timestamps.shape}"
f" -> {embedding.size_bytes} bytes")
print(f" compression_ratio = {embedding.encoding_stats.compression_ratio:.5f}"
f" ({1 / embedding.encoding_stats.compression_ratio:,.0f}x smaller than the audio)")
print(" N embeddings and M timestamps: M == N is frame-aligned, M == 1 is utterance-level.")
print(" `embedding` may also hold N strings instead of vectors - step 8 uses exactly that.")
score = types.Score(metric="Accuracy", description="Overall classification accuracy",
value=0.875, min=0.0, max=1.0)
print(f"\n Score {score.metric}={score.value} in [{score.min}, {score.max}] :: {score.description}")
for bad, why in [(dict(metric="", description="d", value=0.5, min=0.0, max=1.0), "empty metric name"),
(dict(metric="m", description="d", value=0.5, min=1.0, max=0.0), "min > max")]:
try:
types.Score(**bad)
except Exception as e:
print(f" rejected at construction ({why}): {type(e).__name__}: {e}")
return f"Sound {sound.size_bytes:,} B -> embedding {embedding.size_bytes} B"
type_contract()
我们从类型契约开始，因为其他每一层都用它来表达。一个 Sound 携带一段波形，以及 SoundContextParams、标识符、采样率、长度、语言和可选的转录文本，这些会伴随音频走完整个流水线。一个 SoundEmbedding 携带一个包含 N 个嵌入的数组和一个包含 M 个时间戳对的数组，而 N 与 M 之间的关系就是该基准测试的词汇表：M 等于 N 表示每帧一个向量，而 M 等于 1 表示单个话语级向量，这正是我们的编码器所产出的。EncodingStats 记录输入和嵌入的大小，并暴露 compression_ratio，在这里是从音频到向量的一千倍缩减。一个 Score 是一个指标名称、一个值及其边界，它在构造时自我校验，拒绝空的指标名称或最小值大于最大值，因此格式错误的数字无法进入排行榜。embedding 字段也接受 N 个字符串而非 N 个向量，这正是第 8 步所走过的那扇门。
Copy CodeCopiedUse a different Browser
class EnergyEnvelopeEncoder(encoder_lib.MultiModalEncoder):
"""基线：在 `n_bins` 个等长时间切片中的平均能量。只反映响/轻，与音色无关。"""
def __init__(self, n_bins: int = 16):
super().__init__()
self.n_bins = n_bins
def _setup(self):
self._ready = True # 真正的编码器会在这里加载权重
def _check_input_types(self, batch):
for item in batch:
if not isinstance(item, types.Sound):
raise ValueError(f"{type(self).__name__} 接收 types.Sound，但收到 {type(item).__name__}")
def _encode(self, batch) -> list[types.SoundEmbedding]:
out = []
for sound in batch:
slices = np.array_split(sound.waveform.astype(np.float32), self.n_bins)
vec = np.array([[float(np.sqrt(np.mean(s ** 2) + 1e-12)) for s in slices]], dtype=np.float32)
vec /= np.linalg.norm(vec) + 1e-9
out.append(types.SoundEmbedding(
embedding=vec,
timestamps=np.array([[0.0, sound.context.length / sound.context.sample_rate]], dtype=np.float32),
context=sound.context))
return out
class SpectralProfileEncoder(encoder_lib.MultiModalEncoder):
"""竞争方案：将平均对数幅度谱池化到 `n_bands` 个频带。描述音色。"""
def __init__(self, n_bands: int = 16, frame: int = 512):
super().__init__()
self.n_bands, self.frame = n_bands, frame
def _setup(self):
self._window = np.hanning(self.frame).astype(np.float32)
def _check_input_types(self, batch):
for item in batch:
if not isinstance(item, types.Sound):
raise ValueError(f"{type(self).__name__} 接收 types.Sound，但收到 {type(item).__name__}")
def _encode(self, batch) -> list[types.SoundEmbedding]:
out = []
for sound in batch:
w = sound.waveform.astype(np.float32)
n_frames = max(1, len(w) // self.frame)
spectra = [np.abs(np.fft.rfft(w[i * self.frame:(i + 1) * self.frame] * self._window))
for i in range(n_frames)]
mean_spectrum = np.log1p(np.mean(spectra, axis=0))
vec = np.array([[float(b.mean()) for b in np.array_split(mean_spectrum, self.n_bands)]],
dtype=np.float32)
vec /= np.linalg.norm(vec) + 1e-9
out.append(types.SoundEmbedding(
embedding=vec,
timestamps=np.array([[0.0, sound.context.length / sound.context.sample_rate]], dtype=np.float32),
context=sound.context))
return out
@section("2. 编码器契约：三个方法，其余交给框架")
def encoder_contract():
print(" MultiModalEncoder 抽象方法，子类必须实现：")
for name in sorted(encoder_lib.MultiModalEncoder.__abstractmethods__):
print(f" {name}")
print(" final（框架所有，不要覆盖）：setup()、encode()")
t = np.arange(SR) / SR
fade = np.exp(-2.5 * t).astype(np.float32) # 一个衰减的音符，这样包络就不是平的
sound = types.Sound(waveform=(0.5 * fade * np.sin(2 * np.pi * 440 * t)).astype(np.float32),
context=types.SoundContextParams(id="demo_000", sample_rate=SR, length=SR))
for enc in (EnergyEnvelopeEncoder(), SpectralProfileEncoder()):
enc.setup()
emb = enc.encode([sound])[0]
stats = emb.encoding_stats # 由 encode() 附加，不是我们的代码附加的
print(f"\n {type(enc).__name__:24s} -> {emb.embedding.shape} {emb.embedding.dtype}"
f" output_type={enc.output_type().__name__}")
print(f" {'':24s} EncodingStats(input={stats.input_size_bytes:,} B, "
f"embedding={stats.embedding_size_bytes} B, flops={stats.flops})")
print(f" {'':24s} 前 6 维：{np.round(emb.embedding[0][:6], 3)}")
print("\n 包络编码器看到音符衰减；频谱编码器在 440 Hz 处看到一个峰值。")
try:
EnergyEnvelopeEncoder().encode(["not a Sound"])
except ValueError as e:
print(f"\n 错误的输入类型被 _check_input_types 捕获：{e}")
return "两个满足 MultiModalEncoder 的编码器"
encoder_contract()
我们通过继承 MultiModalEncoder 编写两个编码器，它的抽象方法恰好有三个：_setup 加载模型所需的任何东西，_check_input_types 拒绝任何不是 Sound 的输入，_encode 将一批数据转换为 SoundEmbedding 对象。框架负责 setup 和 encode，而 encode 会为每个结果附加 EncodingStats，因此我们的代码从不手动填充它。EnergyEnvelopeEncoder 在十六个等长的时间切片上对能量取平均，因此只描述响度如何变化；SpectralProfileEncoder 将平均对数幅度谱池化为十六个频带，因此描述音色。两者都对输出做 L2 归一化，因此点积即为余弦相似度。通过每个编码器对一个衰减音符进行编码，差异立刻显现：包络编码器看到衰减，而频谱编码器在 440 Hz 处看到一个单峰。
复制代码已复制使用其他浏览器
CLASSES = ["tone", "chirp", "noise"]
N_PER_CLASS = 12
def synthesize(kind: str, index: int, take: int) -> types.Sound:
"""一秒的音频。`take` 0 是文档，take 1 是同一段音频的噪声更大的录音。
两条线索被刻意分开：频谱说明它属于哪一类，而幅度
包络——按每个条目绘制，与类别无关——说明它是哪一个条目。
"""
item = np.random.default_rng(1000 + CLASSES.index(kind) * 100 + index)
control = 0.25 + 0.75 * item.random(8)
envelope = np.interp(np.linspace(0, 7, SR), np.arange(8), control).astype(np.float32)
t = np.arange(SR) / SR
if kind == "tone":
w = np.sin(2 * np.pi * (380 + 80 * item.random()) * t)
elif kind == "chirp":
f0, f1 = 200 + 50 * item.random(), 3200 + 400 * item.random()
w = np.sin(2 * np.pi * (f0 * t + 0.5 * (f1 - f0) * t ** 2))
else:
w = item.standard_normal(SR)
w /= np.sqrt(np.mean(w ** 2)) + 1e-9 # 单位 RMS：包络是唯一的响度线索
take_rng = np.random.default_rng(50_000 + take * 10_000 + CLASSES.index(kind) * 100 + index)
w = (0.4 + 0.2 * take_rng.random()) * envelope * (w + 0.02 * take_rng.standard_normal(SR))
return types.Sound(waveform=w.astype(np.float32), context=types.SoundContextParams(
id=f"{kind}_{index:02d}" + ("" if take == 0 else "_take2"), sample_rate=SR,
length=SR, language="en_us", text=kind))
@section("3. 一个合成语料库，编码为 MSEB 嵌入缓存")
def build_corpus():
corpus = [synthesize(k, i, 0) for k in CLASSES for i in range(N_PER_CLASS)]
queries = [synthesize(k, i, 1) for k in CLASSES for i in range(N_PER_CLASS)]
labels = {s.context.id: s.context.text for s in corpus + queries}
print(f" {len(corpus)} documents + {len(queries)} second takes of the same clips,"
f" {len(CLASSES)} classes, 1.0s each @ {SR} Hz")
caches, query_caches = {}, {}
for enc in (EnergyEnvelopeEncoder(), SpectralProfileEncoder()):
enc.setup()
embeddings = enc.encode(corpus) # one batched call, like a real runner
caches[type(enc).__name__] = {e.context.id: e for e in embeddings}
query_caches[type(enc).__name__] = {e.context.id: e for e in enc.encode(queries)}
matrix = np.vstack([e.embedding for e in embeddings])
within, between = [], []
for i in range(len(corpus)):
for j in range(i + 1, len(corpus)):
sim = float(matrix[i] @ matrix[j])
(within if labels[corpus[i].context.id] == labels[corpus[j].context.id] else between).append(sim)
print(f" {type(enc).__name__:24s} cache of {len(embeddings)} embeddings, dim {matrix.shape[1]}"
f" mean cosine: same-class {np.mean(within):.3f} vs other-class {np.mean(between):.3f}"
f" (gap {np.mean(within) - np.mean(between):+.3f})")
print("\n Read that gap as a prediction: only the spectral encoder separates the classes at all.")
print(" Steps 4-6 check whether the evaluators agree - and whether the gap is the whole story.")
globals().update(CORPUS=corpus, QUERIES=queries, LABELS=labels, CACHES=caches, QCACHES=query_caches)
return f"{len(corpus)} documents + {len(queries)} queries encoded by 2 encoders"
build_corpus()
We synthesize a corpus in which two cues are deliberately separated. The spectrum says which class a clip belongs to, a tone, a chirp or noise, while the amplitude envelope is drawn per item and is independent of class, so it identifies which clip it is without saying anything about what it is. We normalize every waveform to unit RMS before applying the envelope, leaving the envelope as the only loudness cue. We render each of the thirty-six items twice, once as the document and once as a noisier second take of the same clip, and encode both sets with both encoders into MSEB embedding caches, the plain dictionaries from sound id to SoundEmbedding that every evaluator consumes. The mean same-class and other-class cosine similarities printed here read as a prediction about the next three steps: only the spectral encoder separates the classes at all.
Copy CodeCopiedUse a different Browser
def class_prototypes(cache, labels):
"""Class embedding table (C, D): the mean unit vector of each class, as the evaluator's `weights`."""
rows = []
for name in CLASSES:
vecs = np.vstack([cache[i].embedding for i in cache if labels[i] == name])
mean = vecs.mean(axis=0)
rows.append(mean / (np.linalg.norm(mean) + 1e-9))
return np.vstack(rows).astype(np.float32)
@section("4. ClassificationEvaluator: prototypes in, Score objects out")
def classification():
table, example = {}, None
for name, cache in CACHES.items():
evaluator = classification_evaluator.ClassificationEvaluator(
class_labels=CLASSES,
weights=class_prototypes(cache, LABELS),
distance_fn=evaluator_lib.dot_product, # embeddings are L2-normalised -> cosine
top_k_value=2,
)
predictions = evaluator.compute_predictions(cache) # {id: per-class score vector}
references = [classification_evaluator.ClassificationReference(i, LABELS[i]) for i in cache]
table[name] = {s.metric: s.value for s in evaluator.compute_metrics(predictions, references)}
if name == "SpectralProfileEncoder":
key = next(iter(predictions))
example = (key, np.round(list(predictions[key]), 3))
metric_names = list(next(iter(table.values())))[:6]
print(f" {'encoder':26s}" + "".join(f"{m[:14]:>16s}" for m in metric_names))
for name, row in table.items():
print(f" {name:26s}" + "".join(f"{row[m]:16.3f}" for m in metric_names))
print(f"\n compute_predictions 为每个类别返回一个原始分数，例如 {example[0]!r} -> {example[1]}")
print(f" ({CLASSES} - argmax 即为预测结果，设置 top_k_value=2 还会计算 Top-2 Accuracy。)")
print(" compute_metrics 会将这些转换为 types.Score 对象，这正是排行榜所存储的内容。")
for name, row in table.items():
BENCH.setdefault(name, {})["Accuracy"] = row["Accuracy"]
winner = max(table, key=lambda k: table[k]["Accuracy"])
return "Accuracy: " + ", ".join(f"{k} {v['Accuracy']:.3f}" for k, v in table.items()) + f" (winner {winner})"
classification()
ClassificationEvaluator 以类别嵌入表作为其权重，并接收一个距离函数，我们将权重构建为类别原型，即每个类别的平均单位向量。它的两个方法职责清晰分离：compute_predictions 为每个缓存嵌入返回每个类别的原始分数，compute_metrics 将这些分数与 ClassificationReference 标签一起转换为排行榜所存储的 Score 对象列表。将 top_k_value 设为 2 会在 accuracy、balanced accuracy 以及加权的 precision、recall 和 F1 之外，额外加入 Top-2 Accuracy。谱编码器完美地对语料库进行了分类，而包络编码器的表现远高于随机水平，但远低于前者，这正是余弦差距所预测的排序。
Copy CodeCopiedUse a different Browser
@section("5. ClusteringEvaluator：编码时无标签，评分时使用 V-measure")
def clustering():
evaluator = clustering_evaluator.ClusteringEvaluator()
examples = [clustering_evaluator.ClusteringExample(sound_id=i, label=LABELS[i])
for i in next(iter(CACHES.values()))]
print(f" {len(examples)} examples, KMeans with k = {len(CLASSES)} (inferred from the labels)")
for name, cache in CACHES.items():
np.random.seed(0) # MiniBatchKMeans takes no random_state here:
scores = evaluator(cache, examples) # it falls back to NumPy's global RNG, so pin that
# 或一个无结构的嵌入空间得分在 0.01-0.08
# 之间，属于随机水平。该 evaluator 是可调用的。
BENCH.setdefault(name, {})["VMeasure"] = scores[0].value
print(f" {name:26s} {scores[0].metric:12s} {scores[0].value:6.3f}"
f" [{scores[0].min}, {scores[0].max}] :: {scores[0].description}")
print("\n V-measure 是 homogeneity 和 completeness 的调和平均数：1.0 表示聚类
print(" 完全还原了类别，0.0 表示它们不携带任何关于类别的信息。注意它"
print(" 对 envelope encoder 的评判比 accuracy 严厉得多——聚类没有标签可以依赖。")
return ", ".join(f"{k} V={v['VMeasure']:.3f}" for k, v in BENCH.items())
clustering()
ClusteringEvaluator 问的是同一个问题的更难版本，因为它在编码时从来看不到标签：它在缓存上运行 KMeans。它用 V-measure（homogeneity 和 completeness 的调和平均数）对聚类结果与标签进行评分。与分类相比，两个编码器之间的差距在这里急剧拉大，因为有监督的原型读出可以利用一个微弱的线索，而无监督聚类无法自行发现它。有一个实用细节值得复制到任何可复现的基准测试运行中：该评估器构造 MiniBatchKMeans 时没有设置 random_state，因此它会回退到 NumPy 的全局生成器，而不为该生成器设置种子时，一个无结构的嵌入空间在每次运行之间的得分大约在 0.01 到 0.08 之间。
Copy CodeCopiedUse a different Browser
@section("6. RetrievalEvaluator：为语料库建立索引，用第二个 take 查询它，对排名进行评分")
def retrieval():
print(" 任务：每个查询都是某一份文档的带噪录音，且恰好有一份文档是正确的。")
print(" 这是身份识别，不是类别识别——与步骤 4 和 5 是不同的问题。\n")
out = {}
for name, cache in CACHES.items():
doc_ids = list(cache)
docs = np.vstack([cache[i].embedding for i in doc_ids]).astype(np.float32)
queries = QCACHES[name]
searcher = retrieval_evaluator.BruteForceSearcher(candidates=docs, num_neighbors=10)
evaluator = retrieval_evaluator.RetrievalEvaluator(searcher=searcher, id_by_index_id=doc_ids, top_k=5)
predictions = evaluator.compute_predictions(queries)
references = [retrieval_evaluator.RetrievalReferenceId(
sound_id=q, reference_id=q.removesuffix("_take2")) for q in queries]
out[name] = {s.metric: s.value for s in evaluator.compute_metrics(predictions, references)}
q0 = next(iter(queries))
top = [item["id"] for item in predictions[q0].items[:5]]
print(f" top-5 for query {q0!r} under {name}:")
print(f" {top}")
print(f" correct document at rank {top.index(q0.removesuffix('_take2')) + 1}"
f" | neighbours of the same class: {sum(LABELS[i] == LABELS[q0] for i in top)}/5\n")
metric_names = ["MRR", "EM", "RecallAt5", "NDCG@10"]
print(f" {'encoder':26s}" + "".join(f"{m:>14s}" for m in metric_names))
for name, row in out.items():
print(f" {name:26s}" + "".join(f"{row[m]:14.3f}" for m in metric_names))
BENCH.setdefault(name, {})["MRR"] = row["MRR"]
print("\n MRR is 1/rank of the correct document, EM is 'it was rank 1', RecallAt5 is 'it was in the")
print(" top 5'. NDCG@10 here is graded credit for the same single relevant document.")
return ", ".join(f"{k} MRR={v['MRR']:.3f}" for k, v in out.items())
retrieval()
RetrievalEvaluator 回答的问题与前面两个不同，我们特意把任务设置成让这种差异可见。每个查询都恰好是某一篇文档更嘈杂的第二版，因此目标是身份识别而非类别判断。我们把文档嵌入索引进一个 BruteForceSearcher，在查询缓存上请求预测，并为每个查询传入一个 RetrievalReferenceId，指明其唯一正确的文档。评估器返回 MRR、精确匹配、我们 top_k 下的召回率以及十位的 NDCG。结果与前面两步相反：envelope 编码器把每个片段都检索到第一位，因为 envelope 是条目的指纹。相比之下，spectral 编码器的排名略差，因为同类片段在它看来很相似。打印出的 top-5 列表让机制一目了然，一个邻域类别随机，另一个类别纯净。
Copy CodeCopiedUse a different Browser
@section("7. The metric layer on its own: WER, CER, exact match, MRR, nDCG")
def metric_layer():
truth = "the quick brown fox jumps over the lazy dog"
for hypothesis in [truth, "the quick brown fox jumped over a lazy dog", "quick brown fox over lazy dog"]:
werrors, wtotal = metrics.compute_word_errors(truth, hypothesis)
cerrors, ctotal = metrics.compute_character_errors(truth, hypothesis)
print(f" WER {werrors / wtotal:5.3f} ({werrors}/{wtotal} words) "
f"CER {cerrors / ctotal:5.3f} ({cerrors}/{ctotal} chars) {hypothesis!r}")
print("\n ranking metrics take (reference, ranked_ids):")
ranked = ["doc_b", "doc_a", "doc_c", "doc_d"]
for reference in ["doc_b", "doc_a", "doc_c", "doc_z"]:
rank = ranked.index(reference) + 1 if reference in ranked else None
print(f" reference {reference!r:8s} rank {str(rank):4s}"
f" EM {metrics.compute_exact_match(reference, ranked):.1f}"
f" MRR {metrics.compute_reciprocal_rank(reference, ranked):.3f}"
f" nDCG@4 {metrics.compute_ndcg_at_k(reference, ranked, k=4):.3f}")
print(" compute_ndcg_at_k assumes ONE relevant document and compares it by equality, so pass a")
print(" string, not a list - a list reference silently scores 0.0 while MRR still looks fine.")
print("\n embedding-space distances used by the reconstruction and stability tasks:")
a = np.random.default_rng(1).standard_normal((8, 4)).astype(np.float32)
for label, b in [("identical", a), ("noisy", a + 0.1 * np.random.default_rng(2).standard_normal(a.shape))]:
lp = metrics.compute_lp_norm(a, b, p=2)
dtw = metrics.compute_dynamic_time_warping_distance(a, b)
print(f" {label:10s} L2 {json.dumps({k: round(float(v), 3) for k, v in lp.items()})}"
f" DTW {json.dumps({k: round(float(v), 3) for k, v in dtw.items()})}")
return "WER/CER, EM/MRR/nDCG, Lp and DTW distances"
metric_layer()
我们直接调用度量函数，不在它们外面套一个 evaluator，因为它们是任务族共享的那一层。compute_word_errors 和 compute_character_errors 接收两个字符串，分别返回错误数和总数，因此由调用方决定如何对语料库进行聚合。排序度量接收一个参考项和一个标识符排序列表，在同一排序上比较精确匹配、倒数排名和 nDCG，可以看出各自为位置付出了什么代价。有一个尖锐的边界值得指出：compute_ndcg_at_k 假设只有一个相关文档，并按相等性进行比较，因此传入一个相关 id 列表会静默地得零分。相比之下，MRR 确实接受列表，看起来仍然正确。最后我们介绍 compute_lp_norm 和 compute_dynamic_time_warping_distance，它们是重建和稳定性任务背后的嵌入空间距离。
Copy CodeCopiedUse a different Browser
@section("8. SegmentationEvaluator：分别对说了什么和在哪里进行评分")
def segmentation():
evaluator = segmentation_evaluator.SegmentationEvaluator(tau=0.05)
print(" 这里一个 'segment' 携带的是一个 TERM，而不是一个向量：SoundEmbedding.embedding 保存 N 个字符串")
print(" 而 timestamps 保存它们对应的 N 个 [start, end] 区间。tau=0.05 -> 边界可以偏差 50 ms。\n")
TERMS = [("weather", 0.00, 0.30), ("in", 0.30, 0.65), ("boston", 0.65, 1.00)]
truth = [segmentation_evaluator.Segment(embedding=term, start_time=s, end_time=e, confidence=1.0)
for term, s, e in TERMS]
references = [segmentation_evaluator.SegmentationReference(example_id="utt_0", segments=truth)]
def prediction(spans):
return {"utt_0": types.SoundEmbedding(
embedding=np.array([term for term, _, _ in spans]), # N 个字符串
timestamps=np.array([[s, e] for _, s, e in spans], dtype=np.float32), # N 个 [start, end]
context=types.SoundContextParams(id="utt_0", sample_rate=SR, length=SR),
scores=np.ones(len(spans), dtype=np.float32))} # 置信度
candidates = {
"exact": TERMS,
"50 ms out": [("weather", 0.00, 0.28), ("in", 0.28, 0.67), ("boston", 0.67, 1.00)],
"right words, wrong places": [("weather", 0.00, 0.45), ("in", 0.45, 0.80), ("boston", 0.80, 1.00)],
"right places, wrong words": [("weather", 0.00, 0.30), ("on", 0.30, 0.65), ("austin", 0.65, 1.00)],
}
shown = ["TimestampsAccuracy", "EmbeddingsAccuracy", "TimestampsAndEmbeddingsAccuracy", "WordErrorRate", "mAP"]
print(f" {'prediction':28s}" + "".join(f"{m[:13]:>15s}" for m in shown))
for label, spans in candidates.items():
result = evaluator.compute_scores(prediction(spans), references) # per-example scores
scores = {s.metric: s.value for s in evaluator.compute_metrics(result)} # aggregated Scores
print(f" {label:28s}" + "".join(f"{scores[m]:15.3f}" for m in shown))
print("\n The last two rows are the point: one metric cannot tell 'knew the words, missed the timing'")
print(" from 'nailed the timing, heard the wrong words'. Timestamps and embeddings are scored apart,")
print(" and only TimestampsAndEmbeddings credits getting both right at once.")
return "boundary + term scoring at tau=50 ms"
segmentation()
SegmentationEvaluator 把说了什么和说在哪里作为两个独立的量来评分，它使用第 1 步提到的 SoundEmbedding 的字符串形式：embedding 数组为每个分段保存一个词条，timestamps 数组保存它们的时间跨度。它的流程分两个阶段，先对预测和参考计算 compute_scores，再对该结果计算 compute_metrics。我们用 50 毫秒的容差，将同一短语的四种候选分段与一个标准答案进行评分。完全准确和偏差 50 毫秒都得到满分，这正是容差的用途。最后两行承载着教训：词对但位置错，embeddings 得一分、timestamps 得零分；位置对但词错则相反；只有组合指标才会同时认可两者都正确。
Copy CodeCopiedUse a different Browser
@section("9. TaskMetadata and a leaderboard that disagrees with itself")
def task_metadata():
cache = CACHES["SpectralProfileEncoder"]
evaluator = classification_evaluator.ClassificationEvaluator(
class_labels=CLASSES, weights=class_prototypes(cache, LABELS), top_k_value=2)
references = [classification_evaluator.ClassificationReference(i, LABELS[i]) for i in cache]
scores = [s for s in evaluator.compute_metrics(evaluator.compute_predictions(cache), references)
if s.metric in ("Accuracy", "Weighted F1-Score")]
metadata = types.TaskMetadata(
name="SyntheticToneClassification",
description="合成音调、啁啾声和噪声的三分类",
reference="https://github.com/google-research/mseb",
type="Classification",
category="sound",
main_score="Accuracy",
revision="1",
dataset=types.Dataset(path="synthetic/in-notebook", revision="1"),
scores=scores,
eval_splits=["test"],
eval_langs=["en_us"],
)
print(f" TaskMetadata: {metadata.name} type={metadata.type} main_score={metadata.main_score!r}")
print(f" dataset={metadata.dataset.path!r} rev {metadata.dataset.revision}"
f" splits={metadata.eval_splits} langs={metadata.eval_langs}")
print(f" scores={[f'{s.metric}={s.value:.3f}' for s in metadata.scores]}")
try:
types.TaskMetadata(**{**{f.name: getattr(metadata, f.name) for f in metadata.__dataclass_fields__.values()},
"scores": []})
except Exception as e:
print(f" validated at construction: {type(e).__name__}: {e}")
columns = ["Accuracy", "VMeasure", "MRR"]
print(f"\n One row per encoder, one column per task family:")
print(f" {'encoder':26s}" + "".join(f"{c:>12s}" for c in columns) + " what it measures")
for name, row in BENCH.items():
print(f" {name:26s}" + "".join(f"{row[c]:12.3f}" for c in columns)
+ (" timbre -> class" if "Spectral" in name else " loudness over time -> identity"))
flips = [c for c in columns
if (max(BENCH, key=lambda n: BENCH[n][c]) != max(BENCH, key=lambda n: BENCH[n]["Accuracy"]))]
print(f"\n The winner changes column to column ({', '.join(flips)} goes the other way). That is the whole")
print(" argument for a MASSIVE benchmark: a single headline number would have hidden it. An encoder")
print(" that cannot name a sound can still recognise it, and vice versa.")
print("\n A real submission runs mseb.runner over an mseb.task against a published dataset and writes")
print(" these same Score objects to JSON; the layers above are exactly what it exercises.")
return f"TaskMetadata + {len(BENCH)} encoders x {len(columns)} task families"
task_metadata()
我们组装真实提交所携带的 TaskMetadata，包括名称、类型、类别、主分数、数据集路径和版本、评估划分和语言，以及 Score 对象本身，它在构造时以与 Score 相同的方式进行验证，拒绝空的分数列表。然后我们把目前所有结果放进一张表：每个编码器一行，每个任务族一列。赢家逐列变化：无法说出声音名称的编码器仍能识别它，而能正确说出每个声音名称的编码器却会混淆本应归为一类的片段。单一的头条数字会完全掩盖这一点，这正是需要一个在任务上而非仅在数据上庞大的基准测试的理由。
复制代码已复制使用其他浏览器
banner("SUMMARY")
for name, res in RESULTS.items():
print(f" {name:<74s} {res}")
print("""
下一步去哪里
- 换入真实编码器：mseb/encoders/ 提供了 wav2vec、Whisper、CLAP、EnCodec 和 SoundStream
包装器，以及用于语音到文本到嵌入链的 CascadeEncoder。只有步骤 2 中的三个方法
发生变化；上面的每个评估器都继续工作。
- 运行已发布的任务：mseb.runner 使用 apache-beam 在真实数据集上驱动 mseb.task；
任务族位于 mseb/tasks/（分类、检索、重排序、转录、
分割、聚类、推理、brain_encoding、稳定性）。
- 与排行榜比较：https://huggingface.co/spaces/google/mseb-leaderboard
- 阅读你实现的契约：mseb/encoder.py 和 mseb/evaluator.py 总共约 500 行。
""")
摘要打印每个部分返回的单行结果，然后指出这个 notebook 打开的三个方面：换入包中附带的某个真实编码器，wav2vec、Whisper、CLAP、EnCodec、SoundStream 或级联包装器，这只会改变步骤 2 中的三个方法，并让每个评估器继续工作；通过 mseb.runner 在真实数据集上运行已发布的任务；以及将结果与公开排行榜进行比较。
In conclusion, we treated MSEB as what it is, a contract plus a set of evaluators, and drove it end to end without downloading a dataset or touching an accelerator. Implementing three methods was enough to make our own code a first-class citizen of the benchmark, and the framework handled batching, statistics, and validation from there. The evaluators asked genuinely different questions of the same embeddings: classification and clustering ask what a sound is, retrieval asks which sound it is, and segmentation asks what was said and where, scored apart so a timing failure and a recognition failure never hide inside one average. Our two encoders traded places depending on the question asked, and we carry that result forward, because a single number cannot rank a sound embedding. The next step is to substitute a real encoder for our toy ones and re-run the same evaluators, since none of the scoring code above changes when the embeddings improve.
Check out the GitHub Repo with Full Codes. All credit goes to the researcher of this project. Also, feel free to follow us on Twitter and don’t forget to join our 150k+ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
Need to partner with us for promoting your GitHub Repo OR Hugging Face Page OR Product Release OR Webinar etc.? Connect with us
The post A Coding Guide to Google Research’s MSEB: Writing Sound Encoders to the Benchmark Contract and Scoring Them Across Classification, Clustering, Retrieval and Segmentation appeared first on MarkTechPost.