Google Research 的 MSEB 編碼指南：依照基準契約撰寫健全的編碼器，並在分類、聚類、檢索與分割任務中評分

In this tutorial, we work with MSEB, the Massive Sound Embedding Benchmark from Google Research, and approach it from the perspective of what a leaderboard number actually means: the evaluator surface.

We install the package and map its three layers, then write two deliberately different encoders against the framework’s own abstract base class: one that measures loudness over time and one that measures timbre, and encode a small synthetic corpus we generate in the notebook so nothing has to be downloaded.

We drive the classification, clustering, retrieval, and segmentation evaluators over those embeddings, call the metric functions directly to see what each one rewards, and finish by assembling the TaskMetadata a real submission carries.

The result is a comparison in which the two encoders trade places depending on which evaluator is asked, which is the argument for a multi-task benchmark made in numbers rather than in prose.

Copy CodeCopiedUse a different Browserimport 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.printexc(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 encoderlib, evaluator as evaluatorlib, metrics from mseb.evaluators import ( classificationevaluator, clusteringevaluator, retrievalevaluator, segmentationevaluator, ) 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 [(classificationevaluator, "ClassificationEvaluator"), (clusteringevaluator, "ClusteringEvaluator"), (retrievalevaluator, "RetrievalEvaluator"), (segmentationevaluator, "SegmentationEvaluator")]: print(f" {module.name.split('.

')[-1]:28s} {cls}") print("\n Everything below runs on CPU with no dataset download: we synthesise the audio.") We install mseb and import the three layers that a benchmark run walks down.

The types module holds the shapes every task speaks, Sound, SoundEmbedding, Score and TaskMetadata; the encoder module holds MultiModalEncoder, the contract our own model implements; and the evaluators package holds one module per task family.

We import only the four evaluators this notebook drives, because the classification, clustering, retrieval, and segmentation modules depend on nothing heavier than NumPy and scikit-learn.

In contrast, the reranking and transcription evaluators pull in Whisper and the task runner pulls in TensorFlow and apache-beam.Everything below therefore runs on a free CPU runtime with no dataset download and no accelerator.Copy CodeCopiedUse a different BrowserSR = 16000 @section("1.

The type contract: Sound, SoundEmbedding, Score") def typecontract(): 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="demo000", samplerate=SR, length=len(waveform), language="enus", text="a 440 Hz tone"), ) print(f" Sound id={sound.context.id!r} {sound.waveform.shape} @ {sound.context.samplerate} Hz" f" -> {sound.sizebytes:,} 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, encodingstats=types.EncodingStats(inputsizebytes=sound.

sizebytes, embeddingsizebytes=16 * 4), ) print(f" SoundEmbedding embedding{embedding.embedding.shape} timestamps{embedding.timestamps.shape}" f" -> {embedding.sizebytes} bytes") print(f" compressionratio = {embedding.encodingstats.compressionratio:.5f}" f" ({1 / embedding.encodingstats.

compressionratio:,.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.sizebytes:,} B -> embedding {embedding.

sizebytes} B" typecontract() We start with the type contract, because every other layer is expressed in it.A Sound carries a waveform, along with SoundContextParams, the identifier, sample rate, length, language, and optional transcript, which follow the audio through the whole pipeline.

A SoundEmbedding carries an array of N embeddings and an array of M timestamp pairs, and the relation between N and M is the benchmark’s vocabulary: M equal to N means one vector per frame, while M equal to one means a single utterance-level vector, which is what our encoders produce.

EncodingStats records the input and embedding sizes and exposes compressionratio, here a thousandfold reduction from audio to vector.

A Score is a metric name, a value and its bounds, and it validates itself at construction, rejecting an empty metric name or a minimum above its maximum, so a malformed number cannot reach a leaderboard.

The embedding field also accepts N strings instead of N vectors, which is the door that step 8 walks through.Copy CodeCopiedUse a different Browserclass EnergyEnvelopeEncoder(encoderlib.MultiModalEncoder): """Baseline: average energy in nbins equal time slices.Loud/quiet, nothing about timbre.

""" def init(self, nbins: int = 16): super().init() self.nbins = nbins def setup(self): self.ready = True # a real encoder loads weights here def checkinputtypes(self, batch): for item in batch: if not isinstance(item, types.Sound): raise ValueError(f"{type(self).name} takes types.

Sound, got {type(item).name}") def encode(self, batch) -> list[types.SoundEmbedding]: out = [] for sound in batch: slices = np.arraysplit(sound.waveform.astype(np.float32), self.nbins) 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.samplerate]], dtype=np.float32), context=sound.context)) return out class SpectralProfileEncoder(encoderlib.

MultiModalEncoder): """Contender: mean log-magnitude spectrum pooled into nbands bands.Describes timbre.""" def init(self, nbands: int = 16, frame: int = 512): super().init() self.nbands, self.frame = nbands, frame def setup(self): self.window = np.hanning(self.frame).astype(np.

float32) def checkinputtypes(self, batch): for item in batch: if not isinstance(item, types.Sound): raise ValueError(f"{type(self).name} takes types.Sound, got {type(item).name}") def encode(self, batch) ->