import numpy as np
 
def self_attention(Q, K, V, mask=None):
    """スケーリングドット積アテンション
 
    Args:
        Q: クエリ行列 (seq_len, d_k)
        K: キー行列 (seq_len, d_k)
        V: バリュー行列 (seq_len, d_v)
        mask: オプションのマスク行列（因果マスクなど）
 
    Returns:
        出力行列とアテンション重み
    """
    d_k = Q.shape[-1]
 
    # スケーリングドット積
    scores = np.matmul(Q, K.T) / np.sqrt(d_k)
 
    # マスクの適用（因果的言語モデルの場合）
    if mask is not None:
        scores = np.where(mask == 0, -1e9, scores)
 
    # Softmax で正規化
    exp_scores = np.exp(scores - np.max(scores, axis=-1, keepdims=True))
    weights = exp_scores / exp_scores.sum(axis=-1, keepdims=True)
 
    # 加重和
    output = np.matmul(weights, V)
    return output, weights
 
# 例: 4トークン、次元8
np.random.seed(42)
seq_len = 4
d_k = 8
 
Q = np.random.randn(seq_len, d_k)
K = np.random.randn(seq_len, d_k)
V = np.random.randn(seq_len, d_k)
 
# 因果マスク（下三角行列）- デコーダ用
causal_mask = np.tril(np.ones((seq_len, seq_len)))
print("因果マスク:")
print(causal_mask)
 
output, weights = self_attention(Q, K, V, mask=causal_mask)
print(f"\n出力形状: {output.shape}")  # (4, 8)
print(f"アテンション重み形状: {weights.shape}")  # (4, 4)
print(f"\nアテンション重み（各行の合計=1）:")
print(weights.round(3))

import numpy as np
 
class MultiHeadAttention:
    """Multi-Head Attention の NumPy 実装"""
 
    def __init__(self, d_model: int, num_heads: int):
        self.d_model = d_model
        self.num_heads = num_heads
        self.d_k = d_model // num_heads
 
        # 重み行列の初期化（Xavier初期化）
        scale = np.sqrt(2.0 / d_model)
        self.W_q = np.random.randn(d_model, d_model) * scale
        self.W_k = np.random.randn(d_model, d_model) * scale
        self.W_v = np.random.randn(d_model, d_model) * scale
        self.W_o = np.random.randn(d_model, d_model) * scale
 
    def split_heads(self, x):
        """(seq_len, d_model) → (num_heads, seq_len, d_k)"""
        seq_len = x.shape[0]
        x = x.reshape(seq_len, self.num_heads, self.d_k)
        return x.transpose(1, 0, 2)
 
    def forward(self, x, mask=None):
        """Multi-Head Attention の前方計算"""
        # 線形変換
        Q = x @ self.W_q
        K = x @ self.W_k
        V = x @ self.W_v
 
        # ヘッド分割
        Q = self.split_heads(Q)  # (num_heads, seq_len, d_k)
        K = self.split_heads(K)
        V = self.split_heads(V)
 
        # 各ヘッドでアテンション計算
        d_k = self.d_k
        scores = np.matmul(Q, K.transpose(0, 2, 1)) / np.sqrt(d_k)
 
        if mask is not None:
            scores = np.where(mask == 0, -1e9, scores)
 
        exp_scores = np.exp(scores - np.max(scores, axis=-1, keepdims=True))
        weights = exp_scores / exp_scores.sum(axis=-1, keepdims=True)
 
        attended = np.matmul(weights, V)  # (num_heads, seq_len, d_k)
 
        # ヘッド結合
        seq_len = attended.shape[1]
        concat = attended.transpose(1, 0, 2).reshape(seq_len, self.d_model)
 
        # 出力射影
        output = concat @ self.W_o
        return output, weights
 
# 使用例
np.random.seed(42)
mha = MultiHeadAttention(d_model=64, num_heads=8)
x = np.random.randn(10, 64)  # 10トークン、次元64
output, weights = mha.forward(x)
print(f"入力形状: {x.shape}")       # (10, 64)
print(f"出力形状: {output.shape}")   # (10, 64)
print(f"重み形状: {weights.shape}")  # (8, 10, 10) - 8ヘッド

import torch
import torch.nn as nn
 
class TransformerBlock(nn.Module):
    """LLM スタイルの Transformer ブロック（Pre-LN 構成）"""
 
    def __init__(self, d_model=512, nhead=8, dim_ff=2048, dropout=0.1):
        super().__init__()
 
        # Pre-LayerNorm 構成（GPT-2 以降の標準）
        self.ln1 = nn.LayerNorm(d_model)
        self.attn = nn.MultiheadAttention(
            d_model, nhead, dropout=dropout, batch_first=True
        )
 
        self.ln2 = nn.LayerNorm(d_model)
        self.ffn = nn.Sequential(
            nn.Linear(d_model, dim_ff),
            nn.GELU(),              # ReLU → GELU が現代の標準
            nn.Dropout(dropout),
            nn.Linear(dim_ff, d_model),
            nn.Dropout(dropout),
        )
 
    def forward(self, x, attn_mask=None):
        # Pre-LN + 残差接続（Self-Attention）
        normed = self.ln1(x)
        attn_out, _ = self.attn(normed, normed, normed, attn_mask=attn_mask)
        x = x + attn_out
 
        # Pre-LN + 残差接続（FFN）
        normed = self.ln2(x)
        ffn_out = self.ffn(normed)
        x = x + ffn_out
 
        return x
 
# 使用例
block = TransformerBlock(d_model=512, nhead=8, dim_ff=2048)
src = torch.randn(2, 10, 512)  # (バッチ=2, シーケンス長=10, 次元=512)
output = block(src)
print(f"出力形状: {output.shape}")  # torch.Size([2, 10, 512])
 
# パラメータ数の確認
total_params = sum(p.numel() for p in block.parameters())
print(f"パラメータ数: {total_params:,}")  # 約4.2M

import torch
 
def create_causal_mask(seq_len: int) -> torch.Tensor:
    """因果的アテンションマスクの生成
 
    デコーダモデルでは、各位置は自分自身と
    それ以前の位置のみ参照可能。
    """
    mask = torch.triu(torch.ones(seq_len, seq_len), diagonal=1)
    mask = mask.masked_fill(mask == 1, float('-inf'))
    return mask
 
# 8トークンのマスク
mask = create_causal_mask(8)
print("因果マスク（0=参照可能、-inf=マスク）:")
print(mask)
 
# Sliding Window Attention のマスク（Mistral 方式）
def create_sliding_window_mask(seq_len: int, window_size: int) -> torch.Tensor:
    """Sliding Window Attention マスク
 
    各位置は window_size 以内のトークンのみ参照可能。
    長いシーケンスでもメモリ使用量を O(n*w) に抑制。
    """
    mask = torch.zeros(seq_len, seq_len)
    for i in range(seq_len):
        # 因果マスク（未来は見えない）+ ウィンドウ制限
        start = max(0, i - window_size + 1)
        mask[i, :start] = float('-inf')
        mask[i, i+1:] = float('-inf')
    return mask
 
sw_mask = create_sliding_window_mask(8, window_size=3)
print("\nSliding Window マスク (window=3):")
print(sw_mask)

import numpy as np
 
def compute_rope_embeddings(seq_len: int, d_model: int, base: float = 10000.0):
    """RoPE (Rotary Position Embedding) の計算
 
    各次元ペアに対して異なる周波数の回転を適用。
    位置 p のトークンの次元 (2i, 2i+1) に対して:
        cos(p * theta_i), sin(p * theta_i)
    を掛け合わせる。
    """
    # 各次元ペアの周波数 theta
    dim_pairs = d_model // 2
    theta = 1.0 / (base ** (np.arange(0, d_model, 2) / d_model))
 
    # 位置 × 周波数
    positions = np.arange(seq_len)
    angles = np.outer(positions, theta)  # (seq_len, dim_pairs)
 
    cos_vals = np.cos(angles)
    sin_vals = np.sin(angles)
 
    return cos_vals, sin_vals
 
def apply_rope(x, cos_vals, sin_vals):
    """RoPE をクエリ/キーベクトルに適用"""
    d = x.shape[-1]
    x1 = x[..., :d//2]
    x2 = x[..., d//2:]
 
    # 回転の適用
    rotated = np.concatenate([
        x1 * cos_vals - x2 * sin_vals,
        x1 * sin_vals + x2 * cos_vals
    ], axis=-1)
 
    return rotated
 
# 使用例
seq_len = 16
d_model = 64
cos_vals, sin_vals = compute_rope_embeddings(seq_len, d_model)
 
# クエリとキーに RoPE を適用
Q = np.random.randn(seq_len, d_model)
K = np.random.randn(seq_len, d_model)
 
Q_rotated = apply_rope(Q, cos_vals, sin_vals)
K_rotated = apply_rope(K, cos_vals, sin_vals)
 
print(f"元のクエリ形状: {Q.shape}")
print(f"回転後のクエリ形状: {Q_rotated.shape}")
 
# 相対位置の内積が保持されることを確認
# 位置 i と j の内積は |i - j| のみに依存
dot_original = np.sum(Q[3] * K[5])
dot_rotated = np.sum(Q_rotated[3] * K_rotated[5])
print(f"\n元の内積 (pos 3, 5): {dot_original:.4f}")
print(f"RoPE後の内積 (pos 3, 5): {dot_rotated:.4f}")

import numpy as np
 
def estimate_loss(params_billions, tokens_trillions):
    """Chinchilla 式での Loss 推定（簡易版）"""
    E = 1.69     # 不可約損失
    A = 406.4
    B = 410.7
    alpha = 0.34
    beta = 0.28
 
    N = params_billions * 1e9
    D = tokens_trillions * 1e12
 
    loss = E + A / (N ** alpha) + B / (D ** beta)
    return loss
 
def compute_optimal_allocation(compute_budget_flops):
    """与えられた計算予算での最適なパラメータ数・データ量を推定
 
    Chinchilla の知見: N ≈ D / 20 (トークン/パラメータ比 ≈ 20)
    FLOPs ≈ 6 * N * D
    """
    # C = 6 * N * D, D = 20 * N
    # C = 6 * N * 20 * N = 120 * N^2
    N_opt = np.sqrt(compute_budget_flops / 120)
    D_opt = 20 * N_opt
    return N_opt, D_opt
 
# 各モデルサイズでの推定 Loss
print("=" * 60)
print(f"{'Params':>8} {'Tokens':>8} {'Loss':>8} {'最適比':>10}")
print("=" * 60)
for params in [7, 13, 70, 405]:
    for tokens in [1, 2, 5, 15]:
        loss = estimate_loss(params, tokens)
        ratio = tokens * 1e12 / (params * 1e9)
        optimal = "***" if 15 <= ratio <= 25 else ""
        print(f"{params:>5}B  {tokens:>5}T  {loss:>8.3f}  {ratio:>6.0f}:1 {optimal}")
 
# 計算予算に対する最適配分
print("\n" + "=" * 60)
print("計算予算に対する最適配分（Chinchilla則）")
print("=" * 60)
budgets_names = [
    ("1e21 (小規模実験)", 1e21),
    ("1e23 (7B級)", 1e23),
    ("1e24 (70B級)", 1e24),
    ("1e25 (GPT-4級)", 1e25),
]
for name, budget in budgets_names:
    N_opt, D_opt = compute_optimal_allocation(budget)
    print(f"\n計算予算: {name}")
    print(f"  最適パラメータ数: {N_opt/1e9:.1f}B")
    print(f"  最適トークン数: {D_opt/1e12:.2f}T")
    print(f"  推定Loss: {estimate_loss(N_opt/1e9, D_opt/1e12):.3f}")

import torch
import torch.nn as nn
import torch.nn.functional as F
 
class MoELayer(nn.Module):
    """Mixture of Experts レイヤー（簡易実装）"""
 
    def __init__(
        self,
        d_model: int = 512,
        num_experts: int = 8,
        top_k: int = 2,
        dim_ff: int = 2048,
    ):
        super().__init__()
        self.num_experts = num_experts
        self.top_k = top_k
 
        # ゲートネットワーク
        self.gate = nn.Linear(d_model, num_experts, bias=False)
 
        # エキスパート（各エキスパートは FFN）
        self.experts = nn.ModuleList([
            nn.Sequential(
                nn.Linear(d_model, dim_ff),
                nn.GELU(),
                nn.Linear(dim_ff, d_model),
            )
            for _ in range(num_experts)
        ])
 
    def forward(self, x):
        """
        Args:
            x: (batch_size, seq_len, d_model)
        Returns:
            output: (batch_size, seq_len, d_model)
        """
        batch_size, seq_len, d_model = x.shape
        x_flat = x.view(-1, d_model)  # (B*S, d_model)
 
        # ゲート計算
        gate_logits = self.gate(x_flat)  # (B*S, num_experts)
        gate_probs = F.softmax(gate_logits, dim=-1)
 
        # Top-K エキスパート選択
        top_k_probs, top_k_indices = torch.topk(gate_probs, self.top_k, dim=-1)
        top_k_probs = top_k_probs / top_k_probs.sum(dim=-1, keepdim=True)
 
        # 各エキスパートの出力を加重合算
        output = torch.zeros_like(x_flat)
        for k in range(self.top_k):
            expert_indices = top_k_indices[:, k]
            expert_weights = top_k_probs[:, k].unsqueeze(-1)
 
            for i in range(self.num_experts):
                mask = (expert_indices == i)
                if mask.any():
                    expert_input = x_flat[mask]
                    expert_output = self.expertsi
                    output[mask] += expert_weights[mask] * expert_output
 
        return output.view(batch_size, seq_len, d_model)
 
# 使用例
moe = MoELayer(d_model=512, num_experts=8, top_k=2)
x = torch.randn(2, 10, 512)
output = moe(x)
print(f"入力: {x.shape} → 出力: {output.shape}")
 
total_params = sum(p.numel() for p in moe.parameters())
print(f"総パラメータ数: {total_params:,}")
# 8エキスパートの FFN + ゲート

from transformers import AutoModelForCausalLM, AutoTokenizer
import torch
 
# モデルのロード（量子化オプション付き）
model_name = "meta-llama/Llama-3.1-8B-Instruct"
tokenizer = AutoTokenizer.from_pretrained(model_name)
 
# 量子化なし（フル精度）
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    torch_dtype=torch.bfloat16,  # BF16 で半分のメモリ使用
    device_map="auto",            # 自動デバイス配置
    attn_implementation="sdpa",   # PyTorch の最適化アテンション
)
 
# チャットテンプレートを使用（推奨）
messages = [
    {"role": "system", "content": "あなたは親切なAIアシスタントです。"},
    {"role": "user", "content": "大規模言語モデルの仕組みを簡潔に説明してください。"}
]
 
# テンプレート適用
input_ids = tokenizer.apply_chat_template(
    messages,
    add_generation_prompt=True,
    return_tensors="pt"
).to(model.device)
 
# 生成パラメータの設定
outputs = model.generate(
    input_ids,
    max_new_tokens=200,
    temperature=0.7,        # 多様性の制御
    top_p=0.9,              # Nucleus sampling
    repetition_penalty=1.1, # 繰り返し抑制
    do_sample=True,
)
 
# 入力部分を除いて出力
response = tokenizer.decode(
    outputs[0][input_ids.shape[-1]:],
    skip_special_tokens=True
)
print(response)

from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
import torch
 
# 4bit 量子化設定
quantization_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",          # NormalFloat4 量子化
    bnb_4bit_compute_dtype=torch.bfloat16,
    bnb_4bit_use_double_quant=True,     # 二重量子化でさらに圧縮
)
 
model_name = "meta-llama/Llama-3.1-70B-Instruct"
 
# 70B モデルを 4bit で読み込み（約35GB → 約17.5GB）
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    quantization_config=quantization_config,
    device_map="auto",
)
tokenizer = AutoTokenizer.from_pretrained(model_name)
 
# メモリ使用量の確認
total_memory = sum(
    p.nelement() * p.element_size() for p in model.parameters()
)
print(f"モデルメモリ使用量: {total_memory / 1e9:.2f} GB")
 
# 量子化の品質比較
# FP16:  70B × 2 bytes = 140 GB
# INT8:  70B × 1 byte  = 70 GB
# INT4:  70B × 0.5 byte = 35 GB
# NF4 + Double Quant ≈ 17.5 GB

def estimate_training_cost(
    params_billions: float,
    tokens_trillions: float,
    gpu_type: str = "H100",
    mfu: float = 0.4,  # Model FLOPs Utilization
):
    """学習コストの詳細な概算
 
    Args:
        params_billions: パラメータ数（十億単位）
        tokens_trillions: 学習トークン数（兆単位）
        gpu_type: GPU の種類
        mfu: Model FLOPs Utilization（通常 30-50%）
    """
    gpu_specs = {
        "A100_40GB": {"bf16_flops": 312e12, "cost_per_hour": 1.5, "memory_gb": 40},
        "A100_80GB": {"bf16_flops": 312e12, "cost_per_hour": 2.0, "memory_gb": 80},
        "H100_SXM": {"bf16_flops": 989e12, "cost_per_hour": 3.5, "memory_gb": 80},
        "H200":     {"bf16_flops": 989e12, "cost_per_hour": 4.5, "memory_gb": 141},
    }
    spec = gpu_specs.get(gpu_type, gpu_specs["H100_SXM"])
 
    # 理論的 FLOPs: 6 * N * D（前方 + 後方）
    total_flops = 6 * params_billions * 1e9 * tokens_trillions * 1e12
 
    # MFU を考慮した実効スループット
    effective_flops_per_gpu = spec["bf16_flops"] * mfu
 
    # 必要 GPU 時間
    gpu_seconds = total_flops / effective_flops_per_gpu
    gpu_hours = gpu_seconds / 3600
 
    # コスト計算
    cost = gpu_hours * spec["cost_per_hour"]
 
    # 電力消費概算（H100: ~700W）
    power_kwh = gpu_hours * 0.7  # kWh
    co2_tons = power_kwh * 0.4 / 1000  # CO2 トン（米国平均）
 
    print(f"{'='*60}")
    print(f"学習コスト概算: {params_billions}B params × {tokens_trillions}T tokens")
    print(f"{'='*60}")
    print(f"GPU: {gpu_type} (MFU: {mfu*100:.0f}%)")
    print(f"理論 FLOPs: {total_flops:.2e}")
    print(f"必要 GPU 時間: {gpu_hours:,.0f} 時間")
    print(f"推定コスト: ${cost:,.0f}")
    print(f"電力消費: {power_kwh:,.0f} kWh")
    print(f"CO2排出量: {co2_tons:,.0f} トン")
    print(f"\nGPU台数別の学習日数:")
    for num_gpus in [64, 256, 1024, 4096, 16384]:
        days = gpu_hours / num_gpus / 24
        if days >= 1:
            print(f"  {num_gpus:>6}台: {days:>8.1f} 日 (${cost/1e6:.1f}M)")
    print()
 
# 各モデル規模の概算
estimate_training_cost(8, 15, "H100_SXM", mfu=0.4)    # LLaMA 3.1 8B
estimate_training_cost(70, 15, "H100_SXM", mfu=0.35)   # LLaMA 3.1 70B
estimate_training_cost(405, 15, "H100_SXM", mfu=0.3)   # LLaMA 3.1 405B

import torch
import torch.nn as nn
 
class CausalSelfAttentionWithKVCache(nn.Module):
    """KV Cache 付き Self-Attention の概念実装"""
 
    def __init__(self, d_model: int, num_heads: int):
        super().__init__()
        self.d_model = d_model
        self.num_heads = num_heads
        self.d_k = d_model // num_heads
 
        self.W_q = nn.Linear(d_model, d_model)
        self.W_k = nn.Linear(d_model, d_model)
        self.W_v = nn.Linear(d_model, d_model)
        self.W_o = nn.Linear(d_model, d_model)
 
    def forward(self, x, kv_cache=None):
        """
        初回（プリフィル）: 全トークンを処理、KV Cache を構築
        2回目以降（デコード）: 新トークンのみ処理、KV Cache を更新
        """
        batch_size, seq_len, _ = x.shape
 
        Q = self.W_q(x).view(batch_size, seq_len, self.num_heads, self.d_k)
        K = self.W_k(x).view(batch_size, seq_len, self.num_heads, self.d_k)
        V = self.W_v(x).view(batch_size, seq_len, self.num_heads, self.d_k)
 
        if kv_cache is not None:
            # デコードフェーズ: 過去の KV を結合
            past_K, past_V = kv_cache
            K = torch.cat([past_K, K], dim=1)
            V = torch.cat([past_V, V], dim=1)
 
        # 新しい KV Cache
        new_kv_cache = (K, V)
 
        # アテンション計算（Q は現在のトークンのみ）
        Q = Q.transpose(1, 2)  # (B, H, S_q, D)
        K = K.transpose(1, 2)  # (B, H, S_kv, D)
        V = V.transpose(1, 2)  # (B, H, S_kv, D)
 
        scores = torch.matmul(Q, K.transpose(-2, -1)) / (self.d_k ** 0.5)
        weights = torch.softmax(scores, dim=-1)
        output = torch.matmul(weights, V)
 
        output = output.transpose(1, 2).reshape(batch_size, -1, self.d_model)
        output = self.W_o(output)
 
        return output, new_kv_cache
 
# KV Cache の効果を概算
def kv_cache_memory_estimate(
    num_layers: int,
    d_model: int,
    max_seq_len: int,
    batch_size: int = 1,
    dtype_bytes: int = 2,  # BF16
):
    """KV Cache のメモリ使用量を概算"""
    # 各層で K, V の2つのテンソル
    # 形状: (batch_size, seq_len, d_model)
    memory_per_layer = 2 * batch_size * max_seq_len * d_model * dtype_bytes
    total_memory = memory_per_layer * num_layers
 
    print(f"KV Cache メモリ推定:")
    print(f"  層数: {num_layers}, 次元: {d_model}, 最大長: {max_seq_len}")
    print(f"  バッチサイズ: {batch_size}")
    print(f"  合計: {total_memory / 1e9:.2f} GB")
    return total_memory
 
# 各モデルの KV Cache サイズ
kv_cache_memory_estimate(32, 4096, 8192, batch_size=1)    # 8B モデル
kv_cache_memory_estimate(80, 8192, 8192, batch_size=1)    # 70B モデル
kv_cache_memory_estimate(126, 16384, 8192, batch_size=1)  # 405B モデル

from vllm import LLM, SamplingParams
 
# vLLM でモデルロード（PagedAttention 自動適用）
llm = LLM(
    model="meta-llama/Llama-3.1-8B-Instruct",
    dtype="bfloat16",
    max_model_len=8192,
    gpu_memory_utilization=0.9,  # GPU メモリの90%を使用
    tensor_parallel_size=1,       # GPU 数
    enable_prefix_caching=True,   # 共通プレフィックスのキャッシュ
)
 
# サンプリングパラメータ
sampling_params = SamplingParams(
    temperature=0.7,
    top_p=0.9,
    max_tokens=512,
    repetition_penalty=1.1,
)
 
# バッチ推論（Continuous Batching が自動適用）
prompts = [
    "機械学習と深層学習の違いを説明してください。",
    "Pythonでクイックソートを実装してください。",
    "日本の経済政策の課題を3つ挙げてください。",
]
 
outputs = llm.generate(prompts, sampling_params)
 
for output in outputs:
    prompt = output.prompt
    generated_text = output.outputs[0].text
    print(f"プロンプト: {prompt[:50]}...")
    print(f"出力: {generated_text[:100]}...")
    print(f"トークン/秒: {len(output.outputs[0].token_ids) / output.metrics.finished_time:.1f}")
    print()

from dataclasses import dataclass
from enum import Enum
from typing import Optional
 
class ModelTier(Enum):
    SMALL = "small"    # 8B 級 / Haiku
    MEDIUM = "medium"  # 70B 級 / Sonnet
    LARGE = "large"    # 405B 級 / Opus
 
@dataclass
class RoutingDecision:
    model_tier: ModelTier
    reason: str
    estimated_cost_per_1k: float  # $/1K tokens
 
class ModelRouter:
    """タスクの複雑さに応じてモデルを振り分け"""
 
    # 複雑さの指標
    COMPLEXITY_SIGNALS = {
        "simple_keywords": [
            "分類", "抽出", "フォーマット", "変換",
            "翻訳", "要約", "ラベル付け"
        ],
        "complex_keywords": [
            "分析", "推論", "比較", "評価", "設計",
            "コード生成", "デバッグ", "創作", "戦略"
        ],
    }
 
    def route(self, task_description: str, max_latency_ms: Optional[int] = None) -> RoutingDecision:
        """タスク記述からモデルティアを決定"""
        desc_lower = task_description.lower()
 
        # レイテンシ制約がある場合
        if max_latency_ms and max_latency_ms < 200:
            return RoutingDecision(
                model_tier=ModelTier.SMALL,
                reason="レイテンシ制約により小型モデルを選択",
                estimated_cost_per_1k=0.0001,
            )
 
        # 複雑さの判定
        simple_score = sum(
            1 for kw in self.COMPLEXITY_SIGNALS["simple_keywords"]
            if kw in desc_lower
        )
        complex_score = sum(
            1 for kw in self.COMPLEXITY_SIGNALS["complex_keywords"]
            if kw in desc_lower
        )
 
        # テキスト長による追加判定
        if len(task_description) > 2000:
            complex_score += 1
 
        if complex_score > simple_score:
            if complex_score >= 3:
                return RoutingDecision(
                    model_tier=ModelTier.LARGE,
                    reason=f"高複雑度タスク (score: {complex_score})",
                    estimated_cost_per_1k=0.015,
                )
            return RoutingDecision(
                model_tier=ModelTier.MEDIUM,
                reason=f"中複雑度タスク (score: {complex_score})",
                estimated_cost_per_1k=0.003,
            )
 
        return RoutingDecision(
            model_tier=ModelTier.SMALL,
            reason=f"低複雑度タスク (score: {simple_score})",
            estimated_cost_per_1k=0.0001,
        )
 
# 使用例
router = ModelRouter()
 
tasks = [
    "このテキストをポジティブ/ネガティブに分類してください",
    "この設計書のセキュリティ脆弱性を分析し、改善策を推論してください",
    "複数のマイクロサービス間のAPI設計を比較・評価し、最適なアーキテクチャを提案してください",
]
 
for task in tasks:
    decision = router.route(task)
    print(f"タスク: {task[:50]}...")
    print(f"  → {decision.model_tier.value} ({decision.reason})")
    print(f"  → 推定コスト: ${decision.estimated_cost_per_1k}/1Kトークン")
    print()