文字から単語へ：日本語BPEトークナイザーの構築

# 日本語ニュースコーパスのダウンロードと読み込み
import tarfile
import urllib.request
import os

filename = "ldcc-20140209.tar.gz"
if not os.path.exists(filename):
    print("ダウンロード中...")
    urllib.request.urlretrieve("https://www.rondhuit.com/download/ldcc-20140209.tar.gz", filename)

text = ""
articles = []
article_count_limit = 1000

with tarfile.open(filename, "r") as tar:
    for member in tar.getmembers():
        if member.isfile() and member.name.endswith(".txt"):
            file_content = tar.extractfile(member).read().decode("utf-8")
            lines = file_content.split("\n")
            if len(lines) >= 4:  # URL、日付、タイトルをスキップ
                content = "\n".join(lines[3:]).strip()
                if content:
                    articles.append(content)
                    text += content + "\n\n"

print(f"テキスト長: {len(text):,}")
print(f"記事数: {len(articles):,}")
print(f"初期語彙サイズ: {len(set(text)):,}")

def get_pairs(tokens):
    """隣接する文字ペアとその頻度を見つける"""
    pairs = defaultdict(int)
    for i in range(len(tokens) - 1):
        pairs[(tokens[i], tokens[i + 1])] += 1
    return pairs

def merge_tokens(tokens, pair):
    """最も頻度の高いペアを単一のトークンにマージ"""
    new_tokens = []
    i = 0
    while i < len(tokens):
        if (i < len(tokens) - 1 and 
            tokens[i] == pair[0] and tokens[i + 1] == pair[1]):
            new_tokens.append(pair[0] + pair[1])  # ペアをマージ
            i += 2
        else:
            new_tokens.append(tokens[i])
            i += 1
    return new_tokens

# 文字レベルのトークンで初期化
tokens = list(text)  # 個別の文字に分割
vocab = set(tokens)
target_vocab_size = 5000

while len(vocab) < target_vocab_size:
    # 最も頻度の高い隣接ペアを見つける
    pairs = get_pairs(tokens)
    if not pairs:
        break
    
    best_pair = max(pairs, key=pairs.get)
    
    # このペアのすべてのインスタンスをマージ
    tokens = merge_tokens(tokens, best_pair)
    vocab.add("".join(best_pair))
    
    if merge_count % 100 == 0:
        print(f"マージ {merge_count}: 語彙 {len(vocab)}, トークン {len(tokens)}")

def tokenize_with_bpe(text):
    """学習したBPEマージを新しいテキストに適用"""
    tokens = list(text)  # 文字から開始
    changed = True
    
    while changed:
        changed = False
        new_tokens = []
        i = 0
        while i < len(tokens):
            # 隣接ペアが語彙に存在するかチェック
            if (i < len(tokens) - 1 and 
                tokens[i] + tokens[i + 1] in vocab):
                new_tokens.append(tokens[i] + tokens[i + 1])
                i += 2
                changed = True
            else:
                new_tokens.append(tokens[i])
                i += 1
        tokens = new_tokens
    return tokens

# 学習結果:
テキスト長: 659,423
記事数: 1,000
初期語彙サイズ: 2,913
BPE後語彙サイズ: 5,000

# 圧縮結果:
元の文字数: 659,423
BPEトークン: 379,352
圧縮率: 42.5%