# Pre-tokenization の各手法
import re
 
text = "Hello, World! 大規模言語モデル（LLM）は2024年に急速に発展した。"
 
# 方法1: 空白分割（最もシンプル）
whitespace_tokens = text.split()
print(f"空白分割: {whitespace_tokens}")
# → ['Hello,', 'World!', '大規模言語モデル（LLM）は2024年に急速に発展した。']
 
# 方法2: GPT-2/GPT-4o スタイル（正規表現ベース）
gpt2_pattern = re.compile(
    r"""'s|'t|'re|'ve|'m|'ll|'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+""",
    re.UNICODE
)
 
# 方法3: バイトレベル（GPT-4o, Claude）
# 全てのテキストをUTF-8バイト列として処理
byte_sequence = text.encode("utf-8")
print(f"バイト数: {len(byte_sequence)}")
# 日本語は1文字あたり3バイト（UTF-8）
 
# 方法4: SentencePiece スタイル（空白を特殊文字に）
# 空白を "▁" (U+2581) に変換して扱う
sp_text = "▁" + text.replace(" ", "▁")
print(f"SentencePiece形式: {sp_text}")

import tiktoken
 
# GPT-4o 用エンコーダ
enc = tiktoken.encoding_for_model("gpt-4o")
 
text = "大規模言語モデルは素晴らしい技術です。"
tokens = enc.encode(text)
 
print(f"テキスト: {text}")
print(f"トークン数: {len(tokens)}")
print(f"トークンID: {tokens}")
 
# 各トークンの内容を確認
for token_id in tokens:
    token_bytes = enc.decode_single_token_bytes(token_id)
    print(f"  ID {token_id:>6} → {token_bytes}")

from transformers import AutoTokenizer
 
models = {
    "GPT-4o": "Xenova/gpt-4o",
    "Claude": "anthropic/claude-tokenizer",  # 仮想例
    "Llama-3": "meta-llama/Llama-3.1-8B-Instruct",
    "Gemma-2": "google/gemma-2-9b",
}
 
text = "東京タワーの高さは333メートルです。The height is 333 meters."
 
for name, model_id in models.items():
    try:
        tokenizer = AutoTokenizer.from_pretrained(model_id)
        tokens = tokenizer.encode(text)
        print(f"{name:>10}: {len(tokens):>3} トークン")
    except Exception as e:
        print(f"{name:>10}: (ロード不可)")

import tiktoken
from collections import Counter
 
class TokenAnalyzer:
    """テキストのトークン化を詳細に分析するツール"""
 
    def __init__(self, model: str = "gpt-4o"):
        self.enc = tiktoken.encoding_for_model(model)
        self.model = model
 
    def analyze(self, text: str) -> dict:
        """テキストのトークン化を詳細に分析する"""
        tokens = self.enc.encode(text)
        token_strings = [
            self.enc.decode([t]) for t in tokens
        ]
 
        # トークンの種類を分類
        categories = {
            "ascii": 0,
            "japanese": 0,
            "number": 0,
            "punctuation": 0,
            "whitespace": 0,
            "other": 0,
        }
 
        for ts in token_strings:
            if ts.strip() == "":
                categories["whitespace"] += 1
            elif ts.isascii() and ts.isalpha():
                categories["ascii"] += 1
            elif ts.isdigit():
                categories["number"] += 1
            elif any(ord(c) > 0x3000 for c in ts):
                categories["japanese"] += 1
            elif not ts.isalnum():
                categories["punctuation"] += 1
            else:
                categories["other"] += 1
 
        return {
            "text_length_chars": len(text),
            "text_length_bytes": len(text.encode("utf-8")),
            "token_count": len(tokens),
            "chars_per_token": len(text) / len(tokens),
            "bytes_per_token": len(text.encode("utf-8")) / len(tokens),
            "token_categories": categories,
            "unique_tokens": len(set(tokens)),
            "token_ids": tokens,
            "token_strings": token_strings,
        }
 
    def compare_texts(self, texts: dict[str, str]) -> None:
        """複数テキストのトークン効率を比較する"""
        print(f"{'テキスト':<20} {'文字数':>6} {'バイト数':>8} "
              f"{'トークン数':>8} {'文字/トークン':>12}")
        print("-" * 70)
        for name, text in texts.items():
            result = self.analyze(text)
            print(f"{name:<20} {result['text_length_chars']:>6} "
                  f"{result['text_length_bytes']:>8} "
                  f"{result['token_count']:>8} "
                  f"{result['chars_per_token']:>12.2f}")
 
    def estimate_cost(self, text: str, model_pricing: dict) -> dict:
        """テキストのAPIコストを見積もる"""
        tokens = len(self.enc.encode(text))
        input_cost = (tokens / 1_000_000) * model_pricing["input"]
        return {
            "tokens": tokens,
            "input_cost_usd": input_cost,
        }
 
# 使用例
analyzer = TokenAnalyzer("gpt-4o")
 
# 日英比較
texts = {
    "英語（短文）": "The quick brown fox jumps over the lazy dog.",
    "日本語（短文）": "素早い茶色の狐が怠惰な犬を飛び越える。",
    "英語（技術文）": "Transformer architecture uses self-attention mechanism.",
    "日本語（技術文）": "Transformerアーキテクチャは自己注意機構を使用する。",
    "コード": "def hello(): return 'Hello, World!'",
    "混合": "GPT-4oは2024年にリリースされた最新のLLMです。",
}
 
analyzer.compare_texts(texts)

from collections import Counter
 
def get_pairs(word_freqs):
    """全ペアの出現頻度を計算"""
    pairs = Counter()
    for word, freq in word_freqs.items():
        symbols = word.split()
        for i in range(len(symbols) - 1):
            pairs[(symbols[i], symbols[i + 1])] += freq
    return pairs
 
def merge_pair(pair, word_freqs):
    """最頻ペアをマージ"""
    new_word_freqs = {}
    bigram = " ".join(pair)
    replacement = "".join(pair)
    for word, freq in word_freqs.items():
        new_word = word.replace(bigram, replacement)
        new_word_freqs[new_word] = freq
    return new_word_freqs
 
# 学習データの単語頻度
word_freqs = {
    "l o w": 5,
    "l o w e r": 2,
    "n e w e s t": 6,
    "w i d e s t": 3,
}
 
num_merges = 10
merges = []
 
for i in range(num_merges):
    pairs = get_pairs(word_freqs)
    if not pairs:
        break
    best_pair = max(pairs, key=pairs.get)
    word_freqs = merge_pair(best_pair, word_freqs)
    merges.append(best_pair)
    print(f"Merge {i+1}: {best_pair} → {''.join(best_pair)}")
 
print(f"\n最終語彙の一部: {list(word_freqs.keys())}")

from collections import Counter, defaultdict
from typing import Optional
 
class ByteLevelBPE:
    """Byte-Level BPE の教育用実装"""
 
    def __init__(self, vocab_size: int = 1000):
        self.vocab_size = vocab_size
        self.merges: list[tuple[bytes, bytes]] = []
        self.vocab: dict[int, bytes] = {}
 
    def _get_stats(self, ids_list: list[list[int]]) -> Counter:
        """全てのバイトペアの出現頻度を計算"""
        counts = Counter()
        for ids in ids_list:
            for i in range(len(ids) - 1):
                counts[(ids[i], ids[i + 1])] += 1
        return counts
 
    def _merge(self, ids: list[int], pair: tuple[int, int],
               new_id: int) -> list[int]:
        """指定ペアをマージして新しいIDに置換"""
        new_ids = []
        i = 0
        while i < len(ids):
            if i < len(ids) - 1 and ids[i] == pair[0] and ids[i + 1] == pair[1]:
                new_ids.append(new_id)
                i += 2
            else:
                new_ids.append(ids[i])
                i += 1
        return new_ids
 
    def train(self, texts: list[str]) -> None:
        """テキストコーパスから BPE 語彙を学習する"""
        # 初期語彙: 0-255 のバイト値
        self.vocab = {i: bytes([i]) for i in range(256)}
        next_id = 256
 
        # テキストをバイト列に変換
        ids_list = [list(text.encode("utf-8")) for text in texts]
 
        # 語彙サイズに達するまでマージを繰り返す
        while next_id < self.vocab_size:
            stats = self._get_stats(ids_list)
            if not stats:
                break
 
            # 最頻ペアを選択
            best_pair = max(stats, key=stats.get)
 
            # 全テキストでマージを実行
            ids_list = [
                self._merge(ids, best_pair, next_id)
                for ids in ids_list
            ]
 
            # 語彙に追加
            self.vocab[next_id] = self.vocab[best_pair[0]] + self.vocab[best_pair[1]]
            self.merges.append(best_pair)
 
            if next_id % 100 == 0:
                print(f"Merge {next_id - 256}: "
                      f"{self.vocab[best_pair[0]]!r} + "
                      f"{self.vocab[best_pair[1]]!r} → "
                      f"{self.vocab[next_id]!r}")
 
            next_id += 1
 
        print(f"\n学習完了: {len(self.vocab)} トークン, "
              f"{len(self.merges)} マージ")
 
    def encode(self, text: str) -> list[int]:
        """テキストをトークンIDのリストに変換"""
        ids = list(text.encode("utf-8"))
        for pair in self.merges:
            new_id = 256 + self.merges.index(pair)
            ids = self._merge(ids, pair, new_id)
        return ids
 
    def decode(self, ids: list[int]) -> str:
        """トークンIDのリストをテキストに復元"""
        byte_sequence = b"".join(self.vocab[i] for i in ids)
        return byte_sequence.decode("utf-8", errors="replace")
 
# 使用例
bpe = ByteLevelBPE(vocab_size=500)
corpus = [
    "The quick brown fox jumps over the lazy dog.",
    "大規模言語モデルは自然言語処理の革命です。",
    "Machine learning and deep learning are transforming AI.",
] * 100  # コーパスを繰り返して頻度を上げる
 
bpe.train(corpus)
 
test_text = "The quick fox"
encoded = bpe.encode(test_text)
decoded = bpe.decode(encoded)
print(f"原文: {test_text}")
print(f"トークン数: {len(encoded)}")
print(f"復元: {decoded}")

import sentencepiece as spm
 
# モデルの学習
spm.SentencePieceTrainer.train(
    input="corpus.txt",
    model_prefix="my_tokenizer",
    vocab_size=32000,
    model_type="bpe",           # "unigram" も選択可能
    character_coverage=0.9995,  # 日本語は高めに設定
    pad_id=3,
    unk_id=0,
    bos_id=1,
    eos_id=2,
)
 
# 学習済みモデルの使用
sp = spm.SentencePieceProcessor()
sp.load("my_tokenizer.model")
 
text = "大規模言語モデルの性能はトークナイザに大きく依存する"
tokens = sp.encode(text, out_type=str)
ids = sp.encode(text, out_type=int)
 
print(f"テキスト: {text}")
print(f"トークン: {tokens}")
print(f"ID: {ids}")
print(f"復元: {sp.decode(ids)}")

import sentencepiece as spm
 
# 日本語に最適化した SentencePiece 学習
spm.SentencePieceTrainer.train(
    input="japanese_corpus.txt",
    model_prefix="jp_tokenizer",
    vocab_size=32000,
    model_type="unigram",        # 日本語には Unigram が適している場合が多い
    character_coverage=0.9995,   # 日本語の文字カバレッジ
    byte_fallback=True,          # 未知文字をバイト表現にフォールバック
    split_by_unicode_script=True,  # Unicode スクリプト境界で分割
    split_by_number=True,         # 数字の境界で分割
    split_by_whitespace=True,     # 空白で分割
    split_digits=True,            # 各桁を個別のトークンにする
    treat_whitespace_as_suffix=False,
    allow_whitespace_only_pieces=True,
    normalization_rule_name="nfkc",  # Unicode 正規化
    num_threads=8,
    # 特殊トークンの定義
    user_defined_symbols=["<code>", "</code>", "<math>", "</math>"],
    control_symbols=["<sep>", "<cls>", "<mask>"],
)
 
# 学習済みモデルの詳細な使用法
sp = spm.SentencePieceProcessor()
sp.load("jp_tokenizer.model")
 
text = "GPT-4oの性能は2024年に大幅に向上した。"
 
# 通常のエンコード
tokens_str = sp.encode(text, out_type=str)
tokens_id = sp.encode(text, out_type=int)
print(f"トークン(文字列): {tokens_str}")
print(f"トークン(ID): {tokens_id}")
 
# N-best エンコード（複数の分割候補を取得）
nbest = sp.nbest_encode(text, nbest_size=5, out_type=str)
print(f"\nN-best 分割候補:")
for i, candidate in enumerate(nbest):
    print(f"  候補{i+1}: {candidate}")
 
# サンプリングエンコード（正則化効果）
for i in range(3):
    sampled = sp.encode(text, out_type=str, enable_sampling=True,
                        alpha=0.1, nbest_size=-1)
    print(f"サンプル{i+1}: {sampled}")

class ContextWindowManager:
    """コンテキストウィンドウを管理するユーティリティ"""
 
    MODEL_LIMITS = {
        "gpt-4o": 128_000,
        "gpt-4o-mini": 128_000,
        "claude-3-5-sonnet": 200_000,
        "claude-3-5-haiku": 200_000,
        "gemini-1.5-pro": 1_000_000,
        "llama-3.1-8b": 128_000,
    }
 
    def __init__(self, model: str, max_output_tokens: int = 4096,
                 system_prompt_tokens: int = 0):
        self.model = model
        self.context_limit = self.MODEL_LIMITS.get(model, 128_000)
        self.max_output_tokens = max_output_tokens
        self.system_prompt_tokens = system_prompt_tokens
 
    @property
    def available_input_tokens(self) -> int:
        """入力に使える残りトークン数"""
        return (self.context_limit
                - self.max_output_tokens
                - self.system_prompt_tokens)
 
    def can_fit(self, input_tokens: int) -> bool:
        """入力がコンテキストウィンドウに収まるか"""
        return input_tokens <= self.available_input_tokens
 
    def truncate_messages(self, messages: list[dict],
                          token_counter,
                          strategy: str = "sliding_window") -> list[dict]:
        """メッセージ履歴をコンテキストに収まるよう切り詰める"""
        if strategy == "sliding_window":
            return self._sliding_window(messages, token_counter)
        elif strategy == "summarize_old":
            return self._summarize_old(messages, token_counter)
        else:
            raise ValueError(f"Unknown strategy: {strategy}")
 
    def _sliding_window(self, messages: list[dict],
                         token_counter) -> list[dict]:
        """古いメッセージから削除する（最新を優先）"""
        result = []
        total_tokens = 0
        limit = self.available_input_tokens
 
        # 最新のメッセージから逆順に追加
        for msg in reversed(messages):
            msg_tokens = token_counter(msg["content"])
            if total_tokens + msg_tokens > limit:
                break
            result.insert(0, msg)
            total_tokens += msg_tokens
 
        return result
 
    def _summarize_old(self, messages: list[dict],
                        token_counter) -> list[dict]:
        """古いメッセージを要約して圧縮する"""
        # 実装例: 古い部分を要約 + 新しい部分をそのまま保持
        limit = self.available_input_tokens
        half_limit = limit // 2
 
        # 新しいメッセージ（後半）
        recent = []
        recent_tokens = 0
        for msg in reversed(messages):
            msg_tokens = token_counter(msg["content"])
            if recent_tokens + msg_tokens > half_limit:
                break
            recent.insert(0, msg)
            recent_tokens += msg_tokens
 
        # 古いメッセージ（前半）を要約
        old_messages = messages[:len(messages) - len(recent)]
        if old_messages:
            summary_msg = {
                "role": "system",
                "content": f"[以前の会話の要約: {len(old_messages)}件のメッセージ]"
            }
            return [summary_msg] + recent
 
        return recent
 
# 使用例
manager = ContextWindowManager(
    model="claude-3-5-sonnet",
    max_output_tokens=4096,
    system_prompt_tokens=500,
)
 
print(f"モデル: {manager.model}")
print(f"コンテキスト上限: {manager.context_limit:,} tokens")
print(f"入力使用可能: {manager.available_input_tokens:,} tokens")

import tiktoken
 
def estimate_cost(
    text: str,
    model: str = "gpt-4o",
    max_output_tokens: int = 1000
):
    """APIコストの見積もり"""
    pricing = {
        "gpt-4o":          {"input": 2.50, "output": 10.00},
        "gpt-4o-mini":     {"input": 0.15, "output": 0.60},
        "claude-3.5-sonnet": {"input": 3.00, "output": 15.00},
        "claude-3.5-haiku": {"input": 0.80, "output": 4.00},
    }
 
    enc = tiktoken.encoding_for_model("gpt-4o")
    input_tokens = len(enc.encode(text))
 
    if model in pricing:
        p = pricing[model]
        input_cost = input_tokens / 1_000_000 * p["input"]
        output_cost = max_output_tokens / 1_000_000 * p["output"]
        total = input_cost + output_cost
 
        print(f"モデル: {model}")
        print(f"入力トークン: {input_tokens:,}")
        print(f"出力トークン(最大): {max_output_tokens:,}")
        print(f"入力コスト: ${input_cost:.4f}")
        print(f"出力コスト: ${output_cost:.4f}")
        print(f"合計見積もり: ${total:.4f}")
 
text = "ここに長いプロンプトが入ります。" * 100
estimate_cost(text, model="claude-3.5-sonnet")

import tiktoken
from typing import Generator
 
class BatchTokenOptimizer:
    """大量テキストを処理する際のトークン最適化"""
 
    def __init__(self, model: str = "gpt-4o",
                 max_tokens_per_batch: int = 100_000):
        self.enc = tiktoken.encoding_for_model(model)
        self.max_tokens_per_batch = max_tokens_per_batch
 
    def create_batches(self, texts: list[str],
                        max_tokens: int = None
                        ) -> Generator[list[str], None, None]:
        """テキストをトークン数ベースでバッチに分割する"""
        max_tokens = max_tokens or self.max_tokens_per_batch
        current_batch = []
        current_tokens = 0
 
        for text in texts:
            text_tokens = len(self.enc.encode(text))
 
            if current_tokens + text_tokens > max_tokens and current_batch:
                yield current_batch
                current_batch = []
                current_tokens = 0
 
            current_batch.append(text)
            current_tokens += text_tokens
 
        if current_batch:
            yield current_batch
 
    def truncate_to_tokens(self, text: str,
                            max_tokens: int) -> str:
        """テキストを指定トークン数以内に切り詰める"""
        tokens = self.enc.encode(text)
        if len(tokens) <= max_tokens:
            return text
        truncated_tokens = tokens[:max_tokens]
        return self.enc.decode(truncated_tokens)
 
    def split_by_tokens(self, text: str,
                         chunk_size: int,
                         overlap: int = 0) -> list[str]:
        """テキストをトークン数ベースでチャンクに分割する"""
        tokens = self.enc.encode(text)
        chunks = []
        start = 0
 
        while start < len(tokens):
            end = min(start + chunk_size, len(tokens))
            chunk_tokens = tokens[start:end]
            chunks.append(self.enc.decode(chunk_tokens))
            start += chunk_size - overlap
 
        return chunks
 
# 使用例
optimizer = BatchTokenOptimizer("gpt-4o")
 
# 大量テキストをバッチ処理
documents = [f"Document {i}: " + "テスト文章。" * 100 for i in range(50)]
 
for batch_idx, batch in enumerate(optimizer.create_batches(documents)):
    print(f"バッチ {batch_idx + 1}: {len(batch)} ドキュメント")
 
# テキストのトークンベース分割
long_text = "これは非常に長いテキストです。" * 500
chunks = optimizer.split_by_tokens(long_text, chunk_size=512, overlap=64)
print(f"チャンク数: {len(chunks)}")

import tiktoken
 
enc = tiktoken.encoding_for_model("gpt-4o")
 
# 特殊文字のトークン化を検証
test_cases = {
    "絵文字": "🎉🚀💡🔥",
    "数式記号": "∑∫∂∇∞",
    "CJK拡張": "𠮷𩸽𠀋",
    "制御文字": "タブ\tと\n改行",
    "ゼロ幅文字": "hello\u200bworld",  # ゼロ幅スペース
    "結合文字": "がぎぐげご",  # 半濁点・濁点
    "URL": "https://example.com/path?q=test&lang=ja",
    "JSON": '{"key": "value", "num": 42}',
    "コード": 'def hello():\n    print("Hello")',
}
 
for name, text in test_cases.items():
    tokens = enc.encode(text)
    print(f"{name:<12}: {len(text):>3}文字 → {len(tokens):>3}トークン "
          f"(効率: {len(text)/len(tokens):.2f}文字/トークン)")

import tiktoken
 
enc = tiktoken.encoding_for_model("gpt-4o")
 
# トークン境界の問題を可視化
def visualize_tokenization(text: str):
    """トークン分割を視覚的に表示する"""
    tokens = enc.encode(text)
    result = []
    for token_id in tokens:
        token_str = enc.decode([token_id])
        result.append(f"[{token_str}]")
    print(f"原文: {text}")
    print(f"分割: {''.join(result)}")
    print(f"トークン数: {len(tokens)}")
    print()
 
# 問題のある例
visualize_tokenization("unhappiness")      # 正常: [un][happiness]
visualize_tokenization("123456789")         # 数字の分割
visualize_tokenization("user@example.com")  # メールアドレス
visualize_tokenization("2024-03-15T10:30:00Z")  # ISO日時
visualize_tokenization("192.168.1.1")       # IPアドレス
visualize_tokenization("東京都千代田区丸の内1-1-1")  # 日本語住所

# トークナイゼーションを利用したプロンプトインジェクション攻撃の例と対策
 
class TokenSanitizer:
    """トークンレベルでの入力サニタイズ"""
 
    DANGEROUS_TOKEN_PATTERNS = [
        b"<|im_start|>",   # ChatML インジェクション
        b"<|im_end|>",
        b"<|endoftext|>",
        b"[INST]",          # Llama フォーマット
        b"[/INST]",
        b"<s>",
        b"</s>",
    ]
 
    def __init__(self, model: str = "gpt-4o"):
        import tiktoken
        self.enc = tiktoken.encoding_for_model(model)
 
    def sanitize(self, text: str) -> str:
        """危険なトークンパターンを除去する"""
        sanitized = text
        for pattern in self.DANGEROUS_TOKEN_PATTERNS:
            pattern_str = pattern.decode("utf-8", errors="ignore")
            if pattern_str in sanitized:
                sanitized = sanitized.replace(pattern_str, "")
        return sanitized
 
    def validate_token_count(self, text: str,
                              max_tokens: int) -> tuple[bool, int]:
        """トークン数が上限以内か検証する"""
        tokens = self.enc.encode(text)
        return len(tokens) <= max_tokens, len(tokens)

def debug_tokenization(text: str, models: list[str] = None):
    """複数モデルのトークン化結果をデバッグ表示する"""
    import tiktoken
 
    if models is None:
        models = ["gpt-4o", "gpt-4o-mini"]
 
    print(f"テキスト: {text[:50]}{'...' if len(text) > 50 else ''}")
    print(f"文字数: {len(text)}, バイト数: {len(text.encode('utf-8'))}")
    print("-" * 60)
 
    for model in models:
        try:
            enc = tiktoken.encoding_for_model(model)
            tokens = enc.encode(text)
            decoded_tokens = [enc.decode([t]) for t in tokens]
 
            print(f"\n{model}:")
            print(f"  トークン数: {len(tokens)}")
            print(f"  文字/トークン: {len(text)/len(tokens):.2f}")
            print(f"  先頭5トークン: {decoded_tokens[:5]}")
            print(f"  末尾5トークン: {decoded_tokens[-5:]}")
        except Exception as e:
            print(f"  {model}: エラー - {e}")
 
# デバッグ実行
debug_tokenization("Transformerアーキテクチャは自然言語処理に革命をもたらした。")

import time
import tiktoken
 
def benchmark_tokenizer(text: str, iterations: int = 1000):
    """トークナイザの速度をベンチマークする"""
    enc = tiktoken.encoding_for_model("gpt-4o")
 
    # エンコード速度
    start = time.perf_counter()
    for _ in range(iterations):
        tokens = enc.encode(text)
    encode_time = (time.perf_counter() - start) / iterations
 
    # デコード速度
    tokens = enc.encode(text)
    start = time.perf_counter()
    for _ in range(iterations):
        enc.decode(tokens)
    decode_time = (time.perf_counter() - start) / iterations
 
    text_bytes = len(text.encode("utf-8"))
    print(f"テキストサイズ: {text_bytes:,} bytes")
    print(f"トークン数: {len(tokens):,}")
    print(f"エンコード: {encode_time*1000:.3f} ms "
          f"({text_bytes/encode_time/1024/1024:.1f} MB/s)")
    print(f"デコード: {decode_time*1000:.3f} ms "
          f"({text_bytes/decode_time/1024/1024:.1f} MB/s)")
 
# ベンチマーク実行
short_text = "Hello, World!" * 10
long_text = "大規模言語モデルの性能は素晴らしい。" * 1000
 
print("=== 短いテキスト ===")
benchmark_tokenizer(short_text)
print("\n=== 長いテキスト ===")
benchmark_tokenizer(long_text, iterations=100)

class StreamingTokenCounter:
    """ストリーミング方式でメモリ効率的にトークン数をカウントする"""
 
    def __init__(self, model: str = "gpt-4o"):
        import tiktoken
        self.enc = tiktoken.encoding_for_model(model)
 
    def count_file(self, filepath: str,
                    chunk_size: int = 1024 * 1024) -> int:
        """ファイルをチャンク読みしてトークン数をカウント"""
        total_tokens = 0
 
        with open(filepath, "r", encoding="utf-8") as f:
            while True:
                chunk = f.read(chunk_size)
                if not chunk:
                    break
                total_tokens += len(self.enc.encode(chunk))
 
        return total_tokens
 
    def count_streaming(self, text_generator) -> int:
        """ジェネレータからストリーミングでカウント"""
        total_tokens = 0
        for text in text_generator:
            total_tokens += len(self.enc.encode(text))
        return total_tokens

# 演習1: 基本実装のテンプレート
class Exercise1:
    """基本的な実装パターンの演習"""
 
    def __init__(self):
        self.data = []
 
    def validate_input(self, value):
        """入力値の検証"""
        if value is None:
            raise ValueError("入力値がNoneです")
        return True
 
    def process(self, value):
        """データ処理のメインロジック"""
        self.validate_input(value)
        self.data.append(value)
        return self.data
 
    def get_results(self):
        """処理結果の取得"""
        return {
            'count': len(self.data),
            'data': self.data
        }
 
# テスト
def test_exercise1():
    ex = Exercise1()
    assert ex.process(1) == [1]
    assert ex.process(2) == [1, 2]
    assert ex.get_results()['count'] == 2
 
    try:
        ex.process(None)
        assert False, "例外が発生するべき"
    except ValueError:
        pass
 
    print("全テスト合格!")
 
test_exercise1()

# 演習2: 応用パターン
from typing import List, Dict, Optional
from datetime import datetime
 
class AdvancedExercise:
    """応用パターンの演習"""
 
    def __init__(self, max_size: int = 100):
        self._items: List[Dict] = []
        self._max_size = max_size
        self._created_at = datetime.now()
 
    def add(self, key: str, value: any) -> bool:
        """アイテムの追加（サイズ制限付き）"""
        if len(self._items) >= self._max_size:
            return False
        self._items.append({
            'key': key,
            'value': value,
            'timestamp': datetime.now().isoformat()
        })
        return True
 
    def find(self, key: str) -> Optional[Dict]:
        """キーによる検索"""
        for item in reversed(self._items):
            if item['key'] == key:
                return item
        return None
 
    def remove(self, key: str) -> bool:
        """キーによる削除"""
        for i, item in enumerate(self._items):
            if item['key'] == key:
                self._items.pop(i)
                return True
        return False
 
    def stats(self) -> Dict:
        """統計情報"""
        return {
            'total_items': len(self._items),
            'max_size': self._max_size,
            'usage_percent': len(self._items) / self._max_size * 100,
            'uptime': str(datetime.now() - self._created_at)
        }
 
# テスト
def test_advanced():
    ex = AdvancedExercise(max_size=3)
    assert ex.add("a", 1) == True
    assert ex.add("b", 2) == True
    assert ex.add("c", 3) == True
    assert ex.add("d", 4) == False  # サイズ制限
    assert ex.find("b")['value'] == 2
    assert ex.remove("b") == True
    assert ex.find("b") is None
    stats = ex.stats()
    assert stats['total_items'] == 2
    print("応用テスト全合格!")
 
test_advanced()

# 演習3: パフォーマンス最適化
import time
from functools import lru_cache
 
# 最適化前（O(n^2)）
def slow_search(data: list, target: int) -> int:
    """非効率な検索"""
    for i in range(len(data)):
        for j in range(i + 1, len(data)):
            if data[i] + data[j] == target:
                return (i, j)
    return (-1, -1)
 
# 最適化後（O(n)）
def fast_search(data: list, target: int) -> tuple:
    """ハッシュマップを使った効率的な検索"""
    seen = {}
    for i, num in enumerate(data):
        complement = target - num
        if complement in seen:
            return (seen[complement], i)
        seen[num] = i
    return (-1, -1)
 
# ベンチマーク
def benchmark():
    import random
    data = list(range(5000))
    random.shuffle(data)
    target = data[100] + data[4000]
 
    start = time.time()
    result1 = slow_search(data, target)
    slow_time = time.time() - start
 
    start = time.time()
    result2 = fast_search(data, target)
    fast_time = time.time() - start
 
    print(f"非効率版: {slow_time:.4f}秒")
    print(f"効率版:   {fast_time:.6f}秒")
    print(f"高速化率: {slow_time/fast_time:.0f}倍")
 
benchmark()

# 設計判断の記録テンプレート
class ArchitectureDecisionRecord:
    """ADR (Architecture Decision Record) の作成"""
 
    def __init__(self, title: str):
        self.title = title
        self.context = ""
        self.decision = ""
        self.consequences = []
        self.alternatives = []
 
    def set_context(self, context: str):
        """背景と課題の記述"""
        self.context = context
        return self
 
    def set_decision(self, decision: str):
        """決定内容の記述"""
        self.decision = decision
        return self
 
    def add_consequence(self, consequence: str, positive: bool = True):
        """結果の追加"""
        self.consequences.append({
            'description': consequence,
            'type': 'positive' if positive else 'negative'
        })
        return self
 
    def add_alternative(self, name: str, reason_rejected: str):
        """却下した代替案の追加"""
        self.alternatives.append({
            'name': name,
            'reason_rejected': reason_rejected
        })
        return self
 
    def to_markdown(self) -> str:
        """Markdown形式で出力"""
        md = f"# ADR: {self.title}\n\n"
        md += f"## 背景\n{self.context}\n\n"
        md += f"## 決定\n{self.decision}\n\n"
        md += "## 結果\n"
        for c in self.consequences:
            icon = "✅" if c['type'] == 'positive' else "⚠️"
            md += f"- {icon} {c['description']}\n"
        md += "\n## 却下した代替案\n"
        for a in self.alternatives:
            md += f"- **{a['name']}**: {a['reason_rejected']}\n"
        return md