thai_hybrid: dictionary fast-path + model fallback (WO03)

src/interscript/ml/thai_hybrid.py

@@ -0,0 +1,91 @@
+"""Thai lexicon fast-path + neural fallback (WO03).
+
+The FastThaiG2P response: dictionary hits decode at dict speed,
+out-of-vocabulary runs go to the neural model as whole spans. Maximal-
+munch longest-match segmentation; spaces/punct pass through untouched.
+"""
+
+from __future__ import annotations
+
+_THAI = "฀-๿"
+
+
+class Lexicon(dict):
+    """word -> IPA; longest key wins via sorted lookup."""
+
+
+def _longest_at(text: str, i: int, keys: list[str], max_len: int) -> str | None:
+    for n in range(min(max_len, len(text) - i), 0, -1):
+        w = text[i:i + n]
+        if w in keys:
+            return w
+    return None
+
+
+def segment(text: str, lex: Lexicon) -> list[tuple[str, str | None]]:
+    """-> [(chunk, ipa-or-None)]; None = OOV run for the model."""
+    keys = lex.keys()
+    max_len = max((len(k) for k in keys), default=0)
+    out: list[tuple[str, str | None]] = []
+    oov: list[str] = []
+    i = 0
+    while i < len(text):
+        ch = text[i]
+        if not _is_thai(ch):
+            if oov:
+                out.append(("".join(oov), None))
+                oov = []
+            out.append((ch, ch))
+            i += 1
+            continue
+        w = _longest_at(text, i, keys, max_len)
+        if w:
+            if oov:
+                out.append(("".join(oov), None))
+                oov = []
+            out.append((w, lex[w]))
+            i += len(w)
+        else:
+            oov.append(ch)
+            i += 1
+    if oov:
+        out.append(("".join(oov), None))
+    return out
+
+
+def _is_thai(ch: str) -> bool:
+    return "฀" <= ch <= "๿"
+
+
+def hybrid_translate(text: str, lex: Lexicon, model) -> str:
+    """Dict hits splice verbatim; each maximal OOV run is one model call."""
+    parts: list[str] = []
+    for chunk, ipa in segment(text, lex):
+        if ipa is None:
+            parts.append(model.translate(chunk))  # Thai OOV run
+        elif not _is_thai(chunk[0]):
+            parts.append(chunk)  # spaces/punct pass through
+        else:
+            parts.append(ipa)  # dictionary hit
+    return "".join(parts)
+
+
+def build_lexicon_from_kaikki(entries, min_head_len: int = 1) -> Lexicon:
+    """kaikki Thai entries (dicts with 'word' + 'sounds') -> Lexicon."""
+    counts: dict[str, dict[str, int]] = {}
+    for e in entries:
+        word = e.get("word") or ""
+        if len(word) < min_head_len:
+            continue
+        for s in e.get("sounds") or []:
+            ipa = s.get("ipa")
+            if not ipa:
+                continue
+            ipa = ipa.strip("/")
+            if ipa:
+                counts.setdefault(word, {ipa: 0})
+                counts[word][ipa] = counts[word].get(ipa, 0) + 1
+    lex = Lexicon()
+    for word, ipas in counts.items():
+        lex[word] = max(ipas.items(), key=lambda kv: kv[1])[0]
+    return lex

tests/test_thai_hybrid.py

@@ -0,0 +1,56 @@
+"""WO03: Thai lexicon fast-path + hybrid (maximal-munch, model fallback)."""
+
+from __future__ import annotations
+
+from interscript.ml.thai_hybrid import Lexicon, segment, hybrid_translate
+
+
+def make_lexicon() -> Lexicon:
+    return Lexicon({"กิน": "kin", "ข้าว": "kʰâːw", "กินข้าว": "kinkʰâːw"})
+
+
+class FakeModel:
+    def __init__(self, mark="|"):
+        self.mark = mark
+        self.calls: list[str] = []
+
+    def translate(self, text: str, max_len: int = 256) -> str:
+        self.calls.append(text)
+        return self.mark + text + self.mark
+
+
+def test_maximal_munch_prefers_longest_match():
+    assert segment("กินข้าว", make_lexicon()) == [("กินข้าว", "kinkʰâːw")]
+
+
+def test_segment_splits_dict_and_oov_runs():
+    # "นอน" is OOV: the run extends until the next dictionary hit
+    segs = segment("นอนกิน", make_lexicon())
+    assert segs == [("นอน", None), ("กิน", "kin")]
+
+
+def test_segment_preserves_spaces_and_punct():
+    assert segment("กิน ข้าว!", make_lexicon()) == [
+        ("กิน", "kin"), (" ", " "), ("ข้าว", "kʰâːw"), ("!", "!"),
+    ]
+
+
+def test_hybrid_uses_lexicon_then_model_for_oov():
+    model = FakeModel()
+    out = hybrid_translate("กินนอน", make_lexicon(), model)
+    assert out == "kin|นอน|"
+    assert model.calls == ["นอน"]  # one batched fallback span, not per char
+
+
+def test_hybrid_all_dict_never_touches_model():
+    model = FakeModel()
+    out = hybrid_translate("กินข้าว", make_lexicon(), model)
+    assert out == "kinkʰâːw"
+    assert model.calls == []
+
+
+def test_hybrid_all_oov_single_span():
+    model = FakeModel()
+    out = hybrid_translate("นอนๆ", make_lexicon(), model)
+    assert out == "|นอนๆ|"
+    assert model.calls == ["นอนๆ"]