| """Per-word language ID + language-mix estimate for a user turn (and for a reply). |
| Reuses the Dynamic Accent release tagger (rule-based lexicons/diacritics/Pidgin markers, optional GlotLID clause scores). |
| estimate("Ndewo, how are you doing today?") -> |
| {"mix": {"eng": 0.67, "ibo": 0.33}, "matrix": "eng", "pattern": ["greeting:ibo"], "words": [...], "labels": [...]}""" |
| import re, unicodedata |
| from tagger import Tagger, fold |
|
|
| LANGS = ["eng", "pcm", "ibo", "yor", "hau"] |
| NAMES = {"eng": "English", "pcm": "Nigerian Pidgin", "ibo": "Igbo", "yor": "Yoruba", "hau": "Hausa"} |
| |
| GREET = {"ibo": ["nde wo", "ndewoo", "ndeewo", "kedu", "nnoo", "ututu oma", "ehihie oma", "mgbede oma", "ka chi fo", "kedu ka i mere", "i bola chi"], |
| "yor": ["e kaaro", "e kaasan", "e kurole", "e ku irole", "bawo", "se daadaa ni", "e nle", "pele", "e kaabo", "e ku abo", "ekaaro", "ekaasan", "ekaro", "e karo", "e kaaro o"], |
| "hau": ["sannu", "sanu", "sanno", "baka da yamma", "baka da safiya", "barka", "barka da safiya", "barka da yamma", "ina kwana", "ina wuni", "yaya dai", "lafiya lau", "maraba"], |
| "pcm": ["how far", "how body", "wetin dey", "how you dey", "how una dey", "i hail"]} |
| CLOSE = {"ibo": ["daalu", "dalu", "dalo", "daalo", "dalow", "ka o di", "ka emesia", "chukwu gozie"], "yor": ["e se", "o se", "o dabo", "e seun", "ese", "odabo"], |
| "hau": ["na gode", "sai an jima", "sai anjima", "nagode", "mun gode", "na godi", "nagodi"], "pcm": ["we go see", "thank you well well", "make we yarn later"]} |
| DISC = {"pcm": ["abeg", "abi", "sha", "o", "na so", "omo", "wahala"], "yor": ["sha", "o", "ehen", "abi"], "ibo": ["biko", "nna", "nne", "chai", "ehen", "ngwa"], |
| "hau": ["to", "kai", "wallahi", "haba"]} |
|
|
| _T = {} |
| def tagger(lid="glotlid"): |
| if lid not in _T: |
| try: |
| _T[lid] = Tagger(lid if lid != "none" else None) |
| except Exception as e: |
| print("tagger: GlotLID unavailable, rules only:", repr(e)[:120]); _T[lid] = Tagger(None) |
| return _T[lid] |
|
|
| def _find(fs, phrases): |
| for p in sorted(phrases, key=len, reverse=True): |
| if re.search(r"(^|\s)" + re.escape(p) + r"(\s|$)", fs): |
| return p |
| return None |
|
|
| PCM_MARK = {"dey", "don", "abeg", "wetin", "una", "gree", "wahala", "sabi", "comot", "chop", "oya", "abi", "shey", "wey", "dem", "wan", |
| "pikin", "oga", "sef", "jare", "naim", "dis", "dat", "wahala", "kpatakpata", "ehn", "sha", "nor", "abeg", "jor", "ehen", "na"} |
| PCM_BI = {("no", "gree"), ("no", "be"), ("e", "no"), ("make", "we"), ("how", "far"), ("na", "im"), ("i", "go"), ("you", "go"), ("we", "go"), ("e", "go"), |
| ("no", "go"), ("i", "fit"), ("you", "fit"), ("e", "don"), ("i", "don"), ("my", "guy"), ("na", "so"), ("well", "well"), ("no", "dey")} |
|
|
| def pidgin_relabel(words, labs, text): |
| """sentence-level Pidgin: a sentence of English-labelled words with Pidgin markers is Pidgin (the word tagger only flips the marker's clause).""" |
| if not labs: return labs |
| labs = list(labs); f = [fold(w).strip(".,!?;:'\"") for w in words] |
| ends = [i for i, w in enumerate(words) if re.search(r"[.!?]$", w)] + [len(words) - 1] |
| s0 = 0 |
| for e in sorted(set(ends)): |
| idx = list(range(s0, e + 1)); s0 = e + 1 |
| if not idx: continue |
| hits = sum(1 for i in idx if f[i] in PCM_MARK or f[i] == "omo") + sum(1 for a, b in zip(idx, idx[1:]) if (f[a], f[b]) in PCM_BI and not ((f[a], f[b]) == ("how", "far") and a != idx[0])) |
| hits += 1 if f[idx[-1]] == "o" and len(idx) > 2 else 0 |
| lat = [i for i in idx if labs[i] in ("eng", "pcm")] |
| if hits == 0 and len(lat) > 4 and all(labs[i] == "pcm" for i in lat): |
| for i in lat: labs[i] = "eng" |
| continue |
| if lat and (hits >= 2 or (hits >= 1 and len(idx) <= 8)) and sum(labs[i] == "pcm" for i in lat) < len(lat): |
| for i in lat: labs[i] = "pcm" |
| return labs |
|
|
| def window_relabel(words, labs, t, win=4, thr=0.5): |
| """ASR text has no punctuation/diacritics, so the clause-level tagger sees one long 'English' clause. Slide a GlotLID window |
| over the words and let confident window votes override the rule labels (rule label keeps a weight-1 vote).""" |
| from collections import Counter |
| n = len(words) |
| if t.lid is None or n < 3: return labs |
| votes = [Counter({labs[i]: 1.0}) for i in range(n)] |
| for s0 in range(0, max(1, n - win + 1)): |
| sc = t._lid_scores(" ".join(words[s0:s0 + win])); l = max(sc, key=sc.get) |
| if sc[l] >= thr: |
| for i in range(s0, min(n, s0 + win)): votes[i][l] += sc[l] |
| return [v.most_common(1)[0][0] for v in votes] |
|
|
| def estimate(text, lid="glotlid", prior=None, prior_w=0.0, asr=False): |
| """prior: a previous mix dict (conversation-level style), blended with weight prior_w.""" |
| t = tagger(lid) |
| toks, labs = t.word_langs(text) if text.strip() else ([], []) |
| words = [w for _, _, w in toks] |
| if asr: labs = window_relabel(words, list(labs), t) |
| labs = pidgin_relabel(words, labs, text) |
| n = max(1, len(labs)) |
| mix = {l: labs.count(l) / n for l in LANGS if labs.count(l)} |
| if prior and prior_w > 0: |
| keys = set(mix) | set(prior) |
| mix = {k: (1 - prior_w) * mix.get(k, 0) + prior_w * prior.get(k, 0) for k in keys} |
| s = sum(mix.values()) or 1; mix = {k: v / s for k, v in mix.items() if v / s >= 0.03} |
| fs = " " + re.sub(r"[^\w\s']", " ", fold(text)) + " " |
| fs = re.sub(r"\s+", " ", fs) |
| pat = [] |
| head = " ".join(fs.split()[:5]); tail = " ".join(fs.split()[-5:]) |
| for l in ("ibo", "yor", "hau", "pcm"): |
| g = _find(head, GREET[l]); c = _find(tail, CLOSE[l]) |
| if g: pat.append(f"greeting:{l}") |
| if c: pat.append(f"closing:{l}") |
| d = [p for p in DISC[l] if re.search(r"(^|\s)" + re.escape(p) + r"(\s|$)", fs)] |
| if d and l in mix: pat.append(f"discourse:{l}") |
| |
| toks_f = fs.split() |
| for p in pat: |
| k, _, l = p.partition(":") |
| if k in ("greeting", "closing") and mix.get(l, 0) < 0.1 and labs: |
| phr = _find(head if k == "greeting" else tail, (GREET if k == "greeting" else CLOSE)[l]) or "" |
| share = min(0.5, len(phr.split()) / max(1, len(labs))) |
| m2 = max(mix, key=mix.get) |
| mix[m2] = max(0.0, mix[m2] - share); mix[l] = mix.get(l, 0) + share |
| mix = {a: b for a, b in mix.items() if b > 0.01} |
| |
| sw = sum(1 for a, b in zip(labs, labs[1:]) if a != b) |
| if len(mix) > 1: |
| pat.append("intra-sentential" if sw >= 3 else "inter-sentential") |
| matrix = max(mix, key=mix.get) if mix else "eng" |
| return dict(mix={k: round(v, 2) for k, v in sorted(mix.items(), key=lambda x: -x[1])}, matrix=matrix, pattern=pat, |
| words=words, labels=labs, n_words=len(labs), switches=sw) |
|
|
| def describe(est): |
| """natural-language mix instruction for the LLM.""" |
| mix, pat = est["mix"], est["pattern"] |
| if not mix: |
| return "Reply in English." |
| parts = [f"{NAMES[l]} about {round(100 * p / 5) * 5}%" for l, p in mix.items()] |
| s = f"Match the user's language mix: {', '.join(parts)} of the words. The main (matrix) language is {NAMES[est['matrix']]}." |
| for p in pat: |
| k, _, l = p.partition(":") |
| if k == "greeting": s += f" The user greeted in {NAMES[l]}: open with a short {NAMES[l]} greeting." |
| if k == "closing": s += f" The user closed in {NAMES[l]}: you may close in {NAMES[l]}." |
| if k == "discourse": s += f" Sprinkle a {NAMES[l]} discourse word or two the way the user did." |
| if "intra-sentential" in pat: s += " The user switches inside sentences; do the same naturally." |
| if len(mix) == 1: s += f" Reply only in {NAMES[est['matrix']]}." |
| return s |
|
|
| def tvd(a, b): |
| k = set(a) | set(b) |
| return 0.5 * sum(abs(a.get(x, 0) - b.get(x, 0)) for x in k) |
|
|
| if __name__ == "__main__": |
| import sys, json |
| for s in sys.argv[1:] or ["Ndewo! How are you doing today? I wan ask you something.", "Ẹ káàárọ̀, ṣé dáadáa ni? I need help with my account."]: |
| e = estimate(s); print(json.dumps({k: e[k] for k in ("mix", "matrix", "pattern")}, ensure_ascii=False), "|", describe(e)) |
|
|
