32 · IBM Models 2–5 + HMM Alignment#
Dependencies: numpy, scipy
Runtime: ~1 min CPU
Chapter stub — implementation coming in a future commit.
Time-machine sentences#
Every chapter runs the same 20 held-out test sentences so the quality arc across chapters is directly comparable.
# Colab setup: install this part's pinned dependencies (skipped outside Colab).
import sys
if "google.colab" in sys.modules:
get_ipython().system("wget -q -O requirements_part3.txt https://raw.githubusercontent.com/eduardosanchezg/mthistory/main/requirements_part3.txt")
get_ipython().system("pip install -q -r requirements_part3.txt")
IBM Models 2–5 + HMM Alignment#
This chapter builds on IBM Model 1 (chapter 31) and shows how adding positional and sequential information dramatically sharpens word alignments. We implement:
IBM Model 2 — position-based alignment a(i|j,l,m)
IBM Model 3 — conceptual sketch of fertility
HMM Alignment (Vogel et al. 1996) — first-order Markov chain over positions
Side-by-side alignment comparison
Time-machine: align TIME_MACHINE_EN sentences
CORPUS = [
("the cat sat on the mat", "die Katze sass auf der Matte"),
("the dog runs fast", "der Hund laeuft schnell"),
("a man walks in the park", "ein Mann geht im Park"),
("the woman reads a book", "die Frau liest ein Buch"),
("children play in the garden", "Kinder spielen im Garten"),
("the cat and the dog", "die Katze und der Hund"),
("a fast dog runs", "ein schneller Hund laeuft"),
("the man reads the book", "der Mann liest das Buch"),
("a woman walks fast", "eine Frau geht schnell"),
("children run in the park", "Kinder laufen im Park"),
]
TIME_MACHINE_EN = [
"A man is walking his dog in the park .",
"Two children are playing near a fountain .",
"A woman in a red coat is reading a book .",
"Several people are waiting at a bus stop .",
"A cyclist is riding through a crowded street .",
"A dog is chasing a ball on the beach .",
"A group of tourists is taking photographs .",
"A young girl is feeding ducks by a pond .",
"Two men are playing chess in a cafe .",
"A street musician is playing the violin .",
"Children are running through a field of flowers .",
"An old man is sitting on a bench reading a newspaper .",
"A woman is carrying groceries up a flight of stairs .",
"A boy is kicking a football against a wall .",
"A couple is dancing on an empty street .",
"A chef is preparing food in an outdoor kitchen .",
"A cat is sleeping on a warm windowsill .",
"Workers are repairing a road in the rain .",
"A small boat is sailing on a calm lake .",
"Firefighters are climbing a tall ladder .",
]
print(f"Corpus: {len(CORPUS)} sentence pairs")
print(f"Time-machine sentences: {len(TIME_MACHINE_EN)}")
# IBM Model 1 (re-implemented here for self-containment)
from collections import defaultdict
import math
def ibm_model1(corpus, n_iters=20):
# collect vocab
src_vocab = set()
tgt_vocab = set()
for src, tgt in corpus:
src_vocab.update(src.split())
tgt_vocab.update(tgt.split())
# uniform init
t = defaultdict(lambda: 1.0 / (len(tgt_vocab) + 1))
for _ in range(n_iters):
count = defaultdict(float)
total = defaultdict(float)
for src, tgt in corpus:
sw = src.split()
tw = tgt.split()
for e in sw:
Z = sum(t[(e, f)] for f in tw)
if Z == 0:
continue
for f in tw:
delta = t[(e, f)] / Z
count[(e, f)] += delta
total[e] += delta
for (e, f), c in count.items():
t[(e, f)] = c / total[e] if total[e] > 0 else 0.0
return t
def viterbi_align_m1(src, tgt, t):
sw = src.split()
tw = tgt.split()
alignment = []
for j, e in enumerate(sw):
best_i = max(range(len(tw)), key=lambda i: t[(e, tw[i])])
alignment.append((j, best_i))
return alignment
t_m1 = ibm_model1(CORPUS, n_iters=20)
print("Model 1 trained.")
# Quick sanity check
src_ex, tgt_ex = CORPUS[0]
al = viterbi_align_m1(src_ex, tgt_ex, t_m1)
sw = src_ex.split()
tw = tgt_ex.split()
print(f"Example alignment (Model 1):")
print(f" SRC: {sw}")
print(f" TGT: {tw}")
for j, i in al:
print(f" {sw[j]} -> {tw[i]}")
IBM Model 2#
Model 1 ignores word positions — the probability of aligning source position j to target position i is the same regardless of where j sits in the sentence.
Model 2 adds an alignment table a(i | j, l, m) where:
i = target position (0-indexed)
j = source position (0-indexed)
l = source sentence length
m = target sentence length
The generative model becomes:
P(e, a | f) = prod_j t(e_j | f_{a_j}) * a(a_j | j, l, m)
EM for Model 2#
E-step: For each sentence pair, compute posterior alignment probabilities
delta(i, j) = t(e_j | f_i) * a(i | j, l, m) / Z_j
M-step: Normalise to get new t and a tables.
def ibm_model2(corpus, n_iters=20, t_init=None):
# Warm-start from Model 1 if provided
if t_init is None:
t = defaultdict(lambda: 0.01)
else:
t = defaultdict(lambda: 0.01)
t.update(t_init)
# alignment table: a[(i, j, l, m)]
a = defaultdict(lambda: 0.1)
for iteration in range(n_iters):
count_t = defaultdict(float)
total_t = defaultdict(float)
count_a = defaultdict(float)
total_a = defaultdict(float)
for src, tgt in corpus:
sw = src.split()
tw = tgt.split()
l = len(sw)
m = len(tw)
for j, e in enumerate(sw):
# normalisation constant
Z = sum(t[(e, tw[i])] * a[(i, j, l, m)] for i in range(m))
if Z < 1e-300:
continue
for i, f in enumerate(tw):
delta = t[(e, f)] * a[(i, j, l, m)] / Z
count_t[(e, f)] += delta
total_t[e] += delta
count_a[(i, j, l, m)] += delta
total_a[(j, l, m)] += delta
# M-step
for key, c in count_t.items():
t[key] = c / total_t[key[0]] if total_t[key[0]] > 0 else 0.0
for key, c in count_a.items():
j, l, m = key[1], key[2], key[3]
t_key = (j, l, m)
a[key] = c / total_a[t_key] if total_a[t_key] > 0 else 0.0
return t, a
def viterbi_align_m2(src, tgt, t, a):
sw = src.split()
tw = tgt.split()
l = len(sw)
m = len(tw)
alignment = []
for j, e in enumerate(sw):
scores = [t[(e, tw[i])] * a[(i, j, l, m)] for i in range(m)]
best_i = max(range(m), key=lambda i: scores[i])
alignment.append((j, best_i))
return alignment
# Train Model 2 warm-started from Model 1
t_m2, a_m2 = ibm_model2(CORPUS, n_iters=20, t_init=dict(t_m1))
print("Model 2 trained.")
al2 = viterbi_align_m2(src_ex, tgt_ex, t_m2, a_m2)
print(f"\nExample alignment (Model 2):")
print(f" SRC: {sw}")
print(f" TGT: {tw}")
for j, i in al2:
print(f" {sw[j]:15s} -> {tw[i]}")
# Print alignment matrices side by side
def print_alignment_matrix(src, tgt, alignment, label):
sw = src.split()
tw = tgt.split()
# Build matrix
mat = [[" . " for _ in tw] for _ in sw]
for j, i in alignment:
if 0 <= i < len(tw):
mat[j][i] = " X "
header = " " + "".join(f"{w[:6]:^8}" for w in tw)
print(f"\n=== {label} ===")
print(header)
for j, row in enumerate(mat):
print(f"{sw[j][:8]:>8} " + "".join(row))
for idx in range(3):
s, t_s = CORPUS[idx]
al1 = viterbi_align_m1(s, t_s, t_m1)
al2i = viterbi_align_m2(s, t_s, t_m2, a_m2)
print_alignment_matrix(s, t_s, al1, f"Model 1 | {s}")
print_alignment_matrix(s, t_s, al2i, f"Model 2 | {s}")
IBM Model 3 — Fertility (Conceptual Sketch)#
Model 3 introduces fertility: source word f_i generates phi_i target words.
The generative story:
For each source word f_i, sample fertility phi_i ~ n(phi | f_i)
For each generated target word, sample translation t(e | f_i)
Place each generated word at a target position via distortion d(pi | i, l, m)
Fill remaining positions with “spurious” target words
Parameter space explosion: Model 3 requires
Fertility table n(phi | f) for phi in {0, 1, 2, …}
Distortion table d(j | i, l, m)
Deficiency corrections (null word)
Training requires approximations (hill-climbing / pegged EM) because the exact E-step is NP-hard (sums over all n^n alignments).
In practice, Model 3 is initialised from Model 2, then only the top-k alignments are considered (neighbourhood sampling).
# Model 3 parameter space illustration
print("Model 3 parameter counts for a typical corpus:")
src_vocab_size = len(set(w for s, _ in CORPUS for w in s.split()))
tgt_vocab_size = len(set(w for _, t_s in CORPUS for w in t_s.split()))
max_len = max(len(s.split()) for s, _ in CORPUS)
max_fertility = 4 # typical cap
n_translation = src_vocab_size * tgt_vocab_size
n_fertility = src_vocab_size * (max_fertility + 1)
n_distortion = max_len * max_len * max_len * max_len
print(f" |src_vocab| = {src_vocab_size}")
print(f" |tgt_vocab| = {tgt_vocab_size}")
print(f" Translation params: {n_translation:,}")
print(f" Fertility params: {n_fertility:,}")
print(f" Distortion params: {n_distortion:,} (max_len={max_len})")
print()
print("For real corpora (50k vocab, sentences up to 40 words):")
print(f" Translation: {50000*50000:,}")
print(f" Distortion: {40*40*40*40:,}")
print("=> Hill-climbing / neighbourhood sampling is necessary.")
HMM Alignment (Vogel, Ney & Tillmann, 1996)#
Rather than a tabular a(i|j,l,m), HMM alignment models the jump between consecutive alignment positions as a Markov chain:
P(a_1...a_m | f_1...f_l) = prod_j p(a_j | a_{j-1}, l) * t(e_j | f_{a_j})
where p(delta | l) depends only on the jump size delta = a_j - a_{j-1}.
This captures:
Local monotonicity (small jumps are more likely)
Long-range reordering (large jumps are possible but penalised)
EM via Forward-Backward#
The E-step computes marginals gamma(j, i) = P(a_j = i | e, f) using the standard HMM forward-backward algorithm over alignment positions.
def hmm_align(corpus, n_iters=20, t_init=None):
# translation table
if t_init is None:
t = defaultdict(lambda: 0.01)
else:
t = defaultdict(lambda: 0.01)
t.update(t_init)
# jump distribution p[delta] for delta in range(-max_len, max_len+1)
max_len = max(len(s.split()) for s, _ in corpus)
jump_range = list(range(-max_len, max_len + 1))
jump_count = defaultdict(float)
for d in jump_range:
jump_count[d] = 1.0 # uniform init
def jump_prob(delta):
total = sum(jump_count.values())
return jump_count.get(delta, 0.0) / total if total > 0 else 1e-10
for iteration in range(n_iters):
count_t = defaultdict(float)
total_t = defaultdict(float)
new_jump = defaultdict(float)
for src, tgt in corpus:
sw = src.split()
tw = tgt.split()
l = len(sw) # source length
m = len(tw) # target length
# emission: emit[j][i] = t(e_j | f_i)
emit = [[t[(sw[j], tw[i])] for i in range(m)] for j in range(l)]
# forward pass: alpha[j][i] = P(e_1..e_j, a_j=i)
alpha = [[0.0] * m for _ in range(l)]
# init: uniform start
for i in range(m):
alpha[0][i] = (1.0 / m) * emit[0][i]
for j in range(1, l):
for i in range(m):
s_sum = sum(alpha[j-1][i_prev] * jump_prob(i - i_prev)
for i_prev in range(m))
alpha[j][i] = s_sum * emit[j][i]
# backward pass: beta[j][i] = P(e_{j+1}..e_l | a_j=i)
beta = [[0.0] * m for _ in range(l)]
for i in range(m):
beta[l-1][i] = 1.0
for j in range(l-2, -1, -1):
for i in range(m):
beta[j][i] = sum(
jump_prob(i_next - i) * emit[j+1][i_next] * beta[j+1][i_next]
for i_next in range(m)
)
# gamma[j][i] = P(a_j=i | e, f)
gamma = [[0.0] * m for _ in range(l)]
for j in range(l):
Z = sum(alpha[j][i] * beta[j][i] for i in range(m))
if Z < 1e-300:
continue
for i in range(m):
gamma[j][i] = alpha[j][i] * beta[j][i] / Z
# xi[j][i_prev][i] = P(a_{j-1}=i_prev, a_j=i | e, f)
for j in range(1, l):
Z = sum(
alpha[j-1][ip] * jump_prob(i - ip) * emit[j][i] * beta[j][i]
for ip in range(m) for i in range(m)
)
if Z < 1e-300:
continue
for ip in range(m):
for i in range(m):
xi = (alpha[j-1][ip] * jump_prob(i - ip) *
emit[j][i] * beta[j][i]) / Z
new_jump[i - ip] += xi
# accumulate translation counts
for j in range(l):
for i in range(m):
count_t[(sw[j], tw[i])] += gamma[j][i]
total_t[sw[j]] += gamma[j][i]
# M-step translation
for key, c in count_t.items():
t[key] = c / total_t[key[0]] if total_t[key[0]] > 0 else 0.0
# M-step jump
jump_count = new_jump if new_jump else jump_count
def viterbi(src, tgt):
sw_v = src.split()
tw_v = tgt.split()
lv = len(sw_v)
mv = len(tw_v)
vit = [[-1e18] * mv for _ in range(lv)]
back = [[0] * mv for _ in range(lv)]
for i in range(mv):
vit[0][i] = math.log(max(1.0/mv, 1e-300)) + math.log(max(t[(sw_v[0], tw_v[i])], 1e-300))
for j in range(1, lv):
for i in range(mv):
best_score = -1e18
best_prev = 0
for ip in range(mv):
score = vit[j-1][ip] + math.log(max(jump_prob(i - ip), 1e-300))
if score > best_score:
best_score = score
best_prev = ip
vit[j][i] = best_score + math.log(max(t[(sw_v[j], tw_v[i])], 1e-300))
back[j][i] = best_prev
alignment = [0] * lv
alignment[lv-1] = max(range(mv), key=lambda i: vit[lv-1][i])
for j in range(lv-2, -1, -1):
alignment[j] = back[j+1][alignment[j+1]]
return [(j, alignment[j]) for j in range(lv)]
return t, viterbi
t_hmm, viterbi_hmm = hmm_align(CORPUS, n_iters=20, t_init=dict(t_m2))
print("HMM alignment trained.")
# Compare all three models on the same sentence
print("Alignment comparison across three models")
print("=" * 60)
for idx in [0, 2, 4]:
s, ts = CORPUS[idx]
sw_c = s.split()
tw_c = ts.split()
al1 = viterbi_align_m1(s, ts, t_m1)
al2i = viterbi_align_m2(s, ts, t_m2, a_m2)
al_hmm = viterbi_hmm(s, ts)
print(f"\nSRC: {s}")
print(f"TGT: {ts}")
print(f" Model1: {[(sw_c[j], tw_c[i]) for j,i in al1]}")
print(f" Model2: {[(sw_c[j], tw_c[i]) for j,i in al2i]}")
print(f" HMM: {[(sw_c[j], tw_c[i]) for j,i in al_hmm]}")
Time Machine#
We now run Model 2 alignment on the first four TIME_MACHINE_EN sentences. Since we have no German translations for these sentences, we use our trained t-table to show the most probable target word for each source word, simulating what a full MT pipeline would produce at the alignment stage.
# Time-machine: align TIME_MACHINE_EN sentences using Model 2
# We pair them with their own words (EN->EN) to show alignment structure
# and also show top translation candidates from training vocab
tgt_vocab_list = list(set(w for _, ts in CORPUS for w in ts.split()))
def translate_word_m2(word, tgt_vocab_l, t_table):
scores = [(t_table[(word, tw_w)], tw_w) for tw_w in tgt_vocab_l]
scores.sort(reverse=True)
return scores[:3]
print("Time Machine: Top German translations per English word (Model 2)")
print("=" * 60)
for sent in TIME_MACHINE_EN[:4]:
words = sent.split()
print(f"\n{sent}")
for w in words:
top3 = translate_word_m2(w, tgt_vocab_list, t_m2)
guesses = ", ".join(f"{tw_w}({sc:.3f})" for sc, tw_w in top3)
print(f" {w:20s} -> {guesses}")