34 · n-gram Language Models with Kneser-Ney Smoothing#
Dependencies: stdlib + numpy Runtime: < 1 min CPU
“Any sequence of words is possible; our task is to assign probabilities that reflect how likely each sequence is.” — Jurafsky & Martin, Speech and Language Processing
Language models assign probability to sequences of words. In the noisy channel framework (Chapter 30), \(P(e)\) is the language model — it favours fluent, grammatical English over word salad.
This chapter implements n-gram language models from scratch, culminating in Kneser-Ney smoothing — the gold standard for n-gram LMs used by all SMT systems through the 2010s.
# Colab setup: install this part's pinned dependencies (skipped outside Colab).
import sys
if "google.colab" in sys.modules:
import subprocess
subprocess.run(["wget", "-q", "-O", "requirements_part3.txt",
"https://raw.githubusercontent.com/eduardosanchezg/mthistory/main/requirements_part3.txt"])
subprocess.run(["pip", "install", "-q", "-r", "requirements_part3.txt"])
import math
import os
import re
from collections import Counter, defaultdict
from pathlib import Path
import numpy as np
# Reproducibility
np.random.seed(42)
1 · Corpus Loading#
We use Multi30k (English side) as our training corpus. If it’s not available, a built-in fallback of 50 sentences provides a self-contained demo.
# ---------------------------------------------------------------------------
# Fallback corpus — 50 English sentences from Multi30k style
# ---------------------------------------------------------------------------
FALLBACK_EN = [
"A man in an orange hat starring at something .",
"A Boston Terrier is running on lush green grass in front of a white fence .",
"A girl in karate uniform striking a rock .",
"Two women pushing a cart full of cans across a field .",
"A man is skateboarding down a flight of concrete stairs .",
"A man in a black shirt and black shorts is doing a trick on a skateboard .",
"Two girls are posing together for a photo .",
"A man and a woman are dancing in a street .",
"A young boy jumps off a diving board into a pool .",
"A group of people are gathered around a fire .",
"A woman is sitting on a bench reading a book .",
"A dog is running through the snow .",
"Several children are playing on a playground .",
"A man is riding a bicycle on a dirt road .",
"A woman in a red dress is walking down the street .",
"A boy and a girl are playing in the sand .",
"A group of hikers are climbing a steep mountain trail .",
"A man is playing guitar on a stage .",
"Two dogs are playing fetch in a park .",
"A child is blowing out candles on a birthday cake .",
"A man is jumping over a hurdle on a track .",
"Two women are having a conversation at a cafe .",
"A child is feeding ducks at a pond .",
"A man in a blue jacket is painting a wall .",
"Several people are watching a street performer .",
"A dog is fetching a ball from the water .",
"A woman is performing on stage with a microphone .",
"Two men are arm wrestling at a table .",
"A child is building a sandcastle on the beach .",
"A group of cyclists are racing down a hill .",
"A man is chopping wood in a forest .",
"Two women are jogging along a riverside path .",
"A child is learning to ride a bicycle .",
"A man is fishing from a wooden dock .",
"Several people are gathered around a street musician .",
"A woman is hanging laundry on a line .",
"Two boys are playing basketball in a driveway .",
"A man is fixing a car engine in a garage .",
"A child is swinging on a tire swing .",
"A woman is watering plants in her garden .",
"Two men are playing chess in a park .",
"A girl is painting a picture at an easel .",
"A man is rowing a boat on a calm lake .",
"Several children are playing with a sprinkler on a lawn .",
"A woman is knitting in a rocking chair .",
"A man is doing push-ups in a gym .",
"Two dogs are chasing each other in a yard .",
"A child is reading a book under a tree .",
"A woman is walking along a beach at sunset .",
"A man is grilling food at an outdoor barbecue .",
]
def load_corpus(path=None):
"""Load corpus from file path or fall back to built-in data."""
if path is None:
base = os.path.join(os.path.dirname(os.path.abspath("__file__")),
"..", "..", "data", "multi30k")
path = os.path.join(base, "train.en")
if os.path.exists(path):
with open(path) as f:
sents = [l.strip() for l in f if l.strip()]
print(f"Loaded {len(sents):,} sentences from {path}")
return sents
else:
print(f"File not found: {path!r}")
print(f"Using built-in fallback corpus ({len(FALLBACK_EN)} sentences).")
return list(FALLBACK_EN)
def tokenize(text):
"""Lowercase, strip punctuation, add BOS/EOS markers."""
tokens = re.sub(r"[^\w\s]", " ", text.lower()).split()
return ["<s>"] + tokens + ["</s>"]
# Load and split corpus
all_sents = load_corpus()
# Hold out last 10% for evaluation
split = int(0.9 * len(all_sents))
train_sents = all_sents[:split]
dev_sents = all_sents[split:]
print(f"Train: {len(train_sents):,} sentences")
print(f"Dev: {len(dev_sents):,} sentences")
print()
print("Example tokenized sentence:")
print(tokenize(train_sents[0]))
2 · N-gram Counting#
An n-gram is a contiguous sequence of \(n\) tokens. We count every n-gram in the training corpus to estimate how often word sequences occur.
def count_ngrams(sentences, n):
"""
Count all n-grams in a list of sentences.
Each sentence is tokenized (with <s> / </s> markers).
Returns a Counter mapping n-gram tuples to counts.
"""
counts = Counter()
for sent in sentences:
toks = tokenize(sent)
# Pad start with n-1 BOS tokens so every word has a full context
toks = ["<s>"] * (n - 1) + toks
for i in range(len(toks) - n + 1):
counts[tuple(toks[i : i + n])] += 1
return counts
# Build counts for unigram through trigram
unigram_counts = count_ngrams(train_sents, 1)
bigram_counts = count_ngrams(train_sents, 2)
trigram_counts = count_ngrams(train_sents, 3)
total_tokens = sum(unigram_counts.values())
vocab_size = len(unigram_counts)
print(f"Vocabulary size: {vocab_size:,}")
print(f"Total tokens: {total_tokens:,}")
print(f"Unique bigrams: {len(bigram_counts):,}")
print(f"Unique trigrams: {len(trigram_counts):,}")
print()
print("Top 10 unigrams:")
for tok, cnt in unigram_counts.most_common(10):
print(f" {tok:20s} {cnt:6d}")
3 · Maximum Likelihood Estimation#
The simplest LM estimates probabilities by dividing observed counts:
\[
P_{\text{MLE}}(w_n \mid w_1 \ldots w_{n-1}) = \frac{\text{count}(w_1 \ldots w_n)}{\text{count}(w_1 \ldots w_{n-1})}
\]
Problem: MLE assigns probability 0 to any n-gram not seen in training. A single unseen word makes the entire sentence probability 0. This is the zero-frequency problem.
class MLE_LM:
"""Maximum Likelihood Estimate language model."""
def __init__(self, sentences, n):
self.n = n
self.n_counts = count_ngrams(sentences, n)
self.n1_counts = count_ngrams(sentences, n - 1) if n > 1 else None
self.vocab = set(tok for (tok,) in count_ngrams(sentences, 1))
self.V = len(self.vocab)
def prob(self, word, context):
"""
P_MLE(word | context[-n+1:])
context: tuple of preceding tokens (any length)
"""
ctx = tuple(context[-(self.n - 1):]) if self.n > 1 else ()
ngram = ctx + (word,)
if self.n == 1:
total = sum(self.n_counts.values())
return self.n_counts.get(ngram, 0) / total
ctx_count = self.n1_counts.get(ctx, 0)
if ctx_count == 0:
return 0.0
return self.n_counts.get(ngram, 0) / ctx_count
def log_prob_sentence(self, sentence):
"""Log probability of a sentence (sum of log word probs)."""
toks = tokenize(sentence)
toks = ["<s>"] * (self.n - 1) + toks
lp = 0.0
for i in range(self.n - 1, len(toks)):
ctx = tuple(toks[i - self.n + 1 : i])
word = toks[i]
p = self.prob(word, ctx)
lp += math.log2(p) if p > 0 else -float("inf")
return lp
# Train MLE models
mle1 = MLE_LM(train_sents, 1)
mle2 = MLE_LM(train_sents, 2)
mle3 = MLE_LM(train_sents, 3)
# Quick demo
test_sent = train_sents[0]
print(f"Sentence: '{test_sent}'")
print()
print(f" Unigram log-prob : {mle1.log_prob_sentence(test_sent):.2f} bits")
print(f" Bigram log-prob : {mle2.log_prob_sentence(test_sent):.2f} bits")
print(f" Trigram log-prob : {mle3.log_prob_sentence(test_sent):.2f} bits")
# Show zero-probability problem
unseen = "The president is flying to the moon ."
print(f"\nUnseen sentence: '{unseen}'")
print(f" Trigram MLE log-prob: {mle3.log_prob_sentence(unseen)}")
print(" (−inf because some trigram has zero count)")
4 · Kneser-Ney Smoothing#
Kneser-Ney (1995) is the gold standard n-gram smoothing method. It addresses the zero-frequency problem elegantly by redistributing probability mass.
The key formula (interpolated KN)#
\[
P_{\text{KN}}(w \mid h) = \frac{\max(\text{count}(h, w) - d,\; 0)}{\text{count}(h)} + \lambda(h) \cdot P_{\text{KN}}(w \mid h_{1:})
\]
where:
\(d \in (0, 1)\) is the discount (typically 0.75) — we “shave off” \(d\) from every count
\(\lambda(h) = \frac{d \cdot |\{w : \text{count}(h,w) > 0\}|}{\text{count}(h)}\) is the interpolation weight (ensures normalisation)
\(h_{1:}\) is the context with the first word dropped (back-off to shorter context)
The Kneser-Ney base distribution (the key insight)#
The base distribution (unigram level) uses continuation probability instead of raw frequency:
\[
P_{\text{KN}}(w) \propto |\{u : \text{count}(u, w) > 0\}|
\]
This counts the number of unique contexts \(w\) appears in. “Francisco” always follows “San”, so its continuation probability is low — it shouldn’t dominate the backoff. “the” appears after many words, so it gets high probability.
This is why “San Francisco” gets a better probability estimate than naive frequency would predict for “Francisco”.
class KneserNey_LM:
"""
Interpolated Kneser-Ney smoothing, order n.
Implements:
P_KN(w | h) = max(C(h,w) - d, 0) / C(h) + lambda(h) * P_KN(w | h[1:])
Base case:
P_KN(w) ∝ |{u : C(u,w) > 0}| (continuation probability)
"""
def __init__(self, sentences, n=3, d=0.75):
self.n = n
self.d = d
# Build n-gram counts for all orders 1..n
self.counts = {}
for order in range(1, n + 1):
self.counts[order] = count_ngrams(sentences, order)
# Vocabulary from unigrams
self.vocab = [tok for (tok,) in self.counts[1].keys()]
self.V = len(self.vocab)
# Continuation counts: for each word w, how many unique left contexts?
# Used only at the unigram base level.
self._build_continuation_counts()
def _build_continuation_counts(self):
"""
continuation_count[w] = |{u : bigram (u, w) has count > 0}|
"""
self.continuation_count = Counter()
for (u, w), c in self.counts[2].items():
if c > 0:
self.continuation_count[w] += 1
self._continuation_total = sum(self.continuation_count.values())
def _p_continuation(self, word):
"""Base distribution: P_KN(w) ∝ continuation_count(w)."""
return self.continuation_count.get(word, 0) / max(self._continuation_total, 1)
def prob(self, word, context, order=None):
"""
Recursive Kneser-Ney probability.
word : string
context : tuple of preceding tokens
order : which LM order to use (defaults to self.n)
"""
if order is None:
order = self.n
# Base case: unigram Kneser-Ney
if order == 1:
return self._p_continuation(word)
ctx = tuple(context[-(order - 1):])
ngram = ctx + (word,)
ctx_count = self.counts[order - 1].get(ctx, 0)
ngram_count = self.counts[order].get(ngram, 0)
if ctx_count == 0:
# Back off to lower order entirely
return self.prob(word, context, order - 1)
# Number of unique words that follow this context (for lambda)
n_unique_followers = sum(
1 for (c, w2), cnt in self.counts[order].items()
if c == ctx and cnt > 0
)
# Discounted probability
p_disc = max(ngram_count - self.d, 0.0) / ctx_count
# Interpolation weight
lam = (self.d * n_unique_followers) / ctx_count
# Recursive lower-order probability
p_lower = self.prob(word, context, order - 1)
return p_disc + lam * p_lower
def log_prob_sentence(self, sentence):
"""Sum of log2 word probabilities for a sentence."""
toks = tokenize(sentence)
toks = ["<s>"] * (self.n - 1) + toks
lp = 0.0
for i in range(self.n - 1, len(toks)):
ctx = tuple(toks[i - self.n + 1 : i])
word = toks[i]
p = self.prob(word, ctx)
if p > 0:
lp += math.log2(p)
else:
lp -= 40.0 # effectively -inf but finite for display
return lp
def top_k_predictions(self, context, k=10):
"""Return top-k predicted words given context."""
ctx = tuple(tokenize(" ".join(context)))[1:-1] # strip BOS/EOS
scored = [(self.prob(w, ctx), w) for w in self.vocab]
scored.sort(reverse=True)
return scored[:k]
print("Training Kneser-Ney LMs (bigram and trigram)...")
kn2 = KneserNey_LM(train_sents, n=2, d=0.75)
kn3 = KneserNey_LM(train_sents, n=3, d=0.75)
print("Done.")
print()
# Demo: continuation probability vs raw frequency
print("Kneser-Ney continuation probability vs. raw unigram frequency:")
print()
words_to_check = ["francisco", "the", "a", "man", "running", "is"]
# Get raw unigram probs
total_tok = sum(unigram_counts.values())
print(f"{'Word':>12} {'Unigram freq':>14} {'KN continuation':>16} {'Note'}")
print("-" * 65)
for w in words_to_check:
raw = unigram_counts.get((w,), 0) / total_tok
kn_cont = kn3._p_continuation(w)
note = ""
if raw > 0 and kn_cont / (raw + 1e-12) < 0.5:
note = "(low continuation — appears in few contexts)"
elif raw > 0 and kn_cont / (raw + 1e-12) > 1.5:
note = "(high continuation — appears in many contexts)"
print(f"{w:>12} {raw:>14.5f} {kn_cont:>16.5f} {note}")
5 · Perplexity Evaluation#
Perplexity is the standard metric for language models. It measures how “surprised” the model is by the test data:
\[
\text{PP}(\text{test}) = 2^{-\frac{1}{N} \sum_{i} \log_2 P(w_i \mid w_{<i})}
\]
Lower perplexity = better model. A perplexity of \(k\) means the model is, on average, as uncertain as choosing uniformly among \(k\) options.
def perplexity(lm, test_sentences):
"""
Compute perplexity on test sentences.
PP = 2^(-1/N * sum log2 P(w|context))
"""
total_log_prob = 0.0
total_tokens = 0
for sent in test_sentences:
toks = tokenize(sent)
n = lm.n
toks_padded = ["<s>"] * (n - 1) + toks
for i in range(n - 1, len(toks_padded)):
ctx = tuple(toks_padded[i - n + 1 : i])
word = toks_padded[i]
p = lm.prob(word, ctx)
if p > 0:
total_log_prob += math.log2(p)
else:
total_log_prob -= 40.0
total_tokens += 1
avg_log_prob = total_log_prob / max(total_tokens, 1)
return 2 ** (-avg_log_prob)
# ---- Evaluate all models ----
print("Computing perplexities on dev set...")
print()
results = []
# MLE models
for name, lm in [("Unigram MLE", mle1), ("Bigram MLE", mle2), ("Trigram MLE", mle3)]:
pp = perplexity(lm, dev_sents)
results.append((name, pp))
# KN models
for name, lm in [("Bigram KN", kn2), ("Trigram KN", kn3)]:
pp = perplexity(lm, dev_sents)
results.append((name, pp))
# Print table
print(f"{'Model':>20} {'Dev Perplexity':>16}")
print("-" * 42)
for name, pp in results:
marker = " ← best" if pp == min(p for _, p in results) else ""
print(f"{name:>20} {pp:>16.1f}{marker}")
print()
print("Expected approximate ordering: Trigram KN < Bigram KN < Trigram MLE")
print("< Bigram MLE < Unigram MLE")
print("(Actual values depend on corpus size — fallback corpus is very small.)")
6 · Qualitative Analysis#
Top predicted words given context#
Let’s see what the Kneser-Ney trigram model predicts in different contexts.
contexts_to_probe = [
("a", "man"),
("a", "woman"),
("two", "dogs"),
("is", "running"),
("a", "group"),
]
print("Top-10 word predictions by Kneser-Ney trigram model:")
print()
for ctx in contexts_to_probe:
preds = kn3.top_k_predictions(list(ctx), k=10)
pred_str = ", ".join(f"{w}({p:.3f})" for p, w in preds[:5])
print(f" P(? | '{' '.join(ctx)}')")
print(f" {pred_str}")
print()
# -------------------------------------------------------------------
# Show how KN handles unseen contexts vs MLE
# -------------------------------------------------------------------
print("KN vs MLE for unseen contexts:")
print()
test_contexts = [
(("a", "president"), "is"),
(("two", "astronauts"), "are"),
(("the", "robot"), "walks"),
]
for ctx, word in test_contexts:
mle_p = mle3.prob(word, ctx)
kn_p = kn3.prob(word, ctx)
print(f" P('{word}' | '{' '.join(ctx)}')")
print(f" MLE: {mle_p:.6f} {'(zero — unseen context)' if mle_p == 0 else ''}")
print(f" Kneser-Ney: {kn_p:.6f}")
print()
print("KN never assigns zero probability — it always backs off to lower-order")
print("distributions and redistributes the discounted probability mass.")
7 · ARPA Format Output#
ARPA format is the standard file format for n-gram language models, used by all SMT systems including Moses and KenLM. Every SMT decoder (Chapter 33) loads its language model from an ARPA file.
Format:
\data\
ngram 1=N1
ngram 2=N2
\1-grams:
log10_prob word log10_backoff
...
\2-grams:
log10_prob word1 word2 log10_backoff
...
\end\
def write_arpa(lm, filename, max_order=None):
"""
Write a KneserNey_LM to ARPA format.
For each n-gram, log10(P(w|context)) is the n-gram log-probability.
Backoff weights are approximated as log10(lambda(context)).
"""
if max_order is None:
max_order = lm.n
# Collect all n-grams we'll write
ngrams_by_order = {}
for order in range(1, max_order + 1):
ngrams_by_order[order] = {}
for ngram, cnt in lm.counts[order].items():
if cnt == 0:
continue
word = ngram[-1]
context = ngram[:-1]
p = lm.prob(word, context, order=order)
ngrams_by_order[order][ngram] = p
with open(filename, "w") as f:
f.write("\\data\\\n")
for order in range(1, max_order + 1):
f.write(f"ngram {order}={len(ngrams_by_order[order])}\n")
f.write("\n")
for order in range(1, max_order + 1):
f.write(f"\\{order}-grams:\n")
for ngram, p in sorted(ngrams_by_order[order].items()):
log_p = math.log10(p) if p > 0 else -99.0
ngram_str = " ".join(ngram)
if order < max_order:
# Write a placeholder backoff weight of 0 (log10(1))
f.write(f"{log_p:.4f}\t{ngram_str}\t0.0000\n")
else:
f.write(f"{log_p:.4f}\t{ngram_str}\n")
f.write("\n")
f.write("\\end\\\n")
# Write bigram KN model to ARPA
arpa_path = "/tmp/ngram_lm_bigram_kn.arpa"
write_arpa(kn2, arpa_path, max_order=2)
print(f"Written ARPA file: {arpa_path}")
# Show first 25 lines
with open(arpa_path) as f:
lines = f.readlines()
print(f"File size: {len(lines)} lines")
print()
print("First 25 lines:")
print("".join(lines[:25]))
Time Machine · Sentence Fluency Scores#
Every chapter runs the same 20 held-out test sentences so the quality arc across chapters is directly comparable.
Here we score each English sentence with the Kneser-Ney trigram LM. The LM measures fluency, not translation correctness. Sentences the model considers more natural (shorter, more common patterns) will score higher.
TIME_MACHINE_EN = [
"A man is walking his dog in the park .",
"Two children are playing near a fountain .",
"A woman is reading a book on a bench .",
"The dog is running through the snow .",
"A group of people are watching a soccer game .",
"A young girl is jumping rope in the street .",
"Several men are fishing at the lake .",
"A woman in a red dress is dancing .",
"Two boys are riding bicycles on a path .",
"A cat is sleeping on a windowsill .",
"A man is playing guitar at a concert .",
"Children are swimming in the ocean .",
"A woman is cooking dinner in the kitchen .",
"A dog is catching a frisbee in a field .",
"Two men are playing chess in the park .",
"A girl is painting a picture outdoors .",
"A man is climbing a rock face .",
"Several people are hiking through the forest .",
"A woman is pushing a stroller in the park .",
"Two friends are sharing a meal at a table .",
]
print("=" * 68)
print("TIME MACHINE — Chapter 34: KN Trigram Language Model")
print("=" * 68)
print()
print("Scoring 20 test sentences with Kneser-Ney trigram LM")
print("(English LM trained on Multi30k EN side)")
print()
scored = []
for sent in TIME_MACHINE_EN:
lp = kn3.log_prob_sentence(sent)
toks = tokenize(sent)
n_toks = len(toks) - 1 # excluding initial <s> padding
pp = 2 ** (-lp / max(n_toks, 1))
scored.append((pp, lp, sent))
scored.sort() # best (lowest perplexity) first
print(f"{'Rank':>4} {'Perplexity':>12} Sentence")
print("-" * 75)
for rank, (pp, lp, sent) in enumerate(scored, 1):
truncated = sent[:55] + "..." if len(sent) > 55 else sent
print(f" {rank:2d} {pp:12.1f} {truncated}")
print()
best_pp, _, best_sent = scored[0]
worst_pp, _, worst_sent = scored[-1]
print(f"Best (most fluent): '{best_sent}'")
print(f" Perplexity = {best_pp:.1f}")
print()
print(f"Worst (least fluent): '{worst_sent}'")
print(f" Perplexity = {worst_pp:.1f}")
print()
print("These perplexity scores will be used in Chapter 33 (phrase-based SMT)")
print("as the P(e) component of the noisy channel log-linear model.")
Summary#
Model |
Zero probabilities? |
Handles unseen? |
Standard for SMT? |
|---|---|---|---|
Unigram MLE |
Never |
Yes (seen words) |
No |
Bigram MLE |
Yes (unseen bigrams) |
No |
No |
Trigram MLE |
Yes (unseen trigrams) |
No |
No |
Bigram KN |
Never |
Yes |
Baseline |
Trigram KN |
Never |
Yes |
Yes (Moses default) |
Key takeaways:
MLE assigns zero probability to unseen n-grams → broken for any held-out data
Kneser-Ney smoothing redistributes discounted mass via interpolation → no zeros
Continuation probability (not raw frequency) is the correct base distribution
ARPA format is the interchange format used by all SMT decoders
Perplexity is the intrinsic evaluation metric — lower is better
Next chapter: Chapter 35 (Hiero) uses the n-gram LM built here as the fluency component inside a hierarchical phrase-based decoder. The LM trained here can be plugged directly into Moses via the ARPA file.