1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80

from nltk.stem import PorterStemmer, WordNetLemmatizer
from nltk.tokenize import word_tokenize
from nltk.corpus import wordnet
import nltk
 
# Initialize
stemmer = PorterStemmer()
lemmatizer = WordNetLemmatizer()
 
# Download required resources (run once)
# nltk.download('wordnet')
# nltk.download('averaged_perceptron_tagger')
 
# Test cases demonstrating differences
test_cases = [
    # Regular verbs
    ("running", "v"),
    ("runs", "v"),
    ("ran", "v"),
    
    # Irregular verbs
    ("went", "v"),
    ("goes", "v"),
    ("gone", "v"),
    
    # Irregular plurals
    ("mice", "n"),
    ("feet", "n"),
    ("children", "n"),
    
    # Adjectives
    ("better", "a"),
    ("best", "a"),
    ("happier", "a"),
    
    # Verb/Noun ambiguity
    ("studies", "v"),
    ("studies", "n"),
    ("leaves", "v"),
    ("leaves", "n"),
    
    # Stemming errors
    ("university", "n"),
    ("universe", "n"),
    ("organization", "n"),
]
 
# Map simple POS to WordNet POS
def get_wordnet_pos(simple_pos):
    pos_map = {'v': wordnet.VERB, 'n': wordnet.NOUN, 'a': wordnet.ADJ, 'r': wordnet.ADV}
    return pos_map.get(simple_pos, wordnet.NOUN)
 
print("=== Stemming vs Lemmatization Comparison ===\n")
print(f"{'Word':<15} {'POS':<5} {'Stem':<12} {'Lemma':<12} {'Same?':<6}")
print("-" * 55)
 
for word, pos in test_cases:
    stem = stemmer.stem(word)
    lemma = lemmatizer.lemmatize(word, get_wordnet_pos(pos))
    same = "✓" if stem == lemma else "✗"
    print(f"{word:<15} {pos:<5} {stem:<12} {lemma:<12} {same:<6}")
 
# Highlight key differences
print("\n=== Key Differences ===\n")
 
difference_examples = [
    ("went", "v", "Past irregular verb"),
    ("mice", "n", "Irregular plural"),
    ("better", "a", "Suppletive comparative"),
    ("university", "n", "Unrelated word conflation risk"),
]
 
for word, pos, description in difference_examples:
    stem = stemmer.stem(word)
    lemma = lemmatizer.lemmatize(word, get_wordnet_pos(pos))
    print(f"{description}:")
    print(f"  Word: {word}")
    print(f"  Stem: {stem}")
    print(f"  Lemma: {lemma}")
    print()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113

import nltk
from nltk.stem import WordNetLemmatizer
from nltk.corpus import wordnet
from nltk import pos_tag, word_tokenize
from typing import List, Tuple
 
# Download required resources
# nltk.download('wordnet')
# nltk.download('averaged_perceptron_tagger')
# nltk.download('punkt')
 
class EnhancedWordNetLemmatizer:
    """
    WordNet lemmatizer with automatic POS tagging for improved accuracy.
    """
    
    def __init__(self):
        self.lemmatizer = WordNetLemmatizer()
    
    def _get_wordnet_pos(self, treebank_tag: str) -> str:
        """
        Convert Penn Treebank POS tags to WordNet POS tags.
        
        Penn Treebank uses detailed tags (NN, NNS, VB, VBD, JJ, RB, etc.)
        WordNet uses simple tags (n, v, a, r)
        """
        tag_prefix = treebank_tag[0].upper()
        
        tag_map = {
            'J': wordnet.ADJ,      # JJ, JJR, JJS
            'V': wordnet.VERB,     # VB, VBD, VBG, VBN, VBP, VBZ
            'N': wordnet.NOUN,     # NN, NNS, NNP, NNPS
            'R': wordnet.ADV       # RB, RBR, RBS
        }
        
        return tag_map.get(tag_prefix, wordnet.NOUN)  # Default to noun
    
    def lemmatize_word(self, word: str, pos: str = None) -> str:
        """
        Lemmatize a single word.
        
        Args:
            word: Word to lemmatize
            pos: Optional POS tag (Penn Treebank format).
                 If not provided, word is treated as noun.
        """
        if pos:
            wn_pos = self._get_wordnet_pos(pos)
        else:
            wn_pos = wordnet.NOUN
        
        return self.lemmatizer.lemmatize(word.lower(), wn_pos)
    
    def lemmatize_text(self, text: str) -> List[Tuple[str, str, str]]:
        """
        Lemmatize all words in a text with automatic POS tagging.
        
        Returns:
            List of (word, pos_tag, lemma) tuples
        """
        tokens = word_tokenize(text)
        tagged = pos_tag(tokens)
        
        results = []
        for word, tag in tagged:
            lemma = self.lemmatize_word(word, tag)
            results.append((word, tag, lemma))
        
        return results
    
    def lemmatize_sentence(self, text: str) -> str:
        """Return lemmatized text as a string."""
        results = self.lemmatize_text(text)
        return " ".join(lemma for _, _, lemma in results)
 
# Demonstration
lemmatizer = EnhancedWordNetLemmatizer()
 
# Test sentences
test_sentences = [
    "The children were running and playing in the leaves.",
    "She studies physics and her studies are going well.",
    "The mice ran into better hiding spots.",
    "I am going to the best universities in the world.",
]
 
print("=== Enhanced WordNet Lemmatization ===\n")
 
for sentence in test_sentences:
    print(f"Original: {sentence}")
    results = lemmatizer.lemmatize_text(sentence)
    print(f"Detailed:")
    for word, pos, lemma in results:
        if word.lower() != lemma:  # Show only changed words
            print(f"  {word} ({pos}) → {lemma}")
    print(f"Lemmatized: {lemmatizer.lemmatize_sentence(sentence)}")
    print()
 
# Demonstrate POS importance
print("=== POS Tag Importance ===\n")
 
ambiguous_words = [
    ("saw", "VBD", "past tense of see"),
    ("saw", "NN", "cutting tool"),
    ("leaves", "VBZ", "departs"),
    ("leaves", "NNS", "plural of leaf"),
    ("better", "JJR", "more good"),
    ("better", "VB", "to improve"),
]
 
for word, pos, meaning in ambiguous_words:
    lemma = lemmatizer.lemmatize_word(word, pos)
    print(f"'{word}' as {pos} ({meaning}) → '{lemma}'")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

import spacy
from typing import List, Dict
 
# Load spaCy model (install with: python -m spacy download en_core_web_sm)
nlp = spacy.load("en_core_web_sm")
 
# Process text through spaCy pipeline
text = "The children were running and playing in the fallen leaves."
doc = nlp(text)
 
print("=== spaCy Lemmatization ===\n")
print(f"Original: {text}\n")
 
print(f"{'Token':<12} {'POS':<6} {'Tag':<6} {'Lemma':<12} {'Changed':<8}")
print("-" * 50)
 
for token in doc:
    changed = "✓" if token.text.lower() != token.lemma_ else ""
    print(f"{token.text:<12} {token.pos_:<6} {token.tag_:<6} {token.lemma_:<12} {changed}")
 
# Lemmatized text
lemmatized = " ".join([token.lemma_ for token in doc])
print(f"\nLemmatized: {lemmatized}")
 
# Batch processing for performance
print("\n=== Batch Processing ===\n")
 
texts = [
    "The mice were running from the cats.",
    "She studies hard and her studies pay off.",
    "The better universities have better facilities.",
    "He went to the store but the store was closed.",
]
 
# Process in batch (much faster than one at a time)
docs = list(nlp.pipe(texts))
 
for original, doc in zip(texts, docs):
    lemmas = [t.lemma_ for t in doc if not t.is_punct]
    print(f"Original: {original}")
    print(f"Lemmas:   {' '.join(lemmas)}")
    print()
 
# Disable unnecessary pipeline components for speed
print("=== Optimized Pipeline ===\n")
 
# Create pipeline with only necessary components
nlp_lemma = spacy.load("en_core_web_sm", disable=["ner"])
 
# Or more aggressively:
# nlp_lemma = spacy.load("en_core_web_sm", disable=["ner", "parser"])
 
import time
 
large_text = " ".join(texts * 100)  # 400 sentences
 
# Time full pipeline
start = time.perf_counter()
doc_full = nlp(large_text)
full_time = time.perf_counter() - start
 
# Time optimized pipeline
start = time.perf_counter()
doc_opt = nlp_lemma(large_text)
opt_time = time.perf_counter() - start
 
print(f"Full pipeline: {full_time:.3f}s")
print(f"Optimized (no NER): {opt_time:.3f}s")
print(f"Speedup: {full_time/opt_time:.1f}x")
 
# Access spaCy's lemmatization rules
print("\n=== spaCy Lemmatization Rules ===\n")
 
# Example: show exception table for "be"
doc = nlp("am is are was were be being been")
print("Forms of 'be':")
for token in doc:
    print(f"  {token.text:<8} → {token.lemma_}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90

# === Stanza (Stanford NLP) ===
import stanza
 
# Download English model (run once)
# stanza.download('en')
 
# Initialize pipeline
nlp_stanza = stanza.Pipeline('en', processors='tokenize,pos,lemma')
 
text = "The children were running and the mice were hiding."
doc = nlp_stanza(text)
 
print("=== Stanza Lemmatization ===\n")
print(f"Original: {text}\n")
 
for sentence in doc.sentences:
    for word in sentence.words:
        if word.text.lower() != word.lemma:
            print(f"  {word.text} ({word.upos}) → {word.lemma}")
 
# Lemmatized output
lemmas = [word.lemma for sent in doc.sentences for word in sent.words]
print(f"\nLemmas: {' '.join(lemmas)}")
 
# === Simplemma (Fast, Multilingual) ===
# pip install simplemma
 
from simplemma import lemmatize, text_lemmatizer
 
# Single word lemmatization
words = ["running", "mice", "better", "went", "studies"]
print("\n=== Simplemma Lemmatization ===\n")
for word in words:
    lemma = lemmatize(word, lang='en')
    print(f"  {word} → {lemma}")
 
# Text lemmatization
text = "The children were running and playing."
lemmatized = " ".join(text_lemmatizer(text, lang='en'))
print(f"\nOriginal: {text}")
print(f"Lemmatized: {lemmatized}")
 
# Multilingual support
print("\n=== Simplemma Multilingual ===\n")
multilingual_examples = [
    ('de', ["laufen", "läuft", "gelaufen"]),  # German
    ('fr', ["manger", "mange", "mangé"]),      # French
    ('es', ["correr", "corriendo", "corrido"]), # Spanish
]
 
for lang, words in multilingual_examples:
    print(f"{lang.upper()}:", end=" ")
    for word in words:
        lemma = lemmatize(word, lang=lang)
        print(f"{word}→{lemma}", end=" ")
    print()
 
# === Performance Comparison ===
import time
 
test_text = "The children were running and playing in the leaves. " * 100
 
# NLTK
from nltk.stem import WordNetLemmatizer
from nltk import word_tokenize
nltk_lemmatizer = WordNetLemmatizer()
 
start = time.perf_counter()
tokens = word_tokenize(test_text)
nltk_lemmas = [nltk_lemmatizer.lemmatize(t) for t in tokens]
nltk_time = time.perf_counter() - start
 
# spaCy
import spacy
nlp_spacy = spacy.load("en_core_web_sm", disable=["ner"])
 
start = time.perf_counter()
doc = nlp_spacy(test_text)
spacy_lemmas = [t.lemma_ for t in doc]
spacy_time = time.perf_counter() - start
 
# Simplemma
start = time.perf_counter()
simplemma_lemmas = list(text_lemmatizer(test_text, lang='en'))
simplemma_time = time.perf_counter() - start
 
print("\n=== Performance Comparison ===\n")
print(f"NLTK:      {nltk_time:.3f}s")
print(f"spaCy:     {spacy_time:.3f}s")
print(f"Simplemma: {simplemma_time:.3f}s")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98

import spacy
from nltk.stem import WordNetLemmatizer
from nltk.corpus import wordnet
 
nlp = spacy.load("en_core_web_sm")
nltk_lemmatizer = WordNetLemmatizer()
 
print("=== Edge Cases in Lemmatization ===\n")
 
# 1. Suppletive forms
print("1. Suppletive Forms:")
suppletive = ["went", "gone", "better", "best", "worse", "worst", "am", "is", "are", "was", "were"]
for word in suppletive:
    doc = nlp(word)
    spacy_lemma = doc[0].lemma_
    # NLTK needs POS hint
    nltk_lemma_v = nltk_lemmatizer.lemmatize(word, wordnet.VERB)
    nltk_lemma_a = nltk_lemmatizer.lemmatize(word, wordnet.ADJ)
    print(f"  {word:<8} spaCy: {spacy_lemma:<6} NLTK(v): {nltk_lemma_v:<6} NLTK(a): {nltk_lemma_a}")
 
# 2. Proper nouns
print("\n2. Proper Nouns:")
proper_nouns = "James and Williams went to Paris and London."
doc = nlp(proper_nouns)
print(f"  Original: {proper_nouns}")
print(f"  With NER protection:")
for token in doc:
    if token.ent_type_:  # Is named entity
        print(f"    {token.text} ({token.ent_type_}) → {token.text} (preserved)")
    elif token.text.lower() != token.lemma_:
        print(f"    {token.text} → {token.lemma_}")
 
# 3. Compounds and hyphenated words
print("\n3. Compounds and Hyphenated Words:")
compounds = "toothbrush ex-president well-being mother-in-law data-driven"
doc = nlp(compounds)
for token in doc:
    if not token.is_punct:
        print(f"  {token.text:<15} → {token.lemma_}")
 
# 4. Contractions
print("\n4. Contractions:")
contractions = "I'm can't won't shouldn't it's they've"
doc = nlp(contractions)
for token in doc:
    print(f"  {token.text:<12} → {token.lemma_}")
 
# 5. Numbers and symbols
print("\n5. Numbers and Special Cases:")
special = "2nd 3rd 1990s URLs APIs"
doc = nlp(special)
for token in doc:
    print(f"  {token.text:<10} → {token.lemma_}")
 
# 6. Domain-specific / OOV words
print("\n6. Domain-Specific / OOV Words:")
tech_terms = "Kubernetes microservices APIs serverless containerized"
doc = nlp(tech_terms)
print(f"  Original: {tech_terms}")
for token in doc:
    if token.text.lower() != token.lemma_:
        print(f"    {token.text} → {token.lemma_} (changed)")
    else:
        print(f"    {token.text} → {token.lemma_} (unchanged - likely OOV)")
 
# 7. Creating a robust lemmatizer with fallbacks
print("\n=== Robust Lemmatization Pipeline ===\n")
 
def robust_lemmatize(text: str, nlp, preserve_entities: bool = True) -> str:
    """
    Robust lemmatization with entity preservation and fallback handling.
    """
    doc = nlp(text)
    result = []
    
    for token in doc:
        # Skip punctuation
        if token.is_punct:
            result.append(token.text)
            continue
        
        # Preserve named entities if requested
        if preserve_entities and token.ent_type_:
            result.append(token.text)
            continue
        
        # Use lemma if available and different
        if token.lemma_ != "-PRON-" and token.lemma_.strip():
            result.append(token.lemma_)
        else:
            result.append(token.text.lower())
    
    return " ".join(result)
 
test = "Dr. Smith and James went to Google's headquarters in New York."
print(f"Original: {test}")
print(f"Lemmatized (preserve entities): {robust_lemmatize(test, nlp, preserve_entities=True)}")
print(f"Lemmatized (no preservation): {robust_lemmatize(test, nlp, preserve_entities=False)}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162

import spacy
from typing import List, Dict, Iterator
from functools import lru_cache
import time
from concurrent.futures import ProcessPoolExecutor
import multiprocessing as mp
 
# === Optimized spaCy Pipeline ===
 
def create_lemmatization_pipeline(disable_components: List[str] = None):
    """
    Create an optimized spaCy pipeline for lemmatization.
    """
    if disable_components is None:
        # Disable components not needed for lemmatization
        disable_components = ["ner"]  # Keep parser for better POS accuracy
    
    return spacy.load("en_core_web_sm", disable=disable_components)
 
# === Batched Processing ===
 
def batch_lemmatize(
    texts: List[str], 
    nlp, 
    batch_size: int = 1000,
    n_process: int = 1
) -> Iterator[List[str]]:
    """
    Efficiently lemmatize large numbers of texts.
    
    Args:
        texts: List of texts to lemmatize
        nlp: spaCy pipeline
        batch_size: Batch size for nlp.pipe()
        n_process: Number of processes (1 for single-threaded)
    
    Yields:
        Lists of lemmatized words for each document
    """
    for doc in nlp.pipe(texts, batch_size=batch_size, n_process=n_process):
        yield [token.lemma_ for token in doc if not token.is_punct]
 
# === Vocabulary-Based Caching ===
 
class CachedLemmatizer:
    """
    Lemmatizer with LRU cache for frequently occurring words.
    """
    
    def __init__(self, cache_size: int = 100000):
        self.nlp = create_lemmatization_pipeline()
        self.cache_size = cache_size
        
        @lru_cache(maxsize=cache_size)
        def _cached_lemma(word: str) -> str:
            doc = self.nlp(word)
            if len(doc) == 1:
                return doc[0].lemma_
            return word
        
        self._cached_lemma = _cached_lemma
    
    def lemmatize_word(self, word: str) -> str:
        """Lemmatize a single word with caching."""
        return self._cached_lemma(word.lower())
    
    def lemmatize_text(self, text: str) -> str:
        """Lemmatize text, using full pipeline for context."""
        # For sentences, use full pipeline for POS context
        doc = self.nlp(text)
        return " ".join(token.lemma_ for token in doc if not token.is_punct)
    
    def preload_vocabulary(self, words: List[str]):
        """Pre-populate cache with known vocabulary."""
        for word in words:
            self._cached_lemma(word.lower())
    
    def cache_stats(self):
        """Return cache statistics."""
        return self._cached_lemma.cache_info()
 
# === Consistency Guarantees ===
 
class DeterministicLemmatizer:
    """
    Lemmatizer with versioning for reproducibility.
    """
    
    def __init__(self, model_name: str = "en_core_web_sm"):
        self.model_name = model_name
        self.nlp = spacy.load(model_name)
        
        # Store version information
        self.spacy_version = spacy.__version__
        self.model_meta = self.nlp.meta
    
    def get_version_info(self) -> Dict:
        """Return version information for reproducibility."""
        return {
            "spacy_version": self.spacy_version,
            "model_name": self.model_name,
            "model_version": self.model_meta.get("version", "unknown"),
            "model_lang": self.model_meta.get("lang", "unknown"),
        }
    
    def lemmatize_with_metadata(self, text: str) -> Dict:
        """Lemmatize with version metadata for provenance tracking."""
        doc = self.nlp(text)
        return {
            "input": text,
            "lemmas": [t.lemma_ for t in doc],
            "version_info": self.get_version_info(),
        }
 
# === Demonstration ===
 
if __name__ == "__main__":
    print("=== Production Lemmatization ===\n")
    
    # Create test data
    texts = [
        "The children were running and playing.",
        "She studies hard and her studies are impressive.",
        "The mice ran into better hiding spots.",
    ] * 100  # 300 documents
    
    nlp = create_lemmatization_pipeline()
    
    # Benchmark batch processing
    print("Batch processing benchmark:")
    
    for batch_size in [10, 50, 100, 500]:
        start = time.perf_counter()
        results = list(batch_lemmatize(texts, nlp, batch_size=batch_size))
        elapsed = time.perf_counter() - start
        rate = len(texts) / elapsed
        print(f"  Batch size {batch_size:4}: {elapsed:.3f}s ({rate:.0f} docs/sec)")
    
    # Caching demonstration
    print("\nCaching demonstration:")
    cached = CachedLemmatizer(cache_size=1000)
    
    # First pass - cache misses
    start = time.perf_counter()
    for text in texts[:50]:
        _ = cached.lemmatize_text(text)
    first_pass = time.perf_counter() - start
    print(f"  First pass: {first_pass:.3f}s")
    print(f"  Cache info: {cached.cache_stats()}")
    
    # Second pass - cache hits for individual words
    start = time.perf_counter()
    for text in texts[:50]:
        _ = cached.lemmatize_text(text)
    second_pass = time.perf_counter() - start
    print(f"  Second pass: {second_pass:.3f}s")
    print(f"  Cache info: {cached.cache_stats()}")
    
    # Version info for reproducibility
    print("\nVersion tracking:")
    det_lemmatizer = DeterministicLemmatizer()
    print(f"  Version info: {det_lemmatizer.get_version_info()}")