Phonetic Confusion Training for ASR
Build a CLI tool that generates phonetically-plausible ASR error corruptions.
Introduction
30 Second Summary
Voice assistants mishear us all the time. "Set a timer" becomes "set a tiger." These phonetic mix-ups follow predictable patterns rooted in how sounds are physically produced in our mouths.
In this project, you will build a Python CLI tool that generates realistic speech recognition errors using articulatory phonetics and curriculum learning. You will train NLU systems to handle the exact kinds of mistakes that ASR engines produce in the real world.
What You'll Build
A Python CLI tool called asr_confuser that takes clean NLU training utterances and corrupts them with phonetically-plausible ASR errors across three confusion strategies, controlled by a curriculum scheduler that ramps difficulty across training stages.
By the end of this project, you'll have:
- A working ARPABET phonetic feature system that computes articulatory distances between all 39 CMU phonemes.
- Three distinct ASR error simulation strategies: homophone substitution, phoneme-neighbor swaps, and word boundary confusion.
- A curriculum scheduler that progressively increases corruption difficulty from easy (homophones only) to hard (mixed strategies at high WER).
- Secret Mission: Generate phoneme confusion matrices that validate your error patterns against real ASR behavior from Amazon's research.
Are there any prerequisites?
This project assumes you have Python experience and familiarity with NLP and phonetics concepts. You should be comfortable working with dictionaries, classes, and file I/O in Python.
No prior experience with NLTK or the CMU Pronouncing Dictionary is required. You will learn both during the project.
Build the Phonetic Feature Foundation
For your confusion generator to produce realistic ASR errors, it needs to know which sounds are similar to each other. A system that confuses P with B (which differ only in voicing) is far more convincing than one that confuses P with a vowel.
In this step, you will encode the articulatory properties of all 39 ARPABET phonemes as numeric feature vectors. Then you will build a distance function that quantifies how "close" any two sounds are, grounding your entire corruption pipeline in real phonetics.
In this step, get ready to:
- Define articulatory feature vectors for all 39 CMU phonemes (24 consonants, 15 vowels).
- Implement phonetic distance computation and neighbor search.
- Validate that your distance function matches known phonetic relationships.
Define articulatory feature encodings
Every speech sound can be described by a small set of articulatory features. Consonants are characterized by how the airflow is shaped (manner), where in the mouth it happens (place), and whether the vocal cords vibrate (voicing).
- Create a file called phonetic_features.py in your project directory.
- Add the encoding dictionaries by pasting the following code:
"""ARPABET phonetic feature vectors and distance computation."""
# Manner of articulation encoding
MANNER = {
'stop': 0, 'fricative': 1, 'affricate': 2,
'nasal': 3, 'lateral': 4, 'approximant': 5
}
# Place of articulation encoding
PLACE = {
'bilabial': 0, 'labiodental': 1, 'dental': 2,
'alveolar': 3, 'postalveolar': 4, 'velar': 5,
'glottal': 6, 'palatal': 7, 'labiovelar': 8
}
# Voicing encoding
VOICING = {'voiceless': 0, 'voiced': 1}
# Vowel height encoding
HEIGHT = {'high': 0, 'near-high': 1, 'mid': 2, 'near-low': 3, 'low': 4}
# Vowel backness encoding
BACKNESS = {'front': 0, 'central': 1, 'back': 2}
# Vowel rounding encoding
ROUNDING = {'unrounded': 0, 'rounded': 1}What do these encodings represent?
Each dictionary maps a human-readable articulatory label to an integer. Consonants use three dimensions: manner (how air flows), place (where constriction happens), and voicing (vocal cord vibration).
Vowels use three different dimensions: height (jaw openness), backness (tongue position front-to-back), and rounding (lip shape). These integer encodings let you compute numeric distances between sounds later.
- Below the encoding dictionaries, add the consonant feature definitions:
# Consonant features: (manner, place, voicing)
CONSONANT_FEATURES = {
'B': ('stop', 'bilabial', 'voiced'),
'CH': ('affricate', 'postalveolar', 'voiceless'),
'D': ('stop', 'alveolar', 'voiced'),
'DH': ('fricative', 'dental', 'voiced'),
'F': ('fricative', 'labiodental', 'voiceless'),
'G': ('stop', 'velar', 'voiced'),
'HH': ('fricative', 'glottal', 'voiceless'),
'JH': ('affricate', 'postalveolar', 'voiced'),
'K': ('stop', 'velar', 'voiceless'),
'L': ('lateral', 'alveolar', 'voiced'),
'M': ('nasal', 'bilabial', 'voiced'),
'N': ('nasal', 'alveolar', 'voiced'),
'NG': ('nasal', 'velar', 'voiced'),
'P': ('stop', 'bilabial', 'voiceless'),
'R': ('approximant', 'alveolar', 'voiced'),
'S': ('fricative', 'alveolar', 'voiceless'),
'SH': ('fricative', 'postalveolar', 'voiceless'),
'T': ('stop', 'alveolar', 'voiceless'),
'TH': ('fricative', 'dental', 'voiceless'),
'V': ('fricative', 'labiodental', 'voiced'),
'W': ('approximant', 'labiovelar', 'voiced'),
'Y': ('approximant', 'palatal', 'voiced'),
'Z': ('fricative', 'alveolar', 'voiced'),
'ZH': ('fricative', 'postalveolar', 'voiced'),
}What does this dictionary encode?
Each of the 24 ARPABET consonants maps to a tuple of (manner, place, voicing) labels. For example, P is a voiceless bilabial stop, and B is a voiced bilabial stop. They differ only in voicing, which is exactly what makes P/B confusion realistic in ASR output.
- Below the consonant features, add the vowel feature definitions and the combined phoneme set:
# Vowel features: (height, backness, rounding)
VOWEL_FEATURES = {
'AA': ('low', 'back', 'unrounded'),
'AE': ('near-low', 'front', 'unrounded'),
'AH': ('mid', 'central', 'unrounded'),
'AO': ('mid', 'back', 'rounded'),
'AW': ('low', 'back', 'rounded'),
'AY': ('low', 'front', 'unrounded'),
'EH': ('mid', 'front', 'unrounded'),
'ER': ('mid', 'central', 'rounded'),
'EY': ('mid', 'front', 'unrounded'),
'IH': ('near-high', 'front', 'unrounded'),
'IY': ('high', 'front', 'unrounded'),
'OW': ('mid', 'back', 'rounded'),
'OY': ('mid', 'back', 'rounded'),
'UH': ('high', 'back', 'unrounded'),
'UW': ('high', 'back', 'rounded'),
}
ALL_PHONEMES = set(CONSONANT_FEATURES.keys()) | set(VOWEL_FEATURES.keys())Why OW and AO matter
Look at OW and AO in the table. Both are mid, back, rounded vowels. This is why real ASR systems confuse "coal" (OW) with "call" (AO). The Amazon Interspeech 2020 paper names this as a primary confusion pattern.
- Save phonetic_features.py.
Seeing a syntax error?
Make sure every dictionary entry ends with a comma, including the last entry in each dict. Also check that the file starts with the triple-quoted docstring and that all string values use single quotes consistently.
Help me fix a syntax error in my phonetic_features.py file
Implement distance computation and neighbor search
With the raw feature data in place, you need functions that convert phonemes to vectors and measure how far apart two sounds are. These functions power every confusion strategy you will build later.
- At the bottom of phonetic_features.py, add the utility functions that strip stress markers and classify phonemes:
def strip_stress(phoneme):
"""Remove stress marker digits from a phoneme token."""
return ''.join(c for c in phoneme if not c.isdigit())
def is_vowel(phoneme):
"""Check if a phoneme (stress-stripped) is a vowel."""
return strip_stress(phoneme) in VOWEL_FEATURES
def phoneme_to_vector(phoneme):
"""Convert a phoneme to its numeric feature vector."""
p = strip_stress(phoneme)
if p in CONSONANT_FEATURES:
manner, place, voicing = CONSONANT_FEATURES[p]
return ('consonant', MANNER[manner], PLACE[place], VOICING[voicing])
elif p in VOWEL_FEATURES:
height, backness, rounding = VOWEL_FEATURES[p]
return ('vowel', HEIGHT[height], BACKNESS[backness], ROUNDING[rounding])
else:
return NoneWhat do these functions do?
- strip_stress() removes digit suffixes (like the 1 in AO1) so you can compare phonemes regardless of their stress level.
- is_vowel() classifies a phoneme into the vowel or consonant category.
- phoneme_to_vector() looks up the articulatory labels and converts them to a numeric tuple like ('consonant', 0, 0, 1).
- Below those utilities, add the phonetic_distance() function:
def phonetic_distance(p1, p2):
"""
Compute normalized distance between two phonemes.
Returns a float between 0.0 (identical) and 1.0 (maximally different).
Cross-category (vowel vs consonant) comparisons return 1.0.
"""
v1 = phoneme_to_vector(p1)
v2 = phoneme_to_vector(p2)
if v1 is None or v2 is None:
return 1.0
# Different phoneme categories are maximally distant
if v1[0] != v2[0]:
return 1.0
# Compare feature dimensions (skip the category label at index 0)
features1 = v1[1:]
features2 = v2[1:]
if v1[0] == 'consonant':
max_diffs = (
max(MANNER.values()),
max(PLACE.values()),
max(VOICING.values()),
)
else:
max_diffs = (
max(HEIGHT.values()),
max(BACKNESS.values()),
max(ROUNDING.values()),
)
total_distance = 0.0
for f1, f2, max_d in zip(features1, features2, max_diffs):
if max_d > 0:
total_distance += abs(f1 - f2) / max_d
return total_distance / len(max_diffs)How does the distance calculation work?
The function computes a normalized distance between 0.0 and 1.0. It compares each feature dimension independently, dividing by the maximum possible difference in that dimension. Then it averages across all three dimensions.
If two phonemes are from different categories (one vowel, one consonant), the function returns 1.0 immediately. This prevents your generator from ever confusing a vowel with a consonant.
- Finally, add the get_neighbors() function that finds phonemes within a distance threshold:
def get_neighbors(phoneme, threshold=0.3):
"""
Return all phonemes within the given distance threshold.
Results are sorted by distance (closest first).
"""
p = strip_stress(phoneme)
neighbors = []
for candidate in ALL_PHONEMES:
if candidate == p:
continue
dist = phonetic_distance(p, candidate)
if dist <= threshold:
neighbors.append((candidate, dist))
neighbors.sort(key=lambda x: x[1])
return neighborsWhat does get_neighbors do?
This function scans every phoneme in the system and returns those within the specified distance threshold. With a threshold of 0.3, you get only the closest articulatory neighbors. Your confusion strategies will use these neighbors to pick realistic substitution targets.
- Save phonetic_features.py.
Indentation issues?
Make sure all functions are at the module level (no indentation), not accidentally nested inside a dictionary or another function. Each def should start at column 0.
Help me debug indentation problems in my phonetic_features.py
✔️ Awesome, I've got everything!
Great. Double check you have saved phonetic_features.py with all six encoding dicts, both feature dicts, and all five functions.
ⓧ I'd like to double check the full code
"""ARPABET phonetic feature vectors and distance computation."""
# Manner of articulation encoding
MANNER = {
'stop': 0, 'fricative': 1, 'affricate': 2,
'nasal': 3, 'lateral': 4, 'approximant': 5
}
# Place of articulation encoding
PLACE = {
'bilabial': 0, 'labiodental': 1, 'dental': 2,
'alveolar': 3, 'postalveolar': 4, 'velar': 5,
'glottal': 6, 'palatal': 7, 'labiovelar': 8
}
# Voicing encoding
VOICING = {'voiceless': 0, 'voiced': 1}
# Vowel height encoding
HEIGHT = {'high': 0, 'near-high': 1, 'mid': 2, 'near-low': 3, 'low': 4}
# Vowel backness encoding
BACKNESS = {'front': 0, 'central': 1, 'back': 2}
# Vowel rounding encoding
ROUNDING = {'unrounded': 0, 'rounded': 1}
# Consonant features: (manner, place, voicing)
CONSONANT_FEATURES = {
'B': ('stop', 'bilabial', 'voiced'),
'CH': ('affricate', 'postalveolar', 'voiceless'),
'D': ('stop', 'alveolar', 'voiced'),
'DH': ('fricative', 'dental', 'voiced'),
'F': ('fricative', 'labiodental', 'voiceless'),
'G': ('stop', 'velar', 'voiced'),
'HH': ('fricative', 'glottal', 'voiceless'),
'JH': ('affricate', 'postalveolar', 'voiced'),
'K': ('stop', 'velar', 'voiceless'),
'L': ('lateral', 'alveolar', 'voiced'),
'M': ('nasal', 'bilabial', 'voiced'),
'N': ('nasal', 'alveolar', 'voiced'),
'NG': ('nasal', 'velar', 'voiced'),
'P': ('stop', 'bilabial', 'voiceless'),
'R': ('approximant', 'alveolar', 'voiced'),
'S': ('fricative', 'alveolar', 'voiceless'),
'SH': ('fricative', 'postalveolar', 'voiceless'),
'T': ('stop', 'alveolar', 'voiceless'),
'TH': ('fricative', 'dental', 'voiceless'),
'V': ('fricative', 'labiodental', 'voiced'),
'W': ('approximant', 'labiovelar', 'voiced'),
'Y': ('approximant', 'palatal', 'voiced'),
'Z': ('fricative', 'alveolar', 'voiced'),
'ZH': ('fricative', 'postalveolar', 'voiced'),
}
# Vowel features: (height, backness, rounding)
VOWEL_FEATURES = {
'AA': ('low', 'back', 'unrounded'),
'AE': ('near-low', 'front', 'unrounded'),
'AH': ('mid', 'central', 'unrounded'),
'AO': ('mid', 'back', 'rounded'),
'AW': ('low', 'back', 'rounded'),
'AY': ('low', 'front', 'unrounded'),
'EH': ('mid', 'front', 'unrounded'),
'ER': ('mid', 'central', 'rounded'),
'EY': ('mid', 'front', 'unrounded'),
'IH': ('near-high', 'front', 'unrounded'),
'IY': ('high', 'front', 'unrounded'),
'OW': ('mid', 'back', 'rounded'),
'OY': ('mid', 'back', 'rounded'),
'UH': ('high', 'back', 'unrounded'),
'UW': ('high', 'back', 'rounded'),
}
ALL_PHONEMES = set(CONSONANT_FEATURES.keys()) | set(VOWEL_FEATURES.keys())
def strip_stress(phoneme):
"""Remove stress marker digits from a phoneme token."""
return ''.join(c for c in phoneme if not c.isdigit())
def is_vowel(phoneme):
"""Check if a phoneme (stress-stripped) is a vowel."""
return strip_stress(phoneme) in VOWEL_FEATURES
def phoneme_to_vector(phoneme):
"""Convert a phoneme to its numeric feature vector."""
p = strip_stress(phoneme)
if p in CONSONANT_FEATURES:
manner, place, voicing = CONSONANT_FEATURES[p]
return ('consonant', MANNER[manner], PLACE[place], VOICING[voicing])
elif p in VOWEL_FEATURES:
height, backness, rounding = VOWEL_FEATURES[p]
return ('vowel', HEIGHT[height], BACKNESS[backness], ROUNDING[rounding])
else:
return None
def phonetic_distance(p1, p2):
"""
Compute normalized distance between two phonemes.
Returns a float between 0.0 (identical) and 1.0 (maximally different).
Cross-category (vowel vs consonant) comparisons return 1.0.
"""
v1 = phoneme_to_vector(p1)
v2 = phoneme_to_vector(p2)
if v1 is None or v2 is None:
return 1.0
# Different phoneme categories are maximally distant
if v1[0] != v2[0]:
return 1.0
# Compare feature dimensions (skip the category label at index 0)
features1 = v1[1:]
features2 = v2[1:]
if v1[0] == 'consonant':
max_diffs = (
max(MANNER.values()),
max(PLACE.values()),
max(VOICING.values()),
)
else:
max_diffs = (
max(HEIGHT.values()),
max(BACKNESS.values()),
max(ROUNDING.values()),
)
total_distance = 0.0
for f1, f2, max_d in zip(features1, features2, max_diffs):
if max_d > 0:
total_distance += abs(f1 - f2) / max_d
return total_distance / len(max_diffs)
def get_neighbors(phoneme, threshold=0.3):
"""
Return all phonemes within the given distance threshold.
Results are sorted by distance (closest first).
"""
p = strip_stress(phoneme)
neighbors = []
for candidate in ALL_PHONEMES:
if candidate == p:
continue
dist = phonetic_distance(p, candidate)
if dist <= threshold:
neighbors.append((candidate, dist))
neighbors.sort(key=lambda x: x[1])
return neighborsValidate the feature system
Before building on this foundation, you need to confirm the distance function behaves correctly. Phonetically similar sounds should produce small distances, and dissimilar sounds should produce large ones.
- Test your feature system by running this command:
python -c "
from phonetic_features import phonetic_distance, get_neighbors
print('OW vs AO:', phonetic_distance('OW', 'AO'))
print('P vs B:', phonetic_distance('P', 'B'))
print('P vs ZH:', phonetic_distance('P', 'ZH'))
print()
print('Neighbors of OW:', get_neighbors('OW', 0.3))
"- Check your output matches these expectations:
- OW vs AO distance should be very small (both are mid, back, rounded vowels).
- P vs B distance should be small (both are bilabial stops, differing only in voicing).
- P vs ZH distance should be large (they differ in manner, place, and voicing).
- The neighbors of OW should include AO and OY, since those vowels share similar articulatory positions.
Why these test pairs?
The Amazon Interspeech 2020 paper specifically names OW/AO confusion ("coal" vs "call") as a primary ASR error category. If your distance function marks these as close, your generator will produce the same confusion patterns that real ASR systems exhibit.
Getting an ImportError?
Make sure you are running the command from the same directory where phonetic_features.py lives. If you see a ModuleNotFoundError, use cd to navigate into your project folder before running the test.
Help me fix an import error when testing phonetic_features.py
Your phonetic feature foundation is solid. Next up, you will build the actual confusion strategies that use these distances to corrupt words in realistic ways.
Implement Homophone and Phoneme-Neighbor Strategies
With your phonetic distance system in place, you can now build the confusion strategies that simulate real ASR errors. The Amazon Interspeech 2020 paper identifies two dominant error patterns: homophones being swapped (sail becoming sale) and phonetically similar sounds being confused (the OW in "coal" becoming the AO in "call").
In this step, you will implement both patterns as independent strategy classes. Each strategy takes clean text and introduces controlled corruption at a target word error rate (WER), producing the kind of noisy transcripts an NLU model would encounter in production.
In this step, get ready to:
- Build a homophone index from the CMU Pronouncing Dictionary and implement the HomophoneStrategy class.
- Implement PhonemeNeighborStrategy that swaps phonemes with articulatory neighbors and finds matching real words.
- Test both strategies on sample voice-assistant utterances.
Build the homophone index
Homophones are words that sound identical but have different spellings and meanings. ASR systems frequently confuse these because the acoustic signal is the same. You need an index that maps pronunciation keys to all words sharing that pronunciation.
- Create a new file called confusion_strategies.py in your project folder (the same directory where phonetic_features.py lives).
- Add the following imports, module-level dictionary load, and homophone index builder:
"""ASR error simulation strategies: homophone, phoneme-neighbor, word boundary."""
import random
from nltk.corpus import cmudict
from phonetic_features import strip_stress, phonetic_distance, get_neighbors
# Load CMU dict once at module level
CMU_DICT = cmudict.dict()
def build_homophone_index():
"""
Build an index mapping phoneme sequences (without stress) to word lists.
Only includes entries where multiple words share the same pronunciation.
"""
pron_to_words = {}
for word, pronunciations in CMU_DICT.items():
for pron in pronunciations:
key = tuple(strip_stress(p) for p in pron)
if key not in pron_to_words:
pron_to_words[key] = []
if word not in pron_to_words[key]:
pron_to_words[key].append(word)
# Filter to only groups with 2+ words (actual homophones)
return {k: v for k, v in pron_to_words.items() if len(v) > 1}
HOMOPHONE_INDEX = build_homophone_index()
# Reverse lookup: word -> list of homophones
WORD_HOMOPHONES = {}
for words in HOMOPHONE_INDEX.values():
for w in words:
others = [x for x in words if x != w]
if others:
if w not in WORD_HOMOPHONES:
WORD_HOMOPHONES[w] = []
WORD_HOMOPHONES[w].extend(others)What does this code do?
The build_homophone_index() function iterates every word in the CMU dictionary, strips stress markers from its phoneme sequence, and groups words that share the same stripped pronunciation.
The result is a dictionary where keys are phoneme tuples and values are lists of words. For example, the key ('S', 'EY', 'L') maps to both "sail" and "sale". The WORD_HOMOPHONES reverse lookup lets you instantly find all homophones for any given word.
- Save confusion_strategies.py.
- Verify the index builds correctly by running this in your terminal:
python -c "from confusion_strategies import WORD_HOMOPHONES; print(len(WORD_HOMOPHONES)); print(WORD_HOMOPHONES.get('sail', []))"You should see a count of several thousand homophones and a list that includes "sale" for the word "sail".
Seeing a ModuleNotFoundError?
Make sure you are running the command from the same directory that contains both confusion_strategies.py and phonetic_features.py. The import relies on both files being in the same folder.
Help me debug my import error in confusion_strategies.py
Add the shared helper functions
Both strategies need a way to look up a word's phonemes and to find a real word that matches a given phoneme sequence. These two helpers bridge the gap between phoneme-level manipulation and word-level output.
- In confusion_strategies.py, add the following two functions below the WORD_HOMOPHONES block:
def get_word_phonemes(word):
"""Get the first pronunciation for a word from CMU dict."""
prons = CMU_DICT.get(word.lower())
if prons:
return prons[0]
return None
def find_word_for_phonemes(target_phonemes):
"""
Find the best matching word in CMU dict for a phoneme sequence.
Returns (word, distance) or None if no reasonable match found.
"""
target_stripped = [strip_stress(p) for p in target_phonemes]
target_key = tuple(target_stripped)
# Exact match first
for word, prons in CMU_DICT.items():
for pron in prons:
if tuple(strip_stress(p) for p in pron) == target_key:
return word
# Approximate match: find closest by edit-distance-like metric
best_word = None
best_score = float('inf')
# Sample a subset for performance
candidates = random.sample(list(CMU_DICT.items()), min(5000, len(CMU_DICT)))
for word, prons in candidates:
for pron in prons:
pron_stripped = [strip_stress(p) for p in pron]
if len(pron_stripped) != len(target_stripped):
continue
score = sum(
phonetic_distance(a, b)
for a, b in zip(target_stripped, pron_stripped)
)
if score < best_score:
best_score = score
best_word = word
if best_word and best_score < len(target_stripped) * 0.5:
return best_word
return NoneWhat does this code do?
- get_word_phonemes() is a simple lookup that returns the first pronunciation entry for any word in the CMU dictionary.
- find_word_for_phonemes() does the reverse: given a phoneme sequence, it finds a real English word that matches. It tries an exact match first, then falls back to approximate matching by sampling 5000 candidates from the dictionary and scoring by phonetic distance.
The 5000-word sample is a performance trade-off. Searching all 127,000+ entries for every corrupted phoneme sequence would be slow. Sampling gives good-enough results for this project's scope.
- Save confusion_strategies.py.
You won't see a change yet. These helpers are consumed by the strategy classes you will add next.
Implement the HomophoneStrategy class
The homophone strategy is the simplest corruption: pick a word, check if it has a homophone in the index, and substitute it. This produces errors that are acoustically indistinguishable from the original. Real ASR systems make these mistakes constantly.
- Add the HomophoneStrategy class below find_word_for_phonemes() in confusion_strategies.py:
class HomophoneStrategy:
"""Replace words with their homophones."""
def corrupt(self, words, target_wer):
"""
Corrupt a list of words by substituting homophones.
target_wer: fraction of words to attempt replacing (0.0-1.0).
"""
result = list(words)
num_to_corrupt = max(1, int(len(words) * target_wer))
indices = random.sample(range(len(words)), min(num_to_corrupt, len(words)))
corrupted_count = 0
for i in indices:
w = words[i].lower()
if w in WORD_HOMOPHONES:
replacement = random.choice(WORD_HOMOPHONES[w])
result[i] = replacement
corrupted_count += 1
return result, corrupted_countWhat does this code do?
- target_wer controls how many words to attempt corrupting. A WER of 0.1 means try to corrupt 10% of the words.
- The method randomly selects word indices, checks each against the WORD_HOMOPHONES lookup, and substitutes when a homophone exists. Not every selected word will have a homophone, so actual corruption may be lower than the target.
- It returns both the corrupted word list and a count of how many substitutions were made, letting the caller compute the actual WER achieved.
- Save confusion_strategies.py.
- Verify the strategy works by running:
python -c "from confusion_strategies import HomophoneStrategy; s = HomophoneStrategy(); print(s.corrupt('set a timer for five minutes'.split(), 0.3))"You should see a tuple containing the word list (with some words replaced by homophones) and a count of substitutions made. The exact output varies due to randomness, but words like "for" might become "four" or "fore".
Getting zero corruptions every time?
Check that WORD_HOMOPHONES is populated. If the index is empty, the build_homophone_index() function may have a bug in how it strips stress markers. Confirm that strip_stress is imported correctly from phonetic_features.
Help me debug why my HomophoneStrategy returns zero corruptions
Implement the PhonemeNeighborStrategy class
The phoneme-neighbor strategy is more subtle. Instead of swapping entire words for identical-sounding alternatives, it replaces a single phoneme inside a word with a nearby sound in articulatory space. This replicates the OW-to-AO confusion pattern ("coal" becoming "call") identified in the Amazon Interspeech paper.
- Add the PhonemeNeighborStrategy class below HomophoneStrategy in confusion_strategies.py:
class PhonemeNeighborStrategy:
"""Replace words by swapping phonemes with articulatory neighbors."""
def __init__(self, neighbor_threshold=0.4):
self.neighbor_threshold = neighbor_threshold
def corrupt(self, words, target_wer):
"""
Corrupt words by swapping individual phonemes with close neighbors,
then finding the nearest real word for the corrupted phoneme sequence.
"""
result = list(words)
num_to_corrupt = max(1, int(len(words) * target_wer))
indices = random.sample(range(len(words)), min(num_to_corrupt, len(words)))
corrupted_count = 0
for i in indices:
phonemes = get_word_phonemes(words[i])
if not phonemes or len(phonemes) < 2:
continue
# Pick a random phoneme position and swap with a neighbor
pos = random.randint(0, len(phonemes) - 1)
neighbors = get_neighbors(phonemes[pos], self.neighbor_threshold)
if not neighbors:
continue
# Weighted random selection favoring closer neighbors
weights = [1.0 / (dist + 0.1) for _, dist in neighbors]
chosen_phoneme = random.choices(
[n for n, _ in neighbors], weights=weights, k=1
)[0]
# Create corrupted phoneme sequence
corrupted_phonemes = list(phonemes)
corrupted_phonemes[pos] = chosen_phoneme
# Find a real word matching the corrupted phonemes
new_word = find_word_for_phonemes(corrupted_phonemes)
if new_word and new_word.lower() != words[i].lower():
result[i] = new_word
corrupted_count += 1
return result, corrupted_countWhat does this code do?
- neighbor_threshold=0.4 controls how far away a replacement phoneme can be. A higher threshold allows more distant (less plausible) substitutions.
- For each target word, the strategy looks up its phonemes, picks a random position, finds articulatory neighbors using your get_neighbors() function from Step 1, and uses weighted random selection so closer neighbors are more likely to be chosen.
- After swapping the phoneme, it calls find_word_for_phonemes() to map the corrupted sequence back to a real English word. If no match is found, the word is left unchanged.
Why weighted selection toward closer neighbors?
Real ASR systems confuse sounds that are articulatorily close far more often than distant ones. A system is much more likely to hear P as B (differ only in voicing) than P as ZH (differ in manner, place, and voicing). Weighting by inverse distance makes your generated errors follow this natural pattern.
- Save confusion_strategies.py.
You will test this strategy alongside the homophone strategy in the next substep.
✔️ Awesome, I've got everything!
Great. Double check you have saved confusion_strategies.py with both strategy classes in place.
ⓧ I'd like to double check the full code
Here is the complete confusion_strategies.py so far:
"""ASR error simulation strategies: homophone, phoneme-neighbor, word boundary."""
import random
from nltk.corpus import cmudict
from phonetic_features import strip_stress, phonetic_distance, get_neighbors
# Load CMU dict once at module level
CMU_DICT = cmudict.dict()
def build_homophone_index():
"""
Build an index mapping phoneme sequences (without stress) to word lists.
Only includes entries where multiple words share the same pronunciation.
"""
pron_to_words = {}
for word, pronunciations in CMU_DICT.items():
for pron in pronunciations:
key = tuple(strip_stress(p) for p in pron)
if key not in pron_to_words:
pron_to_words[key] = []
if word not in pron_to_words[key]:
pron_to_words[key].append(word)
# Filter to only groups with 2+ words (actual homophones)
return {k: v for k, v in pron_to_words.items() if len(v) > 1}
HOMOPHONE_INDEX = build_homophone_index()
# Reverse lookup: word -> list of homophones
WORD_HOMOPHONES = {}
for words in HOMOPHONE_INDEX.values():
for w in words:
others = [x for x in words if x != w]
if others:
if w not in WORD_HOMOPHONES:
WORD_HOMOPHONES[w] = []
WORD_HOMOPHONES[w].extend(others)
def get_word_phonemes(word):
"""Get the first pronunciation for a word from CMU dict."""
prons = CMU_DICT.get(word.lower())
if prons:
return prons[0]
return None
def find_word_for_phonemes(target_phonemes):
"""
Find the best matching word in CMU dict for a phoneme sequence.
Returns (word, distance) or None if no reasonable match found.
"""
target_stripped = [strip_stress(p) for p in target_phonemes]
target_key = tuple(target_stripped)
# Exact match first
for word, prons in CMU_DICT.items():
for pron in prons:
if tuple(strip_stress(p) for p in pron) == target_key:
return word
# Approximate match: find closest by edit-distance-like metric
best_word = None
best_score = float('inf')
# Sample a subset for performance
candidates = random.sample(list(CMU_DICT.items()), min(5000, len(CMU_DICT)))
for word, prons in candidates:
for pron in prons:
pron_stripped = [strip_stress(p) for p in pron]
if len(pron_stripped) != len(target_stripped):
continue
score = sum(
phonetic_distance(a, b)
for a, b in zip(target_stripped, pron_stripped)
)
if score < best_score:
best_score = score
best_word = word
if best_word and best_score < len(target_stripped) * 0.5:
return best_word
return None
class HomophoneStrategy:
"""Replace words with their homophones."""
def corrupt(self, words, target_wer):
"""
Corrupt a list of words by substituting homophones.
target_wer: fraction of words to attempt replacing (0.0-1.0).
"""
result = list(words)
num_to_corrupt = max(1, int(len(words) * target_wer))
indices = random.sample(range(len(words)), min(num_to_corrupt, len(words)))
corrupted_count = 0
for i in indices:
w = words[i].lower()
if w in WORD_HOMOPHONES:
replacement = random.choice(WORD_HOMOPHONES[w])
result[i] = replacement
corrupted_count += 1
return result, corrupted_count
class PhonemeNeighborStrategy:
"""Replace words by swapping phonemes with articulatory neighbors."""
def __init__(self, neighbor_threshold=0.4):
self.neighbor_threshold = neighbor_threshold
def corrupt(self, words, target_wer):
"""
Corrupt words by swapping individual phonemes with close neighbors,
then finding the nearest real word for the corrupted phoneme sequence.
"""
result = list(words)
num_to_corrupt = max(1, int(len(words) * target_wer))
indices = random.sample(range(len(words)), min(num_to_corrupt, len(words)))
corrupted_count = 0
for i in indices:
phonemes = get_word_phonemes(words[i])
if not phonemes or len(phonemes) < 2:
continue
# Pick a random phoneme position and swap with a neighbor
pos = random.randint(0, len(phonemes) - 1)
neighbors = get_neighbors(phonemes[pos], self.neighbor_threshold)
if not neighbors:
continue
# Weighted random selection favoring closer neighbors
weights = [1.0 / (dist + 0.1) for _, dist in neighbors]
chosen_phoneme = random.choices(
[n for n, _ in neighbors], weights=weights, k=1
)[0]
# Create corrupted phoneme sequence
corrupted_phonemes = list(phonemes)
corrupted_phonemes[pos] = chosen_phoneme
# Find a real word matching the corrupted phonemes
new_word = find_word_for_phonemes(corrupted_phonemes)
if new_word and new_word.lower() != words[i].lower():
result[i] = new_word
corrupted_count += 1
return result, corrupted_countTest both strategies on sample utterances
Time to validate that both strategies produce realistic corruptions on the kind of utterances a voice assistant would process.
- Run the following test script to exercise both strategies on three sample utterances:
python -c "
from confusion_strategies import HomophoneStrategy, PhonemeNeighborStrategy
utterances = [
'set a timer for five minutes',
'play some calm music',
'what is the weather today',
]
homo = HomophoneStrategy()
neighbor = PhonemeNeighborStrategy()
print('=== Homophone Strategy (WER=0.3) ===')
for u in utterances:
words = u.split()
result, count = homo.corrupt(words, 0.3)
print(f' Original: {u}')
print(f' Corrupted: {\" \".join(result)} ({count} swaps)')
print()
print('=== Phoneme Neighbor Strategy (WER=0.3) ===')
for u in utterances:
words = u.split()
result, count = neighbor.corrupt(words, 0.3)
print(f' Original: {u}')
print(f' Corrupted: {\" \".join(result)} ({count} swaps)')
print()
"What should I see?
The homophone strategy should swap words like "for" to "four" or "fore", "some" to "sum", or "weather" to "whether". These are exact sound-alikes.
The phoneme-neighbor strategy should produce subtler changes. Words get replaced with similar-sounding (but not identical) words. Results vary by run due to randomness. Some words may not corrupt if no valid real-word match is found for the modified phoneme sequence.
Phoneme neighbor strategy running slowly?
The find_word_for_phonemes() function samples 5000 dictionary entries for approximate matching. The first run may take a few seconds per utterance. This is expected for a project-scoped implementation. If it takes more than 30 seconds total, check that you are not accidentally iterating the full dictionary multiple times.
Help me debug slow performance in my phoneme neighbor strategy
You now have two distinct corruption strategies that replicate real ASR error patterns. Next, you will build a word boundary confusion strategy and combine all three into a configurable mixer.
Build the Word Boundary Strategy and Strategy Mixer
Your homophone and phoneme-neighbor strategies corrupt individual words, but real ASR systems also struggle at word boundaries.
When a speaker says "ice cream," the phoneme stream has no silence gap between the two words. An ASR decoder must decide where one word ends and another begins. This ambiguity produces errors like "ice cream" becoming "I scream." In this step, you will build a strategy that exploits this weakness, then combine all three strategies into a configurable mixer.
In this step, get ready to:
- Implement a word boundary confusion strategy that re-segments phoneme streams across adjacent words.
- Build a strategy mixer that selects corruption strategies by configurable probability weights.
- Validate that the mixer produces a blend of all three error types on sample utterances.
Implement the word boundary strategy
The word boundary strategy works differently from the first two. Instead of corrupting a single word in isolation, it targets the seam between two adjacent words.
- In confusion_strategies.py, scroll to the bottom of the file (below the PhonemeNeighborStrategy class).
- Add the WordBoundaryStrategy class by copying the following code:
class WordBoundaryStrategy:
"""Corrupt by re-segmenting phoneme streams across word boundaries."""
def corrupt(self, words, target_wer):
"""
Corrupt by merging adjacent words' phonemes and re-segmenting.
"""
result = list(words)
if len(words) < 2:
return result, 0
# Pick adjacent word pairs to attempt boundary confusion
num_attempts = max(1, int(len(words) * target_wer / 2))
corrupted_count = 0
for _ in range(num_attempts):
if len(result) < 2:
break
idx = random.randint(0, len(result) - 2)
w1_phonemes = get_word_phonemes(result[idx])
w2_phonemes = get_word_phonemes(result[idx + 1])
if not w1_phonemes or not w2_phonemes:
continue
# Concatenate phoneme streams
combined = w1_phonemes + w2_phonemes
total_len = len(combined)
if total_len < 3:
continue
# Try different split points
original_split = len(w1_phonemes)
split_offsets = [-1, 1, -2, 2]
random.shuffle(split_offsets)
for offset in split_offsets:
new_split = original_split + offset
if new_split < 1 or new_split >= total_len:
continue
left_phonemes = combined[:new_split]
right_phonemes = combined[new_split:]
left_word = find_word_for_phonemes(left_phonemes)
right_word = find_word_for_phonemes(right_phonemes)
if left_word and right_word:
new_pair = f"{left_word} {right_word}"
old_pair = f"{result[idx]} {result[idx + 1]}"
if new_pair.lower() != old_pair.lower():
result[idx] = left_word
result[idx + 1] = right_word
corrupted_count += 2
break
return result, corrupted_countWhat does this code do?
This strategy simulates the word-boundary ambiguity that trips up real ASR decoders. It picks a pair of adjacent words and concatenates their phoneme sequences into a single stream, erasing the boundary.
Then it tries re-splitting at a slightly different point (offsets of -1, +1, -2, +2 phonemes from the original boundary). Both resulting chunks must map to real English words in CMU dict for the corruption to succeed. This ensures the output is always a valid word pair, just not the intended one.
Why is this the hardest strategy?
Re-segmented phoneme chunks must each independently map to real words. Most random split points produce nonsense. The implementation shuffles split offsets and tries each one, accepting the first valid pair it finds.
The Amazon Interspeech 2020 paper names compound word boundary errors as a primary ASR confusion category alongside homophones and phoneme-neighbor swaps.
- Save confusion_strategies.py.
Getting an IndentationError?
Make sure the entire WordBoundaryStrategy class is indented consistently. The class body should use 4-space indentation, and the nested for offset in split_offsets: loop should be 12 spaces deep (three levels).
help me fix an IndentationError in my WordBoundaryStrategy class
Add the strategy mixer
With all three strategies built, you need a way to combine them. The mixer selects which strategy to apply on each corruption attempt based on configurable probability weights.
- In confusion_strategies.py, add the StrategyMixer class below WordBoundaryStrategy:
class StrategyMixer:
"""Combines multiple confusion strategies with configurable weights."""
def __init__(self, strategy_weights):
"""
strategy_weights: dict mapping strategy instances to their probability weights.
Example: {HomophoneStrategy(): 0.4, PhonemeNeighborStrategy(): 0.4, WordBoundaryStrategy(): 0.2}
"""
self.strategies = list(strategy_weights.keys())
self.weights = list(strategy_weights.values())
def corrupt(self, words, target_wer):
"""Apply a randomly selected strategy based on weights."""
strategy = random.choices(self.strategies, weights=self.weights, k=1)[0]
return strategy.corrupt(words, target_wer)What does this code do?
The StrategyMixer accepts a dictionary mapping strategy instances to probability weights. On each call to corrupt(), it randomly selects one strategy using random.choices (which respects the weight distribution) and delegates the corruption to that strategy.
This design keeps individual strategies decoupled. The curriculum scheduler you build next will configure different weight distributions for each difficulty stage.
- Save confusion_strategies.py.
Validate the mixer on sample utterances
Now you will confirm that the mixer correctly dispatches across all three strategies and produces varied corruptions on real input.
- Run the following test script in your terminal to exercise the mixer:
python -c "
import random
random.seed(42)
from confusion_strategies import (
HomophoneStrategy, PhonemeNeighborStrategy,
WordBoundaryStrategy, StrategyMixer
)
mixer = StrategyMixer({
HomophoneStrategy(): 0.4,
PhonemeNeighborStrategy(): 0.4,
WordBoundaryStrategy(): 0.2,
})
test_utterances = [
'set a timer for five minutes',
'play some calm music',
'what is the weather today',
]
for utterance in test_utterances:
words = utterance.split()
corrupted, count = mixer.corrupt(words, target_wer=0.3)
print(f'Original: {utterance}')
print(f'Corrupted: {\" \".join(corrupted)} ({count} words changed)')
print()
"What should I see?
You should see three utterance pairs printed. Each "Corrupted" line shows a modified version of the original. Some corruptions will be homophone swaps (e.g., "for" becoming "four"), some will be phoneme-neighbor changes, and some may show word boundary re-segmentations.
The exact output varies because the mixer randomly selects strategies. The key thing to verify is that corruptions are appearing and the word count shows changes.
Seeing ImportError or no corruptions?
If you get an ImportError for WordBoundaryStrategy or StrategyMixer, confirm you saved the file after adding both classes. Check that the class names match exactly (capital W, B, S, M).
If all corrupted counts show 0, the word boundary strategy may not be finding valid re-segmentations for these short utterances. That is expected sometimes. The homophone and phoneme-neighbor strategies should still produce changes.
help me debug why my StrategyMixer is not producing corruptions
✔️ Awesome, I've got everything!
Great. Double check you have saved confusion_strategies.py with both new classes.
ⓧ I'd like to double check the full code
"""ASR error simulation strategies: homophone, phoneme-neighbor, word boundary."""
import random
from nltk.corpus import cmudict
from phonetic_features import strip_stress, phonetic_distance, get_neighbors
# Load CMU dict once at module level
CMU_DICT = cmudict.dict()
def build_homophone_index():
"""
Build an index mapping phoneme sequences (without stress) to word lists.
Only includes entries where multiple words share the same pronunciation.
"""
pron_to_words = {}
for word, pronunciations in CMU_DICT.items():
for pron in pronunciations:
key = tuple(strip_stress(p) for p in pron)
if key not in pron_to_words:
pron_to_words[key] = []
if word not in pron_to_words[key]:
pron_to_words[key].append(word)
# Filter to only groups with 2+ words (actual homophones)
return {k: v for k, v in pron_to_words.items() if len(v) > 1}
HOMOPHONE_INDEX = build_homophone_index()
# Reverse lookup: word -> list of homophones
WORD_HOMOPHONES = {}
for words in HOMOPHONE_INDEX.values():
for w in words:
others = [x for x in words if x != w]
if others:
if w not in WORD_HOMOPHONES:
WORD_HOMOPHONES[w] = []
WORD_HOMOPHONES[w].extend(others)
def get_word_phonemes(word):
"""Get the first pronunciation for a word from CMU dict."""
prons = CMU_DICT.get(word.lower())
if prons:
return prons[0]
return None
def find_word_for_phonemes(target_phonemes):
"""
Find the best matching word in CMU dict for a phoneme sequence.
Returns (word, distance) or None if no reasonable match found.
"""
target_stripped = [strip_stress(p) for p in target_phonemes]
target_key = tuple(target_stripped)
# Exact match first
for word, prons in CMU_DICT.items():
for pron in prons:
if tuple(strip_stress(p) for p in pron) == target_key:
return word
# Approximate match: find closest by edit-distance-like metric
best_word = None
best_score = float('inf')
# Sample a subset for performance
candidates = random.sample(list(CMU_DICT.items()), min(5000, len(CMU_DICT)))
for word, prons in candidates:
for pron in prons:
pron_stripped = [strip_stress(p) for p in pron]
if len(pron_stripped) != len(target_stripped):
continue
score = sum(
phonetic_distance(a, b)
for a, b in zip(target_stripped, pron_stripped)
)
if score < best_score:
best_score = score
best_word = word
if best_word and best_score < len(target_stripped) * 0.5:
return best_word
return None
class HomophoneStrategy:
"""Replace words with their homophones."""
def corrupt(self, words, target_wer):
"""
Corrupt a list of words by substituting homophones.
target_wer: fraction of words to attempt replacing (0.0-1.0).
"""
result = list(words)
num_to_corrupt = max(1, int(len(words) * target_wer))
indices = random.sample(range(len(words)), min(num_to_corrupt, len(words)))
corrupted_count = 0
for i in indices:
w = words[i].lower()
if w in WORD_HOMOPHONES:
replacement = random.choice(WORD_HOMOPHONES[w])
result[i] = replacement
corrupted_count += 1
return result, corrupted_count
class PhonemeNeighborStrategy:
"""Replace words by swapping phonemes with articulatory neighbors."""
def __init__(self, neighbor_threshold=0.4):
self.neighbor_threshold = neighbor_threshold
def corrupt(self, words, target_wer):
"""
Corrupt words by swapping individual phonemes with close neighbors,
then finding the nearest real word for the corrupted phoneme sequence.
"""
result = list(words)
num_to_corrupt = max(1, int(len(words) * target_wer))
indices = random.sample(range(len(words)), min(num_to_corrupt, len(words)))
corrupted_count = 0
for i in indices:
phonemes = get_word_phonemes(words[i])
if not phonemes or len(phonemes) < 2:
continue
# Pick a random phoneme position and swap with a neighbor
pos = random.randint(0, len(phonemes) - 1)
neighbors = get_neighbors(phonemes[pos], self.neighbor_threshold)
if not neighbors:
continue
# Weighted random selection favoring closer neighbors
weights = [1.0 / (dist + 0.1) for _, dist in neighbors]
chosen_phoneme = random.choices(
[n for n, _ in neighbors], weights=weights, k=1
)[0]
# Create corrupted phoneme sequence
corrupted_phonemes = list(phonemes)
corrupted_phonemes[pos] = chosen_phoneme
# Find a real word matching the corrupted phonemes
new_word = find_word_for_phonemes(corrupted_phonemes)
if new_word and new_word.lower() != words[i].lower():
result[i] = new_word
corrupted_count += 1
return result, corrupted_count
class WordBoundaryStrategy:
"""Corrupt by re-segmenting phoneme streams across word boundaries."""
def corrupt(self, words, target_wer):
"""
Corrupt by merging adjacent words' phonemes and re-segmenting.
"""
result = list(words)
if len(words) < 2:
return result, 0
# Pick adjacent word pairs to attempt boundary confusion
num_attempts = max(1, int(len(words) * target_wer / 2))
corrupted_count = 0
for _ in range(num_attempts):
if len(result) < 2:
break
idx = random.randint(0, len(result) - 2)
w1_phonemes = get_word_phonemes(result[idx])
w2_phonemes = get_word_phonemes(result[idx + 1])
if not w1_phonemes or not w2_phonemes:
continue
# Concatenate phoneme streams
combined = w1_phonemes + w2_phonemes
total_len = len(combined)
if total_len < 3:
continue
# Try different split points
original_split = len(w1_phonemes)
split_offsets = [-1, 1, -2, 2]
random.shuffle(split_offsets)
for offset in split_offsets:
new_split = original_split + offset
if new_split < 1 or new_split >= total_len:
continue
left_phonemes = combined[:new_split]
right_phonemes = combined[new_split:]
left_word = find_word_for_phonemes(left_phonemes)
right_word = find_word_for_phonemes(right_phonemes)
if left_word and right_word:
new_pair = f"{left_word} {right_word}"
old_pair = f"{result[idx]} {result[idx + 1]}"
if new_pair.lower() != old_pair.lower():
result[idx] = left_word
result[idx + 1] = right_word
corrupted_count += 2
break
return result, corrupted_count
class StrategyMixer:
"""Combines multiple confusion strategies with configurable weights."""
def __init__(self, strategy_weights):
"""
strategy_weights: dict mapping strategy instances to their probability weights.
Example: {HomophoneStrategy(): 0.4, PhonemeNeighborStrategy(): 0.4, WordBoundaryStrategy(): 0.2}
"""
self.strategies = list(strategy_weights.keys())
self.weights = list(strategy_weights.values())
def corrupt(self, words, target_wer):
"""Apply a randomly selected strategy based on weights."""
strategy = random.choices(self.strategies, weights=self.weights, k=1)[0]
return strategy.corrupt(words, target_wer)All three confusion strategies are implemented and wired together through the mixer. Next up, you will build a curriculum scheduler that controls which strategies fire and at what intensity across progressive training stages.
Build the Curriculum Scheduler and CLI
Your three confusion strategies are ready to produce realistic ASR errors. The missing piece is control over when each strategy activates and how aggressively it corrupts.
In this step, you will build a curriculum scheduler that ramps corruption difficulty across training stages, then wrap the entire system in a CLI tool that processes utterance files and outputs augmented datasets.
In this step, get ready to:
- Create a curriculum scheduler that controls WER targets and strategy weights per stage.
- Build a CLI entry point that orchestrates the full pipeline.
- Run the pipeline on sample utterances and verify progressive corruption output.
Implement the curriculum scheduler
The curriculum scheduler follows the principle of starting with easy corruptions and progressively adding harder ones. Stage 1 uses only homophones (lexically valid swaps). Stage 2 adds phoneme-neighbor substitutions. Stage 3 mixes all three strategies at a higher WER.
- Create a new file called curriculum.py in your project folder.
- Add the imports and stage definitions by pasting this code:
"""Curriculum scheduler for progressive ASR error difficulty."""
from confusion_strategies import (
HomophoneStrategy,
PhonemeNeighborStrategy,
WordBoundaryStrategy,
StrategyMixer,
)
# Default curriculum stage definitions
DEFAULT_STAGES = [
{
'name': 'Stage 1: Homophones Only',
'wer': 0.1,
'weights': {'homophone': 0.9, 'phoneme_neighbor': 0.1, 'boundary': 0.0},
},
{
'name': 'Stage 2: Homophones + Phoneme Neighbors',
'wer': 0.2,
'weights': {'homophone': 0.4, 'phoneme_neighbor': 0.5, 'boundary': 0.1},
},
{
'name': 'Stage 3: All Strategies Mixed',
'wer': 0.3,
'weights': {'homophone': 0.3, 'phoneme_neighbor': 0.4, 'boundary': 0.3},
},
]What does this code do?
Each stage dictionary defines three things: a human-readable name, a target WER (word error rate as a fraction), and the probability weights for each corruption strategy.
Stage 1 targets 10% WER using almost exclusively homophones. Stage 2 bumps to 20% WER and shifts weight toward phoneme neighbors. Stage 3 hits 30% WER with all strategies active. This progression mirrors how curriculum learning exposes an NLU model to increasingly noisy data.
- Now add the CurriculumScheduler class below the DEFAULT_STAGES list:
class CurriculumScheduler:
"""Controls difficulty progression across training stages."""
def __init__(self, stages=None):
self.stages = stages or DEFAULT_STAGES
self._strategies = {
'homophone': HomophoneStrategy(),
'phoneme_neighbor': PhonemeNeighborStrategy(),
'boundary': WordBoundaryStrategy(),
}
def get_stage_config(self, stage_num):
"""Get configuration for a given stage (0-indexed)."""
if stage_num >= len(self.stages):
return self.stages[-1]
return self.stages[stage_num]
def get_mixer(self, stage_num):
"""Create a StrategyMixer configured for the given stage."""
config = self.get_stage_config(stage_num)
weights = config['weights']
strategy_weights = {}
for name, weight in weights.items():
if weight > 0:
strategy_weights[self._strategies[name]] = weight
return StrategyMixer(strategy_weights)
def generate_epoch(self, utterances, stage_num):
"""
Generate a corrupted dataset for a given curriculum stage.
Returns list of (original, corrupted, num_corruptions) tuples.
"""
config = self.get_stage_config(stage_num)
mixer = self.get_mixer(stage_num)
target_wer = config['wer']
results = []
for utterance in utterances:
words = utterance.strip().split()
if not words:
results.append((utterance, utterance, 0))
continue
corrupted_words, num_corrupted = mixer.corrupt(words, target_wer)
corrupted_text = ' '.join(corrupted_words)
results.append((utterance.strip(), corrupted_text, num_corrupted))
return results
def summary(self, stage_num, results):
"""Generate a summary report for a curriculum stage."""
config = self.get_stage_config(stage_num)
total_words = sum(len(orig.split()) for orig, _, _ in results)
total_corrupted = sum(n for _, _, n in results)
actual_wer = total_corrupted / total_words if total_words > 0 else 0.0
lines = [
f"--- {config['name']} ---",
f" Target WER: {config['wer']:.1%}",
f" Actual WER: {actual_wer:.1%}",
f" Strategy weights: {config['weights']}",
f" Utterances processed: {len(results)}",
f" Total words: {total_words}, Corrupted: {total_corrupted}",
"",
" Example corruptions:",
]
# Show up to 5 examples where corruption occurred
examples = [(o, c) for o, c, n in results if n > 0][:5]
for orig, corrupted in examples:
lines.append(f" Original: {orig}")
lines.append(f" Corrupted: {corrupted}")
lines.append("")
return '\n'.join(lines)What does this class do?
- get_stage_config() returns the WER target and strategy weights for a given stage number.
- get_mixer() builds a StrategyMixer with only the strategies that have non-zero weight for that stage.
- generate_epoch() iterates over all utterances, corrupts each one using the stage's mixer, and returns original/corrupted pairs with corruption counts.
- summary() computes the actual WER achieved and formats a human-readable report with example corruptions.
- Save curriculum.py.
Seeing an ImportError?
Make sure confusion_strategies.py is in the same directory as curriculum.py. The import pulls all three strategy classes and the StrategyMixer from that file.
help me debug an ImportError when importing from confusion_strategies
✔️ Awesome, I've got everything!
Great. Double check you have saved curriculum.py before moving on.
ⓧ I'd like to double check the full code
"""Curriculum scheduler for progressive ASR error difficulty."""
from confusion_strategies import (
HomophoneStrategy,
PhonemeNeighborStrategy,
WordBoundaryStrategy,
StrategyMixer,
)
# Default curriculum stage definitions
DEFAULT_STAGES = [
{
'name': 'Stage 1: Homophones Only',
'wer': 0.1,
'weights': {'homophone': 0.9, 'phoneme_neighbor': 0.1, 'boundary': 0.0},
},
{
'name': 'Stage 2: Homophones + Phoneme Neighbors',
'wer': 0.2,
'weights': {'homophone': 0.4, 'phoneme_neighbor': 0.5, 'boundary': 0.1},
},
{
'name': 'Stage 3: All Strategies Mixed',
'wer': 0.3,
'weights': {'homophone': 0.3, 'phoneme_neighbor': 0.4, 'boundary': 0.3},
},
]
class CurriculumScheduler:
"""Controls difficulty progression across training stages."""
def __init__(self, stages=None):
self.stages = stages or DEFAULT_STAGES
self._strategies = {
'homophone': HomophoneStrategy(),
'phoneme_neighbor': PhonemeNeighborStrategy(),
'boundary': WordBoundaryStrategy(),
}
def get_stage_config(self, stage_num):
"""Get configuration for a given stage (0-indexed)."""
if stage_num >= len(self.stages):
return self.stages[-1]
return self.stages[stage_num]
def get_mixer(self, stage_num):
"""Create a StrategyMixer configured for the given stage."""
config = self.get_stage_config(stage_num)
weights = config['weights']
strategy_weights = {}
for name, weight in weights.items():
if weight > 0:
strategy_weights[self._strategies[name]] = weight
return StrategyMixer(strategy_weights)
def generate_epoch(self, utterances, stage_num):
"""
Generate a corrupted dataset for a given curriculum stage.
Returns list of (original, corrupted, num_corruptions) tuples.
"""
config = self.get_stage_config(stage_num)
mixer = self.get_mixer(stage_num)
target_wer = config['wer']
results = []
for utterance in utterances:
words = utterance.strip().split()
if not words:
results.append((utterance, utterance, 0))
continue
corrupted_words, num_corrupted = mixer.corrupt(words, target_wer)
corrupted_text = ' '.join(corrupted_words)
results.append((utterance.strip(), corrupted_text, num_corrupted))
return results
def summary(self, stage_num, results):
"""Generate a summary report for a curriculum stage."""
config = self.get_stage_config(stage_num)
total_words = sum(len(orig.split()) for orig, _, _ in results)
total_corrupted = sum(n for _, _, n in results)
actual_wer = total_corrupted / total_words if total_words > 0 else 0.0
lines = [
f"--- {config['name']} ---",
f" Target WER: {config['wer']:.1%}",
f" Actual WER: {actual_wer:.1%}",
f" Strategy weights: {config['weights']}",
f" Utterances processed: {len(results)}",
f" Total words: {total_words}, Corrupted: {total_corrupted}",
"",
" Example corruptions:",
]
# Show up to 5 examples where corruption occurred
examples = [(o, c) for o, c, n in results if n > 0][:5]
for orig, corrupted in examples:
lines.append(f" Original: {orig}")
lines.append(f" Corrupted: {corrupted}")
lines.append("")
return '\n'.join(lines)Build the CLI entry point
The CLI is the user-facing interface that ties the curriculum scheduler to file I/O. It reads clean utterances from a text file, runs each curriculum stage, and writes both corrupted output files and summary reports.
- Create a new file called asr_confuser.py in your project folder.
- Add the imports and argument parser by pasting this code:
"""CLI entry point for the ASR phonetic confusion generator."""
import argparse
import os
import sys
from curriculum import CurriculumScheduler
def main():
parser = argparse.ArgumentParser(
description='ASR Phonetic Confusion Generator with Curriculum Scheduling'
)
parser.add_argument(
'--input', '-i', required=True,
help='Path to clean utterance file (one utterance per line)'
)
parser.add_argument(
'--stages', '-s', type=int, default=3,
help='Number of curriculum stages to generate (default: 3)'
)
parser.add_argument(
'--output-dir', '-o', default='output',
help='Output directory for corrupted files (default: output)'
)
parser.add_argument(
'--seed', type=int, default=42,
help='Random seed for reproducibility (default: 42)'
)
args = parser.parse_args()What does this code do?
- --input (required) points to a text file with one clean utterance per line.
- --stages controls how many curriculum stages to generate (defaults to 3).
- --output-dir sets where corrupted files and reports get written.
- --seed ensures reproducible corruption output across runs.
- Now add the pipeline logic directly below args = parser.parse_args(), still inside the main() function:
# Set random seed
import random
random.seed(args.seed)
# Read input utterances
if not os.path.exists(args.input):
print(f"Error: Input file not found: {args.input}")
sys.exit(1)
with open(args.input, 'r', encoding='utf-8') as f:
utterances = [line.strip() for line in f if line.strip()]
print(f"Loaded {len(utterances)} utterances from {args.input}")
print()
# Create output directory
os.makedirs(args.output_dir, exist_ok=True)
# Initialize curriculum scheduler
scheduler = CurriculumScheduler()
# Generate corrupted data for each stage
for stage_num in range(min(args.stages, len(scheduler.stages))):
print(f"Generating Stage {stage_num + 1}...")
results = scheduler.generate_epoch(utterances, stage_num)
# Write corrupted utterances to file
output_file = os.path.join(
args.output_dir, f"stage_{stage_num + 1}_corrupted.txt"
)
with open(output_file, 'w', encoding='utf-8') as f:
for _, corrupted, _ in results:
f.write(corrupted + '\n')
# Print summary report
report = scheduler.summary(stage_num, results)
print(report)
# Write report to file
report_file = os.path.join(
args.output_dir, f"stage_{stage_num + 1}_report.txt"
)
with open(report_file, 'w', encoding='utf-8') as f:
f.write(report)
print(f"\nDone! Output written to: {args.output_dir}")
if __name__ == '__main__':
main()What does this pipeline do?
The pipeline reads the input file, initializes the scheduler, then loops through each curriculum stage. For each stage it calls generate_epoch() to corrupt the utterances, writes the corrupted text to a file, prints a summary report, and saves that report to disk.
The output directory ends up with one stage_N_corrupted.txt file and one stage_N_report.txt file per stage.
- Save asr_confuser.py.
Seeing a ModuleNotFoundError for curriculum?
Confirm that curriculum.py is saved in the same directory as asr_confuser.py. Also check that the file is named exactly curriculum.py (not curriculum_scheduler.py or similar).
help me fix a ModuleNotFoundError when running asr_confuser.py
✔️ Awesome, I've got everything!
Great. Make sure asr_confuser.py is saved before running the pipeline.
ⓧ I'd like to double check the full code
"""CLI entry point for the ASR phonetic confusion generator."""
import argparse
import os
import sys
from curriculum import CurriculumScheduler
def main():
parser = argparse.ArgumentParser(
description='ASR Phonetic Confusion Generator with Curriculum Scheduling'
)
parser.add_argument(
'--input', '-i', required=True,
help='Path to clean utterance file (one utterance per line)'
)
parser.add_argument(
'--stages', '-s', type=int, default=3,
help='Number of curriculum stages to generate (default: 3)'
)
parser.add_argument(
'--output-dir', '-o', default='output',
help='Output directory for corrupted files (default: output)'
)
parser.add_argument(
'--seed', type=int, default=42,
help='Random seed for reproducibility (default: 42)'
)
args = parser.parse_args()
# Set random seed
import random
random.seed(args.seed)
# Read input utterances
if not os.path.exists(args.input):
print(f"Error: Input file not found: {args.input}")
sys.exit(1)
with open(args.input, 'r', encoding='utf-8') as f:
utterances = [line.strip() for line in f if line.strip()]
print(f"Loaded {len(utterances)} utterances from {args.input}")
print()
# Create output directory
os.makedirs(args.output_dir, exist_ok=True)
# Initialize curriculum scheduler
scheduler = CurriculumScheduler()
# Generate corrupted data for each stage
for stage_num in range(min(args.stages, len(scheduler.stages))):
print(f"Generating Stage {stage_num + 1}...")
results = scheduler.generate_epoch(utterances, stage_num)
# Write corrupted utterances to file
output_file = os.path.join(
args.output_dir, f"stage_{stage_num + 1}_corrupted.txt"
)
with open(output_file, 'w', encoding='utf-8') as f:
for _, corrupted, _ in results:
f.write(corrupted + '\n')
# Print summary report
report = scheduler.summary(stage_num, results)
print(report)
# Write report to file
report_file = os.path.join(
args.output_dir, f"stage_{stage_num + 1}_report.txt"
)
with open(report_file, 'w', encoding='utf-8') as f:
f.write(report)
print(f"\nDone! Output written to: {args.output_dir}")
if __name__ == '__main__':
main()Run the full pipeline on sample utterances
You need a sample input file to test the pipeline end-to-end. This file contains 20 intent-classification utterances that represent typical voice assistant inputs.
- Create a file called sample_utterances.txt in your project folder and paste these 20 utterances:
set a timer for five minutes
play some calm music
what is the weather today
turn off the living room lights
add milk to my shopping list
remind me to call mom tomorrow
what time is my next meeting
play the latest news podcast
set an alarm for seven thirty
how far is the nearest gas station
order more paper towels
cancel my three o'clock appointment
what song is this
turn up the volume
send a message to dad
what is the capital of france
find me a good italian restaurant
read my latest email
how do you say hello in spanish
check my bank account balance- Save sample_utterances.txt.
- Run the full pipeline by executing this command:
python asr_confuser.py --input sample_utterances.txt --output-dir output --seed 42You should see three stage reports printed to your terminal, each showing target WER, actual WER, strategy weights, and example corruptions.
What should I see?
The output prints three sections, one per stage. Each section shows the stage name, target vs actual WER percentages, and up to 5 example original/corrupted pairs. Stage 1 corruptions should mostly be homophone swaps (like "their" to "there"). Stage 3 should show a mix of all three error types at a higher corruption rate.
The actual WER may not exactly hit the target because not every word has a homophone or valid phoneme-neighbor substitution. This is expected behavior.
- Confirm the output files were created by running:
ls outputYou should see six files: stage_1_corrupted.txt, stage_1_report.txt, stage_2_corrupted.txt, stage_2_report.txt, stage_3_corrupted.txt, and stage_3_report.txt.
Pipeline takes a long time or hangs?
The find_word_for_phonemes() function in confusion_strategies.py samples 5000 words for approximate matching. On slower machines this can take a few seconds per utterance. Wait for at least 30 seconds before assuming it has hung.
If you see no output at all, check that sample_utterances.txt is not empty and is in the same directory where you ran the command.
help me debug my asr_confuser pipeline run
Your phonetic confusion generator is fully operational. You now have a CLI tool that produces progressively harder ASR-style corruptions controlled by a curriculum scheduler. Next up, you will extend this with a confusion matrix visualization in the Secret Mission.
Secret mission
Add Confusion Matrix Visualization
Your generator produces corruptions, but how do you know they match real ASR error patterns? Build a confusion matrix that aggregates phoneme substitution statistics across large batches and validates your output against known ASR research findings.
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely locally using Python and NLTK's bundled CMU Pronouncing Dictionary. There are no ongoing costs or cloud resources to worry about.
Decide whether to keep your project files for future experimentation, or remove everything to free up disk space.
Resources you used:
- Project folder with Python source files (phonetic_features.py, confusion_strategies.py, curriculum.py, asr_confuser.py, confusion_matrix.py).
- Generated output directory with corrupted utterance files and reports.
- NLTK package and downloaded cmudict corpus data.
- Sample utterance input file.
Keep everything running
No action needed. Choose this if you plan to keep experimenting with different utterance sets, adjust curriculum stages, or extend the confusion strategies.
- Your project files stay on your Desktop ready for reuse.
- The NLTK cmudict corpus remains cached in ~/nltk_data for fast access in future NLP projects.
- The output directory contains your generated corrupted datasets and reports.
Pause - I'll come back to this later
Since this is a local project with no running processes, there is nothing to pause. Your files sit on disk until you return.
- Delete the output directory to free space. You can regenerate it any time by re-running the CLI.
- Keep the source files and NLTK data intact so you can pick up where you left off.
Remove the output directory by running:
rm -rf outputDelete - I don't want to use this again
Remove all project files, generated output, and downloaded corpus data.
- Delete your project folder from the Desktop by running:
cd ~/Desktop
rm -rf asr_confuser- Remove the downloaded NLTK corpus data by running:
rm -rf ~/nltk_data- Uninstall the NLTK package by running:
pip uninstall nltk -yWhat about the virtual environment?
If you created a virtual environment for this project, deleting the project folder removes it too (if it lived inside the folder). If you created it elsewhere, deactivate it and delete its directory separately.
Nice Work!
Nice Work!
You just built a phonetic confusion generator grounded in articulatory phonetics and curriculum learning theory. Your tool produces realistic ASR-style corruptions that can harden NLU models against the exact error patterns real speech recognizers make.
You've learned how to:
- Build a complete ARPABET phonetic feature system with articulatory distance computation for all 39 CMU phonemes.
- Implement three distinct ASR error simulation strategies: homophone substitution, phoneme-neighbor swaps, and word boundary confusion.
- Create a curriculum scheduler that progressively increases corruption difficulty across training stages.
- Build a CLI tool that processes clean utterances and outputs augmented training data with corruption reports.
- Secret Mission: Generated phoneme confusion matrices that validate your generator produces phonetically plausible error patterns matching real ASR behavior.
Ready to quiz yourself?