Tokenizer - LLM がテキストを数える単位

"""バイト単位の小さな BPE を学習し、結合が進む様子を表示する（標準ライブラリだけ）。

使い方:
    git show 9bbbeb23:content/notes/blockchain-systems/wallet/index.md > wallet.md
    python3 bpe_train.py wallet.md [結果の JSON]
"""
import json
import re
import sys
from collections import Counter

from texts import SHORT, TEXTS

MERGES = 1000
CHECKPOINTS = [0, 10, 50, 100, 300, 1000]


def pairs_of(word):
    return zip(word, word[1:])


def merge(word, pair, new):
    out, i = [], 0
    while i < len(word):
        if i + 1 < len(word) and (word[i], word[i + 1]) == pair:
            out.append(new)
            i += 2
        else:
            out.append(word[i])
            i += 1
    return tuple(out)


def train(text, n):
    # 空白・記号で区切った断片の中だけで結合する（断片をまたぐトークンは作らない）
    pieces = Counter(re.findall(r"\w+|[^\w\s]+|\s+", text))
    words = {tuple(p.encode()): c for p, c in pieces.items()}  # 最初は 1 バイト = 1 記号
    vocab = {i: bytes([i]) for i in range(256)}
    merges = []
    for _ in range(n):
        counts = Counter()
        for w, c in words.items():
            for p in pairs_of(w):
                counts[p] += c
        if not counts:
            break
        # 最も多く隣り合う組を 1 つの新しい記号にする（同数なら記号の番号が小さい組）
        pair = min(counts, key=lambda p: (-counts[p], p))
        new = len(vocab)
        vocab[new] = vocab[pair[0]] + vocab[pair[1]]
        merges.append((pair, new, counts[pair]))
        words = {merge(w, pair, new): c for w, c in words.items()}
    return vocab, merges


def encode(s, merges):
    rank = {pair: (r, new) for r, (pair, new, _) in enumerate(merges)}
    out = []
    for piece in re.findall(r"\w+|[^\w\s]+|\s+", s):
        w = tuple(piece.encode())
        while True:  # 学習で早く覚えた組から順に結合する
            cands = [rank[p] for p in pairs_of(w) if p in rank]
            if not cands:
                break
            r, new = min(cands)
            w = merge(w, merges[r][0], new)
        out.extend(w)
    return out


def show(b):
    return b.decode("utf-8", "backslashreplace")


def main():
    text = open(sys.argv[1], encoding="utf-8").read()
    vocab, merges = train(text, MERGES)
    print(f"corpus: {len(text):,} chars, {len(text.encode()):,} bytes")
    print("first merges (new symbol <- pair, count)")
    for k, (pair, new, c) in enumerate(merges[:12], 1):
        a, b = (show(vocab[x]) for x in pair)
        print(f"  #{k:<3} [{show(vocab[new])}] <- [{a}] + [{b}]  ({c})")
    for m in CHECKPOINTS:
        print(f"\nafter {m} merges")
        for key in ("ja", "en"):
            ids = encode(SHORT[key], merges[:m])
            print(f"  {key} {len(ids):>3}  " + "|".join(show(vocab[i]) for i in ids))
    # グラフ用: Wallet の 2 文と英訳のトークン数を、結合の回数ごとに数える
    steps = list(range(0, MERGES + 1, 25))
    curve = {k: [len(encode(TEXTS[k], merges[:m])) for m in steps] for k in ("ja", "en")}
    print("\nWallet passage: " + ", ".join(
        f"{k} {v[0]} -> {v[-1]} tokens" for k, v in curve.items()))
    if len(sys.argv) > 2:
        with open(sys.argv[2], "w") as f:
            json.dump({"merges": steps, "tokens": curve}, f)


if __name__ == "__main__":
    main()

"""count_tokens.py と bpe_train.py が使う文。"""

# 日本語は debimate の Wallet ノート冒頭の 2 文。英語は本記事のために作った訳。
TEXTS = {
    "ja": "ウォレット（wallet）は、Bitcoin などの暗号資産を受け取り、残高を確認し、"
          "他の相手に送るためのソフトウェアです。PC やスマートフォンの画面で操作するタイプと、"
          "コマンドを入力して操作するタイプがあります。",
    "en": "A wallet is software for receiving crypto assets such as Bitcoin, checking "
          "the balance, and sending them to others. Some wallets are operated on the "
          "screen of a PC or smartphone, and others by typing commands.",
    # LoRA ノートの lab/count_params.py の関数
    "code": 'def read_header(path):\n'
            '    with open(path, "rb") as f:\n'
            '        (length,) = struct.unpack("<Q", f.read(8))\n'
            '        header = json.loads(f.read(length))\n'
            '    metadata = header.pop("__metadata__", {})\n'
            '    return length, header, metadata\n',
    "digits": "2026-11-10 14:05:09 load average: 100.70, 226.82, 290.79 "
              "size 6938040682 bytes",
    "emoji": "👍 👍🏽 👨‍👩‍👧 🇯🇵 🦉",
}
# 冒頭の図に使う短い文（区切りの位置を出力する）
SHORT = {"ja": "秘密鍵を安全に保管する", "en": "Store the private key safely",
         "digits": "6938040682", "emoji": "👍🏽👨‍👩‍👧"}

$ python3 bpe_train.py wallet.md bpe.json
corpus: 37,097 chars, 67,916 bytes
first merges (new symbol <- pair, count)
  #1   [\xe3\x81] <- [\xe3] + [\x81]  (4631)
  #2   [\xe3\x82] <- [\xe3] + [\x82]  (2324)
  #3   [\xe3\x83] <- [\xe3] + [\x83]  (1718)
  #4   [\xe3\x80] <- [\xe3] + [\x80]  (929)
  #5   [の] <- [\xe3\x81] + [\xae]  (687)
  #6   [、] <- [\xe3\x80] + [\x81]  (487)
  #7   [す] <- [\xe3\x81] + [\x99]  (477)
  #8   [  ] <- [ ] + [ ]  (448)
  #9   [を] <- [\xe3\x82] + [\x92]  (437)
  #10  [or] <- [o] + [r]  (380)
  #11  [。] <- [\xe3\x80] + [\x82]  (374)
  #12  [で] <- [\xe3\x81] + [\xa7]  (349)

after 0 merges
  ja  33  \xe7|\xa7|\x98|\xe5|\xaf|\x86|\xe9|\x8d|\xb5|\xe3|\x82|\x92|\xe5|\xae|\x89|\xe5|\x85|\xa8|\xe3|\x81|\xab|\xe4|\xbf|\x9d|\xe7|\xae|\xa1|\xe3|\x81|\x99|\xe3|\x82|\x8b
  en  28  S|t|o|r|e| |t|h|e| |p|r|i|v|a|t|e| |k|e|y| |s|a|f|e|l|y

after 10 merges
  ja  27  \xe7|\xa7|\x98|\xe5|\xaf|\x86|\xe9|\x8d|\xb5|を|\xe5|\xae|\x89|\xe5|\x85|\xa8|\xe3\x81|\xab|\xe4|\xbf|\x9d|\xe7|\xae|\xa1|す|\xe3\x82|\x8b
  en  27  S|t|or|e| |t|h|e| |p|r|i|v|a|t|e| |k|e|y| |s|a|f|e|l|y

after 50 merges
  ja  23  \xe7|\xa7|\x98|\xe5|\xaf|\x86|\xe9|\x8d|\xb5|を|\xe5|\xae|\x89|\xe5\x85|\xa8|に|\xe4|\xbf|\x9d|\xe7|\xae|\xa1|する
  en  27  S|t|or|e| |t|h|e| |p|r|i|v|a|t|e| |k|e|y| |s|a|f|e|l|y

after 100 merges
  ja  19  \xe7|\xa7|\x98|\xe5\xaf|\x86|鍵|を|\xe5\xae|\x89|\xe5\x85|\xa8|に|\xe4|\xbf|\x9d|\xe7|\xae|\xa1|する
  en  26  S|t|or|e| |t|h|e| |p|r|i|v|a|te| |k|e|y| |s|a|f|e|l|y

after 300 merges
  ja  13  \xe7|\xa7\x98密|鍵|を|\xe5\xae|\x89|\xe5\x85|\xa8|に|保|\xe7|\xae\xa1|する
  en  23  S|t|or|e| |th|e| |pr|i|v|a|te| |k|e|y| |s|af|e|l|y

after 1000 merges
  ja   8  秘密鍵|を|\xe5\xae|\x89|全|に|保管|する
  en  19  S|t|or|e| |the| |pr|i|v|a|te| |ke|y| |safe|l|y

Wallet passage: ja 290 -> 66 tokens, en 209 -> 136 tokens

"""同じ内容の文を 4 つのトークナイザーで区切り、文字数・バイト数・トークン数を比べる。

使い方:
    TIKTOKEN_CACHE_DIR=<保存先>/tiktoken \
      uv run --with tiktoken==0.12.0 --with tokenizers==0.22.1 \
      python count_tokens.py <fetch.sh の保存先> [結果の JSON]
"""
import json
import re
import sys

import tiktoken
from tokenizers import Tokenizer

from texts import SHORT, TEXTS


def byte_level_table():
    """GPT-2 系の ByteLevel が 1 バイトを 1 文字に写す表の逆引き。"""
    bs = list(range(33, 127)) + list(range(161, 173)) + list(range(174, 256))
    cs = bs[:]
    n = 0
    for b in range(256):
        if b not in bs:
            bs.append(b)
            cs.append(256 + n)
            n += 1
    return {chr(c): b for b, c in zip(bs, cs)}


def load(root):
    inv = byte_level_table()

    def tk(enc):
        def f(s):
            ids = enc.encode(s, disallowed_special=())
            return [enc.decode_single_token_bytes(i) for i in ids]
        return f

    def qwen(t):  # ByteLevel: トークンの文字列を元のバイト列に戻す
        def f(s):
            toks = t.encode(s, add_special_tokens=False).tokens
            return [bytes(inv[c] for c in tok) for tok in toks]
        return f

    def spm(t):  # SentencePiece 形式: <0xE3> はバイト、▁ は空白
        def f(s):
            out = []
            for tok in t.encode(s, add_special_tokens=False).tokens:
                m = re.fullmatch(r"<0x([0-9A-F]{2})>", tok)
                out.append(bytes([int(m[1], 16)]) if m else tok.replace("\u2581", " ").encode())
            return out
        return f

    return {
        "o200k_base": tk(tiktoken.get_encoding("o200k_base")),
        "cl100k_base": tk(tiktoken.get_encoding("cl100k_base")),
        "Qwen3": qwen(Tokenizer.from_file(f"{root}/qwen3/tokenizer.json")),
        "LLM-jp-3": spm(Tokenizer.from_file(f"{root}/llmjp3/tokenizer.json")),
    }


def main():
    root = sys.argv[1]
    out_path = sys.argv[2] if len(sys.argv) > 2 else None
    toks = load(root)
    result = {"counts": {}, "short": {}}

    print(f"{'text':<7}{'chars':>6}{'bytes':>6}  " +
          "".join(f"{n:>12}" for n in toks))
    for key, s in TEXTS.items():
        row = {"chars": len(s), "bytes": len(s.encode())}
        for name, fn in toks.items():
            row[name] = len(fn(s))
        result["counts"][key] = row
        print(f"{key:<7}{row['chars']:>6}{row['bytes']:>6}  " +
              "".join(f"{row[n]:>12}" for n in toks))

    print("\nchars per token")
    for key in ("ja", "en"):
        row = result["counts"][key]
        print(f"{key:<7}" + "".join(f"{row['chars'] / row[n]:>12.2f}" for n in toks))

    print("\nsplit of short sentences ('|' = token boundary, \\xNN = part of a char)")
    for key, s in SHORT.items():
        for name, fn in toks.items():
            ps = fn(s)
            shown = "|".join(p.decode("utf-8", "backslashreplace") for p in ps)
            result["short"].setdefault(key, {})[name] = [list(p) for p in ps]
            print(f"{key} {name:<12}{len(ps):>3}  {shown}")

    if out_path:
        with open(out_path, "w") as f:
            json.dump(result, f, indent=1, ensure_ascii=False)


if __name__ == "__main__":
    main()

#!/bin/sh
# トークナイザーのファイルだけを取得する（モデルの重みは取得しない）。
# 使い方: sh fetch.sh <保存先>
set -eu
D=${1:?保存先のディレクトリを指定する}
mkdir -p "$D/qwen3" "$D/llmjp3" "$D/tiktoken"
# Qwen3（Qwen/Qwen3-4B、revision 1cfa9a7208912126459214e8b04321603b3df60c）
curl -sSL -o "$D/qwen3/tokenizer.json" \
  https://huggingface.co/Qwen/Qwen3-4B/resolve/1cfa9a7208912126459214e8b04321603b3df60c/tokenizer.json
# LLM-jp-3（llm-jp/llm-jp-3-1.8b、revision 60acda32e07c72b82ce5a73d60b21b3d65e0c810）
curl -sSL -o "$D/llmjp3/tokenizer.json" \
  https://huggingface.co/llm-jp/llm-jp-3-1.8b/resolve/60acda32e07c72b82ce5a73d60b21b3d65e0c810/tokenizer.json
# tiktoken はこの置き場所をキャッシュとして使い、無ければ自分で取得してハッシュを検証する
echo "TIKTOKEN_CACHE_DIR=$D/tiktoken"
sha256sum "$D/qwen3/tokenizer.json" "$D/llmjp3/tokenizer.json"