# Llama 3 をHugging Face Transformers で利用
from transformers import AutoTokenizer, AutoModelForCausalLM
import torch
 
model_id = "meta-llama/Meta-Llama-3.1-8B-Instruct"
 
tokenizer = AutoTokenizer.from_pretrained(model_id)
model = AutoModelForCausalLM.from_pretrained(
    model_id,
    torch_dtype=torch.bfloat16,
    device_map="auto",
)
 
messages = [
    {"role": "system", "content": "あなたは有能なアシスタントです。"},
    {"role": "user", "content": "Pythonのデコレータを説明してください。"},
]
 
input_ids = tokenizer.apply_chat_template(
    messages, return_tensors="pt"
).to(model.device)
 
outputs = model.generate(input_ids, max_new_tokens=512, temperature=0.7)
print(tokenizer.decode(outputs[0], skip_special_tokens=True))

# LoRA を使った Llama 3 のファインチューニング
from transformers import AutoModelForCausalLM, AutoTokenizer, TrainingArguments
from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training
from trl import SFTTrainer
from datasets import load_dataset
import torch
 
# 4bit量子化でモデルをロード
from transformers import BitsAndBytesConfig
 
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_id = "meta-llama/Meta-Llama-3.1-8B-Instruct"
model = AutoModelForCausalLM.from_pretrained(
    model_id,
    quantization_config=bnb_config,
    device_map="auto",
)
tokenizer = AutoTokenizer.from_pretrained(model_id)
tokenizer.pad_token = tokenizer.eos_token
 
# LoRA 設定
lora_config = LoraConfig(
    r=16,                        # LoRA のランク
    lora_alpha=32,               # スケーリング係数
    target_modules=[             # 適用対象レイヤー
        "q_proj", "k_proj", "v_proj", "o_proj",
        "gate_proj", "up_proj", "down_proj",
    ],
    lora_dropout=0.05,
    bias="none",
    task_type="CAUSAL_LM",
)
 
model = prepare_model_for_kbit_training(model)
model = get_peft_model(model, lora_config)
 
# 訓練可能パラメータの確認
model.print_trainable_parameters()
# → trainable params: 41,943,040 || all params: 8,030,261,248 || 0.52%
 
# データセットの準備
dataset = load_dataset("json", data_files="train_data.jsonl")
 
def format_instruction(example):
    return f"""<|begin_of_text|><|start_header_id|>system<|end_header_id|>
あなたは有能なアシスタントです。<|eot_id|><|start_header_id|>user<|end_header_id|>
{example['instruction']}<|eot_id|><|start_header_id|>assistant<|end_header_id|>
{example['output']}<|eot_id|>"""
 
# 訓練実行
training_args = TrainingArguments(
    output_dir="./llama3-ft",
    num_train_epochs=3,
    per_device_train_batch_size=4,
    gradient_accumulation_steps=4,
    learning_rate=2e-4,
    warmup_ratio=0.03,
    logging_steps=10,
    save_strategy="epoch",
    bf16=True,
)
 
trainer = SFTTrainer(
    model=model,
    args=training_args,
    train_dataset=dataset["train"],
    formatting_func=format_instruction,
    max_seq_length=2048,
)
 
trainer.train()
model.save_pretrained("./llama3-ft-lora")

# Mixtral 8x7B の利用例 (vLLM)
from vllm import LLM, SamplingParams
 
llm = LLM(
    model="mistralai/Mixtral-8x7B-Instruct-v0.1",
    tensor_parallel_size=2,  # 2 GPU で分割
    dtype="bfloat16",
)
 
sampling_params = SamplingParams(
    temperature=0.7,
    top_p=0.9,
    max_tokens=1024,
)
 
outputs = llm.generate(
    ["Rustの所有権システムを解説してください。"],
    sampling_params,
)
 
for output in outputs:
    print(output.outputs[0].text)

# Mistral API (OpenAI互換形式)
from openai import OpenAI
 
client = OpenAI(
    api_key="YOUR_MISTRAL_API_KEY",
    base_url="https://api.mistral.ai/v1",
)
 
response = client.chat.completions.create(
    model="mistral-large-latest",
    messages=[
        {"role": "system", "content": "あなたはプログラミングの専門家です。"},
        {"role": "user", "content": "Go言語のgoroutineとチャネルの使い方を教えてください。"},
    ],
    temperature=0.7,
    max_tokens=2048,
)
 
print(response.choices[0].message.content)
 
# Function Calling も対応
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_stock_price",
            "description": "株価を取得します",
            "parameters": {
                "type": "object",
                "properties": {
                    "symbol": {"type": "string", "description": "銘柄コード"},
                },
                "required": ["symbol"],
            },
        },
    }
]
 
response = client.chat.completions.create(
    model="mistral-large-latest",
    messages=[{"role": "user", "content": "Appleの株価を教えて"}],
    tools=tools,
    tool_choice="auto",
)

# Qwen 2.5 — 日本語性能が高いOSSモデル
from transformers import AutoModelForCausalLM, AutoTokenizer
 
model_name = "Qwen/Qwen2.5-7B-Instruct"
 
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    torch_dtype="auto",
    device_map="auto",
)
 
messages = [
    {"role": "system", "content": "あなたは日本語に堪能なアシスタントです。"},
    {"role": "user", "content": "日本の四季について俳句を3つ作ってください。"},
]
 
text = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
inputs = tokenizer([text], return_tensors="pt").to(model.device)
outputs = model.generate(**inputs, max_new_tokens=256)
print(tokenizer.decode(outputs[0], skip_special_tokens=True))

# Qwen-VL (Vision-Language) の利用
from transformers import Qwen2VLForConditionalGeneration, AutoProcessor
from PIL import Image
import torch
 
model = Qwen2VLForConditionalGeneration.from_pretrained(
    "Qwen/Qwen2-VL-7B-Instruct",
    torch_dtype=torch.bfloat16,
    device_map="auto",
)
processor = AutoProcessor.from_pretrained("Qwen/Qwen2-VL-7B-Instruct")
 
# 画像付きプロンプト
image = Image.open("diagram.png")
messages = [
    {
        "role": "user",
        "content": [
            {"type": "image", "image": image},
            {"type": "text", "text": "この図を詳しく説明してください。"},
        ],
    }
]
 
text = processor.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
inputs = processor(text=text, images=[image], return_tensors="pt").to(model.device)
outputs = model.generate(**inputs, max_new_tokens=512)
print(processor.decode(outputs[0], skip_special_tokens=True))

# Qwen2.5-Coder — コード生成特化モデル
from transformers import AutoModelForCausalLM, AutoTokenizer
 
model_name = "Qwen/Qwen2.5-Coder-7B-Instruct"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(
    model_name, torch_dtype="auto", device_map="auto"
)
 
messages = [
    {"role": "system", "content": "あなたは熟練のソフトウェアエンジニアです。"},
    {"role": "user", "content": """
以下の要件でPythonのクラスを実装してください:
- 名前: AsyncRateLimiter
- Token Bucket アルゴリズムによるレート制限
- asyncio 対応
- 設定可能なレート (リクエスト/秒) とバースト数
"""},
]
 
text = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
inputs = tokenizer([text], return_tensors="pt").to(model.device)
outputs = model.generate(**inputs, max_new_tokens=2048, temperature=0.2)
print(tokenizer.decode(outputs[0], skip_special_tokens=True))

# DeepSeek-R1 は思考過程を明示的に出力
from openai import OpenAI
 
client = OpenAI(
    api_key="YOUR_KEY",
    base_url="https://api.deepseek.com"
)
 
response = client.chat.completions.create(
    model="deepseek-reasoner",
    messages=[
        {"role": "user", "content": "271は素数ですか？証明してください。"}
    ]
)
 
# reasoning_content に思考過程が含まれる
print("思考過程:", response.choices[0].message.reasoning_content)
print("最終回答:", response.choices[0].message.content)

# DeepSeek-R1 の蒸留版をローカルで実行 (Ollama)
import subprocess
import requests
 
# Ollama でモデルをダウンロード・実行
# $ ollama pull deepseek-r1:7b
 
# API 経由で利用
response = requests.post(
    "http://localhost:11434/api/chat",
    json={
        "model": "deepseek-r1:7b",
        "messages": [
            {"role": "user", "content": "フィボナッチ数列の第100項を求めるアルゴリズムを説明してください。"}
        ],
        "stream": False,
    },
)
 
result = response.json()
print(result["message"]["content"])

# Gemma 2 — Google の軽量高性能モデル
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch
 
model = AutoModelForCausalLM.from_pretrained(
    "google/gemma-2-9b-it",
    torch_dtype=torch.bfloat16,
    device_map="auto",
)
tokenizer = AutoTokenizer.from_pretrained("google/gemma-2-9b-it")
 
# Gemma 2 の特徴:
# - 知識蒸留 (Knowledge Distillation) で小型化
# - Sliding Window Attention + Global Attention の交互使用
# - 2B / 9B / 27B のサイズバリエーション
# - Gemma License (研究・商用利用可、再配布時にライセンス添付必要)

# Phi-3/4 — Microsoft の小型高性能モデル
# Phi-3 Mini (3.8B) は同サイズで最高性能クラス
 
from transformers import AutoModelForCausalLM, AutoTokenizer
 
model = AutoModelForCausalLM.from_pretrained(
    "microsoft/Phi-3-mini-4k-instruct",
    torch_dtype="auto",
    device_map="auto",
    trust_remote_code=True,
)
 
# Phi の特徴:
# - 高品質な「教科書レベル」のデータで訓練
# - 小型でもGPT-3.5を上回る性能
# - MIT ライセンス
# - 128K コンテキスト版も存在
# - Phi-4 (14B) は Qwen 2.5 72B と同等性能を主張

# GPTQ 量子化モデルの利用
from transformers import AutoModelForCausalLM, AutoTokenizer
 
model = AutoModelForCausalLM.from_pretrained(
    "TheBloke/Llama-3-8B-Instruct-GPTQ",
    device_map="auto",
    trust_remote_code=True,
)
 
# メモリ使用量の比較
# FP16:  8B モデル → 約 16GB VRAM
# INT8:  8B モデル → 約  8GB VRAM
# INT4:  8B モデル → 約  4GB VRAM (GPTQ/AWQ)
# GGUF:  8B モデル → 約  4-6GB (llama.cpp、CPU可)

# vLLM — 高性能推論サーバー
# $ pip install vllm
# $ vllm serve meta-llama/Meta-Llama-3.1-8B-Instruct --port 8000
 
from openai import OpenAI
 
# vLLM は OpenAI 互換 API を提供
client = OpenAI(
    base_url="http://localhost:8000/v1",
    api_key="dummy",  # vLLM ではダミーでOK
)
 
response = client.chat.completions.create(
    model="meta-llama/Meta-Llama-3.1-8B-Instruct",
    messages=[
        {"role": "user", "content": "Hello!"},
    ],
    max_tokens=256,
)
print(response.choices[0].message.content)

# vLLM を使った本番デプロイ用 Dockerfile
FROM vllm/vllm-openai:latest
 
# モデルのダウンロード (事前ダウンロードも可)
ENV MODEL_NAME=Qwen/Qwen2.5-7B-Instruct
ENV MAX_MODEL_LEN=8192
ENV GPU_MEMORY_UTILIZATION=0.9
 
# ヘルスチェック
HEALTHCHECK --interval=30s --timeout=10s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1
 
EXPOSE 8000
 
CMD ["python", "-m", "vllm.entrypoints.openai.api_server", \
     "--model", "${MODEL_NAME}", \
     "--max-model-len", "${MAX_MODEL_LEN}", \
     "--gpu-memory-utilization", "${GPU_MEMORY_UTILIZATION}", \
     "--host", "0.0.0.0", \
     "--port", "8000"]

# docker-compose.yml
version: '3.8'
services:
  llm-server:
    image: vllm/vllm-openai:latest
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]
    ports:
      - "8000:8000"
    environment:
      - HUGGING_FACE_HUB_TOKEN=${HF_TOKEN}
    command: >
      python -m vllm.entrypoints.openai.api_server
      --model Qwen/Qwen2.5-7B-Instruct
      --max-model-len 8192
      --host 0.0.0.0
      --port 8000
    volumes:
      - model-cache:/root/.cache/huggingface
 
volumes:
  model-cache:

# 自社環境での推論性能ベンチマーク
import time
import statistics
from openai import OpenAI
 
client = OpenAI(base_url="http://localhost:8000/v1", api_key="dummy")
 
def benchmark_model(
    model: str,
    prompts: list[str],
    max_tokens: int = 256,
    num_runs: int = 3,
) -> dict:
    """推論性能のベンチマーク"""
    latencies = []
    throughputs = []
 
    for prompt in prompts:
        for _ in range(num_runs):
            start = time.time()
            response = client.chat.completions.create(
                model=model,
                messages=[{"role": "user", "content": prompt}],
                max_tokens=max_tokens,
            )
            elapsed = time.time() - start
 
            output_tokens = response.usage.completion_tokens
            latencies.append(elapsed)
            throughputs.append(output_tokens / elapsed)
 
    return {
        "model": model,
        "avg_latency_ms": statistics.mean(latencies) * 1000,
        "p50_latency_ms": statistics.median(latencies) * 1000,
        "p95_latency_ms": sorted(latencies)[int(len(latencies) * 0.95)] * 1000,
        "avg_throughput_tps": statistics.mean(throughputs),
        "total_requests": len(latencies),
    }
 
# ベンチマーク実行
test_prompts = [
    "Pythonでバブルソートを実装してください。",
    "機械学習とディープラーニングの違いを説明してください。",
    "日本の歴史について500文字程度で要約してください。",
]
 
result = benchmark_model("Qwen/Qwen2.5-7B-Instruct", test_prompts)
print(f"平均レイテンシ: {result['avg_latency_ms']:.0f}ms")
print(f"P95 レイテンシ: {result['p95_latency_ms']:.0f}ms")
print(f"平均スループット: {result['avg_throughput_tps']:.1f} tokens/s")

# NG: 本番環境で FP16 のまま大型モデルを運用
model = AutoModelForCausalLM.from_pretrained(
    "meta-llama/Meta-Llama-3.1-70B-Instruct",
    torch_dtype=torch.float16,  # 140GB VRAM 必要
)
 
# OK: 用途に応じた量子化を適用
# 品質重視 → INT8 (AWQ): 品質低下 1-2%、メモリ半減
# コスト重視 → INT4 (GPTQ): 品質低下 3-5%、メモリ 1/4
# CPU 実行 → GGUF Q4_K_M: llama.cpp で CPU 推論可能

# NG: 認証なしで LLM API を公開
# vllm serve model --host 0.0.0.0  # インターネットから誰でもアクセス可能
 
# OK: 適切な認証・レート制限・プロンプトインジェクション対策
# 1. リバースプロキシ (nginx) で認証を追加
# 2. API キーによるアクセス制御
# 3. レート制限 (リクエスト数/秒)
# 4. 入力長の制限
# 5. 出力フィルタリング (有害コンテンツ検出)

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