Learn AI Series):Exercise 1: Melody complexity analyzer -- categorize intervals (unison/step/skip/leap), compute pitch range, and measure contour complexity (direction changes per interval). Result: C major scale = 100% steps, zero contour changes. Twinkle Twinkle mixes steps and leaps with several direction changes. Random walk produces high contour complexity because each interval direction is essentially a coin flip.
Exercise 2: Chord progression generator with Markov transitions and consonance scoring. The V->I resolution appears as the strongest cadence -- the transition matrix gives V a 55% probability of resolving to I, reflecting how dominant-tonic resolution is the most fundamental harmonic motion. Major triads (I, IV, V) score higher consonance than the diminished triad (vii).
Exercise 3: Rhythmic pattern evaluator measuring IOI statistics, swing ratio, and tempo stability across metronome, swing, human drummer, and random patterns. The metronome has CV=0 (perfect regularity) and swing ratio=1.0. The swing pattern has ratio near 2.0. The human drummer has small but nonzero CV from Gaussian timing jitter. The random pattern has the highest CV and lowest tempo stability.
Here we go! We've been deep in the audio AI domain over the past five episodes: how machines hear sound (#92), transcribing speech to text (#93), synthesizing speech from text (#94), classifying environmental sounds (#95), and generating music from scratch (#96). All of those dealt with understanding what's in the audio -- the words, the sounds, the melodies. Today we're shifting focus to understanding who is in the audio.
Speaker recognition is the audio equivalent of face recognition (episode #88). Just as FaceNet learns a compact embedding vector for each face, speaker recognition models learn embedding vectors for voices. Two audio clips from the same person produce similar vectors; clips from different people produce different vectors. The math is almost identical -- cosine similarity, angular margin losses, the whole apparatus we already built for faces. The difference is the input representation: spectrograms and Mel filterbanks instead of pixel arrays.
This is one of those topics where the practical applications are everywhere -- from "Hey Siri, is that actually me talking?" to meeting transcription systems that figure out which participant said what. The underlying techniques are elegant and connect back to nearly everything we've covered so far. Let's get into it ;-)
Before we write any code, let me clarify the two main tasks in speaker recognition. They sound similar but they're fundamentally different problems.
Speaker verification (SV) answers a yes/no question: "Is this the person they claim to be?" You have a stored voiceprint (an embedding vector from enrollment) and a new audio sample. Compute the similarity. If it's above a threshold -- accept. Below -- reject. This is a binary classification problem (same speaker or different speaker), and it's what banks use for phone authentication, what smart speakers use for personalization, and what forensic analysts use for voice matching.
Speaker identification (SI) answers a who question: "Which of these N known speakers is talking?" You have a database of enrolled speakers with their voiceprints. Given a new audio clip, find the closest match. This is an N-way classification problem, and it's what meeting transcription systems and surveillance systems use.
Both tasks rely on the same foundation: speaker embeddings -- fixed-dimensional vectors that capture the essential characteristics of a voice while being invariant to what's being said, background noise, and recording conditions. The quality of your embeddings determines the quality of everything downstream:
import numpy as np
class SpeakerTaskDemo:
"""Demonstrate the difference between
speaker verification (1:1) and speaker
identification (1:N)."""
def __init__(self):
self.rng = np.random.RandomState(42)
def simulate_embeddings(self,
n_speakers=5,
dim=192,
n_samples=3):
"""Create synthetic speaker
embeddings with realistic
intra/inter speaker distances."""
speakers = {}
for i in range(n_speakers):
center = self.rng.randn(dim)
center = center / np.linalg.norm(
center)
samples = []
for _ in range(n_samples):
noise = self.rng.randn(
dim) * 0.15
emb = center + noise
emb = emb / np.linalg.norm(
emb)
samples.append(emb)
speakers[f"spk_{i}"] = samples
return speakers
def cosine_sim(self, a, b):
return float(np.dot(a, b) / (
np.linalg.norm(a)
* np.linalg.norm(b)))
def verification(self, speakers):
"""1:1 -- is this the claimed
speaker?"""
print("=== Speaker Verification ===")
# Same speaker comparison
s0 = speakers["spk_0"]
same = self.cosine_sim(
s0[0], s0[1])
# Different speaker comparison
diff = self.cosine_sim(
s0[0],
speakers["spk_1"][0])
threshold = 0.7
print(f"Same speaker sim: "
f"{same:.3f} "
f"({'ACCEPT' if same > threshold else 'REJECT'})")
print(f"Diff speaker sim: "
f"{diff:.3f} "
f"({'ACCEPT' if diff > threshold else 'REJECT'})")
print(f"Threshold: {threshold}")
def identification(self, speakers):
"""1:N -- who is this?"""
print("\n=== Speaker Identification ===")
# Take a sample from spk_2
query = speakers["spk_2"][2]
# Compare against all enrolled
# speakers (using first sample
# as enrollment)
print(f"Query: spk_2 sample #3")
best_score = -1
best_match = None
for name, samples in (
speakers.items()):
sim = self.cosine_sim(
query, samples[0])
match = " <-- MATCH" if (
sim > 0.7
and sim > best_score) else ""
if sim > best_score:
best_score = sim
best_match = name
print(f" vs {name}: "
f"{sim:.3f}{match}")
print(f"Identified as: "
f"{best_match} "
f"(score: {best_score:.3f})")
def run(self):
speakers = (
self.simulate_embeddings())
self.verification(speakers)
self.identification(speakers)
# Show intra vs inter variance
print("\n=== Distance Analysis ===")
intra = []
inter = []
names = list(speakers.keys())
for n in names:
samps = speakers[n]
for i in range(len(samps)):
for j in range(
i+1, len(samps)):
intra.append(
self.cosine_sim(
samps[i],
samps[j]))
for i in range(len(names)):
for j in range(
i+1, len(names)):
inter.append(
self.cosine_sim(
speakers[names[i]][0],
speakers[names[j]][0]))
print(f"Intra-speaker sim: "
f"{np.mean(intra):.3f} "
f"(+/- {np.std(intra):.3f})")
print(f"Inter-speaker sim: "
f"{np.mean(inter):.3f} "
f"(+/- {np.std(inter):.3f})")
print(f"Separation gap: "
f"{np.mean(intra) - np.mean(inter):.3f}")
demo = SpeakerTaskDemo()
demo.run()
The key insight from the distance analysis: for speaker recognition to work, the intra-speaker similarity (same person, different utterances) must be significantly higher than the inter-speaker similarity (different people). This is the same separation criterion that drives all metric learning (episode #63 on embeddings). The wider the gap between intra and inter distances, the easier it is to set a reliable threshold for verification or identification.
A speaker embedding model takes a variable-length audio segment and produces a fixed-size vector that captures vocal identity. Two segments from the same speaker should produce similar vectors; segments from different speakers should produce distant vectors. This is conceptually identical to how FaceNet (episode #88) maps face images to embedding vectors -- same problem, different modality.
D-vectors (Variani et al., 2014; Wan et al., 2018) use the simplest possible approach: process Mel spectrogram frames through an LSTM (episode #49), then average the frame-level hidden states to get a single utterance-level vector:
import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np
class DVectorModel(nn.Module):
"""D-vector: LSTM-based speaker
embedding extractor."""
def __init__(self, n_mels=40,
hidden=256,
embed_dim=256,
n_layers=3):
super().__init__()
self.lstm = nn.LSTM(
n_mels, hidden, n_layers,
batch_first=True)
self.proj = nn.Linear(
hidden, embed_dim)
def forward(self, mel_frames):
"""mel_frames: (batch, time,
n_mels)"""
output, _ = self.lstm(mel_frames)
# Average pooling over time
embedding = output.mean(dim=1)
embedding = self.proj(embedding)
return F.normalize(
embedding, dim=1)
# Demo
model = DVectorModel()
# Simulate 2 utterances: 3 sec
# and 5 sec at ~100 frames/sec
utt_short = torch.randn(1, 300, 40)
utt_long = torch.randn(1, 500, 40)
emb_short = model(utt_short)
emb_long = model(utt_long)
print(f"Short utterance embedding: "
f"{emb_short.shape}")
print(f"Long utterance embedding: "
f"{emb_long.shape}")
print(f"Both produce {emb_short.shape[1]}"
f"-dim vectors regardless of "
f"input length")
total = sum(p.numel()
for p in model.parameters())
print(f"Parameters: {total:,}")
The averaging over time is what makes d-vectors length-invariant -- a 2-second clip and a 10-second clip both produce a 256-dimensional vector. Simple and effective, but the averaging loses temporal information. A speaker who emphasizes certain sounds differently at the beginning versus end of an utterance -- that nuance gets averaged away.
X-vectors (Snyder et al., 2018) replaced the LSTM with a TDNN (Time-Delay Neural Network) -- essentially 1D convolutions at different temporal strides -- followed by a statistics pooling layer that computes both the mean AND standard deviation across time:
class XVectorModel(nn.Module):
"""X-vector: TDNN + statistics
pooling for speaker embeddings."""
def __init__(self, n_mels=40,
embed_dim=512):
super().__init__()
# TDNN layers with increasing
# temporal context
self.tdnn1 = nn.Conv1d(
n_mels, 512, 5, padding=2)
self.bn1 = nn.BatchNorm1d(512)
self.tdnn2 = nn.Conv1d(
512, 512, 3,
dilation=2, padding=2)
self.bn2 = nn.BatchNorm1d(512)
self.tdnn3 = nn.Conv1d(
512, 512, 3,
dilation=3, padding=3)
self.bn3 = nn.BatchNorm1d(512)
self.tdnn4 = nn.Conv1d(
512, 512, 1)
self.bn4 = nn.BatchNorm1d(512)
self.tdnn5 = nn.Conv1d(
512, 1500, 1)
self.bn5 = nn.BatchNorm1d(1500)
# Stats pooling: mean + std
# = 1500 * 2 = 3000 dim
self.embed = nn.Linear(
3000, embed_dim)
def forward(self, mel):
"""mel: (batch, n_mels, time)"""
x = F.relu(self.bn1(
self.tdnn1(mel)))
x = F.relu(self.bn2(
self.tdnn2(x)))
x = F.relu(self.bn3(
self.tdnn3(x)))
x = F.relu(self.bn4(
self.tdnn4(x)))
x = F.relu(self.bn5(
self.tdnn5(x)))
# Statistics pooling
mean = x.mean(dim=2)
std = x.std(dim=2)
pooled = torch.cat(
[mean, std], dim=1)
embedding = self.embed(pooled)
return F.normalize(
embedding, dim=1)
model = XVectorModel()
mel = torch.randn(2, 40, 300)
emb = model(mel)
print(f"Input: {mel.shape}")
print(f"Output: {emb.shape}")
total = sum(p.numel()
for p in model.parameters())
print(f"Parameters: {total:,}")
print(f"\nThe statistics pooling "
f"captures BOTH the mean "
f"spectral profile")
print(f"AND the variability -- "
f"how much a speaker's voice "
f"fluctuates over time.")
The statistics pooling is the key innovation here. By computing the standard deviation across time in addition to the mean, the model captures not just what the voice sounds like on average but how much it varies. Some speakers have very consistent vocal characteristics across an utterance; others modulate pitch and energy dramatically. That variation pattern is itself an identity signal. None the less, the dilation factors in the TDNN layers (2, 3) give later layers progressively wider temporal context without increasing the number of parameters -- same idea as dilated convolutions in WaveNet (episode #94).
Modern speaker embedding models like ECAPA-TDNN (Desplanques et al., 2020) improve on x-vectors with three additions: channel attention (Squeeze-and-Excitation blocks that let the network focus on the most informative frequency channels), residual connections (so gradients flow through deep networks), and multi-layer feature aggregation (combining features from multiple TDNN layers before pooling, rather than using only the last layer):
class ChannelAttention(nn.Module):
"""Squeeze-and-Excitation block."""
def __init__(self, ch, r=8):
super().__init__()
self.pool = nn.AdaptiveAvgPool1d(1)
self.fc = nn.Sequential(
nn.Linear(ch, ch // r),
nn.ReLU(),
nn.Linear(ch // r, ch),
nn.Sigmoid())
def forward(self, x):
w = self.pool(x).squeeze(-1)
return x * self.fc(w).unsqueeze(-1)
class AttentiveStatsPool(nn.Module):
"""Attention-weighted stats pool."""
def __init__(self, ch):
super().__init__()
self.attn = nn.Sequential(
nn.Conv1d(ch, 128, 1),
nn.ReLU(),
nn.Conv1d(128, ch, 1),
nn.Softmax(dim=2))
def forward(self, x):
w = self.attn(x)
mean = (x * w).sum(dim=2)
var = ((x**2*w).sum(dim=2)
- mean**2)
std = torch.sqrt(
var.clamp(min=1e-9))
return torch.cat(
[mean, std], dim=1)
class SimpleECAPATDNN(nn.Module):
def __init__(self, n_mels=80,
embed_dim=192):
super().__init__()
self.layer1 = nn.Conv1d(
n_mels, 512, 5, padding=2)
self.bn1 = nn.BatchNorm1d(512)
self.se1 = ChannelAttention(512)
self.layer2 = nn.Conv1d(
512, 512, 3, dilation=2,
padding=2)
self.bn2 = nn.BatchNorm1d(512)
self.se2 = ChannelAttention(512)
self.layer3 = nn.Conv1d(
512, 512, 3, dilation=3,
padding=3)
self.bn3 = nn.BatchNorm1d(512)
self.se3 = ChannelAttention(512)
self.mfa = nn.Conv1d(
512*3, 1536, 1)
self.bn_mfa = nn.BatchNorm1d(1536)
self.asp = AttentiveStatsPool(1536)
self.embed = nn.Linear(
1536*2, embed_dim)
def forward(self, mel):
x1 = self.se1(F.relu(
self.bn1(self.layer1(mel))))
x2 = self.se2(F.relu(
self.bn2(self.layer2(x1))))
x3 = self.se3(F.relu(
self.bn3(self.layer3(x2))))
cat = F.relu(self.bn_mfa(
self.mfa(torch.cat(
[x1, x2, x3], dim=1))))
pooled = self.asp(cat)
return F.normalize(
self.embed(pooled), dim=1)
model = SimpleECAPATDNN()
mel = torch.randn(2, 80, 300)
emb = model(mel)
total = sum(p.numel()
for p in model.parameters())
print(f"Input: {mel.shape}")
print(f"Output: {emb.shape}")
print(f"Parameters: {total:,}")
The attentive statistics pooling learns which frames are most informative for speaker identity. Frames with clear speech get high attention; noisy or silent frames get low attention. This makes the model robust to variable-length inputs with noisy segments.
These models are trained with angular margin losses like AAM-Softmax -- the same ArcFace loss from episode #88 for face recognition. The loss pushes same-speaker embeddings closer on the unit hypersphere while enforcing angular margin between different speakers. Training datasets like VoxCeleb (7,000+ celebrities, 1M+ utterances from YouTube) provide the scale needed to generalize.
Before you can identify speakers, you need to know where speech actually occurs. Voice Activity Detection (VAD) classifies each audio frame as speech or non-speech. It sounds trivial -- and for clean studio audio, it practically is. But in real-world recordings with background music, air conditioning hum, keyboard typing, and other non-speech sounds, VAD becomes a serious preprocessing challenge:
import numpy as np
class SimpleVAD:
"""Energy-based Voice Activity
Detection with adaptive threshold."""
def __init__(self, sr=16000,
frame_ms=25,
hop_ms=10):
self.sr = sr
self.frame_len = int(
sr * frame_ms / 1000)
self.hop_len = int(
sr * hop_ms / 1000)
def compute_energy(self, audio):
"""Frame-level energy in dB."""
n_frames = (
(len(audio) - self.frame_len)
// self.hop_len + 1)
energy = np.zeros(n_frames)
for i in range(n_frames):
start = i * self.hop_len
frame = audio[
start:start+self.frame_len]
rms = np.sqrt(
np.mean(frame ** 2))
energy[i] = 20 * np.log10(
max(rms, 1e-10))
return energy
def detect(self, audio,
energy_thresh_db=-30,
min_speech_ms=200,
min_silence_ms=100):
"""Detect speech segments."""
energy = self.compute_energy(
audio)
speech = energy > energy_thresh_db
hop_ms = (self.hop_len / self.sr
* 1000)
min_speech = int(
min_speech_ms / hop_ms)
min_silence = int(
min_silence_ms / hop_ms)
# Fill short gaps
result = speech.copy()
in_gap = False
gap_start = 0
for i in range(len(result)):
if not result[i] and (
not in_gap):
gap_start = i
in_gap = True
elif result[i] and in_gap:
if i - gap_start < (
min_silence):
result[
gap_start:i] = True
in_gap = False
# Remove short segments
in_seg = False
seg_start = 0
for i in range(len(result)):
if result[i] and not in_seg:
seg_start = i
in_seg = True
elif not result[i] and in_seg:
if i - seg_start < (
min_speech):
result[
seg_start:i] = False
in_seg = False
return result, energy
def extract_segments(self, vad):
"""Convert frame-level VAD to
time segments."""
segments = []
in_seg = False
start = 0
for i in range(len(vad)):
if vad[i] and not in_seg:
start = i
in_seg = True
elif not vad[i] and in_seg:
t_start = (
start * self.hop_len
/ self.sr)
t_end = (
i * self.hop_len
/ self.sr)
segments.append(
(t_start, t_end))
in_seg = False
if in_seg:
t_start = (
start * self.hop_len
/ self.sr)
t_end = (
len(vad) * self.hop_len
/ self.sr)
segments.append(
(t_start, t_end))
return segments
def run(self):
rng = np.random.RandomState(42)
sr = self.sr
dur = 5.0
t = np.arange(int(sr * dur)) / sr
audio = rng.randn(len(t)) * 0.001
# Two speech segments
for start, end, f0 in [
(0.5, 1.8, 150),
(2.5, 4.0, 180)]:
s = slice(int(start*sr),
int(end*sr))
audio[s] += 0.3 * np.sin(
2*np.pi*f0*t[s])
audio[s] += 0.15 * np.sin(
2*np.pi*f0*2*t[s])
vad, energy = self.detect(audio)
segs = self.extract_segments(vad)
print(f"Audio: {dur}s")
print(f"Detected speech segments:")
for start, end in segs:
print(f" {start:.2f}s - "
f"{end:.2f}s")
print(f"Expected: ~0.5-1.8s "
f"and ~2.5-4.0s")
vad = SimpleVAD()
vad.run()
VAD might seem like a solved problem, but it's absolutly critical for downstream quality. Without it, silence and background noise get processed as speech, which wastes compute and (worse) contaminates speaker embeddings with non-speech features. Production systems like Silero VAD use small neural networks (just a few megabytes) trained on massive amounts of audio with precise speech/non-speech labels. They run in real-time on a CPU, processing hours of audio in seconds.
Let's put the pieces together and build a complete speaker verification system. We'll synthesize multi-speaker audio, extract simple spectral features, compute embeddings, and verify speakers using cosine similarity:
import numpy as np
class SpeakerVerificationSystem:
"""Complete speaker verification
pipeline from scratch."""
def __init__(self, sr=16000,
n_mels=40,
embed_dim=64):
self.sr = sr
self.n_mels = n_mels
self.embed_dim = embed_dim
self.rng = np.random.RandomState(
42)
def synthesize_speaker(self,
f0_range,
formants,
duration=2.0,
n_utterances=5):
"""Synthetic speech-like audio."""
utterances = []
for _ in range(n_utterances):
n = int(self.sr * duration)
t = np.arange(n) / self.sr
f0 = self.rng.uniform(
*f0_range)
signal = np.zeros(n)
for f, a in formants:
signal += a * np.sin(
2*np.pi*f*t)
for h in range(2, 6):
signal += (0.3/h) * np.sin(
2*np.pi*f0*h*t)
signal *= (0.5 + 0.5 * np.sin(
2*np.pi*4.5*t))
signal += self.rng.randn(
n) * 0.02
utterances.append(signal)
return utterances
def mel_spectrogram(self, audio):
"""Simplified Mel spectrogram."""
n_fft = 512
hop = 160
n_frames = (
len(audio) - n_fft) // hop
spec = np.zeros(
(self.n_mels, n_frames))
window = np.hanning(n_fft)
for i in range(n_frames):
start = i * hop
frame = audio[
start:start + n_fft]
fft_mag = np.abs(
np.fft.rfft(
frame * window))
# Simple Mel binning
n_bins = len(fft_mag)
mel_bins = np.linspace(
0, n_bins,
self.n_mels + 1,
dtype=int)
for m in range(self.n_mels):
lo = mel_bins[m]
hi = mel_bins[m + 1]
if hi > lo:
spec[m, i] = np.mean(
fft_mag[lo:hi])
return np.log(spec + 1e-9)
def extract_embedding(self, mel):
"""Simple embedding: mean + std
of mel features, projected."""
mean = np.mean(mel, axis=1)
std = np.std(mel, axis=1)
features = np.concatenate(
[mean, std])
# Random projection (in a real
# system this is a learned
# network)
proj = self.rng.randn(
self.embed_dim,
len(features)) * 0.1
emb = proj @ features
norm = np.linalg.norm(emb)
return emb / max(norm, 1e-9)
def cosine_sim(self, a, b):
return float(
np.dot(a, b)
/ (np.linalg.norm(a)
* np.linalg.norm(b)))
def run(self):
speaker_configs = {
'Alice': {'f0_range': (180, 220),
'formants': [(800,.4),
(1200,.3), (2500,.15)]},
'Bob': {'f0_range': (100, 130),
'formants': [(600,.5),
(1000,.25), (2200,.1)]},
'Carol': {'f0_range': (200, 240),
'formants': [(850,.35),
(1300,.28), (2700,.12)]},
'Dave': {'f0_range': (90, 115),
'formants': [(550,.45),
(950,.2), (2000,.1)]},
}
# Generate utterances
speakers = {}
for name, cfg in (
speaker_configs.items()):
utts = self.synthesize_speaker(
cfg['f0_range'],
cfg['formants'])
speakers[name] = utts
# Extract embeddings
embeddings = {}
for name, utts in (
speakers.items()):
embs = []
for utt in utts:
mel = self.mel_spectrogram(
utt)
emb = self.extract_embedding(
mel)
embs.append(emb)
embeddings[name] = embs
# Enrollment: use first 3
# utterances per speaker
enrolled = {}
for name, embs in (
embeddings.items()):
enrolled[name] = np.mean(
embs[:3], axis=0)
enrolled[name] /= (
np.linalg.norm(
enrolled[name]))
# Verification trials
print("=== Verification Trials ===")
print(f"{'Claim':>8} {'Actual':>8}"
f" {'Score':>7}"
f" {'Result':>8}")
print("-" * 36)
trials = [
('Alice', 'Alice', 3),
('Alice', 'Alice', 4),
('Bob', 'Bob', 3),
('Alice', 'Carol', 3),
('Bob', 'Dave', 4),
('Carol', 'Bob', 3),
]
threshold = 0.6
for claim, actual, idx in trials:
test_emb = embeddings[
actual][idx]
score = self.cosine_sim(
enrolled[claim], test_emb)
accept = score > threshold
correct = (
(accept and
claim == actual)
or (not accept and
claim != actual))
print(f"{claim:>8} {actual:>8}"
f" {score:>7.3f}"
f" {'ACCEPT' if accept else 'REJECT':>8}"
f" {'OK' if correct else 'ERR'}")
system = SpeakerVerificationSystem()
system.run()
In a real system, the "random projection" would be replaced by a trained neural network (d-vector LSTM or ECAPA-TDNN), but the pipeline structure is identical: synthesize/capture audio, extract Mel spectrogram, compute embedding, compare with cosine similarity. The threshold (0.6 in our demo) is the critical operating parameter -- too low and impostors get accepted (high false acceptance rate), too high and legitimate speakers get rejected (high false rejection rate).
Diarization segments a multi-speaker recording into speaker-homogenous regions. The output is a timeline showing which speaker is talking at each moment, without necessarily knowing who the speakers are (they're labeled as Speaker 0, Speaker 1, etc.). Think meeting recordings, podcast transcriptions, or courtroom audio analysis.
The traditional diarization pipeline looks like this:
Audio -> VAD -> Segmentation -> Embeddings -> Clustering -> Speaker labels
Let's build one:
import numpy as np
class SpeakerDiarizer:
"""Pipeline-based speaker
diarization system."""
def __init__(self, sr=16000,
seg_dur=1.5,
seg_hop=0.75):
self.sr = sr
self.seg_dur = seg_dur
self.seg_hop = seg_hop
self.rng = np.random.RandomState(
42)
def generate_meeting(self):
"""Simulate a 20s meeting
with 3 speakers."""
dur = 20.0
n = int(self.sr * dur)
audio = self.rng.randn(n) * 0.005
profiles = {
0: (120, [500, 1000, 2200]),
1: (200, [800, 1200, 2600]),
2: (150, [650, 1100, 2400])}
timeline = [
(0, 0.5, 3.2), (1, 3.5, 6.0),
(0, 6.3, 8.5), (2, 8.8, 12.0),
(1, 12.3, 14.5),
(2, 14.8, 17.0),
(0, 17.2, 19.5)]
for spk, start, end in timeline:
f0, fmts = profiles[spk]
s = int(start * self.sr)
e = int(end * self.sr)
t = np.arange(e-s) / self.sr
sig = sum(0.2 * np.sin(
2*np.pi*f*t) for f in fmts)
for h in range(2, 5):
sig += (0.15/h) * np.sin(
2*np.pi*f0*h*t)
sig *= (0.5 + 0.5 * np.sin(
2*np.pi*3*t))
audio[s:e] += sig
return audio, timeline
def segment_audio(self, audio):
"""Split audio into overlapping
segments."""
seg_samples = int(
self.seg_dur * self.sr)
hop_samples = int(
self.seg_hop * self.sr)
segments = []
starts = []
i = 0
while i + seg_samples <= (
len(audio)):
seg = audio[
i:i + seg_samples]
# Skip near-silent segments
if np.sqrt(np.mean(
seg**2)) > 0.01:
segments.append(seg)
starts.append(
i / self.sr)
i += hop_samples
return segments, starts
def extract_features(self, segment):
"""Simple spectral features
as speaker embedding proxy."""
n_fft = 512
hop = 160
n_frames = (
len(segment) - n_fft) // hop
if n_frames < 1:
return np.zeros(80)
spec = np.zeros(
(40, n_frames))
w = np.hanning(n_fft)
for i in range(n_frames):
s = i * hop
fft = np.abs(np.fft.rfft(
segment[s:s+n_fft] * w))
bins = np.linspace(
0, len(fft), 41,
dtype=int)
for m in range(40):
if bins[m+1] > bins[m]:
spec[m, i] = np.mean(
fft[bins[m]:
bins[m+1]])
spec = np.log(spec + 1e-9)
mean = np.mean(spec, axis=1)
std = np.std(spec, axis=1)
return np.concatenate(
[mean, std])
def cluster_embeddings(self, embs,
n_clusters=3):
"""Simple K-means clustering."""
# Initialize with K-means++
centers = [embs[0].copy()]
for _ in range(1, n_clusters):
dists = np.array([
min(np.linalg.norm(
e - c)**2
for c in centers)
for e in embs])
probs = dists / dists.sum()
idx = self.rng.choice(
len(embs), p=probs)
centers.append(
embs[idx].copy())
centers = np.array(centers)
# Iterate
for _ in range(50):
# Assign
labels = np.array([
np.argmin([
np.linalg.norm(e - c)
for c in centers])
for e in embs])
# Update
new_centers = np.array([
embs[labels == k].mean(
axis=0)
if (labels == k).any()
else centers[k]
for k in range(
n_clusters)])
if np.allclose(
centers, new_centers):
break
centers = new_centers
return labels
def run(self):
audio, gt_timeline = (
self.generate_meeting())
print(f"Audio: {len(audio)/self.sr:.1f}s")
print(f"\nGround truth:")
for spk, s, e in gt_timeline:
print(f" Speaker {spk}: "
f"{s:.1f}s - {e:.1f}s")
# Pipeline
segments, starts = (
self.segment_audio(audio))
embs = np.array([
self.extract_features(seg)
for seg in segments])
labels = self.cluster_embeddings(
embs, n_clusters=3)
print(f"\nDiarization output "
f"({len(segments)} segments):")
prev_label = -1
for i, (start, label) in (
enumerate(
zip(starts, labels))):
end = start + self.seg_dur
if label != prev_label:
print(f" Speaker_{label}: "
f"{start:.1f}s - "
f"{end:.1f}s")
prev_label = label
# Simple DER at 100ms resolution
gt_l = np.zeros(200)
for spk, s, e in gt_timeline:
gt_l[int(s*10):int(e*10)] = (
spk + 1)
pr_l = np.zeros(200)
for start, label in zip(
starts, labels):
s = int(start * 10)
e = min(int((start +
self.seg_dur) * 10), 200)
pr_l[s:e] = label + 1
speech = (gt_l > 0).sum()
confusion = ((gt_l > 0)
& (pr_l > 0)
& (gt_l != pr_l)).sum()
missed = ((gt_l > 0)
& (pr_l == 0)).sum()
fa = ((gt_l == 0)
& (pr_l > 0)).sum()
der = ((confusion + missed + fa)
/ max(speech, 1))
print(f"\nDER: {der:.1%} "
f"(confusion={confusion}, "
f"missed={missed}, "
f"fa={fa})")
diarizer = SpeakerDiarizer()
diarizer.run()
The Diarization Error Rate (DER) is the standard evaluation metric. It has three components: missed speech (speech that was labeled as silence), false alarm (silence that was labeled as speech), and speaker confusion (speech that was correctly detected but assigned to the wrong speaker). State-of-the-art systems achieve 3-7% DER on benchmark datasets like AMI and CALLHOME.
Production diarization systems like pyannote.audio use end-to-end neural approaches that handle overlapping speech (two people talking simultaneously) -- a scenario where our clustering-based pipeline fails because a single time frame contains features from multiple speakers. The pyannote system uses a segmentation network trained to predict speaker activity per-frame for each possible speaker, followed by a neural clustering step.
The ultimate goal for meeting transcription is answering: who said what when? This requires combining speech recognition (episode #93) with speaker diarization. The approach is straightforward: run ASR to get word-level timestamps, run diarization to get speaker-level timestamps, then match each word to its speaker based on time overlap:
import numpy as np
class ASRDiarizationMerger:
"""Merge ASR transcripts with
speaker diarization labels."""
def __init__(self):
self.rng = np.random.RandomState(
42)
def simulate_asr_output(self):
"""Simulated Whisper output."""
words = [
('Okay', .5, .8),
('let', .85, .95),
('me', .95, 1.05),
('start', 1.1, 1.4),
('the', 1.45, 1.55),
('meeting', 1.6, 2.0),
('Thanks', 3.5, 3.8),
('for', 3.85, 3.95),
('joining', 4.0, 4.3),
('everyone', 4.35, 4.8),
('I', 6.3, 6.4),
('have', 6.4, 6.6),
('the', 6.65, 6.75),
('updates', 6.8, 7.2),
('ready', 7.25, 7.5)]
return [{'word': w, 'start': s,
'end': e}
for w, s, e in words]
def simulate_diarization(self):
"""Simulated pyannote output."""
return [
('SPEAKER_00', 0.3, 2.2),
('SPEAKER_01', 3.3, 5.0),
('SPEAKER_00', 6.1, 7.8)]
def assign_speakers(self, words,
diar):
"""Map each word to its speaker
based on timestamp overlap."""
result = []
for w in words:
w_mid = (
w['start'] + w['end']
) / 2
speaker = "UNKNOWN"
for spk, s, e in diar:
if s <= w_mid <= e:
speaker = spk
break
result.append({
'speaker': speaker,
'word': w['word'],
'start': w['start']})
return result
def format_transcript(self, merged):
"""Group words by speaker."""
lines = []
cur_spk = None
words = []
for item in merged:
if item['speaker'] != cur_spk:
if words:
lines.append(
(cur_spk,
' '.join(words)))
cur_spk = item['speaker']
words = [item['word']]
else:
words.append(item['word'])
if words:
lines.append(
(cur_spk,
' '.join(words)))
return lines
def run(self):
words = (
self.simulate_asr_output())
diar = (
self.simulate_diarization())
print("=== Speaker-Attributed "
"Transcript ===\n")
merged = self.assign_speakers(
words, diar)
lines = self.format_transcript(
merged)
for spk, text in lines:
print(f"[{spk}]: {text}")
print(f"\n--- Stats ---")
print(f"Total words: {len(words)}")
spk_counts = {}
for item in merged:
spk = item['speaker']
spk_counts[spk] = (
spk_counts.get(spk, 0)
+ 1)
for spk, count in sorted(
spk_counts.items()):
print(f" {spk}: "
f"{count} words")
merger = ASRDiarizationMerger()
merger.run()
This is exactly how modern meeting transcription services work -- Whisper (or another ASR model from episode #93) provides the "what was said" with precise timestamps, and pyannote (or another diarizer) provides the "who was speaking." The merger assigns each word to the speaker who was active at that word's midpoint timestamp.
The main challenge is alignment: ASR and diarization models sometimes disagree about exact transition boundaries. A word might start during one speaker's turn and end during another's (during a rapid speaker change). Production systems handle this with more sophisticated overlap resolution, but the midpoint heuristic works surprisingly well in practice.
Let's build proper evaluation metrics for both speaker verification and diarization:
import numpy as np
class SpeakerMetrics:
"""Compute EER for verification
and DER for diarization."""
def __init__(self):
self.rng = np.random.RandomState(
42)
def compute_eer(self, pos_scores,
neg_scores):
"""Equal Error Rate: threshold
where FAR == FRR."""
thresholds = np.linspace(
0, 1, 1000)
best_diff = float('inf')
eer = 0.0
eer_thresh = 0.0
for t in thresholds:
# False Rejection Rate
frr = np.mean(
np.array(pos_scores) < t)
# False Acceptance Rate
far = np.mean(
np.array(neg_scores) >= t)
diff = abs(far - frr)
if diff < best_diff:
best_diff = diff
eer = (far + frr) / 2
eer_thresh = t
return eer, eer_thresh
def run(self):
# Simulate verification scores
pos = self.rng.normal(
0.82, 0.08, 200)
neg = self.rng.normal(
0.35, 0.12, 200)
eer, thresh = self.compute_eer(
pos, neg)
print(f"=== Speaker Verification ===")
print(f"EER: {eer:.2%}")
print(f"Threshold: {thresh:.3f}")
print(f"Positive mean: "
f"{np.mean(pos):.3f}")
print(f"Negative mean: "
f"{np.mean(neg):.3f}")
# Compare systems
print(f"\n=== System Comparison ===")
systems = {
'Basic x-vector': (
self.rng.normal(.78,.1,200),
self.rng.normal(.4,.15,200)),
'ECAPA-TDNN': (
self.rng.normal(.88,.05,200),
self.rng.normal(.25,.08,200)),
'ECAPA + LM': (
self.rng.normal(.92,.04,200),
self.rng.normal(.2,.06,200))}
print(f"{'System':<16} {'EER':>8}")
print("-" * 26)
for name, (p, n) in (
systems.items()):
e, _ = self.compute_eer(p, n)
print(f"{name:<16} {e:>7.2%}")
metrics = SpeakerMetrics()
metrics.run()
EER (Equal Error Rate) is the threshold where the False Acceptance Rate (impostors accepted) equals the False Rejection Rate (genuine speakers rejected). Modern systems achieve <1% EER on VoxCeleb benchmarks, which means that at the operating threshold, fewer than 1 in 100 imposter attempts succeed AND fewer than 1 in 100 genuine speakers are locked out.
The system comparison shows how architectural improvements translate to EER gains: basic x-vectors might achieve 5-8% EER, ECAPA-TDNN pushes to 1-3%, and adding score normalization or language-model-based resegmentation (for diarization) pushes further below 1%.
There's more to explore in the audio domain -- understanding natural language intent from spoken utterances, enhancing noisy audio recordings, and combining audio and visual streams into unified representations. The human voice carries a remarkable amount of information beyond just the words.
Exercise 1: Build a speaker enrollment and verification simulator. Create a class SpeakerEnrollment that generates 6 synthetic speakers with distinct f0 ranges (80-250 Hz) and formant patterns, produces 5 enrollment + 3 test utterances each (1.5s, sr=16000), computes spectral embeddings (Mel spectrogram -> mean+std -> L2 normalize), averages enrollment embeddings per speaker, then runs all 108 verification trials (every test vs every enrolled speaker). Sweep thresholds 0.0-1.0 to find the EER. Print the EER, threshold, and a table of same-speaker vs different-speaker scores.
Exercise 2: Build a speaker change detector. Create a class SpeakerChangeDetector that generates 15s of audio with 4 segments (A: 0-3.5s, B: 3.5-7s, A: 7-10.5s, C: 10.5-15s). Use sliding windows (1.5s, hop 0.25s) to extract spectral features, compute cosine distance between consecutive windows, and detect peaks (local maxima above mean + 1.5*std). A change is "detected" if a peak falls within +/-0.5s of the true boundary. Print detected vs ground truth change points, precision, and recall.
Exercise 3: Build a multi-speaker overlap detector. Create a class OverlapDetector that generates 12s of audio: A alone (0-3s), A+B overlap (3-5s), B alone (5-8s), B+C overlap (8-10s), C alone (10-12s). Analyze energy in 4 frequency bands per frame, compute an overlap score (0-1) based on active band count and total energy. Compare predicted overlap frames to ground truth (3-5s and 8-10s), print precision, recall, F1, and average overlap score in overlap vs non-overlap regions.
@scipio