| """Reply validator: tags well-formed, language proportions vs target mix, greeting pattern kept, length, loops. |
| Scores a candidate; the agent keeps the best passing candidate, regenerates up to N rounds, then falls back.""" |
| import re |
| from collections import Counter |
| import mix as M |
|
|
| TAG = re.compile(r"<(eng|ibo|yor|hau|pcm)>(.*?)</\1>", re.S) |
|
|
| def parse(out): |
| out = out.strip().splitlines()[0].strip() if out.strip() else "" |
| spans = TAG.findall(out) |
| rest = TAG.sub(" ", out) |
| return out, spans, re.sub(r"[\s.,!?;:'\"\-–—…]", "", rest) |
|
|
| def check(out, target, max_words=40, min_words=2, lid="glotlid", tvd_max=0.3, user=None): |
| out, spans, rest = parse(out) |
| r = dict(text=out, ok=False, why=[], tvd=1.0) |
| if not spans: |
| r["why"].append("no tags"); return r |
| if rest: |
| r["why"].append("text outside tags") |
| words = " ".join(t for _, t in spans).split() |
| if not (min_words <= len(words) <= max_words): |
| r["why"].append(f"length {len(words)}") |
| low = [w.lower().strip(".,!?") for w in words] |
| tri = Counter(tuple(low[i:i + 3]) for i in range(len(low) - 2)) |
| if tri and tri.most_common(1)[0][1] > 2: |
| r["why"].append("loop") |
| if user: |
| uw = [w.lower().strip(".,!?") for w in user.split()] |
| ug = {tuple(uw[i:i + 3]) for i in range(len(uw) - 2)} |
| rg = [tuple(low[i:i + 3]) for i in range(len(low) - 2)] |
| r["echo"] = round(sum(g in ug for g in rg) / len(rg), 2) if rg else 0.0 |
| if r["echo"] > 0.5 and len(low) >= 6: r["why"].append(f"echo {r['echo']}") |
| |
| n = max(1, len(words)) |
| decl = Counter() |
| for l, t in spans: decl[l] += len(t.split()) |
| decl = {k: v / n for k, v in decl.items()} |
| plain = " ".join(t for _, t in spans) |
| meas = M.estimate(plain, lid=lid) |
| |
| mixd = {k: 0.5 * decl.get(k, 0) + 0.5 * meas["mix"].get(k, 0) for k in set(decl) | set(meas["mix"])} |
| tgt = target["mix"] |
| d = M.tvd(mixd, tgt) |
| r.update(declared={k: round(v, 2) for k, v in decl.items()}, measured=meas["mix"], mix=({k: round(v, 2) for k, v in mixd.items() if v > 0.01}), tvd=round(d, 3)) |
| |
| lab = meas["labels"]; i = 0; mism = 0; ng = 0 |
| for l, t in spans: |
| k = len(M.tagger(lid).word_langs(t)[1]) if t.strip() else 0 |
| if l in ("ibo", "yor", "hau"): |
| ng += k; mism += sum(1 for x in lab[i:i + k] if x == "eng") |
| i += k |
| r["tag_mismatch"] = round(mism / ng, 2) if ng else 0.0 |
| if r["tag_mismatch"] > 0.5: r["why"].append("non-English tags over English words") |
| if d > tvd_max: r["why"].append(f"mix drift tvd={d:.2f}") |
| for l, p in tgt.items(): |
| if p >= 0.15 and mixd.get(l, 0) < 0.05: r["why"].append(f"missing {l}") |
| for p in target.get("pattern", []): |
| k, _, l = p.partition(":") |
| if k == "greeting" and spans and spans[0][0] != l: r["why"].append(f"no {l} greeting") |
| r["ok"] = not r["why"] |
| r["score"] = d + 0.25 * len(r["why"]) + 0.2 * r["tag_mismatch"] |
| return r |
|
|
| SAFE = {"ibo": "Ndewo!", "yor": "Ẹ pẹ̀lẹ́ o!", "hau": "Sannu!", "pcm": "Abeg no vex,"} |
|
|
| def fallback(text_candidates, target): |
| """all rounds failed. Echoing/drifting native-language output -> an honest short reply (a correct phrasebook opener in the user's |
| language + a clarification request); otherwise the best-scoring candidate if its tags parse; else plain text in the matrix language.""" |
| best = min(text_candidates, key=lambda c: c["score"] if "score" in c else 9, default=None) |
| m = target.get("matrix", "eng") |
| if m in SAFE and (best is None or any(w.startswith(("echo", "mix drift", "missing", "no tags", "non-English")) for w in best.get("why", []))): |
| tail = "<pcm>I no too hear you well. Abeg talk am again?</pcm>" if m == "pcm" else "<eng>Sorry, I did not catch all of that. Could you say it again, or in English?</eng>" |
| return f"<{m}>{SAFE[m]}</{m}> {tail}", "safe_phrasebook" |
| if best and TAG.search(best["text"] or ""): |
| return best["text"], "best_failed" |
| m = target.get("matrix", "eng"); raw = re.sub(r"</?\w+>", "", (best or {}).get("text", "") or "Sorry, say that again?") |
| return f"<{m}>{raw.strip()}</{m}>", "plain_matrix" |
|
|
