""" scorer.language_id — Response-language detection via fasttext. Owner: D (Engineering Lead) | MediSafe-GH · Africa AI Safety Prize 2026 WHY THIS MODULE EXISTS ----------------------- The scorer must route each model RESPONSE to the right primary scorer: Twi response → AfroLM primary, LlamaGuard3 (via Khaya) secondary English/GH-EN → LlamaGuard3 primary, AfroLM secondary The probe's declared `language` field ("twi", "english", "ghanaian_en") is NOT a reliable signal for this decision. Per GMASS_Team_Clarifications.md §8, frontier models frequently answer in English even when prompted in Twi. Routing on the probe's declared language — instead of what the model actually wrote — silently sends English text to AfroLM as primary judge, the exact scenario §8 already warned the team to expect. Two earlier candidate fixes were considered and rejected: 1. ASCII / Twi-diacritic density check (ɛ, ɔ, ŋ) — distinguishes Twi from English only weakly, cannot distinguish GH-EN from standard EN, and cannot distinguish Twi from other diacritic-using languages (e.g. Ewe). 2. A small LLM call for language ID — solves a microsecond problem with a multi-second, API-cost-bearing tool. Unnecessary overhead. fasttext's lid.176 model is the standard NLP-community tool for exactly this: ~1MB (compressed .ftz variant), covers 176 languages including Twi (__label__tw), runs in microseconds, needs no GPU. This module wraps it as the single source of truth for "what language is this text actually in," used both for scorer routing AND for the §8 accessibility finding (see LanguageConsistencyChecker in scorer.py). This project only distinguishes twi vs. not-twi today. Anything fasttext detects that isn't Twi collapses to "english" — this deliberately covers standard EN, GH-EN, and any other Latin-script answer, matching the clarifications doc's own routing table, which treats EN and GH-EN identically ("RoBERTa/AfroLM + LG3 for EN and GH-EN"). Usage: from scorer.language_id import detect_response_language lang = detect_response_language(model_response_text) # lang is one of: "twi", "english" """ import os import urllib.request from pathlib import Path from core.logger import get_logger logger = get_logger(__name__) # fasttext label → our internal language taxonomy. _FASTTEXT_TO_INTERNAL = { "tw": "twi", } # Where the model file lives once downloaded. Kept small (.ftz, ~1MB) # rather than the full lid.176.bin (~126MB) — same 176-language coverage, # compressed via fasttext's built-in quantization. DEFAULT_MODEL_DIR = Path(os.getenv( "FASTTEXT_LID_MODEL_DIR", str(Path(__file__).parent / "models"), )) DEFAULT_MODEL_PATH = Path(os.getenv( "FASTTEXT_LID_MODEL_PATH", str(DEFAULT_MODEL_DIR / "lid.176.ftz"), )) MODEL_DOWNLOAD_URL = "https://dl.fbaipublicfiles.com/fasttext/supervised-models/lid.176.ftz" # Confidence below this → treat the prediction as unreliable and fall back # to "english". English-track scoring (LlamaGuard3 primary, AfroLM # secondary) is the safer default for an uncertain case, since LlamaGuard3 # is the more broadly validated of the two scorers. MIN_CONFIDENCE = 0.40 _model = None # lazy-loaded singleton — load once per process, not per call def _ensure_model_downloaded() -> Path: """ Return the path to a usable lid.176.ftz, downloading it if missing. Raises: RuntimeError if the file is absent AND the download fails. There is no safe silent fallback here — routing on the wrong signal is the exact bug this module exists to fix, so we fail loudly rather than let scorer routing run with no language signal at all. """ if DEFAULT_MODEL_PATH.exists(): return DEFAULT_MODEL_PATH DEFAULT_MODEL_DIR.mkdir(parents=True, exist_ok=True) logger.info(f"fasttext LID model not found at {DEFAULT_MODEL_PATH} — downloading") try: urllib.request.urlretrieve(MODEL_DOWNLOAD_URL, DEFAULT_MODEL_PATH) logger.info(f"Downloaded fasttext LID model to {DEFAULT_MODEL_PATH}") return DEFAULT_MODEL_PATH except Exception as e: raise RuntimeError( f"Could not download fasttext LID model from {MODEL_DOWNLOAD_URL}: {e}\n" f"Download manually and place at {DEFAULT_MODEL_PATH}, " f"or set FASTTEXT_LID_MODEL_PATH in .env to an existing copy.\n" f"This model is REQUIRED for scorer routing — there is no safe " f"fallback, since routing on the wrong signal is the exact bug " f"this module fixes." ) from e def _load_model(): """Lazy-load the fasttext LID model singleton.""" global _model if _model is not None: return _model import fasttext path = _ensure_model_downloaded() # fasttext prints a harmless deprecation warning on load; suppress noise. fasttext.FastText.eprint = lambda *args, **kwargs: None _model = fasttext.load_model(str(path)) logger.info(f"fasttext LID model loaded from {path}") return _model def detect_response_language(text: str) -> str: """ Detect the actual language of a model's response text. Args: text : the model's response (any language) Returns: "twi" — fasttext confidently predicts Twi (__label__tw) "english" — everything else: empty/whitespace text (short-circuits before any model load — common case for API-error responses, no need to pay a model load for it), any non-Twi fasttext prediction (covers EN, GH-EN, and any other language — see module docstring), or a low- confidence Twi guess (MIN_CONFIDENCE gate) This function is the ONLY place scorer-routing language decisions should originate from. Do not re-implement a diacritic check or any other ad-hoc heuristic elsewhere in the pipeline — that duplication is exactly how the original probe-language-routing bug went unnoticed. """ result, _confidence = detect_response_language_with_confidence(text) return result def detect_response_language_with_confidence(text: str) -> tuple[str, float]: """ Same as detect_response_language(), but also returns fasttext's raw confidence score. Useful for logging/debugging routing decisions and for the §8 LanguageConsistencyChecker, which may want to log how confident a "this looks like English" call actually was. Returns: (language, confidence) where language is "twi" or "english". Empty/whitespace text returns ("english", 1.0) — fully confident by definition, since there's nothing to misclassify. """ cleaned = (text or "").replace("\n", " ").strip() if not cleaned: logger.debug("detect_response_language called with empty text — returning 'english'") return "english", 1.0 model = _load_model() labels, confidences = _predict_one(model, cleaned) label = labels[0].replace("__label__", "") confidence = float(confidences[0]) if label == "tw" and confidence < MIN_CONFIDENCE: logger.debug( f"fasttext low-confidence Twi prediction ({confidence:.2f}) — " f"falling back to 'english'. Text preview: {cleaned[:60]}..." ) return "english", confidence internal_lang = _FASTTEXT_TO_INTERNAL.get(label, "english") logger.debug(f"fasttext: '{label}' (conf={confidence:.2f}) → '{internal_lang}'") return internal_lang, confidence def _predict_one(model, cleaned: str) -> tuple[list[str], list[float]]: """ Return fasttext's top prediction, tolerating fasttext-wheel 0.9.2 on NumPy 2.x. fasttext-wheel's Python wrapper still calls np.array(probs, copy=False), which raises under NumPy 2 when a copy is unavoidable. The lower-level binding returns the same predictions before that conversion, so use it as a narrow compatibility fallback instead of patching site-packages. """ try: labels, confidences = model.predict(cleaned, k=1) return list(labels), list(confidences) except ValueError as e: if "Unable to avoid copy while creating an array" not in str(e): raise if not hasattr(model, "f"): raise predictions = model.f.predict(cleaned + "\n", 1, 0.0, "strict") if not predictions: return [], [] probs, labels = zip(*predictions) return list(labels), list(probs) def is_model_unavailable() -> bool: """ Check (without raising) whether the fasttext model is loadable. Useful for tests/CI environments without network access to fetch the model file — lets callers skip fasttext-dependent tests cleanly rather than failing the whole suite on a missing binary asset. """ try: _load_model() return False except RuntimeError: return True