Guida pratica a MSEB di Google: scrivere encoder sonori e valutarli su classificazione, clustering, ricerca e segmentazione

Introduzione a MSEB

Il Massive Sound Embedding Benchmark (MSEB) di Google Research è una piattaforma di valutazione multi‑task pensata per confrontare encoder audio su una serie di compiti: classificazione, clustering, ricerca (retrieval) e segmentazione. Il valore di un risultato sul leaderboard non è un semplice numero, ma la combinazione di diversi evaluator surface che misurano aspetti diversi del modello. In questo tutorial, partiamo dall’installazione del pacchetto, descriviamo le tre “layer” fondamentali del benchmark, e costruiamo due encoder deliberatamente diversi per osservare come le loro performance variano a seconda del compito.

Installazione e mappatura delle tre layer

Il primo passo consiste nell’installare la versione 0.1.0 di mseb. L’intera pipeline può essere eseguita su CPU, senza necessità di scaricare dataset esterni, grazie alla generazione sintetica di un piccolo corpus audio.

import os
import sys
import json
import math
import traceback
import subprocess
import numpy as np

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 (
classification_evaluator,
clustering_evaluator,
retrieval_evaluator,
segmentation_evaluator,
)

print(f"  mseb {mseb._version}  |  Python {sys.version.split()[0]}  |  numpy {np.version_}")

Il pacchetto è organizzato in tre livelli:

• types: definisce le strutture dati comuni (Sound, SoundEmbedding, Score, TaskMetadata).
• encoder: contiene la classe astratta MultiModalEncoder che ogni modello deve implementare.
• evaluator: fornisce i valutatori per ciascun task (classificazione, clustering, retrieval, segmentazione, ecc.).

Nel nostro caso utilizziamo solo i quattro valutatori più leggeri, che dipendono esclusivamente da NumPy e scikit‑learn, evitando così dipendenze più pesanti come Whisper o TensorFlow.

Contratto dei tipi: Sound, SoundEmbedding e Score

Per comprendere come funziona il benchmark, è fondamentale analizzare il “type contract”. Un oggetto Sound racchiude la forma d’onda e metadati contestuali, mentre SoundEmbedding contiene i vettori di embedding e le relative timbrature temporali. La relazione tra il numero di vettori N e il numero di timestamp M è la chiave di interpretazione: M == N indica un embedding per frame, M == 1 indica un embedding a livello di utterance.

@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),
timestamps=np.array([[0.0, 1.0]], dtype=np.float32),
context=sound.context,
encoding_stats=types.EncodingStats(
inputsizebytes=sound.size_bytes,
embeddingsizebytes=16 * 4
),
)
print(f"  SoundEmbedding embedding{embedding.embedding.shape}  timestamps{embedding.timestamps.shape}"
f"  -> {embedding.size_bytes} bytes")
print(f"                 compressionratio = {embedding.encodingstats.compression_ratio:.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"

Il risultato stampa le dimensioni dell’audio originale, la compressione ottenuta dall’embedding e una validazione di esempio per l’oggetto Score, che rifiuta nomi vuoti o valori min > max.

Implementazione di un encoder basato sull’energia

Il primo encoder, EnergyEnvelopeEncoder, suddivide l’audio in n_bins segmenti temporali e calcola l’energia media di ciascuno. È un modello molto semplice, privo di capacità timbriche.

class EnergyEnvelopeEncoder(encoder_lib.MultiModalEncoder):
"""Baseline: average energy in `n_bins` 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 would load weights here

def checkinput_types(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.arange(self.nbins + 1) * len(sound.waveform) // self.nbins
energies = []
for i in range(self.n_bins):
segment = sound.waveform[slices[i]:slices[i+1]]
energies.append(np.sqrt(np.mean(segment ** 2)))
embedding = np.array(energies, dtype=np.float32).reshape(1, -1)
timestamps = np.array([[0.0, len(sound.waveform) /