Python実務で必ず使うディープラーニングレシピ100選

# CPU版（GPUなしの環境や動作確認用）
pip install torch torchvision torchaudio

# GPU版（CUDA対応）はCUDAのバージョンによってコマンドが変わります
# 公式サイト（https://pytorch.org/get-started/locally/）のセレクタで
# 環境に合ったコマンドを生成してから実行してください

import torch

# バージョンとGPU利用可否の確認
print("PyTorchバージョン:", torch.__version__)
print("GPU利用可能:", torch.cuda.is_available())

if torch.cuda.is_available():
    print("GPU数:", torch.cuda.device_count())
    print("GPU名:", torch.cuda.get_device_name(0))
    print("CUDAバージョン:", torch.version.cuda)
else:
    print("GPUが検出されませんでした。CPUで実行します。")

# 簡単な動作確認（CPUでも問題なく動きます）
x = torch.rand(3, 3)
print("動作確認用テンソル:\n", x)

import torch

# 代表的なテンソル生成方法
a = torch.tensor([[1.0, 2.0], [3.0, 4.0]])
zeros = torch.zeros(2, 3)
ones = torch.ones(2, 3)
randn = torch.randn(2, 3)  # 標準正規分布からサンプリング

print("shape:", a.shape)
print("dtype:", a.dtype)
print("ndim:", a.ndim)

# 四則演算（要素ごと）
b = a * 2 + 1
print("要素ごとの演算:\n", b)

# 行列積
c = a @ a  # torch.matmul(a, a) と同じ
print("行列積:\n", c)

# 形状変換
d = torch.arange(12)
e = d.reshape(3, 4)
print("reshape後のshape:", e.shape)

import numpy as np
import pandas as pd
import torch

# NumPy配列からテンソルへ
# NumPyの既定はfloat64なので、共有したまま渡すために配列側をfloat32で作る。
# ここで .float() を挟むと型変換のコピーが発生し、メモリ共有は切れる。
arr = np.array([[1.0, 2.0], [3.0, 4.0]], dtype=np.float32)
tensor_from_np = torch.from_numpy(arr)
print("NumPy -> Tensor:", tensor_from_np.dtype)

# テンソルからNumPy配列へ
back_to_np = tensor_from_np.numpy()
print("Tensor -> NumPy:", type(back_to_np))

# DataFrameからテンソルへ
df = pd.DataFrame({
    "身長": [170.0, 165.0, 180.0],
    "体重": [65.0, 55.0, 75.0],
})
tensor_from_df = torch.tensor(df.values, dtype=torch.float32)
print("DataFrame -> Tensor:\n", tensor_from_df)

import torch

# 定番のデバイス定義イディオム
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print("使用デバイス:", device)

# テンソルをデバイスに転送
x = torch.randn(4, 4)
x = x.to(device)
print("転送後のデバイス:", x.device)

# モデルもto(device)で丸ごと転送する
model = torch.nn.Linear(4, 2)
model = model.to(device)

# 演算前に両者のデバイスを揃える必要がある
y = model(x)  # xとmodelが同じdeviceにあるため問題なく実行できる
print("出力:\n", y)

# 結果をCPUに戻してからNumPy変換する（GPU上のテンソルは直接numpy()できない）
y_cpu = y.detach().cpu().numpy()
print("NumPy化後の型:", type(y_cpu))

import torch

# requires_grad=Trueで勾配計算の対象にする
x = torch.tensor([2.0], requires_grad=True)
target = torch.tensor([10.0])

# 簡単な損失関数（二乗誤差）
loss = (x * 3 - target) ** 2
print("損失:", loss.item())

# 逆伝播で勾配を計算
loss.backward()
print("xの勾配:", x.grad)

# 勾配降下法の1ステップを手書きで実装
lr = 0.01
with torch.no_grad():
    x -= lr * x.grad

# 次の反復に備えて勾配をリセット（累積されてしまうため）
x.grad.zero_()
print("更新後のx:", x.item())

import torch
from torch.utils.data import Dataset
import pandas as pd

class TabularDataset(Dataset):
    """テーブルデータ用の最小構成Dataset"""

    def __init__(self, df: pd.DataFrame, feature_cols: list, target_col: str):
        self.X = torch.tensor(df[feature_cols].values, dtype=torch.float32)
        # 回帰の出力層は [バッチ, 1] を返すので、正解も [N, 1] に揃えておく。
        # [N] のままだとMSELossでブロードキャストが起き、意図しない損失になる
        self.y = torch.tensor(df[target_col].values, dtype=torch.float32).view(-1, 1)

    def __len__(self):
        return len(self.X)

    def __getitem__(self, idx):
        return self.X[idx], self.y[idx]

# サンプルデータで動作確認
df = pd.DataFrame({
    "気温": [20.0, 25.0, 15.0, 30.0],
    "湿度": [50.0, 60.0, 40.0, 70.0],
    "来客数": [100.0, 150.0, 80.0, 200.0],
})
dataset = TabularDataset(df, feature_cols=["気温", "湿度"], target_col="来客数")

print("データ件数:", len(dataset))
print("1件目:", dataset[0])

import torch
from torch.utils.data import Dataset, DataLoader
import pandas as pd

class TabularDataset(Dataset):
    def __init__(self, df, feature_cols, target_col):
        self.X = torch.tensor(df[feature_cols].values, dtype=torch.float32)
        # 回帰の出力層は [バッチ, 1] を返すので、正解も [N, 1] に揃えておく。
        # [N] のままだとMSELossでブロードキャストが起き、意図しない損失になる
        self.y = torch.tensor(df[target_col].values, dtype=torch.float32).view(-1, 1)

    def __len__(self):
        return len(self.X)

    def __getitem__(self, idx):
        return self.X[idx], self.y[idx]

# 20件のダミーデータを用意
df = pd.DataFrame({
    "気温": range(20),
    "湿度": range(20, 40),
    "来客数": range(100, 120),
})
dataset = TabularDataset(df, ["気温", "湿度"], "来客数")

# バッチサイズ4、シャッフルありのDataLoader
loader = DataLoader(dataset, batch_size=4, shuffle=True)

for epoch in range(1):
    for batch_idx, (X_batch, y_batch) in enumerate(loader):
        print(f"バッチ{batch_idx}: X shape={X_batch.shape}, y shape={y_batch.shape}")

import random
import numpy as np
import torch

def set_seed(seed: int = 42):
    """再現性確保のためのシード固定"""
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    torch.cuda.manual_seed_all(seed)  # GPU未使用時も無害

    # cudnnの挙動を決定論的にする（速度は低下する場合がある）
    torch.backends.cudnn.deterministic = True
    torch.backends.cudnn.benchmark = False

set_seed(42)

# 同じシードなら同じ乱数列になることを確認
print(torch.randn(3))
set_seed(42)
print(torch.randn(3))  # 上と同じ値になる

pip install torchinfo

import torch
import torch.nn as nn
from torchinfo import summary

class SimpleNet(nn.Module):
    def __init__(self):
        super().__init__()
        self.fc1 = nn.Linear(10, 32)
        self.fc2 = nn.Linear(32, 16)
        self.fc3 = nn.Linear(16, 1)
        self.relu = nn.ReLU()

    def forward(self, x):
        x = self.relu(self.fc1(x))
        x = self.relu(self.fc2(x))
        return self.fc3(x)

model = SimpleNet()

# 自前でパラメータ数を数える方法
total_params = sum(p.numel() for p in model.parameters() if p.requires_grad)
print("総パラメータ数:", total_params)

# torchinfoによる詳細サマリー（層ごとの出力形状も確認できる）
summary(model, input_size=(1, 10))

import numpy as np
import tensorflow as tf

# ダミーデータ（気温・湿度から来客数を予測する回帰タスク）
np.random.seed(42)
X = np.random.rand(200, 2).astype("float32")
y = (X[:, 0] * 100 + X[:, 1] * 50).astype("float32")

# Sequentialで層を積むだけでモデルを定義できる
model = tf.keras.Sequential([
    tf.keras.layers.Input(shape=(2,)),
    tf.keras.layers.Dense(32, activation="relu"),
    tf.keras.layers.Dense(16, activation="relu"),
    tf.keras.layers.Dense(1),
])

# compileで最適化手法・損失関数・評価指標をまとめて指定
model.compile(optimizer="adam", loss="mse", metrics=["mae"])

# fitを呼ぶだけで学習ループが実行される
model.fit(X, y, epochs=5, batch_size=16, verbose=1)

# Colabの手順（メニュー操作）:
# 1. 上部メニュー「ランタイム」->「ランタイムのタイプを変更」
# 2. 「ハードウェアアクセラレータ」で GPU（例: T4）を選択して保存
# 3. 以下のコードセルで割り当てられたGPUを確認

import torch
print("GPU利用可能:", torch.cuda.is_available())
!nvidia-smi

# Googleドライブのマウント（データや学習済みモデルの永続化に使う）
from google.colab import drive
drive.mount("/content/drive")

# 追加ライブラリのインストール（セッションごとに再インストールが必要）
!pip install -q torchinfo

# ドライブ上のフォルダをそのまま作業ディレクトリとして使う例
import os
save_dir = "/content/drive/MyDrive/dl_recipes"
os.makedirs(save_dir, exist_ok=True)
print("保存先:", save_dir)

import os

# キャッシュ先を変更したい場合は、モデルを読み込む前に環境変数を設定する
# (デフォルト: torchvisionは~/.cache/torch/、Hugging Faceは~/.cache/huggingface/)
os.environ["TORCH_HOME"] = "/path/to/custom_cache"      # torchvision系のキャッシュ先
os.environ["HF_HOME"] = "/path/to/custom_hf_cache"      # Hugging Face系(HF_HUB_CACHEでハブのみの指定も可)

import torchvision.models as models

# 初回実行時はインターネットから重みがダウンロードされる
model = models.resnet18(weights=models.ResNet18_Weights.DEFAULT)
print("ResNet18のロード完了（2回目以降はキャッシュから読み込まれます）")

# Hugging Face Hub側のモデルも同様にキャッシュされる
from transformers import AutoModel, AutoTokenizer

model_name = "bert-base-uncased"
tokenizer = AutoTokenizer.from_pretrained(model_name)  # 初回のみダウンロード
bert_model = AutoModel.from_pretrained(model_name)

import torch
import torch.nn as nn

torch.manual_seed(0)

class TwoLayerNet(nn.Module):
    def __init__(self, input_dim, hidden_dim, output_dim):
        super().__init__()
        self.fc1 = nn.Linear(input_dim, hidden_dim)
        self.relu = nn.ReLU()
        self.fc2 = nn.Linear(hidden_dim, output_dim)

    def forward(self, x):
        x = self.fc1(x)
        x = self.relu(x)
        x = self.fc2(x)
        return x

model = TwoLayerNet(input_dim=4, hidden_dim=8, output_dim=1)
print(model)

# ダミー入力で動作確認（バッチサイズ5、特徴量4次元）
dummy_input = torch.randn(5, 4)
output = model(dummy_input)
print(output.shape)  # torch.Size([5, 1])

import torch
import torch.nn as nn

torch.manual_seed(0)

X = torch.randn(100, 4)
y = X.sum(dim=1, keepdim=True) + 0.1 * torch.randn(100, 1)

model = nn.Sequential(nn.Linear(4, 8), nn.ReLU(), nn.Linear(8, 1))
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.01)

n_epochs = 50
for epoch in range(n_epochs):
    optimizer.zero_grad()          # 1. 勾配のリセット
    y_pred = model(X)              # 2. forward
    loss = criterion(y_pred, y)    # 3. 損失計算
    loss.backward()                # 4. backward（勾配計算）
    optimizer.step()               # 5. パラメータ更新

    if (epoch + 1) % 10 == 0:
        print(f"epoch {epoch+1}, loss {loss.item():.4f}")

import torch
import torch.nn as nn

torch.manual_seed(0)

y_pred = torch.tensor([2.5, 0.0, 2.1, 7.8, 5.0])
y_true = torch.tensor([3.0, -0.5, 2.0, 7.0, 10.0])  # 最後の1点だけ大きく外れた値

mse_loss = nn.MSELoss()
l1_loss = nn.L1Loss()
huber_loss = nn.HuberLoss(delta=1.0)

print(f"MSELoss  : {mse_loss(y_pred, y_true).item():.4f}")   # 外れ値の影響を強く受ける
print(f"L1Loss   : {l1_loss(y_pred, y_true).item():.4f}")    # 外れ値の影響が小さい
print(f"HuberLoss: {huber_loss(y_pred, y_true).item():.4f}") # 両者の中間的な挙動

import torch
import torch.nn as nn

torch.manual_seed(0)

# 3クラス分類、バッチサイズ4のロジット（Softmax適用前の生の値）
logits = torch.randn(4, 3)
targets = torch.tensor([0, 2, 1, 1])  # 正解クラスのインデックス

criterion = nn.CrossEntropyLoss()

# 正しい使い方: ロジットをそのまま渡す
loss = criterion(logits, targets)
print(f"CrossEntropyLoss(ロジット直接) : {loss.item():.4f}")

# 誤った使い方の例: Softmaxを適用してから渡すと計算がずれる
softmax_probs = torch.softmax(logits, dim=1)
wrong_loss = criterion(softmax_probs, targets)
print(f"CrossEntropyLoss(Softmax後、誤り): {wrong_loss.item():.4f}")

import torch
import torch.nn as nn

torch.manual_seed(0)

model = nn.Sequential(nn.Linear(4, 8), nn.ReLU(), nn.Linear(8, 1))

optimizers = {
    "SGD": torch.optim.SGD(model.parameters(), lr=0.01),
    "SGD+Momentum": torch.optim.SGD(model.parameters(), lr=0.01, momentum=0.9),
    "Adam": torch.optim.Adam(model.parameters(), lr=0.001),
    "AdamW": torch.optim.AdamW(model.parameters(), lr=0.001, weight_decay=0.01),
}

for name, optimizer in optimizers.items():
    print(f"{name}: lr={optimizer.defaults['lr']}")

# 使い分けの目安
# SGD          : 学習の挙動を素直に把握したいベースライン検証向け
# SGD+Momentum : 収束を安定させたい画像系モデルの定番
# Adam         : 収束が速く、まず試す既定の選択肢
# AdamW        : weight_decayを正しく効かせたいTransformer系で優先

import torch
import torch.nn as nn

torch.manual_seed(0)

model = nn.Sequential(nn.Linear(4, 8), nn.ReLU(), nn.Linear(8, 1))
optimizer = torch.optim.Adam(model.parameters(), lr=0.1)

# StepLR: 10エポックごとに学習率を0.5倍にする
step_scheduler = torch.optim.lr_scheduler.StepLR(optimizer, step_size=10, gamma=0.5)

X = torch.randn(20, 4)
y = torch.randn(20, 1)
criterion = nn.MSELoss()

for epoch in range(30):
    optimizer.zero_grad()
    loss = criterion(model(X), y)
    loss.backward()
    optimizer.step()
    step_scheduler.step()  # optimizer.step()の後、エポック末尾で呼ぶ

    if (epoch + 1) % 10 == 0:
        print(f"epoch {epoch+1}, lr {optimizer.param_groups[0]['lr']:.5f}")

import torch
import torch.nn as nn

torch.manual_seed(0)

model = nn.Sequential(nn.Linear(4, 8), nn.Dropout(0.3), nn.ReLU(), nn.Linear(8, 1))
criterion = nn.MSELoss()

X_train = torch.randn(80, 4)
y_train = torch.randn(80, 1)
X_val = torch.randn(20, 4)
y_val = torch.randn(20, 1)

optimizer = torch.optim.Adam(model.parameters(), lr=0.01)

for epoch in range(5):
    model.train()  # 訓練モード（Dropoutが有効）
    optimizer.zero_grad()
    train_loss = criterion(model(X_train), y_train)
    train_loss.backward()
    optimizer.step()

    model.eval()  # 推論モード（Dropoutが無効）
    with torch.no_grad():  # 勾配計算を無効化
        val_loss = criterion(model(X_val), y_val)

    print(f"epoch {epoch+1}, train {train_loss.item():.4f}, val {val_loss.item():.4f}")

import copy

import torch
import torch.nn as nn

torch.manual_seed(0)

model = nn.Sequential(nn.Linear(4, 8), nn.ReLU(), nn.Linear(8, 1))
optimizer = torch.optim.Adam(model.parameters(), lr=0.01)
criterion = nn.MSELoss()

X_train, y_train = torch.randn(80, 4), torch.randn(80, 1)
X_val, y_val = torch.randn(20, 4), torch.randn(20, 1)

best_val_loss = float("inf")
best_state = None     # 最良だった時点の重みを控えておく
patience = 5          # 改善が止まってから許容するエポック数
patience_counter = 0

for epoch in range(100):
    optimizer.zero_grad()
    loss = criterion(model(X_train), y_train)
    loss.backward()
    optimizer.step()

    model.eval()
    with torch.no_grad():
        val_loss = criterion(model(X_val), y_val).item()
    model.train()

    if val_loss < best_val_loss:
        best_val_loss = val_loss
        best_state = copy.deepcopy(model.state_dict())  # 最良時点の重みを保持
        patience_counter = 0  # 改善したのでカウンタをリセット
    else:
        patience_counter += 1

    if patience_counter >= patience:
        print(f"epoch {epoch+1}で早期終了（best_val_loss={best_val_loss:.4f}）")
        break

# 停止した時点ではなく、検証lossが最良だった時点の重みに戻す
if best_state is not None:
    model.load_state_dict(best_state)

import torch
import torch.nn as nn

torch.manual_seed(0)

dropout = nn.Dropout(p=0.5)
x = torch.ones(1, 10)

dropout.train()  # 訓練モード
print("train:", dropout(x))  # 約半分が0になり、残りは1/(1-p)倍にスケーリングされる

dropout.eval()  # 推論モード
print("eval :", dropout(x))  # 何も変化させずそのまま出力する

# 実際のモデルへの組み込み例
model = nn.Sequential(
    nn.Linear(20, 64),
    nn.ReLU(),
    nn.Dropout(p=0.3),   # 活性化関数の後に挿入するのが定番
    nn.Linear(64, 32),
    nn.ReLU(),
    nn.Dropout(p=0.3),
    nn.Linear(32, 1),
)
print(model)

import torch
import torch.nn as nn

torch.manual_seed(0)

# 全結合層向け: BatchNorm1d
mlp = nn.Sequential(
    nn.Linear(20, 64),
    nn.BatchNorm1d(64),  # 全結合層の直後・活性化関数の前
    nn.ReLU(),
    nn.Linear(64, 1),
)

# 畳み込み層向け: BatchNorm2d
cnn = nn.Sequential(
    nn.Conv2d(3, 16, kernel_size=3, padding=1),
    nn.BatchNorm2d(16),  # チャネル数16に一致させる
    nn.ReLU(),
    nn.MaxPool2d(2),
)

x_tabular = torch.randn(8, 20)
x_image = torch.randn(8, 3, 32, 32)

print(mlp(x_tabular).shape)   # torch.Size([8, 1])
print(cnn(x_image).shape)     # torch.Size([8, 16, 16, 16])

import torch
import torch.nn as nn

torch.manual_seed(0)

model = nn.Sequential(nn.Linear(4, 8), nn.ReLU(), nn.Linear(8, 1))

# 保存: state_dict（パラメータの辞書）のみを保存する
torch.save(model.state_dict(), "model_weights.pth")

# 読み込み: 同じ構造のモデルを用意してstate_dictをロードする
loaded_model = nn.Sequential(nn.Linear(4, 8), nn.ReLU(), nn.Linear(8, 1))

# GPUで保存した重みをCPU環境で読み込む場合はmap_locationを指定する
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# weights_only=True にすると、重み以外のオブジェクトを復元しないので安全に読める
state_dict = torch.load("model_weights.pth", map_location=device, weights_only=True)
loaded_model.load_state_dict(state_dict)
loaded_model.eval()

print("読み込み完了:", loaded_model)

import torch
import torch.nn as nn
import matplotlib.pyplot as plt

torch.manual_seed(0)

model = nn.Sequential(nn.Linear(4, 8), nn.ReLU(), nn.Linear(8, 1))
optimizer = torch.optim.Adam(model.parameters(), lr=0.01)
criterion = nn.MSELoss()

X_train, y_train = torch.randn(80, 4), torch.randn(80, 1)
X_val, y_val = torch.randn(20, 4), torch.randn(20, 1)

train_losses, val_losses = [], []

for epoch in range(50):
    optimizer.zero_grad()
    train_loss = criterion(model(X_train), y_train)
    train_loss.backward()
    optimizer.step()
    train_losses.append(train_loss.item())

    model.eval()
    with torch.no_grad():
        val_loss = criterion(model(X_val), y_val)
    model.train()
    val_losses.append(val_loss.item())

plt.plot(train_losses, label="train loss")
plt.plot(val_losses, label="val loss")
plt.xlabel("epoch")
plt.ylabel("loss")
plt.legend()
plt.savefig("learning_curve.png", dpi=150, bbox_inches="tight")

import torch
import torch.nn as nn
from sklearn.datasets import make_regression
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler

torch.manual_seed(0)

# 1. データ生成と前処理
X, y = make_regression(n_samples=500, n_features=6, noise=10.0, random_state=0)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=0)

scaler = StandardScaler()
X_train = scaler.fit_transform(X_train)
X_test = scaler.transform(X_test)

X_train = torch.tensor(X_train, dtype=torch.float32)
y_train = torch.tensor(y_train, dtype=torch.float32).view(-1, 1)
X_test = torch.tensor(X_test, dtype=torch.float32)
y_test = torch.tensor(y_test, dtype=torch.float32).view(-1, 1)

# 2. モデル定義
model = nn.Sequential(nn.Linear(6, 32), nn.ReLU(), nn.Linear(32, 1))
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.01)

# 3. 学習
for epoch in range(200):
    optimizer.zero_grad()
    loss = criterion(model(X_train), y_train)
    loss.backward()
    optimizer.step()

# 4. 評価
model.eval()
with torch.no_grad():
    test_loss = criterion(model(X_test), y_test)
print(f"テストMSE: {test_loss.item():.2f}")

import torch
import torch.nn as nn
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler

torch.manual_seed(0)

# 1. データ生成と前処理
X, y = make_classification(n_samples=500, n_features=8, n_classes=3,
                            n_informative=5, random_state=0)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=0)

scaler = StandardScaler()
X_train = scaler.fit_transform(X_train)
X_test = scaler.transform(X_test)

X_train = torch.tensor(X_train, dtype=torch.float32)
y_train = torch.tensor(y_train, dtype=torch.long)
X_test = torch.tensor(X_test, dtype=torch.float32)
y_test = torch.tensor(y_test, dtype=torch.long)

# 2. モデル定義（出力層はクラス数分のロジット、Softmaxは不要）
model = nn.Sequential(nn.Linear(8, 32), nn.ReLU(), nn.Linear(32, 3))
criterion = nn.CrossEntropyLoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.01)

# 3. 学習
for epoch in range(200):
    optimizer.zero_grad()
    loss = criterion(model(X_train), y_train)
    loss.backward()
    optimizer.step()

# 4. 評価
model.eval()
with torch.no_grad():
    pred_labels = model(X_test).argmax(dim=1)
    accuracy = (pred_labels == y_test).float().mean()
print(f"テスト精度: {accuracy.item():.3f}")

import tensorflow as tf
from sklearn.datasets import make_regression
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler

tf.random.set_seed(0)

# 1. データ生成と前処理（レシピ25と同一）
X, y = make_regression(n_samples=500, n_features=6, noise=10.0, random_state=0)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=0)

scaler = StandardScaler()
X_train = scaler.fit_transform(X_train)
X_test = scaler.transform(X_test)

# 2. モデル定義（nn.Sequentialに相当）
model = tf.keras.Sequential([
    tf.keras.layers.Input(shape=(6,)),
    tf.keras.layers.Dense(32, activation="relu"),
    tf.keras.layers.Dense(1),
])

# 3. compile: 損失関数・最適化アルゴリズムをまとめて指定する
model.compile(optimizer=tf.keras.optimizers.Adam(learning_rate=0.01), loss="mse")

# 4. fit: forward/backward/stepの学習ループを内部で自動実行する
model.fit(X_train, y_train, epochs=200, verbose=0)

# 5. 評価
test_loss = model.evaluate(X_test, y_test, verbose=0)
print(f"テストMSE: {test_loss:.2f}")

from PIL import Image
from torchvision import transforms

# 画像を読み込んでテンソル化する一連の流れ
img = Image.open("sample.jpg").convert("RGB")  # PILで読み込み、RGBに統一

# 学習済みモデル(ImageNet)向けの標準的な前処理
preprocess = transforms.Compose([
    transforms.Resize(256),          # 短辺を256pxにリサイズ
    transforms.CenterCrop(224),      # 中央224x224を切り出し
    transforms.ToTensor(),           # PIL画像をTensorに変換し、値を[0,1]にスケーリング
    transforms.Normalize(            # ImageNetの平均・標準偏差で正規化
        mean=[0.485, 0.456, 0.406],
        std=[0.229, 0.224, 0.225],
    ),
])

tensor = preprocess(img)             # shape: (3, 224, 224)
batch = tensor.unsqueeze(0)          # shape: (1, 3, 224, 224) バッチ次元を追加

print(tensor.shape, tensor.dtype)
print(tensor.min().item(), tensor.max().item())  # 正規化後は0付近を中心とした分布になる

import torchvision
from torchvision import transforms
from torch.utils.data import DataLoader

transform = transforms.Compose([
    transforms.ToTensor(),
    transforms.Normalize((0.1307,), (0.3081,)),  # MNIST全体の平均・標準偏差
])

# 初回実行時はdownload=Trueによりデータを自動ダウンロード(数十MB程度)
train_set = torchvision.datasets.MNIST(
    root="./data", train=True, download=True, transform=transform
)
test_set = torchvision.datasets.MNIST(
    root="./data", train=False, download=True, transform=transform
)

train_loader = DataLoader(train_set, batch_size=64, shuffle=True)
test_loader = DataLoader(test_set, batch_size=64, shuffle=False)

images, labels = next(iter(train_loader))
print(images.shape, labels.shape)  # (64, 1, 28, 28), (64,)

# CIFAR-10も同じ書き方でダウンロード可能(初回はカラー画像で約170MB)
cifar_set = torchvision.datasets.CIFAR10(
    root="./data", train=True, download=True, transform=transforms.ToTensor()
)

import torch.nn as nn

class SimpleCNN(nn.Module):
    def __init__(self, num_classes=10):
        super().__init__()
        self.conv1 = nn.Conv2d(1, 16, kernel_size=3, padding=1)  # (B,1,28,28) -> (B,16,28,28)
        self.conv2 = nn.Conv2d(16, 32, kernel_size=3, padding=1) # (B,16,14,14) -> (B,32,14,14)
        self.pool = nn.MaxPool2d(kernel_size=2)                  # 高さ・幅をそれぞれ半分に縮小
        self.relu = nn.ReLU()
        self.fc = nn.Linear(32 * 7 * 7, num_classes)             # 28 -> 14 -> 7 と2回poolした結果

    def forward(self, x):
        x = self.pool(self.relu(self.conv1(x)))  # (B,1,28,28)  -> (B,16,14,14)
        x = self.pool(self.relu(self.conv2(x)))  # (B,16,14,14) -> (B,32,7,7)
        x = x.view(x.size(0), -1)                # (B,32,7,7)   -> (B, 1568)
        x = self.fc(x)                           # (B, 1568)    -> (B, num_classes)
        return x

model = SimpleCNN()
print(model)

import torch
import torch.nn as nn
import torch.optim as optim

# 前提: レシピ29のtrain_loader・test_loaderと、レシピ30のSimpleCNNクラスを定義済みとします
device = "cuda" if torch.cuda.is_available() else "cpu"
model = SimpleCNN().to(device)
optimizer = optim.Adam(model.parameters(), lr=1e-3)
criterion = nn.CrossEntropyLoss()

# 学習(わかりやすさのため1エポックのみ)
model.train()
for images, labels in train_loader:
    images, labels = images.to(device), labels.to(device)
    optimizer.zero_grad()
    outputs = model(images)
    loss = criterion(outputs, labels)
    loss.backward()
    optimizer.step()

# 評価
model.eval()
correct, total = 0, 0
with torch.no_grad():
    for images, labels in test_loader:
        images, labels = images.to(device), labels.to(device)
        outputs = model(images)
        preds = outputs.argmax(dim=1)
        correct += (preds == labels).sum().item()
        total += labels.size(0)

print(f"Test Accuracy: {correct / total:.4f}")

from torchvision import transforms

train_transform = transforms.Compose([
    transforms.RandomCrop(32, padding=4),        # ランダムな位置で切り出し、周囲をパディング
    transforms.RandomHorizontalFlip(p=0.5),      # 左右反転(自然画像で有効。文字画像等は要注意)
    transforms.ColorJitter(                       # 明るさ・コントラスト・彩度をランダムに変化
        brightness=0.2, contrast=0.2, saturation=0.2
    ),
    transforms.RandAugment(num_ops=2, magnitude=9),  # 複数の変換を自動選択・自動適用する手法
    transforms.ToTensor(),
    transforms.Normalize((0.4914, 0.4822, 0.4465), (0.2470, 0.2435, 0.2616)),
])

# 検証用データには水増しをかけず、決定的な前処理のみを適用する
val_transform = transforms.Compose([
    transforms.ToTensor(),
    transforms.Normalize((0.4914, 0.4822, 0.4465), (0.2470, 0.2435, 0.2616)),
])

import torch
from torchvision.models import resnet50, ResNet50_Weights
from PIL import Image

# weights引数に列挙型を渡す現行API(pretrained=Trueは非推奨)
weights = ResNet50_Weights.DEFAULT  # 初回はImageNet学習済み重みを自動ダウンロード(約100MB)
model = resnet50(weights=weights)
model.eval()

preprocess = weights.transforms()        # このモデル専用の前処理を取得
categories = weights.meta["categories"]  # ImageNetの1000クラス名一覧

img = Image.open("sample.jpg").convert("RGB")
batch = preprocess(img).unsqueeze(0)

with torch.no_grad():
    logits = model(batch)
    probs = logits.softmax(dim=1)

top5_prob, top5_idx = probs.topk(5)
for prob, idx in zip(top5_prob[0], top5_idx[0]):
    print(f"{categories[idx]}: {prob.item():.3f}")

import torch
import torch.nn as nn
from torchvision.models import resnet18, ResNet18_Weights

weights = ResNet18_Weights.DEFAULT
model = resnet18(weights=weights)

num_classes = 5  # 自社の分類クラス数(例: 製品カテゴリ5種)
model.fc = nn.Linear(model.fc.in_features, num_classes)  # 最終全結合層のみ差し替え

device = "cuda" if torch.cuda.is_available() else "cpu"
model = model.to(device)

# 差し替えた層以外の重みは、ImageNetで学習済みのものを初期値として使う。
# この書き方では全層が更新対象になる（＝全層のファインチューニング）。
# 最終層だけを学習させたい場合はレシピ35の凍結を使う
optimizer = torch.optim.Adam(model.parameters(), lr=1e-4)

# 前提: レシピ34で作ったmodel（fcを差し替えたResNet18）とtorchのimportを引き継ぎます

# 段階1: 全層を凍結し、最終層だけを学習させる
for param in model.parameters():
    param.requires_grad = False
for param in model.fc.parameters():
    param.requires_grad = True

optimizer = torch.optim.Adam(model.fc.parameters(), lr=1e-3)
# ここで数エポック学習し、最終層をデータに合わせて安定させる

# 段階2: 精度が頭打ちになったら、後段の層から順に解凍する
for param in model.layer4.parameters():
    param.requires_grad = True

# 解凍した層は学習率を小さくするのが定石(既存の特徴表現を壊さないため)
optimizer = torch.optim.Adam([
    {"params": model.layer4.parameters(), "lr": 1e-5},
    {"params": model.fc.parameters(), "lr": 1e-4},
])

from torchvision import datasets, transforms
from torch.utils.data import DataLoader, random_split

# フォルダ構成の約束事:
# data/
#   class_a/xxx.jpg, yyy.jpg, ...
#   class_b/zzz.jpg, ...
# サブフォルダ名がそのままクラスラベルになる

# 学習済みモデルにつなぐ場合は、その重みが前提にしている前処理に合わせる。
# torchvisionの重みは weights.transforms() で正しい前処理一式を取り出せる
#   from torchvision.models import ResNet18_Weights
#   transform = ResNet18_Weights.DEFAULT.transforms()
# 自分で書く場合も、リサイズとテンソル化だけでなく正規化まで揃える
transform = transforms.Compose([
    transforms.Resize((224, 224)),
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.485, 0.456, 0.406],   # ImageNetの統計量
                         std=[0.229, 0.224, 0.225]),
])

full_dataset = datasets.ImageFolder(root="./data", transform=transform)
print(full_dataset.classes)  # ['class_a', 'class_b', ...] アルファベット順に並ぶ

# train/valに8:2で分割
n_val = int(len(full_dataset) * 0.2)
n_train = len(full_dataset) - n_val
train_set, val_set = random_split(full_dataset, [n_train, n_val])

train_loader = DataLoader(train_set, batch_size=32, shuffle=True)
val_loader = DataLoader(val_set, batch_size=32, shuffle=False)

from torchvision.models import resnet18, ResNet18_Weights
from PIL import Image

weights = ResNet18_Weights.DEFAULT
model = resnet18(weights=weights)
preprocess = weights.transforms()  # 学習時と同じリサイズ・正規化を再現する

print(preprocess)
# Resize(256) -> CenterCrop(224) -> ToTensor -> Normalize(mean, std) 等が
# モデルごとに適切な値で自動設定されている

# 独自に前処理を書き直すと、リサイズの補間方式や正規化の値がずれて
# 精度が大きく落ちる事故が起きやすい(学習時はBICUBIC、推論時はデフォルトのBILINEARだった、等)
img = Image.open("sample.jpg").convert("RGB")
batch = preprocess(img).unsqueeze(0)  # 学習時と同じ前処理をそのまま利用する

import torch
from sklearn.metrics import confusion_matrix, classification_report

# 前提: レシピ34で学習したmodel・deviceと、レシピ36のval_loader・full_datasetを使います
model.eval()
all_preds, all_labels = [], []

with torch.no_grad():
    for images, labels in val_loader:
        images = images.to(device)
        outputs = model(images)
        preds = outputs.argmax(dim=1).cpu()
        all_preds.extend(preds.tolist())
        all_labels.extend(labels.tolist())

# 検証側に1件も現れなかったクラスがあってもズレないよう、ラベルを明示する
labels = list(range(len(full_dataset.classes)))

cm = confusion_matrix(all_labels, all_preds, labels=labels)
print(cm)  # 行:正解クラス、列:予測クラス

# クラスごとのprecision/recall/f1をまとめて確認する
print(classification_report(all_labels, all_preds, labels=labels,
                            target_names=full_dataset.classes, zero_division=0))

import matplotlib.pyplot as plt
import torch

# 前提: レシピ38で計算したall_preds・all_labelsと、レシピ36の検証用データセットを使います
MEAN = torch.tensor([0.485, 0.456, 0.406]).view(3, 1, 1)
STD = torch.tensor([0.229, 0.224, 0.225]).view(3, 1, 1)

wrong_indices = [i for i, (p, l) in enumerate(zip(all_preds, all_labels)) if p != l]

fig, axes = plt.subplots(2, 4, figsize=(12, 6))
for ax, idx in zip(axes.flatten(), wrong_indices[:8]):
    image, label = val_set[idx]
    # レシピ36で正規化しているので、表示前に元のスケールへ戻す
    image = (image * STD + MEAN).clamp(0, 1)
    ax.imshow(image.permute(1, 2, 0).numpy())  # (C,H,W) -> (H,W,C)に並べ替え
    true_name = full_dataset.classes[label]
    pred_name = full_dataset.classes[all_preds[idx]]
    ax.set_title(f"true:{true_name}\npred:{pred_name}")
    ax.axis("off")

plt.tight_layout()
plt.savefig("misclassified.png")

pip install grad-cam

import numpy as np
from pytorch_grad_cam import GradCAM
from pytorch_grad_cam.utils.model_targets import ClassifierOutputTarget
from pytorch_grad_cam.utils.image import show_cam_on_image

# 前提: レシピ33のResNet(model)と、前処理済みテンソルbatch・元画像img(レシピ33参照)を使います
model.eval()
target_layers = [model.layer4[-1]]  # 最終の畳み込みブロックを可視化対象に指定
cam = GradCAM(model=model, target_layers=target_layers)

input_tensor = batch  # 前処理済みの(1,3,224,224)テンソル
predicted_class_idx = int(model(input_tensor).argmax(dim=1)[0])
targets = [ClassifierOutputTarget(predicted_class_idx)]  # 可視化したいクラスを指定

grayscale_cam = cam(input_tensor=input_tensor, targets=targets)[0]  # (224,224)のヒートマップ

rgb_img = np.array(img.resize((224, 224))) / 255.0  # 0〜1に正規化した元画像
visualization = show_cam_on_image(rgb_img, grayscale_cam, use_rgb=True)

from torchvision.models import (
    mobilenet_v3_small, MobileNet_V3_Small_Weights,
    efficientnet_b0, EfficientNet_B0_Weights,
    resnet50, ResNet50_Weights,
)

candidates = {
    "MobileNetV3-Small": mobilenet_v3_small(weights=MobileNet_V3_Small_Weights.DEFAULT),
    "EfficientNet-B0": efficientnet_b0(weights=EfficientNet_B0_Weights.DEFAULT),
    "ResNet50": resnet50(weights=ResNet50_Weights.DEFAULT),
}

for name, m in candidates.items():
    n_params = sum(p.numel() for p in m.parameters())
    print(f"{name}: {n_params / 1e6:.1f}M params")

# 目安: MobileNetV3-Small(約2.5M params)はエッジ端末・リアルタイム処理向け
# EfficientNet-B0(約5.3M params)は精度と速度のバランス型
# ResNet50(約25.6M params)は精度優先だがGPU推論を前提とすることが多い

import torch

# 前提: レシピ34で学習したmodel・device・full_datasetと、推論対象のDataLoader(inference_loader)を用意済みとします
CONFIDENCE_THRESHOLD = 0.8
model.eval()
results = []

with torch.no_grad():
    # inference_loaderは (画像テンソル, ファイルパス) を返す想定。
    # ImageFolderは (画像, ラベル) を返すので、パスも欲しい場合は
    # __getitem__ で self.samples[idx][0] を一緒に返すDatasetを自作する
    for images, paths in inference_loader:
        images = images.to(device)
        probs = model(images).softmax(dim=1)
        confidences, preds = probs.max(dim=1)

        for path, pred, conf in zip(paths, preds.tolist(), confidences.tolist()):
            results.append({
                "path": path,
                "pred_class": full_dataset.classes[pred],
                "confidence": conf,
                "needs_human_review": conf < CONFIDENCE_THRESHOLD,
            })

# 確信度が閾値未満の画像だけを人間の確認キューに回す業務設計
review_queue = [r for r in results if r["needs_human_review"]]
print(f"要確認: {len(review_queue)} / {len(results)} 件")

import numpy as np
import matplotlib.pyplot as plt

# ダミーの元画像(64x64のグレースケール画像を疑似生成)
np.random.seed(0)
image = np.random.randint(80, 180, size=(64, 64), dtype=np.uint8)

# ダミーのセグメンテーションマスク(0:背景, 1:対象領域, 2:欠陥領域)
mask = np.zeros((64, 64), dtype=np.uint8)
mask[10:50, 15:45] = 1
mask[30:35, 30:38] = 2

# クラスごとに色を割り当てたカラーマスクを作成
color_map = {0: (0, 0, 0), 1: (0, 128, 255), 2: (255, 0, 0)}
color_mask = np.zeros((64, 64, 3), dtype=np.uint8)
for class_id, color in color_map.items():
    color_mask[mask == class_id] = color

# 元画像にマスクを半透明で重ねて表示
fig, axes = plt.subplots(1, 2, figsize=(8, 4))
axes[0].imshow(image, cmap='gray')
axes[0].set_title('元画像')
axes[1].imshow(image, cmap='gray')
axes[1].imshow(color_mask, alpha=0.5)
axes[1].set_title('マスクのオーバーレイ')
for ax in axes:
    ax.axis('off')
plt.tight_layout()
plt.show()

import torch
from torchvision.models.segmentation import deeplabv3_resnet50, DeepLabV3_ResNet50_Weights
from PIL import Image
import numpy as np

# 学習済み重みとモデルの読み込み(現行のweights APIを使用)
weights = DeepLabV3_ResNet50_Weights.DEFAULT
model = deeplabv3_resnet50(weights=weights)
model.eval()

# 前処理は重みに付属するtransformsをそのまま利用
preprocess = weights.transforms()

# ダミー画像を用意(実務ではImage.open('path.jpg')で実画像を読み込む)
dummy_image = Image.fromarray(
    (np.random.rand(320, 480, 3) * 255).astype(np.uint8)
)
input_tensor = preprocess(dummy_image).unsqueeze(0)

# 推論の実行
with torch.no_grad():
    output = model(input_tensor)['out']

# 各画素についてクラススコアが最大のクラスIDを採用
predicted_mask = output.argmax(1).squeeze(0).numpy()
print(f"マスクの形状: {predicted_mask.shape}")
print(f"検出されたクラスID: {np.unique(predicted_mask)}")

import numpy as np
import matplotlib.pyplot as plt
import matplotlib.patches as mpatches

# 44で得られる推論結果に相当するダミーマスクを用意
np.random.seed(1)
predicted_mask = np.random.choice([0, 8, 12, 15], size=(64, 64), p=[0.6, 0.15, 0.1, 0.15])
class_names = {0: '背景', 8: '猫', 12: '犬', 15: '人'}

# クラスIDごとに固定の色を割り当て
palette = {0: (0, 0, 0), 8: (255, 165, 0), 12: (0, 200, 0), 15: (0, 100, 255)}
color_image = np.zeros((*predicted_mask.shape, 3), dtype=np.uint8)
for class_id, color in palette.items():
    color_image[predicted_mask == class_id] = color

# 塗り分け画像と凡例を表示
fig, ax = plt.subplots(figsize=(6, 6))
ax.imshow(color_image)
ax.axis('off')
ax.set_title('クラス別セグメンテーション結果')

legend_patches = [
    mpatches.Patch(color=np.array(color) / 255, label=class_names[class_id])
    for class_id, color in palette.items()
]
ax.legend(handles=legend_patches, bbox_to_anchor=(1.3, 1), loc='upper right')
plt.tight_layout()
plt.show()

import torch
import torch.nn as nn

def conv_block(in_ch, out_ch):
    return nn.Sequential(
        nn.Conv2d(in_ch, out_ch, 3, padding=1),
        nn.ReLU(inplace=True),
        nn.Conv2d(out_ch, out_ch, 3, padding=1),
        nn.ReLU(inplace=True),
    )

class MiniUNet(nn.Module):
    def __init__(self, in_channels=3, num_classes=3):
        super().__init__()
        self.enc1 = conv_block(in_channels, 32)
        self.enc2 = conv_block(32, 64)
        self.pool = nn.MaxPool2d(2)
        self.bottleneck = conv_block(64, 128)
        self.up2 = nn.ConvTranspose2d(128, 64, 2, stride=2)
        self.dec2 = conv_block(128, 64)
        self.up1 = nn.ConvTranspose2d(64, 32, 2, stride=2)
        self.dec1 = conv_block(64, 32)
        self.out_conv = nn.Conv2d(32, num_classes, 1)

    def forward(self, x):
        e1 = self.enc1(x)
        e2 = self.enc2(self.pool(e1))
        b = self.bottleneck(self.pool(e2))
        d2 = self.dec2(torch.cat([self.up2(b), e2], dim=1))
        d1 = self.dec1(torch.cat([self.up1(d2), e1], dim=1))
        return self.out_conv(d1)

# 動作確認(バッチ1枚・3チャンネル・64x64画像を通す)
model = MiniUNet(in_channels=3, num_classes=3)
dummy_input = torch.randn(1, 3, 64, 64)
output = model(dummy_input)
print(f"出力形状: {output.shape}")

import torch
import numpy as np
from torch.utils.data import Dataset
import torchvision.transforms.functional as TF
import random

class SegmentationDataset(Dataset):
    def __init__(self, images, masks, train=True):
        self.images = images  # (N, H, W, 3) のnumpy配列
        self.masks = masks    # (N, H, W) のnumpy配列(クラスID)
        self.train = train

    def __len__(self):
        return len(self.images)

    def __getitem__(self, idx):
        image = TF.to_pil_image(self.images[idx])
        mask = TF.to_pil_image(self.masks[idx].astype(np.uint8))

        if self.train and random.random() > 0.5:
            # 画像とマスクに必ず同じ変換を適用する(左右反転)
            image = TF.hflip(image)
            mask = TF.hflip(mask)

        image_tensor = TF.to_tensor(image)
        mask_tensor = torch.as_tensor(np.array(mask), dtype=torch.long)
        return image_tensor, mask_tensor

# 動作確認用のダミーデータ
dummy_images = np.random.randint(0, 255, (4, 64, 64, 3), dtype=np.uint8)
dummy_masks = np.random.randint(0, 3, (4, 64, 64), dtype=np.uint8)
dataset = SegmentationDataset(dummy_images, dummy_masks)
img, msk = dataset[0]
print(f"画像テンソル形状: {img.shape}, マスクテンソル形状: {msk.shape}")

import torch
import torch.nn as nn
import torch.nn.functional as F

class DiceLoss(nn.Module):
    def __init__(self, num_classes, smooth=1e-6):
        super().__init__()
        self.num_classes = num_classes
        self.smooth = smooth

    def forward(self, logits, targets):
        probs = F.softmax(logits, dim=1)
        targets_onehot = F.one_hot(targets, self.num_classes).permute(0, 3, 1, 2).float()
        dims = (0, 2, 3)
        intersection = torch.sum(probs * targets_onehot, dims)
        union = torch.sum(probs + targets_onehot, dims)
        dice_per_class = (2 * intersection + self.smooth) / (union + self.smooth)
        return 1 - dice_per_class.mean()

# ダミーの出力とラベルで両損失を比較
logits = torch.randn(2, 3, 32, 32)
targets = torch.randint(0, 3, (2, 32, 32))

ce_loss_fn = nn.CrossEntropyLoss()
dice_loss_fn = DiceLoss(num_classes=3)

ce_loss = ce_loss_fn(logits, targets)
dice_loss = dice_loss_fn(logits, targets)
print(f"CrossEntropy損失: {ce_loss.item():.4f}")
print(f"Dice損失: {dice_loss.item():.4f}")

import numpy as np

def compute_iou(pred_mask, true_mask, num_classes):
    ious = []
    for class_id in range(num_classes):
        pred_area = (pred_mask == class_id)
        true_area = (true_mask == class_id)
        intersection = np.logical_and(pred_area, true_area).sum()
        union = np.logical_or(pred_area, true_area).sum()
        if union == 0:
            ious.append(np.nan)  # そのクラスが画像中に存在しない場合
        else:
            ious.append(intersection / union)
    return ious

# ダミーの予測マスクと正解マスクで検証
np.random.seed(2)
pred_mask = np.random.randint(0, 3, (64, 64))
true_mask = np.random.randint(0, 3, (64, 64))

class_names = ['背景', '対象領域', '欠陥領域']
ious = compute_iou(pred_mask, true_mask, num_classes=3)
for name, iou in zip(class_names, ious):
    print(f"{name}のIoU: {iou:.4f}")

mean_iou = np.nanmean(ious)
print(f"mIoU: {mean_iou:.4f}")

import torch
import torch.nn as nn
import torch.optim as optim

# 前提: レシピ46のMiniUNetクラスを定義済みとします(以下は合成データでの最小学習ループ)
torch.manual_seed(0)
model = MiniUNet(in_channels=3, num_classes=3)
optimizer = optim.Adam(model.parameters(), lr=1e-3)
criterion = nn.CrossEntropyLoss()

# 合成データ(中央に対象クラスを配置したダミー画像とマスク)を10枚生成
images = torch.randn(10, 3, 64, 64)
masks = torch.zeros(10, 64, 64, dtype=torch.long)
masks[:, 20:44, 20:44] = 1  # 中央に対象領域を配置
images[:, :, 20:44, 20:44] += 2.0  # 画像側にも対応する信号を載せる（学習できる形にする）

# 5エポックの学習ループ
for epoch in range(5):
    optimizer.zero_grad()
    outputs = model(images)
    loss = criterion(outputs, masks)
    loss.backward()
    optimizer.step()
    print(f"epoch {epoch + 1}: loss = {loss.item():.4f}")

import numpy as np
from scipy import ndimage
from skimage.morphology import remove_small_objects, binary_opening, disk

# ノイズを含むダミーの二値予測マスクを用意(小さな誤検出領域を含む)
np.random.seed(3)
pred_mask = np.zeros((100, 100), dtype=bool)
pred_mask[30:70, 30:70] = True
noise_coords = np.random.randint(0, 100, size=(15, 2))
for y, x in noise_coords:
    pred_mask[y, x] = True  # 孤立したノイズ画素

# モルフォロジー演算(オープニング)で孤立ノイズを除去
opened_mask = binary_opening(pred_mask, footprint=disk(2))

# さらに面積の小さい連結領域を除去
cleaned_mask = remove_small_objects(opened_mask, min_size=50)

print(f"後処理前の有効画素数: {pred_mask.sum()}")
print(f"オープニング後の有効画素数: {opened_mask.sum()}")
print(f"小領域除去後の有効画素数: {cleaned_mask.sum()}")

# 穴埋め処理(領域内部の欠損を補完したい場合)
filled_mask = ndimage.binary_fill_holes(cleaned_mask)
print(f"穴埋め後の有効画素数: {filled_mask.sum()}")

# Meta公式の実装はGitHubから直接入れる
pip install git+https://github.com/facebookresearch/segment-anything.git
# チェックポイント(例: ViT-B、約375MB)は事前に手動でダウンロードしておく
# https://github.com/facebookresearch/segment-anything#model-checkpoints

import numpy as np
from segment_anything import sam_model_registry, SamPredictor

# ダウンロード済みのチェックポイントパスを指定してモデルを構築
sam = sam_model_registry["vit_b"](checkpoint="sam_vit_b_01ec64.pth")
predictor = SamPredictor(sam)

# ダミー画像(実務では読み込んだ実画像を使用)
# set_imageはRGBの並びを前提にしている。cv2.imreadはBGRで返すので、
# cv2.cvtColor(img, cv2.COLOR_BGR2RGB) で並べ替えてから渡すこと
dummy_image = np.random.randint(0, 255, (480, 640, 3), dtype=np.uint8)
predictor.set_image(dummy_image)

# 対象物の中心付近を1点指定してプロンプトベースでマスクを生成
input_point = np.array([[320, 240]])
input_label = np.array([1])  # 1は前景(対象)を意味するラベル

masks, scores, logits = predictor.predict(
    point_coords=input_point,
    point_labels=input_label,
    multimask_output=True,
)
print(f"生成されたマスク数: {masks.shape[0]}")
print(f"各マスクの信頼度スコア: {scores}")

from PIL import Image, ImageDraw

img = Image.open("sample.jpg").convert("RGB")
draw = ImageDraw.Draw(img)

# (x1, y1, x2, y2) = 左上と右下の座標
box = (120, 80, 340, 260)
draw.rectangle(box, outline="red", width=3)

img.save("sample_with_box.jpg")
print("box width:", box[2] - box[0], "box height:", box[3] - box[1])

import torch
from torchvision.io import read_image
from torchvision.models.detection import (
    fasterrcnn_resnet50_fpn_v2,
    FasterRCNN_ResNet50_FPN_V2_Weights,
)

weights = FasterRCNN_ResNet50_FPN_V2_Weights.DEFAULT
model = fasterrcnn_resnet50_fpn_v2(weights=weights, box_score_thresh=0.5)
model.eval()

preprocess = weights.transforms()
img = read_image("sample.jpg")  # uint8 の Tensor (C, H, W)
batch = [preprocess(img)]

with torch.no_grad():
    outputs = model(batch)

result = outputs[0]
score_threshold = 0.7
keep = result["scores"] >= score_threshold

boxes = result["boxes"][keep]
labels = result["labels"][keep]
scores = result["scores"][keep]

print(f"検出数: {len(boxes)}")
for box, label, score in zip(boxes, labels, scores):
    print(box.tolist(), int(label), float(score))

import torch
from torchvision.io import read_image
from torchvision.utils import draw_bounding_boxes
from torchvision.transforms.functional import to_pil_image
from torchvision.models.detection import (
    fasterrcnn_resnet50_fpn_v2,
    FasterRCNN_ResNet50_FPN_V2_Weights,
)

weights = FasterRCNN_ResNet50_FPN_V2_Weights.DEFAULT
categories = weights.meta["categories"]  # COCOのクラス名リスト(背景含む)

model = fasterrcnn_resnet50_fpn_v2(weights=weights)
model.eval()

img = read_image("sample.jpg")
batch = [weights.transforms()(img)]

with torch.no_grad():
    result = model(batch)[0]

keep = result["scores"] >= 0.7
boxes = result["boxes"][keep]
labels = result["labels"][keep]
scores = result["scores"][keep]

label_texts = [
    f"{categories[label]}: {score:.2f}"
    for label, score in zip(labels, scores)
]

annotated = draw_bounding_boxes(
    img, boxes=boxes, labels=label_texts, colors="lime", width=3, font_size=20
)
to_pil_image(annotated).save("sample_annotated.jpg")

pip install ultralytics

from ultralytics import YOLO

model = YOLO("yolo11n.pt")  # 初回はモデルを自動ダウンロード
results = model("sample.jpg", conf=0.5)

result = results[0]
print("検出数:", len(result.boxes))

for box in result.boxes:
    xyxy = box.xyxy[0].tolist()      # [x1, y1, x2, y2]
    cls_id = int(box.cls[0])
    cls_name = result.names[cls_id]  # COCOのクラス名
    conf = float(box.conf[0])
    print(cls_name, conf, xyxy)

# ラベル・矩形を描き込んだ画像を保存
result.save(filename="sample_yolo.jpg")

from pathlib import Path

# フォルダ構成の例
# dataset/
#   images/train/xxx.jpg
#   images/val/yyy.jpg
#   labels/train/xxx.txt   # 1行 = 1物体: class cx cy w h (正規化座標)
#   labels/val/yyy.txt
#   data.yaml

# data.yaml の中身の例
data_yaml = """
path: dataset
train: images/train
val: images/val
names:
  0: box
  1: pallet
  2: forklift
"""
Path("dataset").mkdir(parents=True, exist_ok=True)  # 無いと書き込みで落ちる
with open("dataset/data.yaml", "w", encoding="utf-8") as f:
    f.write(data_yaml)

from ultralytics import YOLO

model = YOLO("yolo11n.pt")  # 学習済み重みから転移学習
model.train(
    data="dataset/data.yaml",
    epochs=100,
    imgsz=640,
    batch=16,
    project="runs_custom",
    name="warehouse_v1",
)

import torch
from torchvision.ops import nms

# 同じ物体に対する重複ボックスを想定したダミーデータ
# nmsはboxesを浮動小数テンソルとして扱うので、dtypeを明示する
boxes = torch.tensor([
    [100, 100, 210, 210],  # ほぼ同じ位置のボックスA
    [105, 102, 215, 208],  # ボックスAとほぼ重複
    [400, 300, 480, 380],  # 別の物体のボックスB
], dtype=torch.float32)
scores = torch.tensor([0.95, 0.80, 0.88])

keep_idx = nms(boxes, scores, iou_threshold=0.5)
print("残るボックスのインデックス:", keep_idx.tolist())
print("残るボックス:", boxes[keep_idx])

from ultralytics import YOLO

model = YOLO("runs_custom/warehouse_v1/weights/best.pt")
metrics = model.val(data="dataset/data.yaml")

print("mAP50:", metrics.box.map50)        # IoU閾値0.5でのmAP
print("mAP50-95:", metrics.box.map)       # IoU 0.5〜0.95を平均したmAP
print("クラスごとのmAP50-95:", metrics.box.maps)

from ultralytics import YOLO

model = YOLO("yolo11n.pt")
result = model("store_floor.jpg", conf=0.5)[0]

PERSON_CLASS_ID = 0  # UltralyticsのCOCOクラスでは0番が"person"
                     # torchvisionの検出モデルは0番が背景でpersonは1番。混同しやすい
# 滞在判定エリア(x1, y1, x2, y2): 例としてレジ前の矩形を定義
zone = (300, 200, 700, 480)

person_count = 0
in_zone_count = 0

for box in result.boxes:
    if int(box.cls[0]) != PERSON_CLASS_ID:
        continue
    person_count += 1

    x1, y1, x2, y2 = box.xyxy[0].tolist()
    foot_x = (x1 + x2) / 2   # 足元のx座標は矩形の中心
    foot_y = y2              # 足元のy座標は矩形の下端

    if zone[0] <= foot_x <= zone[2] and zone[1] <= foot_y <= zone[3]:
        in_zone_count += 1

print(f"フレーム内の人数: {person_count}")
print(f"レジ前エリア内の人数: {in_zone_count}")

import cv2
from ultralytics import YOLO

model = YOLO("yolo11n.pt")

cap = cv2.VideoCapture("input_video.mp4")
fps = cap.get(cv2.CAP_PROP_FPS)
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))

fourcc = cv2.VideoWriter_fourcc(*"mp4v")
writer = cv2.VideoWriter("output_video.mp4", fourcc, fps, (width, height))

while True:
    ok, frame = cap.read()
    if not ok:
        break  # 動画の終端

    result = model(frame, conf=0.5, verbose=False)[0]
    annotated_frame = result.plot()  # 検出結果を描き込んだフレーム(BGR)

    writer.write(annotated_frame)

cap.release()
writer.release()
print("書き出し完了: output_video.mp4")

pip install transformers torch

from transformers import pipeline

# タスク名だけ指定すればデフォルトモデルが自動でダウンロードされる
classifier = pipeline("sentiment-analysis")

results = classifier([
    "This product is amazing, I love it!",
    "The service was terrible and slow."
])

for text, result in zip(
    ["This product is amazing, I love it!", "The service was terrible and slow."],
    results
):
    print(f"{text} -> {result['label']} ({result['score']:.3f})")

from transformers import AutoTokenizer

tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased")

text = "Unhappiness is not the opposite of happiness."

tokens = tokenizer.tokenize(text)
print("トークン:", tokens)

encoded = tokenizer.encode(text)
print("ID列:", encoded)

decoded = tokenizer.decode(encoded)
print("デコード結果:", decoded)

# 実際にモデルへ渡す形（attention_maskなど）まで確認する
inputs = tokenizer(text, return_tensors="pt", padding=True, truncation=True)
print(inputs)

from transformers import pipeline

sentiment_analyzer = pipeline(
    "sentiment-analysis",
    model="llm-book/bert-base-japanese-v3-marc_ja",
    tokenizer="llm-book/bert-base-japanese-v3-marc_ja"
)

texts = [
    "この製品はとても使いやすく、満足しています。",
    "対応が遅く、サポート品質にも不満が残りました。"
]

for text in texts:
    result = sentiment_analyzer(text)[0]
    print(f"{text} -> {result['label']} ({result['score']:.3f})")

from transformers import pipeline

classifier = pipeline(
    "zero-shot-classification",
    model="MoritzLaurer/mDeBERTa-v3-base-mnli-xnli"
)

inquiry = "注文した商品が届かないのですが、配送状況を教えてください。"
candidate_labels = [
    "配送に関する問い合わせ",
    "返品・交換の相談",
    "料金・請求に関する質問",
    "製品の使い方の質問"
]

# 仮説文の雛形も日本語にしておく（既定は英語なので日英が混ざる）
result = classifier(
    inquiry,
    candidate_labels,
    hypothesis_template="この文章は{}に関するものです。",
)

for label, score in zip(result["labels"], result["scores"]):
    print(f"{label}: {score:.3f}")

from transformers import pipeline

ner = pipeline(
    "ner",
    model="tsmatz/xlm-roberta-ner-japanese",
    aggregation_strategy="simple"
)

# 動作確認用の架空の文（実在の企業名は使わない）
text = "田中太郎さんは、サンプル商事株式会社の大阪支社で働いています。"
entities = ner(text)

for entity in entities:
    print(f"{entity['word']} ({entity['entity_group']}): {entity['score']:.3f}")

from transformers import AutoTokenizer, AutoModelForSeq2SeqLM

model_name = "tsmatz/mt5_summarize_japanese"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForSeq2SeqLM.from_pretrained(model_name)

long_text = "ここに要約したい長文のニュース記事やレポートのテキストが入ります。" * 5

def summarize(text, max_new_tokens=100):
    inputs = tokenizer(text, return_tensors="pt", truncation=True, max_length=512)
    output_ids = model.generate(
        **inputs,
        max_new_tokens=max_new_tokens,
        min_new_tokens=20,
        do_sample=False,     # 毎回同じ結果にする
    )
    return tokenizer.decode(output_ids[0], skip_special_tokens=True)

def summarize_long_text(text, max_chunk_chars=800):
    chunks = [text[i:i + max_chunk_chars] for i in range(0, len(text), max_chunk_chars)]
    return " ".join(summarize(chunk) for chunk in chunks)

print(summarize_long_text(long_text))

from transformers import AutoTokenizer, AutoModelForSeq2SeqLM

def load_translator(model_name):
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    model = AutoModelForSeq2SeqLM.from_pretrained(model_name)
    return tokenizer, model

def translate(text, tokenizer, model):
    inputs = tokenizer(text, return_tensors="pt", truncation=True, max_length=512)
    output_ids = model.generate(**inputs, max_new_tokens=256)
    return tokenizer.decode(output_ids[0], skip_special_tokens=True)

ja_en_tok, ja_en_model = load_translator("Helsinki-NLP/opus-mt-ja-en")
en_ja_tok, en_ja_model = load_translator("Helsinki-NLP/opus-mt-en-jap")

japanese_text = "この度は弊社製品をご利用いただき、誠にありがとうございます。"
print(translate(japanese_text, ja_en_tok, ja_en_model))

english_text = "Thank you for choosing our product."
print(translate(english_text, en_ja_tok, en_ja_model))

pip install sentence-transformers

from sentence_transformers import SentenceTransformer, util

model = SentenceTransformer("paraphrase-multilingual-MiniLM-L12-v2")

faq_questions = [
    "パスワードを忘れた場合はどうすればよいですか。",
    "支払い方法にはどのようなものがありますか。",
    "解約の手続き方法を教えてください。"
]
faq_embeddings = model.encode(faq_questions, convert_to_tensor=True)

query = "ログインパスワードがわからなくなりました。"
query_embedding = model.encode(query, convert_to_tensor=True)

scores = util.cos_sim(query_embedding, faq_embeddings)[0]
best_idx = scores.argmax().item()

print(f"最も近いFAQ: {faq_questions[best_idx]} (類似度: {scores[best_idx]:.3f})")

from sentence_transformers import SentenceTransformer
from sklearn.cluster import KMeans

model = SentenceTransformer("paraphrase-multilingual-MiniLM-L12-v2")

inquiries = [
    "商品の配送が遅れています。",
    "注文した商品がまだ届きません。",
    "料金の請求内容について確認したいです。",
    "請求書の金額が違う気がします。",
    "サイトのログインができません。",
    "パスワードの再設定がうまくいきません。"
]

embeddings = model.encode(inquiries)
kmeans = KMeans(n_clusters=3, random_state=42, n_init=10)
labels = kmeans.fit_predict(embeddings)

for label in sorted(set(labels)):
    print(f"--- クラスタ{label} ---")
    for inquiry, l in zip(inquiries, labels):
        if l == label:
            print(f"  {inquiry}")

pip install datasets

from datasets import load_dataset

# 英語の映画レビューデータセットを例に処理の流れを示す
dataset = load_dataset("imdb")

def add_length(example):
    example["text_length"] = len(example["text"])
    return example

dataset = dataset.map(add_length)

small_train = dataset["train"].shuffle(seed=42).select(range(2000))
split = small_train.train_test_split(test_size=0.2, seed=42)

print(split)
print(split["train"][0]["text"][:200])

pip install transformers torch datasets evaluate accelerate scikit-learn

from transformers import AutoTokenizer, AutoModelForSequenceClassification
from transformers import TrainingArguments, Trainer
from datasets import load_dataset
import numpy as np
import evaluate

dataset = load_dataset("imdb")
small_train = dataset["train"].shuffle(seed=42).select(range(1000))
small_eval = dataset["test"].shuffle(seed=42).select(range(200))

model_name = "distilbert-base-uncased"
tokenizer = AutoTokenizer.from_pretrained(model_name)

def tokenize_fn(batch):
    return tokenizer(batch["text"], padding="max_length", truncation=True, max_length=256)

train_tokenized = small_train.map(tokenize_fn, batched=True)
eval_tokenized = small_eval.map(tokenize_fn, batched=True)

model = AutoModelForSequenceClassification.from_pretrained(model_name, num_labels=2)
accuracy = evaluate.load("accuracy")

def compute_metrics(eval_pred):
    logits, labels = eval_pred
    predictions = np.argmax(logits, axis=-1)
    return accuracy.compute(predictions=predictions, references=labels)

training_args = TrainingArguments(
    output_dir="./results",
    num_train_epochs=1,
    per_device_train_batch_size=8,
    per_device_eval_batch_size=8,
    eval_strategy="epoch",
    save_strategy="epoch",
    logging_steps=50
)

trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=train_tokenized,
    eval_dataset=eval_tokenized,
    compute_metrics=compute_metrics
)

trainer.train()

# 前提: レシピ73で学習したtrainerとtokenizerをそのまま引き継ぎます
trainer.save_model("./my_sentiment_model")
tokenizer.save_pretrained("./my_sentiment_model")

from transformers import AutoModelForSequenceClassification, AutoTokenizer, pipeline

loaded_model = AutoModelForSequenceClassification.from_pretrained("./my_sentiment_model")
loaded_tokenizer = AutoTokenizer.from_pretrained("./my_sentiment_model")

classifier = pipeline(
    "text-classification",
    model=loaded_model,
    tokenizer=loaded_tokenizer
)

print(classifier("This movie was absolutely wonderful."))

# f1 は内部で scikit-learn を使うので一緒に入れる
pip install evaluate scikit-learn

import evaluate
import numpy as np

accuracy_metric = evaluate.load("accuracy")
f1_metric = evaluate.load("f1")

predictions = np.array([1, 0, 1, 1, 0, 1, 0, 0])
references = np.array([1, 0, 0, 1, 0, 1, 1, 0])

accuracy_result = accuracy_metric.compute(predictions=predictions, references=references)
f1_result = f1_metric.compute(predictions=predictions, references=references, average="binary")

print("Accuracy:", accuracy_result)
print("F1:", f1_result)

from transformers import AutoTokenizer, AutoModelForSequenceClassification
import torch

model_name = "distilbert-base-uncased-finetuned-sst-2-english"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForSequenceClassification.from_pretrained(model_name)

def classify_long_text(text, max_length=512, stride=50):
    encodings = tokenizer(
        text,
        max_length=max_length,
        truncation=True,
        stride=stride,
        return_overflowing_tokens=True,
        padding="max_length",   # 最後の区間だけ短くなるとテンソルにまとめられない
        return_tensors="pt"
    )
    all_probs = []
    for i in range(encodings["input_ids"].shape[0]):
        inputs = {
            "input_ids": encodings["input_ids"][i].unsqueeze(0),
            "attention_mask": encodings["attention_mask"][i].unsqueeze(0)
        }
        with torch.no_grad():
            outputs = model(**inputs)
        # 区間ごとに確率へ直してから平均する（生のロジットのまま平均しない）
        all_probs.append(torch.softmax(outputs.logits, dim=-1))

    averaged_probs = torch.mean(torch.cat(all_probs, dim=0), dim=0)
    predicted_id = torch.argmax(averaged_probs).item()
    return model.config.id2label[predicted_id], averaged_probs

long_review = "This is a very long product review. " * 100
label, probs = classify_long_text(long_review)
print(f"予測ラベル: {label}  確率: {probs.tolist()}")

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split

texts = [
    "この商品は素晴らしいです。",
    "対応が悪くて残念でした。",
    "とても満足しています。",
    "二度と利用したくありません。",
    "期待どおりの品質で満足です。",
    "説明と違っていて不満です。",
    "梱包も丁寧で好印象でした。",
    "初期不良があり困りました。"
]
labels = [1, 0, 1, 0, 1, 0, 1, 0]

# stratifyを付けないと、分割した学習側が片方のクラスだけになり学習できないことがある
X_train, X_test, y_train, y_test = train_test_split(
    texts, labels, test_size=0.5, random_state=42, stratify=labels
)

# 日本語は空白で単語が区切られないため、既定の単語分割では特徴が粗くなる。
# 簡易にはこのように文字N-gramを使い、本格的にはMeCab等で分かち書きしてから渡す
vectorizer = TfidfVectorizer(analyzer="char", ngram_range=(2, 3))
X_train_vec = vectorizer.fit_transform(X_train)
X_test_vec = vectorizer.transform(X_test)

clf = LogisticRegression()
clf.fit(X_train_vec, y_train)

predictions = clf.predict(X_test_vec)
print("予測:", predictions)
print("正解:", y_test)

import os
from transformers import pipeline

# モデル名は環境変数で切り替えられるようにしておく
MODEL_NAME = os.environ.get("LOCAL_LLM_MODEL", "Qwen/Qwen2.5-0.5B-Instruct")

generator = pipeline(
    "text-generation",
    model=MODEL_NAME,
    dtype="auto",
    device_map="auto",  # GPUがあれば自動的に使用
)

prompt = "在庫管理を効率化するアイデアを3つ挙げてください。"

outputs = generator(
    prompt,
    max_new_tokens=200,
    do_sample=True,
    temperature=0.7,
)

print(outputs[0]["generated_text"])

import os
from transformers import AutoTokenizer, AutoModelForCausalLM

# 小型の指示応答モデルを例に使用(環境変数で差し替え可能)
MODEL_NAME = os.environ.get("LOCAL_LLM_MODEL", "Qwen/Qwen2.5-0.5B-Instruct")

tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
model = AutoModelForCausalLM.from_pretrained(
    MODEL_NAME,
    dtype="auto",
    device_map="auto",
)

messages = [
    {"role": "system", "content": "あなたは社内ヘルプデスクのアシスタントです。丁寧な日本語で回答してください。"},
    {"role": "user", "content": "経費精算システムのパスワードを忘れました。どうすればよいですか。"},
]

input_ids = tokenizer.apply_chat_template(
    messages,
    tokenize=True,
    add_generation_prompt=True,
    return_tensors="pt",
).to(model.device)

output_ids = model.generate(input_ids, max_new_tokens=150)
response = tokenizer.decode(
    output_ids[0][input_ids.shape[-1]:],
    skip_special_tokens=True,
)
print(response)

# 前提: レシピ79で読み込んだtokenizerとmodelを使います
def generate(prompt, temperature=0.7, top_p=0.9, max_new_tokens=200, repetition_penalty=1.1):
    input_ids = tokenizer.apply_chat_template(
        [{"role": "user", "content": prompt}],
        tokenize=True,
        add_generation_prompt=True,
        return_tensors="pt",
    ).to(model.device)
    output_ids = model.generate(
        input_ids,
        do_sample=True,
        temperature=temperature,
        top_p=top_p,
        max_new_tokens=max_new_tokens,
        repetition_penalty=repetition_penalty,
    )
    return tokenizer.decode(
        output_ids[0][input_ids.shape[-1]:],
        skip_special_tokens=True,
    )

# temperatureを変えて挙動を比較する
for temp in [0.2, 0.7, 1.2]:
    print(f"--- temperature={temp} ---")
    print(generate("新製品のキャッチコピーを1つ考えてください。", temperature=temp))

import os
import torch
from transformers import AutoTokenizer, AutoModelForCausalLM, BitsAndBytesConfig

MODEL_NAME_LARGE = os.environ.get("LARGE_LLM_MODEL", "elyza/Llama-3-ELYZA-JP-8B")

quant_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_compute_dtype=torch.bfloat16,
    bnb_4bit_use_double_quant=True,
)

tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME_LARGE)
model = AutoModelForCausalLM.from_pretrained(
    MODEL_NAME_LARGE,
    quantization_config=quant_config,
    device_map="auto",
)

print(f"モデル: {MODEL_NAME_LARGE}")
print(f"確保されたGPUメモリ: {torch.cuda.memory_allocated() / 1024**3:.1f} GB")

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ.get("OPENAI_BASE_URL"),  # 未設定ならOpenAI本体を利用
)

response = client.chat.completions.create(
    model=os.environ["OPENAI_MODEL"],
    messages=[
        {"role": "system", "content": "あなたは丁寧な日本語で回答するアシスタントです。"},
        {"role": "user", "content": "在庫回転率を改善する施策を3つ挙げてください。"},
    ],
    temperature=0.3,
)

print(response.choices[0].message.content)

import json
import os
import pandas as pd

# 前提: レシピ82で作成したclientを使います
response = client.chat.completions.create(
    model=os.environ["OPENAI_MODEL"],  # 利用するモデル名は環境変数で指定
    messages=[
        {"role": "system", "content": "商品レビューを分析し、指定したJSON形式のみで出力してください。"},
        {"role": "user", "content": "とても使いやすいが、価格がやや高いと感じた。"},
    ],
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "review_analysis",
            "strict": True,
            "schema": {
                "type": "object",
                "properties": {
                    "sentiment": {"type": "string", "enum": ["positive", "negative", "neutral"]},
                    "keywords": {"type": "array", "items": {"type": "string"}},
                },
                "required": ["sentiment", "keywords"],
                "additionalProperties": False,
            },
        },
    },
)

choice = response.choices[0]
if getattr(choice.message, "refusal", None) or choice.finish_reason == "length":
    raise RuntimeError("構造化出力を取得できませんでした(拒否または出力打ち切り)")

result = json.loads(choice.message.content)
df = pd.DataFrame([result])
print(df)

import time
import pandas as pd

# 前提: レシピ82のclient(とos.environ["OPENAI_MODEL"])を使います
def classify_with_retry(text, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model=os.environ["OPENAI_MODEL"],
                messages=[
                    {"role": "system", "content": "顧客からの問い合わせを『請求』『技術』『その他』のいずれかに分類してください。分類名のみ出力すること。"},
                    {"role": "user", "content": text},
                ],
                temperature=0,
            )
            return response.choices[0].message.content.strip()
        except Exception as e:
            wait = 2 ** attempt
            print(f"エラーが発生したため{wait}秒後に再試行します: {e}")
            time.sleep(wait)
    return "分類失敗"

df_inquiries = pd.DataFrame({
    "text": ["請求書の金額が間違っています", "アプリが起動しません", "営業時間を教えてください"],
})
df_inquiries["category"] = df_inquiries["text"].apply(classify_with_retry)
print(df_inquiries)

import numpy as np

# 前提: レシピ82のclientを使います(埋め込み取得のAPIを呼び出すため)
texts = [
    "在庫管理システムの導入事例",
    "需要予測モデルの精度改善",
    "配送ルートの最適化",
]

response = client.embeddings.create(
    model=os.environ.get("EMBEDDING_MODEL", "text-embedding-3-small"),
    input=texts,
)

embeddings = np.array([item.embedding for item in response.data])
np.save("document_embeddings.npy", embeddings)
print(embeddings.shape)

import numpy as np

# 前提: レシピ82のclientと、レシピ85と同様の埋め込み取得を使います
def chunk_text(text, chunk_size=200, overlap=50):
    chunks = []
    start = 0
    while start < len(text):
        chunks.append(text[start:start + chunk_size])
        start += chunk_size - overlap
    return chunks

documents = ["長い社内マニュアルの本文がここに入る想定です。実際には複数ファイルを読み込みます。"]
chunks = []
for doc in documents:
    chunks.extend(chunk_text(doc))

chunk_embeddings = np.array([
    item.embedding
    for item in client.embeddings.create(
        model=os.environ.get("EMBEDDING_MODEL", "text-embedding-3-small"),
        input=chunks,
    ).data
])

def search(query, top_k=3):
    query_embedding = np.array(
        client.embeddings.create(
            model=os.environ.get("EMBEDDING_MODEL", "text-embedding-3-small"),
            input=[query],
        ).data[0].embedding
    )
    scores = chunk_embeddings @ query_embedding / (
        np.linalg.norm(chunk_embeddings, axis=1) * np.linalg.norm(query_embedding)
    )
    top_indices = np.argsort(scores)[::-1][:top_k]
    return [(chunks[i], scores[i]) for i in top_indices]

results = search("有給休暇の申請方法")
for chunk, score in results:
    print(score, chunk[:50])

# 前提: レシピ86のsearch関数と、レシピ82のclientを使います
def answer_with_rag(query):
    results = search(query, top_k=3)
    context = "\n\n".join(
        [f"[出典{i+1}] {chunk}" for i, (chunk, _) in enumerate(results)]
    )
    messages = [
        {"role": "system", "content": "以下の資料の範囲内でのみ回答し、根拠にした出典番号を文末に付記してください。資料に記載のない内容は「資料からは分かりません」と答えてください。"},
        {"role": "user", "content": f"資料:\n{context}\n\n質問: {query}"},
    ]
    response = client.chat.completions.create(
        model=os.environ["OPENAI_MODEL"],
        messages=messages,
        temperature=0,
    )
    return response.choices[0].message.content

print(answer_with_rag("有給休暇の申請方法を教えてください"))

import os
from datasets import Dataset
from transformers import (AutoTokenizer, AutoModelForCausalLM,
                          TrainingArguments, Trainer, DataCollatorForLanguageModeling)
from peft import LoraConfig, TaskType, get_peft_model

MODEL_NAME = os.environ.get("LOCAL_LLM_MODEL", "Qwen/Qwen2.5-0.5B-Instruct")
tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
base_model = AutoModelForCausalLM.from_pretrained(MODEL_NAME)

lora_config = LoraConfig(
    r=8,
    lora_alpha=16,
    target_modules=["q_proj", "v_proj"],
    lora_dropout=0.05,
    task_type=TaskType.CAUSAL_LM,
)

model = get_peft_model(base_model, lora_config)
model.print_trainable_parameters()  # 学習対象が全体のごく一部に絞られていることを確認できる

train_texts = [
    "社内用語『KPI』とは重要業績評価指標のことです。",
    "有給休暇の申請は人事システムから行います。",
]
dataset = Dataset.from_dict({"text": train_texts})

def tokenize(batch):
    return tokenizer(batch["text"], truncation=True, padding="max_length", max_length=64)

tokenized_dataset = dataset.map(tokenize, batched=True)

training_args = TrainingArguments(
    output_dir="./lora_output",
    per_device_train_batch_size=2,
    num_train_epochs=3,
    learning_rate=2e-4,
    logging_steps=10,
)

# 因果言語モデルの学習ではlabels(=input_ids)を作るデータコレータを渡す
collator = DataCollatorForLanguageModeling(tokenizer=tokenizer, mlm=False)
trainer = Trainer(model=model, args=training_args,
                  train_dataset=tokenized_dataset, data_collator=collator)
trainer.train()
model.save_pretrained("./lora_adapter")

import json

# 前提: レシピ82のclient(とos.environ["OPENAI_MODEL"])を使います
def judge_response(question, answer, criteria="正確性と分かりやすさ"):
    messages = [
        {
            "role": "system",
            "content": (
                f"あなたは回答品質の評価者です。{criteria}の観点で1〜5点で採点し、"
                "理由を1文で述べてください。"
                '{"score": 整数, "reason": 文字列} という形式のJSONのみを出力してください。'
            ),
        },
        {"role": "user", "content": f"質問: {question}\n回答: {answer}"},
    ]
    response = client.chat.completions.create(
        model=os.environ["OPENAI_MODEL"],
        messages=messages,
        temperature=0,
    )
    return json.loads(response.choices[0].message.content)

result = judge_response(
    "有給休暇の申請方法を教えてください",
    "人事システムから申請書を提出し、上長の承認を得てください。",
)
print(result)

import requests

response = requests.post(
    "http://localhost:11434/api/generate",
    json={
        "model": "llama3.2",
        "prompt": "在庫管理を効率化するアイデアを3つ挙げてください。",
        "stream": False,
    },
    timeout=120,   # サーバー未起動のときに待ち続けないようにする
)
# モデル未取得などの失敗を、その場で例外として気づけるようにする
response.raise_for_status()

print(response.json()["response"])

import torch

def print_gpu_memory(tag: str = "") -> None:
    """現在のGPUメモリ使用状況を表示する"""
    if not torch.cuda.is_available():
        print("GPUが利用できないため、メモリ確認をスキップします")
        return
    allocated = torch.cuda.memory_allocated() / 1024**2
    reserved = torch.cuda.memory_reserved() / 1024**2
    print(f"[{tag}] 使用中: {allocated:.1f}MB / 確保済み: {reserved:.1f}MB")

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = torch.nn.Linear(1024, 1024).to(device)

print_gpu_memory("モデル配置後")

try:
    # バッチサイズを大きくしすぎるとOOMになりやすい
    x = torch.randn(65536, 1024, device=device)
    y = model(x)
    print_gpu_memory("順伝播後")
except torch.cuda.OutOfMemoryError:
    print("OOMが発生しました。バッチサイズを下げて再試行します")
    torch.cuda.empty_cache()  # 未使用のキャッシュを解放
    x_small = torch.randn(1024, 1024, device=device)
    y = model(x_small)
    print_gpu_memory("縮小後の順伝播後")

import torch
import torch.nn as nn

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = nn.Sequential(nn.Linear(512, 256), nn.ReLU(), nn.Linear(256, 10)).to(device)
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
criterion = nn.CrossEntropyLoss()

# fp16用の勾配スケーラー（アンダーフロー防止）
scaler = torch.amp.GradScaler("cuda", enabled=torch.cuda.is_available())

X = torch.randn(256, 512, device=device)
y = torch.randint(0, 10, (256,), device=device)

for epoch in range(3):
    optimizer.zero_grad()

    # autocastの範囲内では演算ごとに最適な精度が自動選択される
    with torch.amp.autocast("cuda", enabled=torch.cuda.is_available()):
        outputs = model(X)
        loss = criterion(outputs, y)

    # スケールした損失で逆伝播し、アンダーフローを防ぐ
    scaler.scale(loss).backward()
    scaler.step(optimizer)
    scaler.update()

    print(f"epoch {epoch}: loss={loss.item():.4f}")

import time
import torch
from torch.utils.data import TensorDataset, DataLoader

def build_loader(num_workers: int, pin_memory: bool) -> DataLoader:
    X = torch.randn(20000, 3, 64, 64)
    y = torch.randint(0, 10, (20000,))
    dataset = TensorDataset(X, y)
    return DataLoader(
        dataset,
        batch_size=128,
        shuffle=True,
        num_workers=num_workers,
        pin_memory=pin_memory,
        persistent_workers=(num_workers > 0),  # エポックごとの再起動を防ぐ
        prefetch_factor=4 if num_workers > 0 else None,  # 先読みするバッチ数
    )

def measure_one_epoch(loader: DataLoader) -> float:
    start = time.time()
    for _ in loader:
        pass
    return time.time() - start

if __name__ == "__main__":
    for workers in [0, 2, 4]:
        loader = build_loader(num_workers=workers, pin_memory=True)
        elapsed = measure_one_epoch(loader)
        print(f"num_workers={workers}: {elapsed:.2f}秒")

import time
import torch
import torch.nn as nn

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

class SimpleCNN(nn.Module):
    def __init__(self):
        super().__init__()
        self.net = nn.Sequential(
            nn.Conv2d(3, 32, 3, padding=1), nn.ReLU(),
            nn.Conv2d(32, 64, 3, padding=1), nn.ReLU(),
            nn.AdaptiveAvgPool2d(1), nn.Flatten(),
            nn.Linear(64, 10),
        )

    def forward(self, x):
        return self.net(x)

model = SimpleCNN().to(device)
compiled_model = torch.compile(model)  # 追加はこの1行のみ

X = torch.randn(64, 3, 64, 64, device=device)

# 初回はコンパイルが走るため時間がかかる
start = time.time()
_ = compiled_model(X)
print(f"初回実行（コンパイル込み）: {time.time() - start:.2f}秒")

# 2回目以降は同じ入力形状であればコンパイル済みの高速なパスが使われる
start = time.time()
for _ in range(10):
    _ = compiled_model(X)
print(f"2回目以降10回の実行: {time.time() - start:.2f}秒")

import torch
import torch.nn as nn
import numpy as np
import onnxruntime as ort

class SimpleNet(nn.Module):
    def __init__(self):
        super().__init__()
        self.fc = nn.Sequential(nn.Linear(10, 32), nn.ReLU(), nn.Linear(32, 1))

    def forward(self, x):
        return self.fc(x)

model = SimpleNet().eval()
dummy_input = torch.randn(1, 10)

# ONNX形式で書き出す（バッチサイズを可変にする設定も含む）
torch.onnx.export(
    model,
    dummy_input,
    "simple_net.onnx",
    input_names=["input"],
    output_names=["output"],
    dynamo=False,  # dynamic_axes は旧経路の引数。新経路では dynamic_shapes を使う
    dynamic_axes={"input": {0: "batch_size"}, "output": {0: "batch_size"}},
    opset_version=17,
)

# onnxruntimeで推論し、PyTorchの出力と一致するか確認する
session = ort.InferenceSession("simple_net.onnx", providers=["CPUExecutionProvider"])
test_input = np.random.randn(4, 10).astype(np.float32)

with torch.no_grad():
    torch_output = model(torch.from_numpy(test_input)).numpy()

onnx_output = session.run(["output"], {"input": test_input})[0]

print("最大誤差:", np.abs(torch_output - onnx_output).max())

import torch
import torch.nn as nn
from torch.utils.tensorboard import SummaryWriter

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = nn.Sequential(nn.Linear(20, 32), nn.ReLU(), nn.Linear(32, 2)).to(device)
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
criterion = nn.CrossEntropyLoss()

# ログの出力先ディレクトリ（実験ごとにサブフォルダを分けるのが定番）
writer = SummaryWriter(log_dir="runs/experiment_001")

X = torch.randn(500, 20, device=device)
y = torch.randint(0, 2, (500,), device=device)

for epoch in range(20):
    optimizer.zero_grad()
    outputs = model(X)
    loss = criterion(outputs, y)
    loss.backward()
    optimizer.step()

    acc = (outputs.argmax(dim=1) == y).float().mean().item()

    # スカラー値をタグ付きで記録する
    writer.add_scalar("Loss/train", loss.item(), epoch)
    writer.add_scalar("Accuracy/train", acc, epoch)

writer.close()
print("記録完了。'tensorboard --logdir runs' で確認できます")

tensorboard --logdir runs

pip install fastapi "uvicorn[standard]"

# main.py
import torch
import torch.nn as nn
from fastapi import FastAPI
from pydantic import BaseModel, Field

N_FEATURES = 4

class SimpleNet(nn.Module):
    def __init__(self):
        super().__init__()
        self.fc = nn.Sequential(nn.Linear(N_FEATURES, 16), nn.ReLU(), nn.Linear(16, 1))

    def forward(self, x):
        return self.fc(x)

# サーバー起動時に一度だけモデルを読み込む
model = SimpleNet()
model.load_state_dict(torch.load("model.pt", map_location="cpu", weights_only=True))
model.eval()

app = FastAPI(title="推論API")

class PredictRequest(BaseModel):
    # 長さを検証しておかないと、要素数が違う入力で500エラーになる
    features: list[float] = Field(min_length=N_FEATURES, max_length=N_FEATURES)

class PredictResponse(BaseModel):
    prediction: float

@app.get("/health")
def health() -> dict:
    return {"status": "ok"}

@app.post("/predict", response_model=PredictResponse)
def predict(request: PredictRequest) -> PredictResponse:
    x = torch.tensor([request.features], dtype=torch.float32)
    with torch.no_grad():
        y = model(x)
    return PredictResponse(prediction=y.item())

uvicorn main:app --host 0.0.0.0 --port 8000

import time
import pandas as pd
import torch

def load_model():
    model = torch.nn.Linear(10, 1)
    model.eval()
    return model

def predict_chunk(model, chunk: pd.DataFrame) -> list:
    x = torch.tensor(chunk.values, dtype=torch.float32)
    with torch.no_grad():
        return model(x).squeeze(-1).tolist()

def run_batch_inference(input_path: str, output_path: str, chunk_size: int = 1000, max_retries: int = 2) -> None:
    model = load_model()
    failed_chunks = []
    wrote_header = False

    reader = pd.read_csv(input_path, chunksize=chunk_size)
    for i, chunk in enumerate(reader):
        preds = None
        for attempt in range(1, max_retries + 1):
            try:
                preds = predict_chunk(model, chunk)
                print(f"チャンク{i}: {len(chunk)}件 完了")
                break
            except Exception as e:
                print(f"チャンク{i}: 失敗（{attempt}回目） {e}")
                if attempt == max_retries:
                    failed_chunks.append(i)
                time.sleep(1)

        # 失敗したチャンクも欠測として同じ行数だけ書き出す。
        # 飛ばしてしまうと入力の行と出力の行の対応が崩れ、後から突き合わせられなくなる
        if preds is None:
            preds = [None] * len(chunk)

        # チャンクごとに追記する。全件をメモリに溜めない
        out = pd.DataFrame({"row_id": chunk.index, "prediction": preds})
        out.to_csv(output_path, mode="w" if not wrote_header else "a",
                   header=not wrote_header, index=False)
        wrote_header = True

    print(f"完了。失敗チャンク数: {len(failed_chunks)} {failed_chunks}")

import numpy as np
import pandas as pd

def compute_baseline_stats(df: pd.DataFrame, columns: list) -> dict:
    """学習データの基準統計量を保存しておく"""
    stats = {}
    for col in columns:
        stats[col] = {
            "mean": df[col].mean(),
            "std": df[col].std(),
            "q10": df[col].quantile(0.1),
            "q90": df[col].quantile(0.9),
        }
    return stats

def check_drift(current_df: pd.DataFrame, baseline_stats: dict, threshold: float = 0.3) -> list:
    """基準統計量からの乖離が大きい列を検出する"""
    alerts = []
    for col, base in baseline_stats.items():
        current_mean = current_df[col].mean()
        # 標準偏差を単位にした乖離幅で判定する
        diff_ratio = abs(current_mean - base["mean"]) / (base["std"] + 1e-8)
        if diff_ratio > threshold:
            alerts.append({
                "column": col,
                "baseline_mean": base["mean"],
                "current_mean": current_mean,
                "diff_ratio": round(diff_ratio, 2),
            })
    return alerts

# 学習時に保存しておいた基準統計量（実運用ではファイルに保存・読み込みする）
baseline_df = pd.DataFrame({
    "気温": np.random.normal(20, 5, 1000),
    "湿度": np.random.normal(50, 10, 1000),
})
baseline_stats = compute_baseline_stats(baseline_df, ["気温", "湿度"])

# 本番投入後、ある日届いた入力データ（分布がシフトしたケースを想定）
current_df = pd.DataFrame({
    "気温": np.random.normal(28, 5, 500),
    "湿度": np.random.normal(50, 10, 500),
})
alerts = check_drift(current_df, baseline_stats, threshold=0.5)

for alert in alerts:
    print("ドリフト検知:", alert)

# これは全体の順序を示す疑似コードです。build_dataloaders などの関数名は、
# 前のレシピで書いた処理をひとまとめにした呼び名で、このままでは実行できません。
# ステップ1: データ準備
# - Dataset/DataLoaderを構築し、num_workers・pin_memoryで読み込みを高速化する（レシピ6, 7, 93）
train_loader, val_loader = build_dataloaders()

# ステップ2: モデル定義と学習
# - CNNまたは転移学習でモデルを構築する（セクション3、レシピ28〜42）
# - 再現性のため乱数シードを固定する（レシピ8）
# - AMPで学習を高速化・省メモリ化する（レシピ92）
# - 入力形状が固定ならtorch.compileで高速化する（レシピ94）
model = build_model()
set_seed(42)
# コンパイル前のモデルを必ず手元に残す。
# 保存とONNX化はコンパイル前のモデルに対して行う（compile後のstate_dictはキーが変わる）
compiled_model = torch.compile(model)

# - 学習ループ中はGPUメモリを定期的に確認する（レシピ91）
# - lossと評価指標をTensorBoard（必要に応じてMLflow）に記録する（レシピ96）
for epoch in range(num_epochs):
    train_one_epoch(compiled_model, train_loader)   # 学習はコンパイル済みで速く回す
    evaluate(compiled_model, val_loader)
    log_metrics_to_tensorboard(epoch)
    print_gpu_memory(tag=f"epoch{epoch}")

# ステップ3: モデルの保存（重みは共有されているので、元のmodelから保存すればよい）
torch.save(model.state_dict(), "model.pt")

# ステップ4: 推論形式への変換とAPI化
# - 必要に応じてONNXにエクスポートする（レシピ95）
# - FastAPIで推論APIとして公開する（レシピ97）
export_to_onnx(model, "model.onnx")
# 用意したmain.pyをuvicorn main:app --host 0.0.0.0 --port 8000 で起動する

# ステップ5: 運用
# - 夜間バッチで大量データを推論する場合はチャンク処理とリトライを組み込む（レシピ98）
# - 入力分布の変化を定点観測し、劣化の兆候を検知する（レシピ99）
run_batch_inference("new_data.csv", "predictions.csv")
alerts = check_drift(current_batch_df, baseline_stats)
if alerts:
    notify_team_and_consider_retraining(alerts)