33 · Phrase-Based SMT — Stack Decoder#
Dependencies: numpy, scipy
Runtime: ~2 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")
Phrase-Based SMT — Phrase Tables and Stack Decoding#
Phrase-based SMT (PBSMT) dominated machine translation from ~1999–2016. The key insight: translate chunks of words rather than single words. This chapter builds the full pipeline:
Word alignment (IBM Model 2)
Phrase pair extraction
Phrase table estimation
Bigram language model
Stack (beam) decoder
Time-machine demo
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")
# Re-implement IBM Model 2 (self-contained)
from collections import defaultdict
import math
def ibm_model1(corpus, n_iters=15):
tgt_vocab = set(w for _, ts in corpus for w in ts.split())
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:
d = t[(e, f)] / Z
count[(e, f)] += d
total[e] += d
for (e, f), c in count.items():
t[(e, f)] = c / total[e] if total[e] > 0 else 0.0
return t
def ibm_model2(corpus, n_iters=15, t_init=None):
t = defaultdict(lambda: 0.01)
if t_init:
t.update(t_init)
a = defaultdict(lambda: 0.1)
for _ 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, m = len(sw), len(tw)
for j, e in enumerate(sw):
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
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 get_viterbi_alignment(src, tgt, t, a):
sw = src.split()
tw = tgt.split()
l, m = len(sw), len(tw)
alignment = set()
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.add((j, best_i)) # (src_pos, tgt_pos)
return alignment
print("Training IBM Model 1...")
t1 = ibm_model1(CORPUS)
print("Training IBM Model 2...")
t2, a2 = ibm_model2(CORPUS, t_init=dict(t1))
print("Done.")
# Collect Viterbi alignments for the whole corpus
corpus_alignments = []
for src, tgt in CORPUS:
al = get_viterbi_alignment(src, tgt, t2, a2)
corpus_alignments.append((src.split(), tgt.split(), al))
print(f"Aligned {len(corpus_alignments)} sentence pairs.")
Phrase Extraction#
A phrase pair (s, t) extracted from a sentence pair is consistent if:
Every source word in s that has an alignment link, aligns only to words in t
Every target word in t that has an alignment link, aligns only to words in s
At least one word on each side has an alignment link
Algorithm (Koehn et al. 2003):
for each target span [t_start, t_end]:
find source span [s_start, s_end] consistent with this target span
if consistent: add (src_phrase, tgt_phrase) to phrase table
def extract_phrases(src_words, tgt_words, alignment, max_phrase_len=4):
# alignment: set of (src_pos, tgt_pos) pairs
phrases = []
l_s = len(src_words)
l_t = len(tgt_words)
# For each target span
for t_start in range(l_t):
for t_end in range(t_start, min(l_t, t_start + max_phrase_len)):
# Find source words that align to this target span
aligned_src = [sp for sp, tp in alignment
if t_start <= tp <= t_end]
if not aligned_src:
continue
s_start = min(aligned_src)
s_end = max(aligned_src)
if s_end - s_start >= max_phrase_len:
continue
# Check consistency: no source word in span aligns outside target span
consistent = True
for sp in range(s_start, s_end + 1):
for sp2, tp2 in alignment:
if sp2 == sp and not (t_start <= tp2 <= t_end):
consistent = False
break
if not consistent:
break
if not consistent:
continue
# Check consistency: no target word in span aligns outside source span
for tp in range(t_start, t_end + 1):
for sp2, tp2 in alignment:
if tp2 == tp and not (s_start <= sp2 <= s_end):
consistent = False
break
if not consistent:
break
if not consistent:
continue
src_phrase = tuple(src_words[s_start:s_end+1])
tgt_phrase = tuple(tgt_words[t_start:t_end+1])
phrases.append((src_phrase, tgt_phrase))
return phrases
# Extract phrases from the whole corpus
phrase_counts = defaultdict(float) # (src, tgt) -> count
src_counts = defaultdict(float) # src -> count
tgt_counts = defaultdict(float) # tgt -> count
for sw_c, tw_c, al in corpus_alignments:
phrases = extract_phrases(sw_c, tw_c, al)
for sp, tp in phrases:
phrase_counts[(sp, tp)] += 1
src_counts[sp] += 1
tgt_counts[tp] += 1
print(f"Extracted {len(phrase_counts)} unique phrase pairs.")
print("\nSample phrase pairs (src -> tgt):")
for (sp, tp), c in sorted(phrase_counts.items(), key=lambda x: -x[1])[:15]:
sp_str = " ".join(sp)
tp_str = " ".join(tp)
print(f" {sp_str:25s} -> {tp_str:25s} count={c:.0f}")
# Build phrase table with log-probabilities
# P(tgt|src) = count(src,tgt) / count(src)
# P(src|tgt) = count(src,tgt) / count(tgt)
phrase_table = {} # (src_phrase, tgt_phrase) -> {"log_p_t_given_s": x, "log_p_s_given_t": y}
for (sp, tp), c in phrase_counts.items():
p_t_s = c / src_counts[sp] if src_counts[sp] > 0 else 1e-10
p_s_t = c / tgt_counts[tp] if tgt_counts[tp] > 0 else 1e-10
phrase_table[(sp, tp)] = {
"log_p_t_s": math.log(max(p_t_s, 1e-10)),
"log_p_s_t": math.log(max(p_s_t, 1e-10)),
}
# Build index: src_phrase -> list of (tgt_phrase, scores)
pt_index = defaultdict(list)
for (sp, tp), scores in phrase_table.items():
pt_index[sp].append((tp, scores))
print(f"Phrase table: {len(phrase_table)} entries")
print(f"Unique source phrases: {len(pt_index)}")
# Show top translations for a few phrases
for query in [("the",), ("a",), ("the", "cat"), ("in", "the", "park")]:
if query in pt_index:
options = sorted(pt_index[query], key=lambda x: -x[1]["log_p_t_s"])[:3]
q_str = " ".join(query)
print(f"\n src='{q_str}' top translations:")
for tp, sc in options:
tp_s = " ".join(tp)
lp_val = sc["log_p_t_s"]
print(f" '{tp_s}' log_p={lp_val:.3f}")
# Bigram language model (inline, no external import)
def train_bigram_lm(corpus, smoothing=0.1):
unigram = defaultdict(float)
bigram = defaultdict(float)
vocab = set()
for _, tgt in corpus:
words = ["<s>"] + tgt.split() + ["</s>"]
vocab.update(words)
for w in words:
unigram[w] += 1.0
for w1, w2 in zip(words, words[1:]):
bigram[(w1, w2)] += 1.0
V = len(vocab)
def log_prob(sentence):
words = ["<s>"] + sentence + ["</s>"]
lp = 0.0
for w1, w2 in zip(words, words[1:]):
num = bigram[(w1, w2)] + smoothing
den = unigram[w1] + smoothing * V
lp += math.log(num / den)
return lp
return log_prob
lm = train_bigram_lm(CORPUS)
print("Bigram LM trained.")
# Test
test1 = ["die", "Katze"]
test2 = ["Katze", "die"]
print(f"LM log P({test1}) = {lm(test1):.3f}")
print(f"LM log P({test2}) = {lm(test2):.3f}")
# Stack (beam) decoder for phrase-based SMT
# Each hypothesis covers a contiguous left-to-right span of source words
# (simplified: monotone decoding + distortion penalty)
DISTORTION_PENALTY = -0.5 # per word jump
LM_WEIGHT = 0.5
TM_WEIGHT = 1.0
BEAM_SIZE = 5
def decode(src_sentence, pt_idx, lm_func, beam_size=BEAM_SIZE, max_phrase_len=4):
src_words = src_sentence.split()
n = len(src_words)
# Each hypothesis: (score, covered_end, translation_so_far)
# covered_end = number of source words covered (monotone)
# stacks[k] = beam of hypotheses covering exactly k source words
init_hyp = (0.0, 0, []) # score, covered, words
stacks = [[] for _ in range(n + 1)]
stacks[0].append(init_hyp)
for k in range(n):
if not stacks[k]:
continue
for score, covered, trans in stacks[k]:
# Try extending with every phrase starting at position covered
for plen in range(1, min(max_phrase_len, n - covered) + 1):
src_phrase = tuple(src_words[covered:covered + plen])
if src_phrase not in pt_idx:
continue
for tgt_phrase, ph_scores in pt_idx[src_phrase]:
new_words = trans + list(tgt_phrase)
new_covered = covered + plen
# Translation model score
tm_score = TM_WEIGHT * ph_scores["log_p_t_s"]
# LM score for new words
if len(trans) == 0:
context = ["<s>"]
else:
context = trans[-1:]
lm_score = 0.0
prev = context[-1]
for w in tgt_phrase:
lm_score += LM_WEIGHT * lm_func([prev, w])
prev = w
new_score = score + tm_score + lm_score
stacks[new_covered].append((new_score, new_covered, new_words))
# Prune
stacks[k+1] = sorted(stacks[k+1], key=lambda x: -x[0])[:beam_size]
# Pick best complete hypothesis
if not stacks[n]:
return ["<no translation>"]
best = max(stacks[n], key=lambda x: x[0])
return best[2]
# Test on corpus sentences
print("Phrase-based decoder test on training sentences:")
for src, ref in CORPUS[:5]:
hyp = decode(src, pt_index, lm)
print(f" SRC: {src}")
print(f" REF: {ref}")
hyp_str = " ".join(hyp)
print(f" HYP: {hyp_str}")
print()
Time Machine#
Let us run the phrase-based system on the first 5 TIME_MACHINE_EN sentences. Note: our phrase table is trained only on the tiny 10-sentence corpus, so coverage will be sparse — but we can see the pipeline in action.
print("Time Machine: Phrase-Based SMT on TIME_MACHINE_EN")
print("=" * 60)
for sent in TIME_MACHINE_EN[:5]:
hyp = decode(sent.lower(), pt_index, lm, beam_size=10)
print(f"EN: {sent}")
de_out = " ".join(hyp) if hyp else "(no translation)"
print(f"DE: {de_out}")
print()