Language Identification › Implementation: Building a Language Detection Pipeline

import subprocess
import sys

# Install required packages
subprocess.run(
    [
        sys.executable,
        "-m",
        "pip",
        "install",
        "fasttext-wheel",
        "langdetect",
        "lingua-language-detector",
        "--quiet",
    ],
    capture_output=True,
)

# Small training corpora for five languages
TRAINING_DATA = {
    "en": [
        "the quick brown fox jumps over the lazy dog",
        "language identification is a fundamental task in natural language processing",
        "machine learning models can process text in many different languages",
        "the algorithm analyzes character sequences to determine the source language",
        "natural language processing has advanced significantly in recent years",
        "deep learning has transformed how we approach language understanding",
        "the model learns statistical patterns from large text corpora",
        "word embeddings capture semantic relationships between vocabulary items",
    ],
    "fr": [
        "le rapide renard brun saute par dessus le chien paresseux",
        "l identification de langue est une tache fondamentale en traitement automatique",
        "les modeles d apprentissage automatique peuvent traiter du texte en plusieurs langues",
        "l algorithme analyse les sequences de caracteres pour determiner la langue source",
        "le traitement automatique des langues a considerablement progresse ces dernieres annees",
        "l apprentissage profond a transforme notre approche de la comprehension du langage",
        "le modele apprend des motifs statistiques a partir de grands corpus de textes",
        "les plongements de mots capturent les relations semantiques entre les mots du vocabulaire",
    ],
    "de": [
        "der schnelle braune fuchs springt ueber den faulen hund",
        "sprachidentifikation ist eine grundlegende aufgabe in der verarbeitung natuerlicher sprache",
        "maschinelle lernmodelle koennen text in vielen verschiedenen sprachen verarbeiten",
        "der algorithmus analysiert zeichenfolgen um die quellsprache zu bestimmen",
        "die verarbeitung natuerlicher sprache hat sich in den letzten jahren erheblich weiterentwickelt",
        "tiefes lernen hat unseren ansatz zum sprachverstaendnis veraendert",
        "das modell lernt statistische muster aus grossen textkorpora",
        "worteinbettungen erfassen semantische beziehungen zwischen vokabulareintraegen",
    ],
    "es": [
        "el rapido zorro marron salta sobre el perro perezoso",
        "la identificacion de idiomas es una tarea fundamental en el procesamiento del lenguaje natural",
        "los modelos de aprendizaje automatico pueden procesar texto en muchos idiomas diferentes",
        "el algoritmo analiza secuencias de caracteres para determinar el idioma fuente",
        "el procesamiento del lenguaje natural ha avanzado significativamente en los ultimos anos",
        "el aprendizaje profundo ha transformado como abordamos la comprension del lenguaje",
        "el modelo aprende patrones estadisticos de grandes corpus de textos",
        "las incrustaciones de palabras capturan relaciones semanticas entre elementos del vocabulario",
    ],
    "it": [
        "la veloce volpe marrone salta sopra il cane pigro",
        "l identificazione della lingua e un compito fondamentale nel trattamento del linguaggio naturale",
        "i modelli di apprendimento automatico possono elaborare testo in molte lingue diverse",
        "l algoritmo analizza sequenze di caratteri per determinare la lingua sorgente",
        "l elaborazione del linguaggio naturale ha fatto notevoli progressi negli ultimi anni",
        "l apprendimento profondo ha trasformato il nostro approccio alla comprensione del linguaggio",
        "il modello apprende schemi statistici da grandi corpus di testi",
        "le incorporazioni di parole catturano le relazioni semantiche tra gli elementi del vocabolario",
    ],
}

import math
from collections import Counter


def get_char_ngrams(text: str, n: int = 3) -> list[str]:
    """Extract all character n-grams from text."""
    text = text.lower()
    # Add padding characters to handle word boundaries
    padded = f"  {text}  "
    return [padded[i : i + n] for i in range(len(padded) - n + 1)]


def build_ngram_model(sentences: list[str], n: int = 3) -> dict[str, float]:
    """Build a smoothed character n-gram language model."""
    counts: Counter = Counter()
    prefix_counts: Counter = Counter()

    for sentence in sentences:
        ngrams = get_char_ngrams(sentence, n)
        for gram in ngrams:
            counts[gram] += 1
            prefix_counts[gram[:-1]] += 1

    # Laplace smoothed probabilities
    vocab_size = len(counts)
    log_probs = {}
    for gram, count in counts.items():
        prefix = gram[:-1]
        log_probs[gram] = math.log(
            (count + 1) / (prefix_counts[prefix] + vocab_size)
        )

    # Store the smoothing parameters for unseen n-grams
    log_probs["__unk_log_prob__"] = math.log(
        1 / (sum(counts.values()) + vocab_size)
    )

    return log_probs


def compute_log_prob(text: str, model: dict[str, float], n: int = 3) -> float:
    """Compute the log-probability of text under a language model."""
    ngrams = get_char_ngrams(text, n)
    unk_log_prob = model.get("__unk_log_prob__", -10.0)
    total = sum(model.get(gram, unk_log_prob) for gram in ngrams)
    # Normalize by document length to avoid length bias
    return total / len(ngrams) if ngrams else float("-inf")


def predict_language(
    text: str, models: dict[str, dict[str, float]]
) -> tuple[str, dict[str, float]]:
    """Predict language using the n-gram models. Returns predicted language and all scores."""
    scores = {
        lang: compute_log_prob(text, model) for lang, model in models.items()
    }
    predicted = max(scores, key=scores.__getitem__)
    return predicted, scores


# Build models for all five languages
ngram_models = {
    lang: build_ngram_model(sentences)
    for lang, sentences in TRAINING_DATA.items()
}

# Test the model on held-out sentences
TEST_SENTENCES = [
    ("en", "the neural network was trained on a large dataset"),
    (
        "fr",
        "le reseau de neurones a ete entraine sur un grand ensemble de donnees",
    ),
    ("de", "das neuronale netz wurde auf einem grossen datensatz trainiert"),
    ("es", "la red neuronal fue entrenada en un gran conjunto de datos"),
    ("it", "la rete neurale e stata addestrata su un grande insieme di dati"),
    ("en", "hello world"),  # Short text challenge
    ("fr", "bonjour monde"),
    ("de", "hallo welt"),
]

results = []
for true_lang, sentence in TEST_SENTENCES:
    predicted, scores = predict_language(sentence, ngram_models)
    correct = predicted == true_lang
    results.append(
        {
            "sentence": sentence[:45] + "..."
            if len(sentence) > 45
            else sentence,
            "true": true_lang,
            "predicted": predicted,
            "correct": correct,
            "confidence": scores[predicted],
        }
    )

n_correct = sum(r["correct"] for r in results)

import os
import urllib.request

# Download FastText language identification model if not present
model_path = "/tmp/lid.176.bin"
model_url = (
    "https://dl.fbaipublicfiles.com/fasttext/supervised-models/lid.176.bin"
)

if not os.path.exists(model_path):
    print("Downloading FastText LID model (126 MB)...")
    urllib.request.urlretrieve(model_url, model_path)
    print("Downloaded successfully.")

try:
    import fasttext

    ft_model = fasttext.load_model(model_path)
    FASTTEXT_AVAILABLE = True
except Exception:
    FASTTEXT_AVAILABLE = False

def detect_language_fasttext(
    text: str, model, top_k: int = 3
) -> list[tuple[str, float]]:
    """Detect language using FastText. Returns top_k (language, probability) pairs."""
    # FastText expects single-line input; use internal f.predict to avoid NumPy 2.x
    # compatibility issue with np.array(probs, copy=False) in fasttext's Python wrapper.
    text_nl = text.replace("\n", " ").strip() + "\n"
    predictions = model.f.predict(text_nl, top_k, 0.0, "strict")
    # predictions is a list of (prob, label) tuples
    result = [
        (label.replace("__label__", ""), prob) for prob, label in predictions
    ]
    return result


# Test documents including a code-switched example
test_documents = [
    "The attention mechanism allows the model to focus on relevant parts of the input sequence.",
    "Le mecanisme d attention permet au modele de se concentrer sur les parties pertinentes de l entree.",
    "Der Aufmerksamkeitsmechanismus ermoglicht es dem Modell, sich auf relevante Teile der Eingabe zu konzentrieren.",
    "No puedo creer que this actually worked, increible!",  # Code-switching
    "안녕하세요, 오늘 날씨가 정말 좋네요.",  # Korean
    "这是一个关于自然语言处理的例子。",  # Chinese
    "مرحبا بالعالم",  # Arabic
    "Hi",  # Ambiguous short text
]

if FASTTEXT_AVAILABLE:
    detection_results = []
    for doc in test_documents:
        results_ft = detect_language_fasttext(doc, ft_model, top_k=3)
        detection_results.append(
            {
                "doc": doc[:50] + "..." if len(doc) > 50 else doc,
                "results": results_ft,
            }
        )

class LanguageFilter:
    """Filter documents by detected language with configurable confidence threshold."""

    def __init__(
        self,
        target_languages: list[str],
        min_confidence: float = 0.8,
        model=None,
    ):
        self.target_languages = set(target_languages)
        self.min_confidence = min_confidence
        self.model = model
        self.stats = {
            "total": 0,
            "accepted": 0,
            "rejected_language": 0,
            "rejected_confidence": 0,
        }

    def filter_document(self, text: str) -> tuple[bool, str, float]:
        """
        Check whether a document passes the language filter.
        Returns (passes, detected_language, confidence).
        """
        self.stats["total"] += 1

        if self.model is not None and FASTTEXT_AVAILABLE:
            results_ft = detect_language_fasttext(text, self.model, top_k=1)
            detected_lang, confidence = results_ft[0]
        else:
            # Fall back to n-gram model (covers only en/fr/de/es/it)
            predicted, scores = predict_language(text, ngram_models)
            detected_lang = predicted
            sorted_scores = sorted(scores.values(), reverse=True)
            if len(sorted_scores) >= 2:
                confidence = math.exp(sorted_scores[0] - sorted_scores[1])
                confidence = min(confidence, 1.0)
            else:
                confidence = 1.0

        if detected_lang not in self.target_languages:
            self.stats["rejected_language"] += 1
            return False, detected_lang, confidence

        if confidence < self.min_confidence:
            self.stats["rejected_confidence"] += 1
            return False, detected_lang, confidence

        self.stats["accepted"] += 1
        return True, detected_lang, confidence

    def filter_corpus(self, documents: list[str]) -> list[str]:
        """Filter a list of documents, returning only those that pass."""
        return [doc for doc in documents if self.filter_document(doc)[0]]

    def report(self) -> dict:
        """Return filtering statistics."""
        total = self.stats["total"]
        return {
            "total": total,
            "accepted": self.stats["accepted"],
            "rejected_language": self.stats["rejected_language"],
            "rejected_confidence": self.stats["rejected_confidence"],
            "acceptance_rate": self.stats["accepted"] / total
            if total > 0
            else 0.0,
        }

# Simulate a mixed-language corpus
mixed_corpus = [
    "The transformer architecture uses self-attention to process sequences in parallel.",
    "L architecture transformer utilise l auto-attention pour traiter les sequences en parallele.",
    "Die Transformer-Architektur verwendet Selbst-Aufmerksamkeit zur parallelen Sequenzverarbeitung.",
    "La arquitectura transformer usa auto-atencion para procesar secuencias en paralelo.",
    "Tokenization splits text into subword units before model processing.",
    "La tokenisation divise le texte en unites de sous-mots avant le traitement par le modele.",
    "For more information, visit our website.",
    "Besuchen Sie unsere Website fur weitere Informationen.",
    "Hi",  # Short ambiguous text
    "ok",  # Very short
    "Natural language processing has many practical applications.",
    "La langue naturelle est complexe et riche en nuances semantiques.",
    "Diese Methode verbessert die Sprachverarbeitung erheblich.",
]

# Filter for English only with 70% confidence threshold
en_filter = LanguageFilter(
    target_languages=["en"],
    min_confidence=0.7,
    model=ft_model if FASTTEXT_AVAILABLE else None,
)

accepted_docs = en_filter.filter_corpus(mixed_corpus)
filter_report = en_filter.report()