Learn AI Series):Exercise 1: The intent classifier maps 7 classes (weather, timer, music, lights, news, reminder, search) using keyword overlap scoring. Each intent has a keyword list; classification computes the fraction of keywords matched per intent, normalizes by total score, and rejects below a 0.4 threshold. "What is the weather like" confidently maps to WEATHER. "Tell me about quantum physics" produces low scores across all intents and correctly returns UNKNOWN. Testing 50 random utterances from the combined vocabulary shows clear separation between in-domain (above 0.4) and out-of-domain (below 0.4) queries.
Exercise 2: The slot filler tracks required slots per intent (city for weather, duration for timer, song_or_artist for music). "Set a timer" triggers FILLING state with missing duration -- the system asks "How long?". After "5 minutes", pattern matching extracts the duration, the slot fills, and the state transitions to COMPLETE. Three test dialogues all complete in 2 turns, with the state machine resetting cleanly between conversations.
Exercise 3: The error recovery pipeline simulates ASR noise at three levels. At clean SNR, all words pass through with 0.95 confidence. At medium noise, 30% of words get single-character substitutions (confidence 0.4) -- edit distance correction against a vocabulary recovers most errors. At high noise, words get deletions (20%) and heavy corruption (30%) -- average confidence drops below threshold, triggering "I didn't quite catch that" clarification rather than executing a wrong command. Substitution errors dominate at medium noise; deletions increase at high noise.
Here we go! We're nearing the end of the audio AI arc and episode 99 brings us to a topic that underpins everything we've built so far. Over the past seven episodes we've covered how machines hear sound (#92), transcribe speech (#93), synthesize it back (#94), classify environmental sounds (#95), generate music (#96), recognize who's speaking (#97), and understand spoken intent (#98). All of those assumed the audio was reasonably clean -- or at least clean enough to work with. Today we deal with the reality that audio in the wild is almost never clean.
Audio enhancement is a family of techniques for taking messy, noisy, degraded recordings and making them sound like they were captured in a proper studio. You're listening to a podcast recorded in someone's kitchen. The fridge hums. Dishes clatter somewhere in the background. Traffic bleeds through the window. The content might be brilliant, but the audio makes you want to reach for the stop button. That is the problem we're solving today.
And here's the thing -- audio enhancement isn't just a nice-to-have cosmetic improvement. It's a critical preprocessing step for every other audio AI system we've built in this series. Speech recognition accuracy drops dramatically in noisy conditions. Speaker verification fails when noise corrupts the voice embeddings we so carefully extracted in episode #97. Intent classification misses commands buried under background chatter. Fix the audio first, and every downstream system works better. The classical methods are elegant and still widely deployed in production. The deep learning methods are dramaticaly better. We'll build both ;-)
The idea behind spectral subtraction is beautifully simple. If you have a noisy signal and can estimate the noise spectrum, just subtract the noise from the signal in the frequency domain. The assumption: noise is additive. The noisy signal Y = S + N, where S is clean speech and N is noise. In the frequency domain: |Y| = |S| + |N| (approximately). So |S| is roughly |Y| - |N|. Let's implement a complete spectral subtractor from scratch using only NumPy:
import numpy as np
class SpectralSubtractor:
"""Classical spectral subtraction
for noise reduction."""
def __init__(self, sr=16000,
n_fft=1024, hop=256):
self.sr = sr
self.n_fft = n_fft
self.hop = hop
def stft(self, audio):
"""Short-time Fourier transform."""
window = np.hanning(self.n_fft)
n_frames = (
(len(audio) - self.n_fft)
// self.hop + 1)
n_bins = self.n_fft // 2 + 1
result = np.zeros(
(n_bins, n_frames),
dtype=complex)
for i in range(n_frames):
start = i * self.hop
frame = audio[
start:start + self.n_fft]
result[:, i] = np.fft.rfft(
frame * window)
return result
def istft(self, stft_matrix, length):
"""Inverse STFT with overlap-add."""
window = np.hanning(self.n_fft)
n_frames = stft_matrix.shape[1]
output = np.zeros(length)
window_sum = np.zeros(length)
for i in range(n_frames):
start = i * self.hop
frame = np.fft.irfft(
stft_matrix[:, i])
end = min(start + self.n_fft,
length)
flen = end - start
output[start:end] += (
frame[:flen] * window[:flen])
window_sum[start:end] += (
window[:flen] ** 2)
mask = window_sum > 1e-8
output[mask] /= window_sum[mask]
return output
def denoise(self, noisy, noise_frames=30,
oversubtract=1.0,
floor=0.002):
"""Spectral subtraction with
spectral flooring."""
spec = self.stft(noisy)
magnitude = np.abs(spec)
phase = np.angle(spec)
# Estimate noise from first
# N frames (assumed noise-only)
noise_est = np.mean(
magnitude[:, :noise_frames],
axis=1, keepdims=True)
# Subtract with oversubtraction
# factor and spectral floor
clean_mag = np.maximum(
magnitude
- oversubtract * noise_est,
floor * noise_est)
# Reconstruct with original phase
clean_spec = clean_mag * np.exp(
1j * phase)
return self.istft(
clean_spec, len(noisy))
def run(self):
rng = np.random.RandomState(42)
dur = 3.0
n = int(self.sr * dur)
t = np.arange(n) / self.sr
# Synthesize "speech": formants
speech = np.zeros(n)
for f0, amp in [(150, 0.3),
(300, 0.2),
(900, 0.1),
(2500, 0.05)]:
speech += amp * np.sin(
2 * np.pi * f0 * t)
# Amplitude modulation (syllables)
speech *= (0.5 + 0.5 * np.sin(
2 * np.pi * 4 * t))
# Only active 0.3s-2.5s
envelope = np.zeros(n)
envelope[int(0.3*self.sr):
int(2.5*self.sr)] = 1.0
speech *= envelope
# Add noise
noise = rng.randn(n) * 0.08
noisy = speech + noise
# Denoise
clean = self.denoise(noisy)
# Measure SNR improvement
speech_power = np.mean(
speech[int(0.5*self.sr):
int(2.0*self.sr)] ** 2)
noise_region = slice(
0, int(0.2 * self.sr))
input_noise = np.mean(
noisy[noise_region] ** 2)
output_noise = np.mean(
clean[noise_region] ** 2)
input_snr = 10 * np.log10(
speech_power
/ max(input_noise, 1e-10))
output_snr = 10 * np.log10(
speech_power
/ max(output_noise, 1e-10))
print(f"Audio: {dur}s at "
f"{self.sr} Hz")
print(f"Input SNR: "
f"{input_snr:.1f} dB")
print(f"Output SNR: "
f"{output_snr:.1f} dB")
print(f"Improvement: "
f"{output_snr - input_snr:.1f}"
f" dB")
sub = SpectralSubtractor()
sub.run()
It works. But it has a well-known artifact: musical noise. When you subtract a noise estimate from a noisy spectrum, small random fluctuations produce isolated spectral peaks that sound like random tonal blips -- almost like a digital xylophone playing random notes underneath the speech. The spectral_floor parameter helps (instead of setting bins to zero, we set them to a small fraction of the noise estimate), but musical noise is never fully eliminated.
The bigger limitation: spectral subtraction assumes stationary noise -- a constant hum, fan, or hiss. The moment noise changes character (someone coughs, a door slams, traffic varies), the noise estimate becomes wrong and artifacts appear everywhere.
The Wiener filter is the mathematically optimal linear filter for noise reduction when you know (or can estimate) the signal and noise power spectra. Instead of hard subtraction, it computes a gain mask -- a value between 0 and 1 for each frequency bin at each time frame. The formula is elegant: gain = SNR / (SNR + 1), where SNR is the estimated signal-to-noise ratio per bin:
class WienerDenoiser:
"""Wiener filter for noise reduction
with gain smoothing."""
def __init__(self, sr=16000,
n_fft=1024, hop=256):
self.sr = sr
self.n_fft = n_fft
self.hop = hop
def stft(self, audio):
window = np.hanning(self.n_fft)
n_frames = (
(len(audio) - self.n_fft)
// self.hop + 1)
n_bins = self.n_fft // 2 + 1
result = np.zeros(
(n_bins, n_frames),
dtype=complex)
for i in range(n_frames):
s = i * self.hop
result[:, i] = np.fft.rfft(
audio[s:s+self.n_fft]
* window)
return result
def istft(self, spec, length):
window = np.hanning(self.n_fft)
n_frames = spec.shape[1]
output = np.zeros(length)
wsum = np.zeros(length)
for i in range(n_frames):
s = i * self.hop
frame = np.fft.irfft(
spec[:, i])
e = min(s + self.n_fft, length)
fl = e - s
output[s:e] += (
frame[:fl] * window[:fl])
wsum[s:e] += window[:fl] ** 2
mask = wsum > 1e-8
output[mask] /= wsum[mask]
return output
def denoise(self, noisy,
noise_frames=30,
smooth_alpha=0.9):
"""Wiener filter with temporal
gain smoothing."""
spec = self.stft(noisy)
power = np.abs(spec) ** 2
phase = np.angle(spec)
# Estimate noise power
noise_power = np.mean(
power[:, :noise_frames],
axis=1, keepdims=True)
# A priori SNR estimate
signal_power = np.maximum(
power - noise_power, 0.0)
# Wiener gain: SNR / (SNR + 1)
snr = signal_power / (
noise_power + 1e-10)
gain = snr / (snr + 1)
# Temporal smoothing to reduce
# musical noise
smoothed = np.zeros_like(gain)
smoothed[:, 0] = gain[:, 0]
for i in range(1, gain.shape[1]):
smoothed[:, i] = (
smooth_alpha
* smoothed[:, i-1]
+ (1 - smooth_alpha)
* gain[:, i])
clean_spec = (np.abs(spec)
* smoothed
* np.exp(1j * phase))
return self.istft(
clean_spec, len(noisy))
def compare_methods(self):
"""Compare Wiener vs spectral
subtraction."""
rng = np.random.RandomState(42)
dur = 3.0
n = int(self.sr * dur)
t = np.arange(n) / self.sr
speech = np.zeros(n)
for f0, a in [(150, 0.3),
(450, 0.15),
(1200, 0.08)]:
speech += a * np.sin(
2*np.pi*f0*t)
speech *= (0.5 + 0.5 * np.sin(
2*np.pi*3.5*t))
env = np.zeros(n)
env[int(0.3*self.sr):
int(2.5*self.sr)] = 1.0
speech *= env
print(f"{'SNR_in':>8} {'Wiener':>10} "
f"{'SpSub':>10} {'Winner':>8}")
print("-" * 40)
for snr_db in [20, 10, 5, 0, -5]:
sp_pow = np.mean(speech ** 2)
noise_pow = sp_pow / (
10 ** (snr_db / 10))
noise = rng.randn(n) * np.sqrt(
noise_pow)
noisy = speech + noise
w_clean = self.denoise(noisy)
w_err = np.mean(
(speech - w_clean) ** 2)
sub = SpectralSubtractor(
self.sr, self.n_fft,
self.hop)
s_clean = sub.denoise(noisy)
s_err = np.mean(
(speech - s_clean) ** 2)
winner = ("Wiener"
if w_err < s_err
else "SpSub")
print(f"{snr_db:>6}dB "
f"{w_err:>10.6f} "
f"{s_err:>10.6f} "
f"{winner:>8}")
wd = WienerDenoiser()
wd.compare_methods()
The Wiener filter produces smoother output than spectral subtraction because the gain transitions continuously rather than hard-clipping at zero. The temporal smoothing (exponential moving average on the gain) further reduces musical noise by preventing the gain from flickering rapidly between frames. But it shares the same fundamental limitation: it needs a good noise estimate, and it assumes the noise doesn't change dramatically over time.
Notice the comparison: Wiener consistently beats spectral subtraction, especially at lower SNR levels where hard subtraction produces the worst musical noise artifacts. The softer gain curve preserves more of the speech signal while still suppressing noise effectively. At -5dB SNR (where noise is louder than speech!) both methods struggle, but Wiener degrades more gracefully.
Neural networks learn to denoise without explicit noise estimation. You train them on pairs of (noisy audio, clean audio) and the network learns a direct mapping. The key insight: the network can handle non-stationary noise because it learns patterns in both speech and noise, adapting its behavior at every time step.
The dominant architecture is the U-Net applied to spectrograms (yes, the same U-Net we built for image segmentation in episode #80). Audio enhancement is, in a way, an image-to-image translation problem -- from a noisy spectrogram image to a clean one:
import torch
import torch.nn as nn
import torch.nn.functional as F
class AudioDenoiseUNet(nn.Module):
"""U-Net for spectrogram mask
prediction."""
def __init__(self, in_ch=1,
base_ch=32):
super().__init__()
self.enc1 = self._block(
in_ch, base_ch)
self.enc2 = self._block(
base_ch, base_ch * 2)
self.enc3 = self._block(
base_ch * 2, base_ch * 4)
self.pool = nn.MaxPool2d(2)
self.bottleneck = self._block(
base_ch * 4, base_ch * 8)
self.up3 = nn.ConvTranspose2d(
base_ch * 8, base_ch * 4,
2, stride=2)
self.dec3 = self._block(
base_ch * 8, base_ch * 4)
self.up2 = nn.ConvTranspose2d(
base_ch * 4, base_ch * 2,
2, stride=2)
self.dec2 = self._block(
base_ch * 4, base_ch * 2)
self.up1 = nn.ConvTranspose2d(
base_ch * 2, base_ch,
2, stride=2)
self.dec1 = self._block(
base_ch * 2, base_ch)
self.out = nn.Sequential(
nn.Conv2d(base_ch, in_ch, 1),
nn.Sigmoid())
def _block(self, in_c, out_c):
return nn.Sequential(
nn.Conv2d(in_c, out_c, 3,
padding=1),
nn.BatchNorm2d(out_c),
nn.ReLU(inplace=True),
nn.Conv2d(out_c, out_c, 3,
padding=1),
nn.BatchNorm2d(out_c),
nn.ReLU(inplace=True))
def forward(self, x):
"""x: (batch, 1, freq, time)
magnitude spectrogram."""
e1 = self.enc1(x)
e2 = self.enc2(self.pool(e1))
e3 = self.enc3(self.pool(e2))
b = self.bottleneck(
self.pool(e3))
d3 = self.dec3(torch.cat(
[self.up3(b), e3], dim=1))
d2 = self.dec2(torch.cat(
[self.up2(d3), e2], dim=1))
d1 = self.dec1(torch.cat(
[self.up1(d2), e1], dim=1))
mask = self.out(d1)
return x * mask
model = AudioDenoiseUNet()
total = sum(p.numel()
for p in model.parameters())
noisy_spec = torch.randn(2, 1, 128, 128)
clean_spec = model(noisy_spec)
print(f"Input: {noisy_spec.shape}")
print(f"Output: {clean_spec.shape}")
print(f"Parameters: {total:,}")
print(f"Mask range: "
f"[{clean_spec.min():.3f}, "
f"{clean_spec.max():.3f}]")
The model predicts a mask -- values between 0 and 1 for each time-frequency bin. Bins dominated by speech get mask values near 1 (pass through). Bins dominated by noise get values near 0 (suppress). This is fundamentally more flexible than spectral subtraction because the mask adapts to the content at every single moment, and the network has learned what speech vs noise looks like in the spectrogram domain.
Training data comes from mixing clean speech datasets (like LibriSpeech or VCTK) with noise datasets (MUSAN, AudioSet noise subset, the DNS Challenge dataset) at various SNR levels. You mix clean + noise -> feed the noisy spectrogram -> predict mask -> compare the masked result with the clean spectrogram -> backpropagate. The model learns to handle everything from gentle fan hum at 20dB SNR to construction noise at -5dB SNR.
Instead of working in the spectrogram domain, some models operate directly on raw waveforms. Conv-TasNet (Time-domain Audio Separation Network) and Meta's DEMUCS architecture (which we mentioned in episode #96 for music source separation) use 1D convolutions on the time-domain signal. This avoids a fundamental problem with spectrogram-based methods: phase reconstruction.
When you predict a magnitude mask and apply it to the noisy magnitude spectrogram, you keep the original (noisy) phase. But phase carries spatial and temporal information that matters for perceptual quality. Waveform-domain models reconstruct everything -- magnitude and phase together:
class SimpleWaveformDenoiser(nn.Module):
"""Simplified waveform-domain
denoiser inspired by Conv-TasNet."""
def __init__(self, n_filters=256,
kernel_size=20,
n_blocks=4,
hidden=128):
super().__init__()
# Learned encoder (replaces STFT)
self.encoder = nn.Conv1d(
1, n_filters, kernel_size,
stride=kernel_size // 2,
bias=False)
# Separator: dilated 1D convs
layers = []
for i in range(n_blocks):
dilation = 2 ** i
layers.append(nn.Sequential(
nn.Conv1d(
n_filters, hidden, 1),
nn.PReLU(),
nn.GroupNorm(1, hidden),
nn.Conv1d(
hidden, hidden, 3,
padding=dilation,
dilation=dilation,
groups=hidden),
nn.PReLU(),
nn.GroupNorm(1, hidden),
nn.Conv1d(
hidden, n_filters, 1)))
self.separator = nn.ModuleList(
layers)
self.mask = nn.Sequential(
nn.Conv1d(
n_filters, n_filters, 1),
nn.Sigmoid())
# Learned decoder (replaces iSTFT)
self.decoder = nn.ConvTranspose1d(
n_filters, 1, kernel_size,
stride=kernel_size // 2,
bias=False)
def forward(self, x):
"""x: (batch, 1, samples)"""
enc = F.relu(self.encoder(x))
sep = enc
for block in self.separator:
sep = sep + block(sep)
masked = enc * self.mask(sep)
return self.decoder(masked)
model = SimpleWaveformDenoiser()
total = sum(p.numel()
for p in model.parameters())
audio = torch.randn(2, 1, 16000)
clean = model(audio)
print(f"Input: {audio.shape}")
print(f"Output: {clean.shape}")
print(f"Parameters: {total:,}")
The learned encoder replaces the STFT -- it's a 1D convolution that transforms the waveform into a representation that the separator network can work with. The learned decoder (transposed convolution) converts back to a waveform. The separator uses dilated convolutions with exponentially increasing dilation rates (1, 2, 4, 8) to capture both local and long-range temporal context -- the same dilation trick we saw in WaveNet-style architectures (episode #94).
The advantage over spectrogram methods is that everything is learned end-to-end. The model discovers whatever intermediate representation works best for separating speech from noise, rather than being constrained to work with human-designed spectrograms. In practice, waveform-domain models consistantly achieve higher perceptual quality scores than spectrogram-mask methods, particularly for speech naturalness.
Acoustic Echo Cancellation (AEC) handles a very specific problem in real-time communication. During a video call, your microphone picks up audio playing through your speakers -- the far-end speaker hears their own voice echoed back with a delay. AEC estimates what the microphone captured from the speaker output and subtracts it, leaving only the near-end speech (your voice):
class AdaptiveEchoCanceller:
"""LMS-based acoustic echo
cancellation."""
def __init__(self, filter_length=512,
mu=0.01, sr=16000):
self.filter_length = filter_length
self.mu = mu
self.sr = sr
def cancel(self, mic, ref):
"""mic: microphone signal
(speech + echo).
ref: reference signal from
speakers (far-end audio)."""
n = len(mic)
w = np.zeros(self.filter_length)
output = np.zeros(n)
erle_history = []
for i in range(
self.filter_length, n):
x = ref[
i - self.filter_length:i
][::-1]
echo_est = np.dot(w, x)
error = mic[i] - echo_est
output[i] = error
norm = np.dot(x, x) + 1e-10
w += self.mu * error * x / norm
if i % 1000 == 0 and i > 0:
win = slice(
max(0, i-1000), i)
echo_pow = np.mean(
mic[win] ** 2)
res_pow = np.mean(
output[win] ** 2)
if res_pow > 1e-10:
erle = 10 * np.log10(
echo_pow / res_pow)
erle_history.append(erle)
return output, erle_history
def run(self):
rng = np.random.RandomState(42)
dur = 3.0
n = int(self.sr * dur)
t = np.arange(n) / self.sr
# Far-end speech
far_end = np.zeros(n)
for f, a in [(200, 0.3),
(400, 0.15),
(800, 0.08)]:
far_end += a * np.sin(
2*np.pi*f*t)
far_end *= (0.5 + 0.5 * np.sin(
2*np.pi*3*t))
# Room impulse response
rir_len = 200
rir = np.zeros(rir_len)
rir[30] = 0.8 # direct path
rir[75] = 0.3 # first reflection
rir[120] = 0.15 # second reflection
rir[180] = 0.05 # late reverb
echo = np.convolve(
far_end, rir)[:n]
# Near-end speech (local speaker)
near_end = np.zeros(n)
for f, a in [(150, 0.2),
(300, 0.1)]:
near_end += a * np.sin(
2*np.pi*f*t)
ne_env = np.zeros(n)
ne_env[int(1.0*self.sr):
int(2.5*self.sr)] = 1.0
near_end *= ne_env
mic = near_end + echo
mic += rng.randn(n) * 0.01
output, erle = self.cancel(
mic, far_end)
print(f"Echo cancellation results:")
print(f" Filter: "
f"{self.filter_length} taps")
print(f" Room impulse: "
f"{rir_len} samples "
f"({rir_len/self.sr*1000:.0f}"
f"ms)")
if erle:
print(f" Final ERLE: "
f"{erle[-1]:.1f} dB")
print(f" ERLE progression: "
f"{[f'{e:.1f}' for e in erle]}")
ne_region = slice(
int(1.2*self.sr),
int(2.0*self.sr))
ne_corr = np.corrcoef(
near_end[ne_region],
output[ne_region])[0, 1]
print(f" Near-end correlation: "
f"{ne_corr:.3f}")
aec = AdaptiveEchoCanceller()
aec.run()
The adaptive filter learns the room impulse response -- how sound bounces from speaker to microphone through the room. The four taps in our synthetic RIR represent the direct path (30 samples delay), first wall reflection, second reflection, and late reverberance. As the room changes (someone moves, a door opens), the LMS filter continuously adapts its weights.
The ERLE (Echo Return Loss Enhancement) metric measures how much echo the canceller removes, in dB. A good AEC achieves 20-40 dB ERLE, meaning the echo is reduced to 1/100th to 1/10000th of its original power. Modern production AEC systems use neural networks instead of LMS filters and handle even double-talk (both people speaking simultaneously) much more robustly -- a situation where the classic LMS filter tends to diverge because it can't distinguish near-end speech from echo estimation error.
Audio super-resolution upsamples low-bandwidth audio to higher fidelity. A phone call sampled at 8kHz (cutting off at 4kHz per Nyquist) sounds muffled because everything above 4kHz is gone -- the sibilants ("s", "sh", "f"), the breathiness, the high harmonics that give voices their crisp, natural quality. Audio super-resolution hallucinates those missing frequencies based on learned correlations between low and high frequency content:
class AudioSuperResDemo:
"""Demonstrate audio super-resolution
concept with spectral analysis."""
def __init__(self, sr_high=16000,
sr_low=8000):
self.sr_high = sr_high
self.sr_low = sr_low
def synthesize_speech(self, sr,
dur=2.0):
"""Generate speech-like signal
with harmonics and formants."""
rng = np.random.RandomState(42)
n = int(sr * dur)
t = np.arange(n) / sr
signal = np.zeros(n)
f0 = 150
for h in range(1, 20):
freq = f0 * h
if freq > sr / 2:
break
amp = 0.3 / h
for fc, bw in [
(500, 100),
(1500, 200),
(2500, 300),
(3500, 400),
(5000, 500)]:
if freq > sr / 2:
break
gain = np.exp(
-0.5 * ((freq - fc)
/ bw) ** 2)
amp *= (1 + 2 * gain)
signal += amp * np.sin(
2*np.pi*freq*t)
# Fricative noise bursts
fric = rng.randn(n) * 0.05
if sr >= 16000:
from_bin = int(
3000 * n / sr)
to_bin = int(
min(7000, sr/2) * n / sr)
fft = np.fft.rfft(fric)
fft[:from_bin] = 0
fft[to_bin:] = 0
fric = np.fft.irfft(
fft, n=n)
signal += fric
env = np.zeros(n)
env[int(0.2*sr):
int(1.8*sr)] = 1.0
signal *= env
return signal
def analyze_bandwidth(self, signal,
sr, label):
fft = np.abs(np.fft.rfft(signal))
freqs = np.fft.rfftfreq(
len(signal), 1.0/sr)
bands = [
('0-1kHz', 0, 1000),
('1-2kHz', 1000, 2000),
('2-4kHz', 2000, 4000),
('4-6kHz', 4000, 6000),
('6-8kHz', 6000, 8000)]
print(f"\n {label}:")
for name, lo, hi in bands:
if hi > sr / 2:
print(f" {name}: "
f"(beyond Nyquist)")
continue
mask = (freqs >= lo) & (
freqs < hi)
energy = np.mean(
fft[mask] ** 2)
print(f" {name}: "
f"{energy:.4f}")
def run(self):
high = self.synthesize_speech(
self.sr_high)
low = self.synthesize_speech(
self.sr_low)
print("Audio super-resolution "
"analysis:")
print(f" High-res: "
f"{self.sr_high} Hz "
f"(Nyquist: "
f"{self.sr_high//2} Hz)")
print(f" Low-res: "
f"{self.sr_low} Hz "
f"(Nyquist: "
f"{self.sr_low//2} Hz)")
self.analyze_bandwidth(
high, self.sr_high,
"16kHz signal")
self.analyze_bandwidth(
low, self.sr_low,
"8kHz signal (phone quality)")
fft_high = np.abs(
np.fft.rfft(high))
freqs = np.fft.rfftfreq(
len(high), 1.0/self.sr_high)
above_4k = freqs >= 4000
missing = np.mean(
fft_high[above_4k] ** 2)
total = np.mean(fft_high ** 2)
print(f"\n Energy above 4kHz: "
f"{missing/total:.1%} "
f"of total")
print(f" This is what super-"
f"resolution must predict")
demo = AudioSuperResDemo()
demo.run()
This is conceptually identical to image super-resolution (episode #86) -- predict missing detail from low-resolution input. The training pairs are trivial to create: take high-quality 16kHz audio, downsample it to 8kHz, and train the model to recover the original high frequencies. Modern audio super-resolution models (typically U-Net or WaveNet variants) achieve remarkebly natural-sounding results -- phone calls that sound closer to studio recordings. The model learns that certain low-frequency patterns (like the onset of a sibilant "s" visible in the lower harmonics) predict specific high-frequency content.
Let me put everything together into a complete pipeline that combines multiple enhancement stages. Real-world production systems don't use just one technique -- they stack several, each attacking a different aspect of the noise:
class NoiseReductionPipeline:
"""Multi-stage noise reduction
combining classical methods."""
def __init__(self, sr=16000,
n_fft=1024, hop=256):
self.sr = sr
self.n_fft = n_fft
self.hop = hop
def stft(self, audio):
window = np.hanning(self.n_fft)
nf = ((len(audio) - self.n_fft)
// self.hop + 1)
nb = self.n_fft // 2 + 1
result = np.zeros(
(nb, nf), dtype=complex)
for i in range(nf):
s = i * self.hop
result[:, i] = np.fft.rfft(
audio[s:s+self.n_fft]
* window)
return result
def istft(self, spec, length):
window = np.hanning(self.n_fft)
nf = spec.shape[1]
out = np.zeros(length)
ws = np.zeros(length)
for i in range(nf):
s = i * self.hop
frame = np.fft.irfft(
spec[:, i])
e = min(s+self.n_fft, length)
fl = e - s
out[s:e] += (
frame[:fl]*window[:fl])
ws[s:e] += window[:fl]**2
m = ws > 1e-8
out[m] /= ws[m]
return out
def vad_gate(self, audio,
frame_ms=25, hop_ms=10,
thresh_db=-25):
"""Simple energy-based VAD gate:
suppress non-speech frames."""
frame_len = int(
self.sr * frame_ms / 1000)
hop_len = int(
self.sr * hop_ms / 1000)
nf = ((len(audio) - frame_len)
// hop_len + 1)
gated = audio.copy()
for i in range(nf):
s = i * hop_len
frame = audio[
s:s + frame_len]
rms = np.sqrt(
np.mean(frame ** 2))
db = 20 * np.log10(
max(rms, 1e-10))
if db < thresh_db:
e = min(
s + frame_len,
len(gated))
gated[s:e] *= 0.01
return gated
def wiener_stage(self, audio,
noise_frames=20):
spec = self.stft(audio)
power = np.abs(spec) ** 2
noise_p = np.mean(
power[:, :noise_frames],
axis=1, keepdims=True)
sig_p = np.maximum(
power - noise_p, 0.0)
gain = sig_p / (
sig_p + noise_p + 1e-10)
for i in range(1, gain.shape[1]):
gain[:, i] = (
0.85 * gain[:, i-1]
+ 0.15 * gain[:, i])
clean = (np.abs(spec) * gain
* np.exp(
1j*np.angle(spec)))
return self.istft(
clean, len(audio))
def spectral_floor_stage(self, audio,
noise_frames=20,
floor=0.01):
spec = self.stft(audio)
mag = np.abs(spec)
noise_est = np.mean(
mag[:, :noise_frames],
axis=1, keepdims=True)
clean_mag = np.maximum(
mag - 1.5 * noise_est,
floor * noise_est)
clean = clean_mag * np.exp(
1j * np.angle(spec))
return self.istft(
clean, len(audio))
def run(self):
rng = np.random.RandomState(42)
dur = 4.0
n = int(self.sr * dur)
t = np.arange(n) / self.sr
speech = np.zeros(n)
for f, a in [(130, 0.25),
(260, 0.15),
(780, 0.08),
(2100, 0.04)]:
speech += a * np.sin(
2*np.pi*f*t)
speech *= (0.5 + 0.5 * np.sin(
2*np.pi*4*t))
env = np.zeros(n)
env[int(0.5*self.sr):
int(3.5*self.sr)] = 1.0
speech *= env
# Multiple noise types
noise = rng.randn(n) * 0.1
noise += 0.05 * np.sin(
2*np.pi*50*t) # mains hum
for _ in range(8): # clicks
pos = rng.randint(n)
noise[pos:pos+50] += (
rng.randn(min(50, n-pos))
* 0.3)
noisy = speech + noise
def snr(clean, test, region):
sp = np.mean(clean[region]**2)
np_ = np.mean(
(test[region]
- clean[region])**2)
return 10*np.log10(
sp / max(np_, 1e-10))
sp_region = slice(
int(1.0*self.sr),
int(3.0*self.sr))
results = {'Input': noisy}
s1 = self.vad_gate(noisy)
results['VAD gate'] = s1
s2 = self.wiener_stage(s1)
results['+ Wiener'] = s2
s3 = self.spectral_floor_stage(s2)
results['+ Spec floor'] = s3
print(f"Multi-stage pipeline:")
print(f"{'Stage':<16} {'SNR (dB)':>10}")
print("-" * 28)
for name, sig in results.items():
s = snr(speech, sig, sp_region)
print(f"{name:<16} {s:>10.1f}")
pipe = NoiseReductionPipeline()
pipe.run()
Each stage attacks a different aspect of the noise. The VAD gate suppresses frames where there's no speech at all (silence or pure noise segments), preventing noise from bleeding through during pauses. The Wiener filter handles stationary broadband noise (fan hum, white noise). The spectral floor catches remaining narrowband noise and musical noise artifacts from the previous stage. The combined pipeline achieves significantly higher SNR improvement than any single stage alone.
This is how production audio enhancement works conceptually -- systems like Krisp, NVIDIA RTX Voice, and the noise suppression built into Zoom all use multi-stage approaches. The key difference is that they replace each classical stage with a neural network component, and often train the entire pipeline end-to-end so the stages learn to complement each other.
Audio enhancement for live applications (video calls, hearing aids, live streaming) has a hard constraint that batch processing doesn't: latency. You can't buffer 5 seconds of audio before processing it -- the delay would make conversation impossible. Acceptable latency is under 40ms, and ideally under 20ms for hearing aids:
class RealtimeAnalysis:
"""Analyze latency vs quality
tradeoffs for real-time audio
enhancement."""
def __init__(self, sr=16000):
self.sr = sr
def latency_analysis(self):
configs = [
('Hearing aid', 5, 64, 32),
('VoIP low', 10, 160, 80),
('VoIP standard', 20, 320, 160),
('VoIP generous', 40, 640, 320),
('Offline batch', 500, 8000,
4000),
]
print(f"{'Config':<18} "
f"{'Latency':>8} "
f"{'Window':>8} "
f"{'Hop':>6} "
f"{'Chunks/s':>10}")
print("-" * 54)
for name, lat, win, hop in configs:
chunks = self.sr // hop
print(f"{name:<18} "
f"{lat:>6}ms "
f"{win:>7} "
f"{hop:>5} "
f"{chunks:>9}")
def quality_metrics(self):
rng = np.random.RandomState(42)
n = 16000
t = np.arange(n) / 16000.0
clean = np.zeros(n)
for f, a in [(150, 0.3),
(300, 0.15),
(900, 0.08)]:
clean += a * np.sin(
2*np.pi*f*t)
clean *= (0.5 + 0.5 * np.sin(
2*np.pi*4*t))
scenarios = {
'Perfect': clean.copy(),
'Mild noise': (
clean + rng.randn(n)*0.03),
'Moderate': (
clean + rng.randn(n)*0.1),
'Heavy noise': (
clean + rng.randn(n)*0.3),
'Distorted': np.clip(
clean * 3, -0.5, 0.5),
}
print(f"\n{'Scenario':<18} "
f"{'SI-SNR':>8} "
f"{'Corr':>8} "
f"{'STOI-like':>10}")
print("-" * 48)
for name, test in (
scenarios.items()):
target = clean
s_target = (
np.dot(test, target)
/ np.dot(target, target)
* target)
e_noise = test - s_target
si_snr = 10 * np.log10(
np.dot(s_target, s_target)
/ (np.dot(e_noise, e_noise)
+ 1e-10))
corr = np.corrcoef(
clean, test)[0, 1]
frame_len = 256
stoi_scores = []
for i in range(0, n-frame_len,
frame_len//2):
c = clean[i:i+frame_len]
d = test[i:i+frame_len]
if np.std(c) > 1e-6:
r = np.corrcoef(
c, d)[0, 1]
stoi_scores.append(
max(0, r))
stoi = np.mean(stoi_scores)
print(f"{name:<18} "
f"{si_snr:>7.1f} "
f"{corr:>8.3f} "
f"{stoi:>10.3f}")
def run(self):
print("=== Real-time Analysis ===\n")
self.latency_analysis()
self.quality_metrics()
analysis = RealtimeAnalysis()
analysis.run()
PESQ (Perceptual Evaluation of Speech Quality) is the ITU standard metric that models human perception of speech quality, with scores ranging from -0.5 to 4.5. STOI (Short-Time Objective Intelligibility) measures how understandable the speech remains on a scale from 0 to 1 -- good enhancement achieves 0.85-0.95. SI-SNR (Scale-Invariant Signal-to-Noise Ratio) measures signal-to-distortion ratio in dB and is the most common training objective for neural speech enhancement because it's differentiable and correlates well with perceptual quality.
The latency table shows the fundamental tradeoff: shorter processing windows mean lower latency but less frequency resolution and less context for the model to work with. Hearing aids must operate at 5ms latency with tiny 64-sample windows. VoIP systems get 20-40ms. Offline processing (podcast cleanup, film post-production) can use arbitrarily large windows for maximum quality.
Modern production systems combine multiple innovations. Meta's Denoiser (based on the DEMUCS architecture) achieves near-studio quality on single-channel noisy speech at real-time speeds. Microsoft's DTLN (Dual-Signal Transformation LSTM Network) targets extreme low-latency use at 5-10ms. Google's research into unified audio models builds on the same foundations we've been exploring throughout this whole audio arc.
The field continues moving rapidly. Recent work focuses on personalized enhancement (adapting to a specific speaker's voice characteristics, using the speaker embeddings from episode #97), multi-channel processing (using multiple microphones for spatial filtering and beamforming), and unified models that handle denoising, dereverberation, echo cancellation, and bandwidth extension all in a single forward pass. The trend is clear: specialized single-task models are being replaced by general-purpose audio enhancement networks that handle whatever degradation the input happens to contain.
We're nearly done with the audio AI domain. The next episode brings audio and vision together into unified multimodal representations -- how models learn to jointly understand what they see and what they hear. Then we'll wrap the entire audio arc with a hands-on mini project before stepping into reinforcement learning territory. Exciting times ahead!
Exercise 1: Build a multi-band noise reducer. Create MultiBandDenoiser that splits audio into 4 bands (0-500Hz, 500-2kHz, 2-4kHz, 4-8kHz) via FFT slicing, estimates noise per band from the first 0.3s, applies Wiener filtering per band with band-specific smoothing (low bands: more smoothing, high bands: less), and recombines. Test on a 3-second signal with formant speech (0.5-2.5s) plus different noise per band (60Hz hum, broadband hiss, high-frequency crackle). Compare per-band and total SNR against single-band Wiener. Multi-band should win because it applies different attenuation per frequency region.
Exercise 2: Build an adaptive noise tracker. Create AdaptiveNoiseTracker using minimum statistics (track minimum power per frequency bin over a 50-frame sliding window). Generate 5 seconds where noise changes at 2.5s (white noise 0.05 -> pink noise 0.08), speech active 1-4s. Implement Wiener filtering with the adaptive estimate (updated every 10 frames). Compare: (1) fixed estimate from first 0.3s, (2) fixed estimate from first 0.3s after the noise change, (3) your adaptive tracker. Print SNR before and after the change. The adaptive tracker should handle both regions well; fixed estimate from region 1 should degrade in region 2.
Exercise 3: Build a perceptual quality comparator. Create QualityComparator that generates 2s clean speech, makes 5 degraded versions (0dB noise, 20dB noise, clipping via np.clip(signal*3, -0.5, 0.5), bandlimited to 4kHz, reverb via 100-tap exponential-decay impulse response). Compute 4 metrics: SI-SNR, segmental SNR (per 32ms frame), spectral distortion (log-spectral distance), short-time correlation (STOI-like). Print comparison table. Verify: clipping has low spectral distortion but poor SI-SNR; bandlimiting has high spectral distortion but decent correlation.
@scipio