#!/usr/bin/env python3 """ Representation-layer probe: what does MiniCPM5-2B's tokenizer actually see? This is the mechanistic half of the study. The behavioural half asks whether the model *reveals* that a question has several legitimate answers. This half asks whether the tokenizer could even carry the distinctions those answers depend on. Everything here is computed from the tokenizer and the vocabulary, so the numbers are exact and independent of quantisation or of prompting. Method note. MiniCPM5-2B uses a GPT-2-style byte-level BPE. Tokens that are genuine text units ("prayer") live in the vocabulary directly; anything else is stored as *bytes* of UTF-8, each mapped into a printable stand-in character. Arabic therefore has to be reassembled from raw byte tokens. To measure that honestly we decode the byte alphabet back to bytes and then to text, rather than inspecting the stand-in characters. Measurements ------------ fertility_by_script tokens per character and per word, by script byte_level_profile what share of the vocabulary and of each script is byte-level fallback rather than a text unit vocab_script_profile how many consecutive Arabic *characters* any single token covers term_morphology how a set of Arabic religious terms actually tokenises: characters per token, fused or detached short vowels diacritic_cost what a vocalised spelling costs over the unvocalised one, and how the vowel is encoded diacritic_discrimination whether two spellings differing only in a short vowel survive as different token sequences """ from __future__ import annotations import json import pathlib from collections import Counter from transformers import AutoTokenizer ROOT = pathlib.Path(__file__).resolve().parent.parent OUT = ROOT / "results" / "repr_probe.json" REPO = "openbpm/MiniCPM5-2B".replace("openbpm", "openbmb") TEXTS = { "en": "A traveller with no water performs tayammum by striking the hands on bare stone, and this is valid in the view of the school.", "zh": "一名没有水的旅行者在裸石上拍打双手进行土净礼，这在教派的观点中是被允许的。", "ar_vocalized": "السَّابِحُ الَّذِي لَا مَاءَ عِنْدَهُ يَتَضَمَّضُ بِالْحِجَارَةِ وَيَجُوزُ فِي مَذْهَبِ الْفَقِيهِ.", "ar_plain": "السابح الذي لا ماء عنده يتضض بالحجارة ويجوز في مذهب الفقيه.", "ur": "سفر کرنے والا مسافر جب پر پتھر پر ہاتھ مارنا اور مذہب کے مطابق یہ جائز ہے۔", "fa": "مسافری که آب ندارد بر سنگ خام با دستهای خود تیمم می‌کند و در مذهب فقیه جایز است.", "hi": "एक यात्री जिसके पास पानी नहीं है वह नंगे पत्थर पर अपने हाथ मारता है और मजहब के अनुसार यह जायज है।", "arabizi": "Al musafir ila laa maa lahu maa yatamassabu bi al-hijarati wa yajuzu fi madhhab al-faqih.", } TERMS = [ ("صَلَاة", "صلاه"), ("زَكَاة", "زكاه"), ("تَيَمُّم", "تيمم"), ("وُضُوء", "وضوء"), ("رَكْعَة", "ركعه"), ("سُجُود", "سجود"), ("خُطْبَة", "خطبه"), ("جَنَازَة", "جنازه"), ("مَذْهَب", "مذهب"), ("حَجّ", "حج"), ("قَصْر", "قصر"), ("نِيَّة", "نية"), ] HARAKAT_PAIRS = [ ("نِيَّة", "نَيَّة"), ("زَكَاة", "زِكَاة"), ("حَلَال", "حِلَال"), ("رَكْعَة", "رَكَعَة"), ("طَهَارَة", "طُهَارَة"), ] HARAKAT = set("ًٌٍَُِّْٰٕٓٔ") ENGLISH_EQUIV = [("prayer", "صلاة"), ("purification", "طهارة"), ("zakat", "زكاة"), ("fasting", "صوم"), ("pilgrimage", "حج")] # ------------------------------------------------------- byte-level alphabet -- def bytes_to_unicode() -> dict[int, str]: bs = (list(range(ord("!"), ord("~") + 1)) + list(range(ord("\xa1"), ord("\xac") + 1)) + list(range(ord("\xae"), ord("\xff") + 1))) cs = bs[:] n = 0 for b in range(256): if b not in bs: bs.append(b) cs.append(256 + n) n += 1 return dict(zip(bs, (chr(c) for c in cs))) B2U = bytes_to_unicode() U2B = {v: k for k, v in B2U.items()} def token_to_bytes(tok_str: str) -> bytes | None: try: return bytes(U2B[c] for c in tok_str) except KeyError: return None def is_byte_token(tok_str: str) -> bool: """A token is byte-level iff its characters decode to bytes that are NOT valid UTF-8 on their own -- i.e. it is a fragment of a character rather than a character (or a word) in the vocabulary.""" b = token_to_bytes(tok_str) if b is None: return False try: b.decode("utf-8") return False except UnicodeDecodeError: return True def byte_token_to_text(tok_str: str) -> str: b = token_to_bytes(tok_str) if b is None: return "" return b.decode("utf-8", errors="replace") def is_arabic_letter(ch: str) -> bool: return "؀" <= ch <= "ۿ" def longest_arabic_run(s: str) -> int: best = run = 0 for ch in s: run = run + 1 if is_arabic_letter(ch) else 0 best = max(best, run) return best def token_texts(tok, ids): """Decode each token id back to the text it contributes.""" out = [] for i in ids: t = tok.convert_ids_to_tokens(i) if is_byte_token(t): out.append(byte_token_to_text(t)) else: out.append(t.replace("Ġ", " ")) return out # ----------------------------------------------------------------- measures -- def fertility(tok, text: str) -> dict: ids = tok(text, add_special_tokens=False)["input_ids"] toks = [tok.convert_ids_to_tokens(i) for i in ids] n_byte = sum(1 for t in toks if is_byte_token(t)) words = len(text.split()) chars = len([c for c in text if not c.isspace()]) return { "n_tokens": len(ids), "n_words": words, "n_chars": chars, "tokens_per_word": round(len(ids) / max(words, 1), 3), "tokens_per_char": round(len(ids) / max(chars, 1), 4), "byte_level_tokens": n_byte, "byte_level_rate": round(n_byte / max(len(ids), 1), 4), } def vocab_script_profile(tok) -> dict: vocab = tok.get_vocab() n_total = len(vocab) n_byte = 0 n_text = 0 n_arabic_units = 0 n_latin_units = 0 n_cjk_units = 0 n_devanagari_units = 0 n_other_units = 0 run_hist: Counter[int] = Counter() for tok_str in vocab: if is_byte_token(tok_str): n_byte += 1 continue n_text += 1 s = tok_str if any(c in U2B for c in tok_str): s = byte_token_to_text(tok_str) s = s.replace("Ġ", " ") if any(is_arabic_letter(c) for c in s): n_arabic_units += 1 run_hist[longest_arabic_run(s)] += 1 elif any("a" <= c <= "z" or "A" <= c <= "Z" for c in s): n_latin_units += 1 elif any(0x4E00 <= ord(c) <= 0x9FFF for c in s): n_cjk_units += 1 elif any(0x0900 <= ord(c) <= 0x097F for c in s): n_devanagari_units += 1 else: n_other_units += 1 tot = max(n_arabic_units, 1) return { "vocab_size": n_total, "n_byte_level_tokens": n_byte, "share_byte_level": round(n_byte / n_total, 4), "n_text_unit_tokens": n_text, "n_arabic_text_units": n_arabic_units, "share_arabic_text_units": round(n_arabic_units / n_total, 6), "n_latin_text_units": n_latin_units, "n_cjk_text_units": n_cjk_units, "n_devanagari_text_units": n_devanagari_units, "n_other_text_units": n_other_units, "arabic_unit_run_histogram": {str(k): v for k, v in sorted(run_hist.items())}, "max_arabic_chars_in_one_unit": max(run_hist) if run_hist else 0, "mean_arabic_chars_per_unit": round( sum(k * v for k, v in run_hist.items()) / tot, 3), } def text_of_ids(tok, ids) -> str: """Reconstruct the original string from token ids by joining their bytes.""" buf = bytearray() for i in ids: t = tok.convert_ids_to_tokens(i) b = token_to_bytes(t) if b is None: buf += t.encode("utf-8") else: buf += b return buf.decode("utf-8", errors="replace") def token_char_attribution(tok, ids) -> list[str]: """Attribute decoded characters to the token that completed them. A byte-level token may hold half a character; the character is attributed to whichever token finally completes it. This lets us ask, for example, whether a short vowel ever rides along with its letter inside one token. """ pending = b"" out = [] for i in ids: t = tok.convert_ids_to_tokens(i) b = token_to_bytes(t) if b is None: b = t.encode("utf-8") pending += b got = "" while pending: try: got += pending.decode("utf-8") pending = b"" except UnicodeDecodeError as e: if e.reason == "unexpected end of data": break got += "�" pending = pending[e.end:] out.append(got) return out def term_morphology(tok) -> dict: rows = [] for voc, plain in TERMS: vi = tok(voc, add_special_tokens=False)["input_ids"] pi = tok(plain, add_special_tokens=False)["input_ids"] vt = token_char_attribution(tok, vi) pt = token_char_attribution(tok, pi) runs = [longest_arabic_run(t) for t in vt] fused = sum(1 for t in vt if any(is_arabic_letter(c) for c in t) and any(c in HARAKAT for c in t)) detached_v = sum(1 for t in vt if t and all(c in HARAKAT for c in t)) rows.append({ "vocalized": voc, "plain": plain, "n_tokens_vocalized": len(vi), "n_tokens_plain": len(pi), "diacritic_cost_ratio": round(len(vi) / max(len(pi), 1), 3), "extra_tokens_for_vocalization": len(vi) - len(pi), "max_arabic_chars_in_one_token": max(runs) if runs else 0, "n_tokens_fusing_letter_and_vowel": fused, "n_tokens_that_are_only_a_vowel": detached_v, "tokens_vocalized": vt, "tokens_plain": pt, }) return { "per_term": rows, "n_terms": len(rows), "mean_tokens_plain": round(sum(r["n_tokens_plain"] for r in rows) / len(rows), 2), "mean_tokens_vocalized": round(sum(r["n_tokens_vocalized"] for r in rows) / len(rows), 2), "mean_diacritic_cost_ratio": round( sum(r["diacritic_cost_ratio"] for r in rows) / len(rows), 3), "max_arabic_chars_in_one_token": max( r["max_arabic_chars_in_one_token"] for r in rows), "n_terms_with_fused_letter_and_vowel": sum( r["n_tokens_fusing_letter_and_vowel"] > 0 for r in rows), "n_terms_with_detached_vowel_tokens": sum( r["n_tokens_that_are_only_a_vowel"] > 0 for r in rows), } def diacritic_cost(tok) -> dict: fused = detached = total = 0 examples = [] for voc, _ in TERMS: for t in token_char_attribution(tok, tok(voc, add_special_tokens=False)["input_ids"]): if any(c in HARAKAT for c in t): total += 1 if any(is_arabic_letter(c) for c in t): fused += 1 else: detached += 1 if len(examples) < 8: examples.append(t) return { "tokens_containing_a_harakat": total, "fused_with_letter": fused, "detached_harakat_tokens": detached, "share_detached": round(detached / max(total, 1), 4), "examples_detached": examples, } def diacritic_discrimination(tok) -> dict: rows = [] for a, b in HARAKAT_PAIRS: ai = tok(a, add_special_tokens=False)["input_ids"] bi = tok(b, add_special_tokens=False)["input_ids"] rows.append({"a": a, "b": b, "a_tokens": ai, "b_tokens": bi, "identical": ai == bi}) return { "pairs": rows, "n_pairs": len(rows), "n_merged": sum(1 for r in rows if r["identical"]), "merge_rate": round(sum(1 for r in rows if r["identical"]) / len(rows), 3), } def english_equivalent_cost(tok) -> list[dict]: rows = [] for en, ar in ENGLISH_EQUIV: e = tok(en, add_special_tokens=False)["input_ids"] a = tok(ar, add_special_tokens=False)["input_ids"] rows.append({"english": en, "arabic": ar, "english_tokens": len(e), "arabic_tokens": len(a), "arabic_over_english": round(len(a) / max(len(e), 1), 2)}) return rows def main() -> None: tok = AutoTokenizer.from_pretrained(REPO, trust_remote_code=True) report = { "repo": REPO, "vocab_size": len(tok), "fertility_by_script": {k: fertility(tok, v) for k, v in TEXTS.items()}, "vocab_script_profile": vocab_script_profile(tok), "term_morphology": term_morphology(tok), "diacritic_cost": diacritic_cost(tok), "diacritic_discrimination": diacritic_discrimination(tok), "english_equivalent_cost": english_equivalent_cost(tok), } OUT.parent.mkdir(parents=True, exist_ok=True) OUT.write_text(json.dumps(report, ensure_ascii=False, indent=2)) print("=== fertility (byte-level share = share of tokens that are raw bytes) ===") for k, v in report["fertility_by_script"].items(): print(f" {k:14s} tok/char={v['tokens_per_char']:.4f} tok/word={v['tokens_per_word']:6.2f}" f" byte-level={v['byte_level_rate']*100:5.1f}%") vp = report["vocab_script_profile"] print("\n=== vocabulary profile ===") print(f" vocab={vp['vocab_size']} byte-level tokens={vp['n_byte_level_tokens']} " f"({vp['share_byte_level']*100:.1f}%)") print(f" tokens containing Arabic: {vp['n_arabic_text_units']} " f"({vp['share_arabic_text_units']*100:.3f}%) Latin {vp['n_latin_text_units']} " f"CJK {vp['n_cjk_text_units']}") print(f" max Arabic characters in one unit: {vp['max_arabic_chars_in_one_unit']} " f"mean {vp['mean_arabic_chars_per_unit']}") tm = report["term_morphology"] print("\n=== religious terms ===") print(f" mean tokens: plain {tm['mean_tokens_plain']} vocalised {tm['mean_tokens_vocalized']} " f"(ratio {tm['mean_diacritic_cost_ratio']})") print(f" max Arabic characters in one token across terms: {tm['max_arabic_chars_in_one_token']}") print(f" terms where any token fuses a letter with its vowel: " f"{tm['n_terms_with_fused_letter_and_vowel']}/{tm['n_terms']}") dc = report["diacritic_cost"] print(f" harakat tokens: {dc['tokens_containing_a_harakat']} detached: " f"{dc['detached_harakat_tokens']} ({dc['share_detached']*100:.0f}%)") dd = report["diacritic_discrimination"] print(f" harakat-only pairs merged by the tokenizer: {dd['n_merged']}/{dd['n_pairs']}") print("\n=== same content, two scripts ===") for r in report["english_equivalent_cost"]: print(f" {r['english']:14s} {r['english_tokens']} tok | {r['arabic']} " f"{r['arabic_tokens']} tok ({r['arabic_over_english']}x)") if __name__ == "__main__": main()