from peft import LoraConfig, get_peft_model, TaskType
from transformers import AutoModelForCausalLM, AutoTokenizer, TrainingArguments
from trl import SFTTrainer
 
# ベースモデルのロード
model_name = "meta-llama/Llama-3.1-8B-Instruct"
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    torch_dtype="auto",
    device_map="auto",
)
tokenizer = AutoTokenizer.from_pretrained(model_name)
 
# LoRA 設定
lora_config = LoraConfig(
    task_type=TaskType.CAUSAL_LM,
    r=16,                          # ランク (8-64 が一般的)
    lora_alpha=32,                 # スケーリング係数
    lora_dropout=0.05,             # ドロップアウト率
    target_modules=[               # 適用する層
        "q_proj", "k_proj", "v_proj", "o_proj",
        "gate_proj", "up_proj", "down_proj",
    ],
)
 
# LoRA を適用
model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
# trainable params: 41,943,040 || all params: 8,072,204,288 || trainable%: 0.52%

from transformers import BitsAndBytesConfig
import torch
 
# 4bit 量子化設定
bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",        # NormalFloat4
    bnb_4bit_compute_dtype=torch.bfloat16,
    bnb_4bit_use_double_quant=True,    # 二重量子化
)
 
# QLoRA: 4bit量子化モデル + LoRA
model = AutoModelForCausalLM.from_pretrained(
    "meta-llama/Llama-3.1-8B-Instruct",
    quantization_config=bnb_config,
    device_map="auto",
)
 
# LoRA を適用（上記と同じ設定）
model = get_peft_model(model, lora_config)
# VRAM: 70B → ~40GB (QLoRA) vs ~140GB (フル精度)

from peft import LoraConfig, get_peft_model, TaskType
from transformers import AutoModelForCausalLM
 
# 異なる target_modules 設定の比較
configs = {
    "minimal": {
        "target_modules": ["q_proj", "v_proj"],
        "r": 16,
        "lora_alpha": 32,
    },
    "standard": {
        "target_modules": ["q_proj", "k_proj", "v_proj", "o_proj"],
        "r": 16,
        "lora_alpha": 32,
    },
    "full": {
        "target_modules": [
            "q_proj", "k_proj", "v_proj", "o_proj",
            "gate_proj", "up_proj", "down_proj",
        ],
        "r": 16,
        "lora_alpha": 32,
    },
}
 
for name, config_params in configs.items():
    lora_config = LoraConfig(
        task_type=TaskType.CAUSAL_LM,
        lora_dropout=0.05,
        **config_params,
    )
 
    model = AutoModelForCausalLM.from_pretrained(
        "meta-llama/Llama-3.1-8B-Instruct",
        torch_dtype="auto",
        device_map="auto",
    )
    model = get_peft_model(model, lora_config)
 
    trainable = sum(p.numel() for p in model.parameters() if p.requires_grad)
    total = sum(p.numel() for p in model.parameters())
    print(f"{name:10s}: trainable={trainable:>12,} ({trainable/total:.2%})")
 
# 出力例:
# minimal   : trainable=  13,107,200 (0.16%)
# standard  : trainable=  26,214,400 (0.32%)
# full      : trainable=  41,943,040 (0.52%)

import torch
import torch.nn as nn
 
class LoRALayer(nn.Module):
    """LoRA レイヤーの教育的実装"""
 
    def __init__(
        self,
        original_layer: nn.Linear,
        r: int = 16,
        alpha: float = 32.0,
        dropout: float = 0.05,
    ):
        super().__init__()
        self.original = original_layer
        self.r = r
        self.alpha = alpha
        self.scaling = alpha / r
 
        # 元の重みを凍結
        for param in self.original.parameters():
            param.requires_grad = False
 
        in_dim = original_layer.in_features
        out_dim = original_layer.out_features
 
        # 低ランク行列 A と B
        self.lora_A = nn.Linear(in_dim, r, bias=False)
        self.lora_B = nn.Linear(r, out_dim, bias=False)
        self.dropout = nn.Dropout(dropout)
 
        # A はランダム初期化、B はゼロ初期化
        # → 学習開始時は ΔW = 0 (元のモデルと同一)
        nn.init.kaiming_uniform_(self.lora_A.weight)
        nn.init.zeros_(self.lora_B.weight)
 
    def forward(self, x: torch.Tensor) -> torch.Tensor:
        # 元の出力 + LoRA の低ランク近似
        original_output = self.original(x)
        lora_output = self.lora_B(self.lora_A(self.dropout(x)))
        return original_output + self.scaling * lora_output
 
    def merge_weights(self):
        """推論時に LoRA 重みを元の重みにマージ (推論高速化)"""
        delta_w = self.scaling * (self.lora_B.weight @ self.lora_A.weight)
        self.original.weight.data += delta_w
        return self.original
 
 
# 使用例
linear = nn.Linear(4096, 4096)
lora_linear = LoRALayer(linear, r=16, alpha=32)
 
# 訓練可能パラメータ数
trainable = sum(p.numel() for p in lora_linear.parameters() if p.requires_grad)
total = sum(p.numel() for p in lora_linear.parameters())
print(f"Trainable: {trainable:,} / {total:,} ({trainable/total:.4%})")
# Trainable: 131,072 / 16,908,288 (0.7754%)

from datasets import Dataset, load_dataset
from transformers import AutoTokenizer
 
tokenizer = AutoTokenizer.from_pretrained("meta-llama/Llama-3.1-8B-Instruct")
 
# 方法1: ローカルデータから作成
raw_data = [
    {
        "instruction": "以下のPythonコードのバグを修正してください。",
        "input": "def add(a, b):\n    return a - b",
        "output": "def add(a, b):\n    return a + b\n\n# 修正: 減算(-) を加算(+) に変更しました。"
    },
    {
        "instruction": "SQLクエリを最適化してください。",
        "input": "SELECT * FROM users WHERE name LIKE '%田中%'",
        "output": (
            "SELECT id, name, email FROM users WHERE name LIKE '%田中%'\n\n"
            "-- 改善点:\n"
            "-- 1. SELECT * を必要なカラムに限定\n"
            "-- 2. LIKE前方一致の場合はインデックスが効くが、中間一致は全件走査"
        )
    },
    # ... 数百〜数千件
]
 
# チャットテンプレートに変換
def format_chat(example):
    messages = [
        {"role": "system", "content": "あなたは優秀なプログラミングアシスタントです。"},
        {"role": "user", "content": f"{example['instruction']}\n\n{example['input']}"},
        {"role": "assistant", "content": example["output"]},
    ]
    return {"text": tokenizer.apply_chat_template(messages, tokenize=False)}
 
dataset = Dataset.from_list(raw_data).map(format_chat)
 
# 方法2: Hugging Face Hub からロード
dataset_hf = load_dataset("kunishou/databricks-dolly-15k-ja")
 
 
# 方法3: JSONL ファイルから
import json
 
def load_jsonl(filepath: str) -> Dataset:
    data = []
    with open(filepath) as f:
        for line in f:
            data.append(json.loads(line))
    return Dataset.from_list(data)

from peft import LoraConfig, TaskType
from transformers import (
    AutoModelForCausalLM, AutoTokenizer,
    TrainingArguments, BitsAndBytesConfig,
)
from trl import SFTTrainer, SFTConfig
import torch
 
# モデルとトークナイザ
model_name = "meta-llama/Llama-3.1-8B-Instruct"
tokenizer = AutoTokenizer.from_pretrained(model_name)
tokenizer.pad_token = tokenizer.eos_token
 
# QLoRA 用量子化設定
bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_compute_dtype=torch.bfloat16,
    bnb_4bit_use_double_quant=True,
)
 
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    quantization_config=bnb_config,
    device_map="auto",
    attn_implementation="flash_attention_2",  # Flash Attention 2
)
 
# LoRA 設定
lora_config = LoraConfig(
    task_type=TaskType.CAUSAL_LM,
    r=16,
    lora_alpha=32,
    lora_dropout=0.05,
    target_modules=[
        "q_proj", "k_proj", "v_proj", "o_proj",
        "gate_proj", "up_proj", "down_proj",
    ],
)
 
# 学習設定
training_args = SFTConfig(
    output_dir="./output",
    num_train_epochs=3,
    per_device_train_batch_size=4,
    per_device_eval_batch_size=4,
    gradient_accumulation_steps=4,     # 実効バッチサイズ = 4 × 4 = 16
    learning_rate=2e-4,
    lr_scheduler_type="cosine",
    warmup_ratio=0.1,
    weight_decay=0.01,
    bf16=True,
    logging_steps=10,
    save_steps=100,
    eval_steps=100,
    eval_strategy="steps",
    save_total_limit=3,
    load_best_model_at_end=True,
    max_seq_length=2048,
    dataset_text_field="text",
    gradient_checkpointing=True,       # メモリ節約
    gradient_checkpointing_kwargs={"use_reentrant": False},
    optim="paged_adamw_8bit",          # メモリ効率的なオプティマイザ
    report_to="wandb",                 # Weights & Biases でモニタリング
)
 
# トレーナー
trainer = SFTTrainer(
    model=model,
    args=training_args,
    train_dataset=dataset["train"],
    eval_dataset=dataset["test"],
    tokenizer=tokenizer,
    peft_config=lora_config,
)
 
# 学習実行
trainer.train()
 
# モデル保存
trainer.save_model("./output/final")
tokenizer.save_pretrained("./output/final")

from transformers import TrainerCallback
import matplotlib.pyplot as plt
 
class LossMonitorCallback(TrainerCallback):
    """学習曲線をリアルタイムで監視"""
 
    def __init__(self):
        self.train_losses = []
        self.eval_losses = []
        self.steps = []
        self.eval_steps = []
 
    def on_log(self, args, state, control, logs=None, **kwargs):
        if "loss" in logs:
            self.train_losses.append(logs["loss"])
            self.steps.append(state.global_step)
 
        if "eval_loss" in logs:
            self.eval_losses.append(logs["eval_loss"])
            self.eval_steps.append(state.global_step)
 
            # 過学習検出: eval_loss が連続3回上昇
            if len(self.eval_losses) >= 3:
                if (self.eval_losses[-1] > self.eval_losses[-2] >
                    self.eval_losses[-3]):
                    print("WARNING: 過学習の兆候を検出。学習停止を検討してください。")
 
    def plot(self, save_path: str = "loss_curve.png"):
        plt.figure(figsize=(10, 6))
        plt.plot(self.steps, self.train_losses, label="Train Loss", alpha=0.7)
        if self.eval_losses:
            plt.plot(self.eval_steps, self.eval_losses, label="Eval Loss",
                     marker="o", linewidth=2)
        plt.xlabel("Steps")
        plt.ylabel("Loss")
        plt.title("Fine-tuning Loss Curve")
        plt.legend()
        plt.grid(True, alpha=0.3)
        plt.savefig(save_path, dpi=150, bbox_inches="tight")
        print(f"学習曲線を {save_path} に保存しました")
 
 
# 使用例
loss_monitor = LossMonitorCallback()
 
trainer = SFTTrainer(
    model=model,
    args=training_args,
    train_dataset=dataset["train"],
    eval_dataset=dataset["test"],
    tokenizer=tokenizer,
    peft_config=lora_config,
    callbacks=[loss_monitor],
)
 
trainer.train()
loss_monitor.plot()

from trl import DPOTrainer, DPOConfig
from datasets import load_dataset
 
# 好みデータセットの形式
# {"prompt": "...", "chosen": "良い応答", "rejected": "悪い応答"}
dataset = load_dataset("your-org/preference-dataset")
 
# DPO 設定
training_args = DPOConfig(
    output_dir="./dpo-output",
    num_train_epochs=3,
    per_device_train_batch_size=4,
    gradient_accumulation_steps=4,
    learning_rate=5e-7,
    beta=0.1,              # KLペナルティの強さ
    loss_type="sigmoid",   # "sigmoid", "hinge", "ipo"
    logging_steps=10,
    save_steps=100,
    bf16=True,
)
 
# DPO トレーナー
trainer = DPOTrainer(
    model=model,
    ref_model=None,  # PEFT使用時は自動でリファレンスモデル生成
    args=training_args,
    train_dataset=dataset["train"],
    tokenizer=tokenizer,
    peft_config=lora_config,
)
 
trainer.train()
trainer.save_model("./dpo-final")

from datasets import Dataset
import json
 
def create_preference_dataset(
    sft_model,
    tokenizer,
    prompts: list[str],
    n_responses: int = 4,
    temperature: float = 0.8,
) -> Dataset:
    """SFT モデルから好みデータセットを自動生成"""
    preference_data = []
 
    for prompt in prompts:
        # 複数の応答を生成
        responses = []
        for _ in range(n_responses):
            inputs = tokenizer(prompt, return_tensors="pt").to(sft_model.device)
            outputs = sft_model.generate(
                **inputs,
                max_new_tokens=512,
                temperature=temperature,
                do_sample=True,
            )
            response = tokenizer.decode(outputs[0], skip_special_tokens=True)
            responses.append(response)
 
        # LLM-as-a-Judge で評価 (GPT-4o を使用)
        from openai import OpenAI
        client = OpenAI()
 
        judge_prompt = f"""
以下の質問に対する{n_responses}つの回答を評価し、
最も良い回答と最も悪い回答を選んでください。
 
質問: {prompt}
 
回答:
{chr(10).join(f'{i+1}. {r}' for i, r in enumerate(responses))}
 
JSON形式で出力: {{"best": <番号>, "worst": <番号>, "reason": "<理由>"}}
"""
 
        judge_response = client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": judge_prompt}],
            response_format={"type": "json_object"},
            temperature=0,
        )
 
        result = json.loads(judge_response.choices[0].message.content)
 
        preference_data.append({
            "prompt": prompt,
            "chosen": responses[result["best"] - 1],
            "rejected": responses[result["worst"] - 1],
        })
 
    return Dataset.from_list(preference_data)
 
 
# 手動でのデータ作成テンプレート
manual_preference = [
    {
        "prompt": "Pythonでリストの重複を除去する方法は？",
        "chosen": (
            "リストの重複除去にはいくつかの方法があります:\n\n"
            "1. `set()` を使う方法（最もシンプル）:\n"
            "```python\n"
            "unique = list(set(original_list))\n"
            "```\n"
            "注意: 順序が保持されません。\n\n"
            "2. `dict.fromkeys()` で順序を保持:\n"
            "```python\n"
            "unique = list(dict.fromkeys(original_list))\n"
            "```\n"
        ),
        "rejected": "set使えばいいです。",
    },
    # ... 数百件
]

from trl import ORPOTrainer, ORPOConfig
 
# ORPO: SFT と DPO を同時に行う手法
# → SFT ステップが不要、より効率的
orpo_config = ORPOConfig(
    output_dir="./orpo-output",
    num_train_epochs=3,
    per_device_train_batch_size=4,
    gradient_accumulation_steps=4,
    learning_rate=8e-6,
    beta=0.1,              # ORPO のオッズ比パラメータ
    logging_steps=10,
    bf16=True,
    optim="paged_adamw_8bit",
    gradient_checkpointing=True,
)
 
trainer = ORPOTrainer(
    model=model,
    args=orpo_config,
    train_dataset=dataset["train"],
    tokenizer=tokenizer,
    peft_config=lora_config,
)
 
trainer.train()

from openai import OpenAI
import json
 
client = OpenAI()
 
# 1. 学習データの準備 (JSONL 形式)
training_data = [
    {
        "messages": [
            {"role": "system", "content": "技術文書を日本語で要約するアシスタント"},
            {"role": "user", "content": "以下の文書を要約してください: ..."},
            {"role": "assistant", "content": "要約: ..."}
        ]
    },
    # ... 数十〜数百例
]
 
with open("training_data.jsonl", "w") as f:
    for item in training_data:
        f.write(json.dumps(item, ensure_ascii=False) + "\n")
 
# 2. ファイルアップロード
file = client.files.create(
    file=open("training_data.jsonl", "rb"),
    purpose="fine-tune"
)
 
# 3. ファインチューニングジョブ作成
job = client.fine_tuning.jobs.create(
    training_file=file.id,
    model="gpt-4o-mini-2024-07-18",
    hyperparameters={
        "n_epochs": 3,
        "learning_rate_multiplier": 1.8,
        "batch_size": 4,
    }
)
 
# 4. ステータス確認
status = client.fine_tuning.jobs.retrieve(job.id)
print(f"Status: {status.status}")
# ファインチューニング済みモデル: ft:gpt-4o-mini:org-name::job-id

import json
from pathlib import Path
 
class OpenAIFTDataValidator:
    """OpenAI ファインチューニングデータの検証ツール"""
 
    def __init__(self, filepath: str):
        self.filepath = filepath
        self.data = []
        with open(filepath) as f:
            for line in f:
                self.data.append(json.loads(line))
 
    def validate(self) -> dict:
        """データの品質チェック"""
        issues = []
        stats = {
            "total_examples": len(self.data),
            "total_tokens": 0,
            "avg_tokens": 0,
            "max_tokens": 0,
            "min_tokens": 0,
        }
 
        token_counts = []
        for i, example in enumerate(self.data):
            messages = example.get("messages", [])
 
            # 必須フィールドチェック
            if not messages:
                issues.append(f"行 {i}: messages が空")
                continue
 
            roles = [m["role"] for m in messages]
 
            # システムプロンプトの一貫性
            if roles[0] == "system":
                system_content = messages[0]["content"]
            else:
                issues.append(f"行 {i}: system メッセージなし")
 
            # assistant 応答の存在確認
            if "assistant" not in roles:
                issues.append(f"行 {i}: assistant 応答なし")
 
            # トークン数の概算 (1トークン ≈ 4文字)
            total_chars = sum(len(m["content"]) for m in messages)
            est_tokens = total_chars // 4
            token_counts.append(est_tokens)
 
        if token_counts:
            stats["total_tokens"] = sum(token_counts)
            stats["avg_tokens"] = sum(token_counts) // len(token_counts)
            stats["max_tokens"] = max(token_counts)
            stats["min_tokens"] = min(token_counts)
 
        # 推奨チェック
        if len(self.data) < 10:
            issues.append("WARNING: 例が10件未満。最低50件を推奨")
        elif len(self.data) < 50:
            issues.append("NOTE: 例が50件未満。100件以上を推奨")
 
        # コスト概算 (gpt-4o-mini の FT 料金: $3.00/1M training tokens)
        cost_per_epoch = (stats["total_tokens"] / 1_000_000) * 3.00
        stats["estimated_cost_per_epoch"] = f"${cost_per_epoch:.2f}"
        stats["estimated_cost_3_epochs"] = f"${cost_per_epoch * 3:.2f}"
 
        return {"stats": stats, "issues": issues}
 
 
# 使用例
validator = OpenAIFTDataValidator("training_data.jsonl")
report = validator.validate()
print("=== データ検証レポート ===")
for key, value in report["stats"].items():
    print(f"  {key}: {value}")
if report["issues"]:
    print("\n問題点:")
    for issue in report["issues"]:
        print(f"  - {issue}")

from openai import OpenAI
 
client = OpenAI()
 
def compare_base_vs_ft(
    base_model: str,
    ft_model: str,
    test_prompts: list[dict],
) -> list[dict]:
    """ベースモデルとFTモデルの比較評価"""
    results = []
 
    for test in test_prompts:
        # ベースモデルの回答
        base_resp = client.chat.completions.create(
            model=base_model,
            messages=test["messages"],
            max_tokens=500,
            temperature=0,
        )
 
        # FTモデルの回答
        ft_resp = client.chat.completions.create(
            model=ft_model,
            messages=test["messages"],
            max_tokens=500,
            temperature=0,
        )
 
        # LLM-as-a-Judge で比較
        judge_resp = client.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "user",
                "content": f"""以下の2つの回答を比較評価してください。
 
質問: {test['messages'][-1]['content']}
 
回答A (ベース): {base_resp.choices[0].message.content}
 
回答B (FT): {ft_resp.choices[0].message.content}
 
JSON: {{"winner": "A" | "B" | "tie", "reason": "<理由>", "score_a": 1-5, "score_b": 1-5}}"""
            }],
            response_format={"type": "json_object"},
            temperature=0,
        )
 
        import json
        result = json.loads(judge_resp.choices[0].message.content)
        result["prompt"] = test["messages"][-1]["content"][:100]
        results.append(result)
 
    # 集計
    wins = {"A": 0, "B": 0, "tie": 0}
    for r in results:
        wins[r["winner"]] += 1
 
    print(f"ベース勝利: {wins['A']}, FT勝利: {wins['B']}, 引き分け: {wins['tie']}")
    return results

from peft import AutoPeftModelForCausalLM
from transformers import AutoTokenizer
 
# LoRA アダプタをベースモデルにマージ
model = AutoPeftModelForCausalLM.from_pretrained(
    "./output/final",             # LoRA アダプタのパス
    torch_dtype="auto",
    device_map="auto",
)
 
merged_model = model.merge_and_unload()
 
# マージ後のモデルを保存
merged_model.save_pretrained("./merged_model")
tokenizer = AutoTokenizer.from_pretrained("./output/final")
tokenizer.save_pretrained("./merged_model")
 
print("マージ完了。LoRA なしで推論可能になりました。")

# llama.cpp の convert スクリプトで GGUF に変換
python llama.cpp/convert_hf_to_gguf.py \
    ./merged_model \
    --outtype bf16 \
    --outfile ./merged_model.gguf
 
# 量子化 (4bit)
./llama.cpp/build/bin/llama-quantize \
    ./merged_model.gguf \
    ./merged_model-q4_k_m.gguf \
    Q4_K_M
 
# Ollama で利用
cat > Modelfile << 'EOF'
FROM ./merged_model-q4_k_m.gguf
 
SYSTEM """あなたは専門的なアシスタントです。"""
 
PARAMETER temperature 0.3
PARAMETER num_ctx 4096
EOF
 
ollama create my-finetuned -f Modelfile
ollama run my-finetuned

from huggingface_hub import HfApi
 
api = HfApi()
 
# リポジトリ作成
api.create_repo("your-org/my-finetuned-model", private=True)
 
# アップロード
api.upload_folder(
    folder_path="./merged_model",
    repo_id="your-org/my-finetuned-model",
    commit_message="Upload fine-tuned Llama 3.1 8B",
)
 
# または LoRA アダプタのみアップロード (軽量)
api.upload_folder(
    folder_path="./output/final",
    repo_id="your-org/my-lora-adapter",
    commit_message="Upload LoRA adapter",
)

def debug_training_data(dataset, tokenizer, n_samples: int = 5):
    """学習データのデバッグ"""
    print("=== 学習データ確認 ===")
    for i, example in enumerate(dataset.select(range(n_samples))):
        text = example.get("text", "")
        tokens = tokenizer.encode(text)
 
        print(f"\n--- Example {i+1} ---")
        print(f"文字数: {len(text)}")
        print(f"トークン数: {len(tokens)}")
        print(f"最初の200文字: {text[:200]}")
        print(f"最後の200文字: {text[-200:]}")
 
        # 特殊トークンの確認
        special_tokens = [
            t for t in tokens
            if t in tokenizer.all_special_ids
        ]
        print(f"特殊トークン数: {len(special_tokens)}")
 
    # 統計情報
    all_lengths = [len(tokenizer.encode(e["text"])) for e in dataset]
    print(f"\n=== 統計 ===")
    print(f"データ件数: {len(all_lengths)}")
    print(f"平均トークン数: {sum(all_lengths) / len(all_lengths):.0f}")
    print(f"最大トークン数: {max(all_lengths)}")
    print(f"最小トークン数: {min(all_lengths)}")

# NG: 全タスクで同じ学習率を使用
learning_rate = 2e-4  # 常にこの値
 
# OK: タスクとモデルサイズに応じて調整
learning_rates = {
    "sft_7b_lora":   2e-4,   # 小〜中モデルの LoRA SFT
    "sft_70b_lora":  5e-5,   # 大モデルの LoRA SFT
    "dpo_7b":        5e-7,   # DPO は低学習率
    "dpo_70b":       1e-7,   # 大モデルの DPO はさらに低く
    "openai_ft":     1.8,    # OpenAI API の multiplier
}
 
# ベストプラクティス: 学習率スケジュール
# 1. warmup (5-10% のステップ) で線形に上昇
# 2. cosine decay で徐々に低下
# 3. 最終学習率は初期の 10% 程度

# NG: 学習完了 → 即デプロイ
trainer.train()
deploy(model)  # 品質未確認
 
# OK: 段階的な評価プロセス
trainer.train()
 
# 1. 定量評価 (自動)
eval_results = evaluate_on_test_set(model, test_dataset)
if eval_results["score"] < baseline_score:
    raise ValueError("品質がベースラインを下回っています")
 
# 2. 定性評価 (人手サンプル)
samples = generate_samples(model, sample_prompts, n=20)
# 人手で確認
 
# 3. A/B テスト
# 既存モデルとの比較を実施
 
# 4. 段階的ロールアウト
# 10% のトラフィックで開始 → 問題なければ拡大

# 演習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