【トークナイザー完全解説】仕組みから実装まで自然言語処理の基盤技術

入力: "東京エレクトロンデバイスは、日本の会社です"
出力: ['東京', 'エレクト', 'ロン', 'デバイス', 'は', '、', '日本の', '会社', 'です']

# 初期状態（文字レベル）
文字: ['h', 'u', 'g', 's']
語彙: ['h', 'u', 'g', 's']

# Step 1: "u"+"g" が最頻出 → "ug"を作成
文字: ['h', 'ug', 's'] 
語彙: ['h', 'u', 'g', 's', 'ug']

# Step 2: "h"+"ug" が最頻出 → "hug"を作成
文字: ['hug', 's']
語彙: ['h', 'u', 'g', 's', 'ug', 'hug']

# BPEの場合
選択基準: 頻度が最も高いペア

# WordPieceの場合  
選択基準: P(AB) / (P(A) × P(B)) が最大のペア

# WordPieceのトークン表現
トークン例: ['Tra', '##nsform', '##ers']
# ##マークで非語頭を示す

# スペースの扱い
入力: "Hello World"
内部表現: "Hello▁World"
# ▁（U+2581）でスペースを表現

# BPE/WordPiece: 小さい語彙から拡張
初期語彙 → 徐々に拡張 → 最終語彙

# Unigram: 大きい語彙から削減
初期語彙（大） → 徐々に削減 → 最終語彙

# 可逆的な変換が可能
元テキスト ⟷ トークン列
# 情報の損失なし

# 英語の例
"Hello world" → ["Hello", "world"]  # 明確な区切り

# 日本語の例
"こんにちは世界" → ?
# 考えられる分割:
# ["こんにちは", "世界"]
# ["こん", "にちは", "世界"]  
# ["こ", "ん", "に", "ち", "は", "せ", "か", "い"]

文字体系の例:
ひらがな: "ひらがな"    # 音韻文字
カタカナ: "カタカナ"    # 外来語表記
漢字:     "漢字"       # 表意文字
混合:     "日本語処理"  # 複合語

語根: "食べ"
活用例: ["食べる", "食べた", "食べない", "食べられる", "食べさせる"]

# MeCab + SentencePiece のアプローチ
import MeCab
import sentencepiece as spm

# Step 1: 形態素解析で単語分割
wakati = MeCab.Tagger("-O wakati")
words = wakati.parse("自然言語処理は面白い").strip().split()
# ['自然', '言語', '処理', 'は', '面白い']

# Step 2: SentencePieceでサブワード化
sp = spm.SentencePieceProcessor()
tokens = sp.encode_as_pieces(" ".join(words))

# NovelAI Tokenizerを採用
# 日本語と英語を効率的に処理
語彙サイズ: 約100,000
対応言語: 日本語 + 英語
特徴: 日本語の語彙を大幅に拡張

# 既存モデルの語彙拡張
ベースモデル: Llama2 (英語中心)
拡張後: 日本語語彙を追加
学習データ: 約180億トークンの日本語テキスト

from transformers import AutoTokenizer

# BERT日本語モデルのトークナイザー
tokenizer = AutoTokenizer.from_pretrained("cl-tohoku/bert-base-japanese-v2")

# トークン化の実行
text = "自然言語処理は面白い技術です。"
tokens = tokenizer.tokenize(text)
print(f"トークン: {tokens}")
# ['自然', '言語', '処理', 'は', '面白い', '技術', 'です', '。']

# ID変換
input_ids = tokenizer.encode(text)
print(f"ID: {input_ids}")
# [2, 12345, 23456, 34567, ...]

models = [
    "bert-base-uncased",           # 英語BERT（WordPiece）
    "roberta-base",                # 英語RoBERTa（BPE）  
    "cl-tohoku/bert-base-japanese-v2", # 日本語BERT
    "microsoft/graphcodebert-base" # コード用BERT
]

text = "Hello, 世界! def hello_world():"

for model_name in models:
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    tokens = tokenizer.tokenize(text)
    print(f"{model_name}: {tokens}")

texts = [
    "機械学習は人工知能の一分野です。",
    "深層学習はニューラルネットワークを使用します。",
    "Transformerアーキテクチャが注目されています。"
]

# バッチエンコーディング
encoded = tokenizer(
    texts,
    padding=True,           # 長さを揃える
    truncation=True,        # 最大長で切り詰め
    max_length=128,         # 最大長設定
    return_tensors="pt"     # PyTorchテンソルで返す
)

print(f"Input IDs shape: {encoded['input_ids'].shape}")
print(f"Attention mask: {encoded['attention_mask'].shape}")

from datasets import load_dataset

# WikiTextデータセットの使用
dataset = load_dataset("wikitext", "wikitext-103-raw-v1", split="train")

def get_training_corpus():
    """学習用コーパスのジェネレータ"""
    for i in range(0, len(dataset), 1000):
        yield dataset[i:i+1000]["text"]

from tokenizers import (
    Tokenizer,
    models,
    pre_tokenizers,
    decoders,
    trainers,
    processors,
)

# BPEモデルの初期化
tokenizer = Tokenizer(models.BPE(unk_token="<unk>"))

# 前処理設定
tokenizer.pre_tokenizer = pre_tokenizers.ByteLevel(add_prefix_space=False)

# トレーナー設定
trainer = trainers.BpeTrainer(
    vocab_size=30000,                    # 語彙サイズ
    min_frequency=2,                     # 最小出現頻度
    special_tokens=["<unk>", "<pad>", "<s>", "</s>"]
)

# 学習実行
tokenizer.train_from_iterator(get_training_corpus(), trainer)

# WordPieceモデルの初期化
tokenizer = Tokenizer(models.WordPiece(unk_token="[UNK]"))

# 前処理（BERT風）
tokenizer.pre_tokenizer = pre_tokenizers.BertPreTokenizer()

# トレーナー設定
trainer = trainers.WordPieceTrainer(
    vocab_size=30000,
    special_tokens=["[UNK]", "[CLS]", "[SEP]", "[PAD]", "[MASK]"]
)

tokenizer.train_from_iterator(get_training_corpus(), trainer)

# Fast tokenizerでのオフセット取得
tokenizer = AutoTokenizer.from_pretrained("bert-base-cased", use_fast=True)

text = "Hello world!"
encoding = tokenizer(text, return_offsets_mapping=True)

print("トークン -> 元テキスト位置:")
for i, (start, end) in enumerate(encoding.offset_mapping):
    token = tokenizer.convert_ids_to_tokens(encoding.input_ids[i])
    original_text = text[start:end] if start != end else ""
    print(f"{token} -> '{original_text}' [{start}:{end}]")

# 特別トークンの設定
special_tokens_dict = {
    "pad_token": "<pad>",
    "unk_token": "<unk>",
    "bos_token": "<s>",
    "eos_token": "</s>",
    "sep_token": "<sep>",
    "cls_token": "<cls>",
    "mask_token": "<mask>"
}

# 特別トークンの追加
tokenizer.add_special_tokens(special_tokens_dict)

# カスタムトークンの追加
new_tokens = ["<PERSON>", "<DATE>", "<LOCATION>"]
tokenizer.add_tokens(new_tokens)

import time
from transformers import AutoTokenizer

def measure_tokenization_speed(tokenizer, texts, num_runs=100):
    """トークン化速度の測定"""
    start_time = time.time()
    
    for _ in range(num_runs):
        for text in texts:
            tokens = tokenizer.tokenize(text)
    
    elapsed_time = time.time() - start_time
    return elapsed_time / (num_runs * len(texts))

# 異なるトークナイザーでの性能比較
tokenizers = {
    "BERT": "bert-base-uncased",
    "RoBERTa": "roberta-base", 
    "Japanese BERT": "cl-tohoku/bert-base-japanese-v2"
}

test_texts = [
    "自然言語処理は機械学習の重要な分野です。",
    "Deep learning has revolutionized artificial intelligence.",
    "Transformer models are state-of-the-art in NLP tasks."
] * 100

for name, model in tokenizers.items():
    tokenizer = AutoTokenizer.from_pretrained(model)
    speed = measure_tokenization_speed(tokenizer, test_texts)
    print(f"{name}: {speed:.4f}秒/テキスト")

# 未使用語彙の削除
def compress_vocabulary(tokenizer, corpus):
    """実際に使用される語彙のみを保持"""
    used_tokens = set()
    
    for text in corpus:
        tokens = tokenizer.tokenize(text)
        used_tokens.update(tokens)
    
    # 新しい語彙作成
    compressed_vocab = {token: i for i, token in enumerate(used_tokens)}
    return compressed_vocab

# トークンIDの量子化
import numpy as np

def quantize_token_ids(token_ids, bits=16):
    """トークンIDを低精度で表現"""
    max_id = max(token_ids)
    
    if bits == 16:
        dtype = np.uint16
        max_val = 65535
    elif bits == 8:
        dtype = np.uint8  
        max_val = 255
    
    if max_id <= max_val:
        return np.array(token_ids, dtype=dtype)
    else:
        raise ValueError(f"語彙サイズが{bits}bitを超えています")

def calculate_compression_ratio(original_text, tokens):
    """圧縮率の計算"""
    original_chars = len(original_text)
    token_count = len(tokens)
    return original_chars / token_count

# 例
text = "自然言語処理の研究開発"
tokens = tokenizer.tokenize(text)
ratio = calculate_compression_ratio(text, tokens)
print(f"圧縮率: {ratio:.2f} (文字数/トークン数)")

def calculate_oov_rate(tokenizer, test_corpus):
    """Out-of-Vocabulary率の計算"""
    total_tokens = 0
    oov_tokens = 0
    
    for text in test_corpus:
        tokens = tokenizer.tokenize(text)
        total_tokens += len(tokens)
        
        for token in tokens:
            if token == tokenizer.unk_token:
                oov_tokens += 1
                
    return oov_tokens / total_tokens if total_tokens > 0 else 0

def vocabulary_overlap(tokenizer1, tokenizer2):
    """2つのトークナイザー間の語彙重複度"""
    vocab1 = set(tokenizer1.get_vocab().keys())
    vocab2 = set(tokenizer2.get_vocab().keys())
    
    intersection = vocab1 & vocab2
    union = vocab1 | vocab2
    
    jaccard_similarity = len(intersection) / len(union)
    return jaccard_similarity

def evaluate_domain_adaptation(tokenizer, domains):
    """ドメイン別パフォーマンス評価"""
    results = {}
    
    for domain, texts in domains.items():
        compression_ratios = []
        oov_rates = []
        
        for text in texts:
            tokens = tokenizer.tokenize(text)
            
            # 圧縮率
            ratio = calculate_compression_ratio(text, tokens)
            compression_ratios.append(ratio)
            
            # 未知語率  
            oov_count = sum(1 for t in tokens if t == tokenizer.unk_token)
            oov_rate = oov_count / len(tokens) if tokens else 0
            oov_rates.append(oov_rate)
        
        results[domain] = {
            "avg_compression_ratio": np.mean(compression_ratios),
            "avg_oov_rate": np.mean(oov_rates)
        }
    
    return results

# 使用例
domains = {
    "医療": medical_texts,
    "法律": legal_texts, 
    "技術": technical_texts,
    "日常会話": casual_texts
}

performance = evaluate_domain_adaptation(tokenizer, domains)
for domain, metrics in performance.items():
    print(f"{domain}: 圧縮率 {metrics['avg_compression_ratio']:.2f}, "
          f"未知語率 {metrics['avg_oov_rate']:.3f}")

# 用途別推奨トークナイザー
recommendations = {
    "英語一般": "roberta-base",           # BPE
    "日本語一般": "cl-tohoku/bert-base-japanese-v2",  # WordPiece
    "多言語": "xlm-roberta-base",         # SentencePiece
    "コード": "microsoft/codebert-base",  # BPE
    "科学技術": "allenai/scibert_scivocab_uncased",  # WordPiece
    "医療": "dmis-lab/biobert-v1.1",     # WordPiece
}

def select_tokenizer(use_case, language="en"):
    """用途に応じたトークナイザーの選択"""
    if language == "ja":
        if use_case in ["general", "news", "web"]:
            return "cl-tohoku/bert-base-japanese-v2"
        elif use_case == "casual":
            return "rinna/japanese-gpt2-medium"
    
    elif language == "en":
        return recommendations.get(use_case, "bert-base-uncased")
    
    else:  # 多言語
        return "xlm-roberta-base"

import re
import unicodedata

def preprocess_text(text, language="ja"):
    """言語に応じたテキスト前処理"""
    
    # Unicode正規化
    text = unicodedata.normalize('NFKC', text)
    
    if language == "ja":
        # 全角英数字を半角に変換
        text = text.translate(str.maketrans(
            '０１２３４５６７８９ＡＢＣＤＥＦＧＨＩＪＫＬＭＮＯＰＱＲＳＴＵＶＷＸＹＺａｂｃｄｅｆｇｈｉｊｋｌｍｎｏｐｑｒｓｔｕｖｗｘｙｚ',
            '0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz'
        ))
        
        # 不要な空白の除去
        text = re.sub(r'\s+', ' ', text)
        
    elif language == "en":
        # 基本的なクリーニング
        text = re.sub(r'\s+', ' ', text)
        text = text.strip()
    
    return text

# 使用例
texts = [
    "これは　　テスト　　です。",  # 余分な空白
    "１２３ＡＢＣ",              # 全角文字
    "Hello    World!",           # 英語の余分な空白
]

for text in texts:
    cleaned = preprocess_text(text, "ja")
    print(f"元: '{text}' -> 後: '{cleaned}'")

class TokenizerError(Exception):
    """トークナイザー関連のエラー"""
    pass

def safe_tokenize(tokenizer, text, max_length=512):
    """安全なトークン化処理"""
    try:
        # 前処理
        if not isinstance(text, str):
            raise TokenizerError(f"入力はstrである必要があります: {type(text)}")
        
        if not text.strip():
            return []
        
        # トークン化
        tokens = tokenizer.tokenize(text)
        
        # 長さチェック
        if len(tokens) > max_length:
            print(f"警告: トークン長が最大長を超えています ({len(tokens)} > {max_length})")
            tokens = tokens[:max_length]
        
        return tokens
        
    except Exception as e:
        raise TokenizerError(f"トークン化中にエラーが発生: {str(e)}")

# 使用例
try:
    tokens = safe_tokenize(tokenizer, problematic_text)
    print(f"成功: {len(tokens)}個のトークンを生成")
except TokenizerError as e:
    print(f"エラー: {e}")

# 2025年のトレンド
model_trends = {
    "GPT-4": {"vocab_size": 100000+, "approach": "BPE"},
    "Claude": {"vocab_size": 100000+, "approach": "BPE"}, 
    "Gemini": {"vocab_size": 250000+, "approach": "SentencePiece"},
    "Japanese LLMs": {"vocab_size": 50000-100000, "approach": "Mixed"}
}

# 多言語対応の例
multilingual_examples = {
    "日本語": "自然言語処理の研究",
    "English": "Natural language processing research", 
    "中文": "自然语言处理研究",
    "한국어": "자연어 처리 연구",
    "العربية": "بحوث معالجة اللغة الطبيعية"
}

# 同一トークナイザーで全言語を処理
tokenizer = AutoTokenizer.from_pretrained("xlm-roberta-large")
for lang, text in multilingual_examples.items():
    tokens = tokenizer.tokenize(text)
    print(f"{lang}: {len(tokens)}トークン")

domain_specialized = {
    "科学技術": "allenai/scibert_scivocab_uncased",
    "医療": "dmis-lab/biobert-v1.1", 
    "金融": "ProsusAI/finbert",
    "法律": "nlpaueb/legal-bert-base-uncased",
    "プログラミング": "microsoft/codebert-base"
}

# 概念的な適応的トークナイザー
class AdaptiveTokenizer:
    def __init__(self, base_tokenizer):
        self.base = base_tokenizer
        self.domain_cache = {}
    
    def tokenize_with_domain(self, text, domain=None):
        if domain and domain in self.domain_cache:
            # ドメイン特化の語彙を優先使用
            specialized_tokens = self.domain_cache[domain]
            return self.apply_domain_vocabulary(text, specialized_tokens)
        else:
            return self.base.tokenize(text)
    
    def learn_domain_vocabulary(self, domain, texts):
        # ドメイン特有の語彙を学習
        domain_vocab = self.extract_domain_terms(texts)
        self.domain_cache[domain] = domain_vocab

# 将来の文脈考慮型トークナイザーの概念
class ContextAwareTokenizer:
    def tokenize_with_context(self, text, context=None):
        """文脈に応じてトークン化方法を調整"""
        if context:
            # 文脈に基づいてトークン化戦略を変更
            if context.get("formal", False):
                return self.formal_tokenize(text)
            elif context.get("casual", False):
                return self.casual_tokenize(text)
        
        return self.default_tokenize(text)

# 解決策: 効率化技術
optimization_techniques = {
    "語彙剪定": "使用頻度の低い語彙を削除",
    "量子化": "低精度での表現", 
    "キャッシュ": "頻出パターンの事前計算",
    "並列化": "バッチ処理の最適化"
}

# 解決の方向性
solutions = {
    "言語固有エンベッディング": "言語ごとの特殊化層",
    "転移学習": "高リソース言語から低リソース言語へ", 
    "合成データ": "不足データの人工生成",
    "クロスリンガル事前学習": "多言語同時学習"
}