【Python】pytorchでのモデルの作り方｜初心者向け

pytorchでは、最低限以下の４つを意識すれば大丈夫です。

①Datasetの作成

load_boston

今回はload_bostonデータを使います。

from sklearn.datasets import load_boston
import numpy as np
import pandas as pd

data = load_boston()
df = pd.DataFrame(data["data"], columns = data["feature_names"])
df["target"] = data["target"]
df

こんな感じです。

“CRIM”～”LSTAT”の特徴を使って”target”を予測します。

回帰タスクですね。

feature_cols = [c for c in df.columns if c != "target"]
print(feature_cols)

特徴量の列名を保存しておきました。

from sklearn.preprocessing import StandardScaler
scaler = StandardScaler()
for col in [c for c in df.columns if c not in ["CHAS", "target"]]:
    df[col] = scaler.fit_transform(df[col].values.reshape(-1, 1))

StandardScalerで標準化しておきましょう。

スケールが違うと、同じ１の変化でも意味合いが変わるので、分散１、平均０にします。

Dataset

pytorchモデルに必須であるDatasetを作りましょう。

import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader

こいつらはpytroch三種（四種？）の神器です。ほぼ必須。

class BostonData(Dataset):
    def __init__(self, input_df):
        self.data = input_df
    
    def __len__(self):
        return self.data.shape[0]
    
    def __getitem__(self, idx):
        row = self.data.loc[idx]
        features = row[feature_cols].values
        target = row["target"]
        return features, target

これがDatasetです。クラスで作成します。

selfはクラス自身を指しています。よくわからなくても書いとけばOK。

getitemのidxはどんな名称でもいいので、人によってiと書いたり様々です。

モデルには特徴量だけを入れたいので、特徴量(feature_cols)と正解(target)が別々に出てくるようにしましょう。

ds = BostonData(df)
ds[0]

Datasetを作りました。()にデータを入れるとinitが起動します。

ds[0]とはidx = 0のデータのことです。

出力とdfの１行目を見てみると同じ値のはずです。

もっと言うと、ds[0][0]が特徴量、ds[0][1]がtargetになってます。

行番号(idx)を入力するとデータが返ってくれば大丈夫です。

②DataLoaderの作成

dl = DataLoader(ds, batch_size = 4)

DataLoaderにさっき作ったDatasetつまりdsを入れます。

batch_sizeは一回の起動で何個データを出すか、ということです。

batch = next(iter(dl))
print(len(batch))
print(batch[0].shape)
batch[0]

next(iter())で１batchごとのデータを取り出せます。

batchの長さは２です。それぞれ特徴量とtargetのことです。

別々になるのは、Datasetの出力で特徴量とtargetを分けたからです。

batch[0]が特徴量で、サイズは４行13列です。

さっきbatch_size = 4にしたので、一回で４行分のデータを出しているわけですね。

試しにbatch[0]を見てみると、４行分の特徴量が出てくるはずです。

from tqdm.notebook import tqdm
for batch in tqdm(dl):
    pass

tqdmは進捗を表示してくれるライブラリです。

for文でbatchを取り出す作業だけをしてみると、127回実行されていますね。

127 * 4 = 508で、元々データは506行あったので、実行回数127はあっています。

最後のbatchだけ２個しか取り出しません。(126 * 4 + 2)

こんな感じで、batch_sizeごとのデータを出してくれたら完成です。

③Modelの作成

Modelの記述

class BostonModel(nn.Module):
    def __init__(self):
        super().__init__()
        self.linear1 = nn.Linear(in_features = len(feature_cols), out_features = 16)
        self.linear2 = nn.Linear(in_features = 16, out_features = 32)
        self.linear3 = nn.Linear(in_features = 32, out_features = 1)
    
    def forward(self, inputs):
        x = nn.ReLU()(self.linear1(inputs))
        x = nn.ReLU()(self.linear2(x))
        logits = self.linear3(x)
        return logits

モデルもクラスで作成します。

まずはinitでモデルの条件を書きます。

super().__init__()は、ただただ必須なので書いておけばいいです。

self.linear1とかは学習の層を決めています。

ニューラルネットの基礎知識は解説しませんので、下の本を読んでください。

Linearは全結合するだけの層で、in_featuresが入力サイズ、out_featuresが出力サイズです。

今回は特徴量が13なので、最初の層は13個入力されます。

len(feature_cols)にしておけば13になりますね。

out_featuresは適当です。大きくすればするほど複雑なモデルになります。

２層目のlinear2では、in_featuresが１層目のout_featuresと同じにならないとエラーになります。

    def forward(self, inputs):
        x = self.linear1(inputs)
        x = self.linear2(x)
        logits = self.linear3(x)
        return logits

このように、forward関数ではlinear1⇒linear2⇒linear3の順でデータが渡されるように設定したためです。

forward関数では特徴量を入力して、どのように計算が実行されるかを決めます。

入力がinputsで、１層目の出力がx、２層目はxを入力して新たなxとして更新。

最後にlogitsとして３層目が出力します。logitsとかxとか名前は適当です。

Modelの実行テスト

試しに１batchだけモデルに入れてみましょう。

model = BostonModel()
batch = next(iter(dl))
print(batch[0].shape)
logits = model(batch[0].float())
print(logits.shape)

クラス名を書けばinitが実行されてモデルが作られます。

ただし、学習していない作り立てのモデルはアホです。

１個だけbatchを取り出して、特徴量を入れます。

このときfloatと書いているのは、データの型をモデルに合わせないといけないからです。

floatにしたりlongにしたり色々あるので、エラーが起きたら柔軟に対応してください。

batch[0].shapeとは、１つのbatchのデータサイズです。

次の出力はモデルが予測した値です。

４行１列になっていて、これはbatch_sizeが４で予測したい列数つまりtargetが１個だからです。

こんな感じで特徴を入力したらモデルの予測値が出てきます。

あとは実際のtargetの値との誤差を計算して、モデルを修正していきます。

④学習

データ分割

全体のデータを学習用と評価用に分けます。

from sklearn.model_selection import KFold
df["fold"] = -1
kf = KFold(n_splits = 10, shuffle = True)
for fold, (_, valid_idx) in enumerate(kf.split(df)):
    df.loc[valid_idx, "fold"] = fold
df

10分割して、該当する番号をfold列にしました。

fold = 0
train = df.loc[df["fold"] != fold].reset_index(drop = True)
valid = df.loc[df["fold"] == fold].reset_index(drop = True)
train_ds = BostonData(train)
valid_ds = BostonData(valid)
train_dl = DataLoader(train_ds, batch_size = 16, shuffle = True)
valid_dl = DataLoader(valid_ds, batch_size = 16 * 2, shuffle = False)

fold0を評価用データにしておきます。

DatasetとDataLoaderをそれぞれ作成しましょう。

shuffleはシャッフルです。学習データはランダムに渡します。

評価データでのbatch_sizeは２倍にしておきました。

これは評価時は学習時よりもメモリ消費が少ないからです。

損失関数と最適化手法

model = BostonModel()
criterion = nn.MSELoss() #損失関数
optimizer = torch.optim.Adam(model.parameters()) #最適化手法

optimizerにはモデルのパラメータを入れます。

ほかにも学習率や減衰率を入れたりもしますが、長くなるので割愛。

学習ループ

model.train()
for batch in train_dl:
    optimizer.zero_grad() #必須。忘れやすい。
    logits = model(batch[0].float()).squeeze(-1) #予測値
    loss = criterion(logits, batch[1].float()) #誤差計算
    loss.backward() #誤差伝播
    optimizer.step() #パラメータ更新
    print(loss.item())

これらをfor文でループしていきます。

全batch繰り返したら１回の学習が終了です。

ちなみにcriterionの（）に入れるのは（予測値, 正解）の順ですのでご注意を。

評価ループ

model.eval()
with torch.no_grad(): #必須。忘れやすい。
    for batch in valid_dl:
        logits = model(batch[0].float()).squeeze(-1)
        loss = criterion(logits, batch[1].float())
        print(loss.item())

評価ループではパラメータ更新などが不要です。

誤差を計算して、評価データでの性能を見るだけ。

torch.no_grad()を忘れてメモリが吹っ飛ぶことがあるので注意。

エポックループ

これまでに設定したものを繰り返し実行します。

model = BostonModel()
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters())

history = {"train": [], "valid": []}

for epoch in range(30):
    model.train()
    train_loss = 0
    for batch in train_dl:
        optimizer.zero_grad()
        logits = model(batch[0].float()).squeeze(-1)
        loss = criterion(logits, batch[1].float())
        loss.backward()
        optimizer.step()
        train_loss += loss.item()
    train_loss /= len(train_dl)

    model.eval()
    valid_loss = 0
    with torch.no_grad():
        for batch in valid_dl:
            logits = model(batch[0].float()).squeeze(-1)
            loss = criterion(logits, batch[1].float())
            valid_loss += loss.item()
    valid_loss /= len(valid_dl)
    
    history["train"].append(train_loss)
    history["valid"].append(valid_loss)

この１単位をエポックと言ったりします。

今回は30エポック実行しましょう。

各学習と評価の誤差を合計し、historyに保存しました。

import matplotlib.pyplot as plt
plt.plot(history["train"], label = "train")
plt.plot(history["valid"], label = "valid")
plt.legend()
plt.show()

こんな感じで誤差がどんどん下がっていきます。

１エポック目ではとんでもなく誤差が大きいですが、10エポックくらいで落ち着いていますね。

基本的に学習データでの誤差の方が小さくなるはずですが、今回は逆になりました。

予測値出力

30エポック後の性能を見てみましょう。

model.eval()
pred = []
truth = []
with torch.no_grad():
    for batch in valid_dl:
        logits = model(batch[0].float()).squeeze(-1)
        pred.extend(logits.tolist())
        truth.extend(batch[1].tolist())

これで予測値を取り出します。

plt.scatter(truth, pred)
plt.plot(range(50), range(50), "red")
plt.xlabel("truth")
plt.ylabel("pred")
plt.show()

from sklearn.metrics import mean_squared_error
rmse = np.sqrt(mean_squared_error(truth, pred))
print("RMSE:", rmse)

予測値と正解をプロットしてみると、だいたいy = xの直線上にあります。

平均二乗誤差の平方根（RMSE）は5.8くらいです。

まとめ：pytorchに挑戦しよう

今回はpytorchでモデルを作る方法を解説しました。

簡単にまとめると以上。

あとはモデルの層を変えたり学習率やスケジューラーを設定したり、改善方法は無限にあります。

最後に全体のコードをまとめておきます。

from sklearn.datasets import load_boston
import numpy as np
import pandas as pd
data = load_boston()
df = pd.DataFrame(data["data"], columns = data["feature_names"])
df["target"] = data["target"]
feature_cols = [c for c in df.columns if c != "target"]

from sklearn.preprocessing import StandardScaler
scaler = StandardScaler()
for col in [c for c in df.columns if c not in ["CHAS", "target"]]:
    df[col] = scaler.fit_transform(df[col].values.reshape(-1, 1))

import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader
class BostonData(Dataset):
    def __init__(self, input_df):
        self.data = input_df
    
    def __len__(self):
        return self.data.shape[0]
    
    def __getitem__(self, idx):
        row = self.data.loc[idx]
        features = row[feature_cols].values
        target = row["target"]
        return features, target

class BostonModel(nn.Module):
    def __init__(self):
        super().__init__()
        self.linear1 = nn.Linear(in_features = len(feature_cols), out_features = 16)
        self.linear2 = nn.Linear(in_features = 16, out_features = 32)
        self.linear3 = nn.Linear(in_features = 32, out_features = 1)
    
    def forward(self, inputs):
        x = nn.ReLU()(self.linear1(inputs))
        x = nn.ReLU()(self.linear2(x))
        logits = self.linear3(x)
        return logits

from sklearn.model_selection import KFold
df["fold"] = -1
kf = KFold(n_splits = 10, shuffle = True)
for fold, (_, valid_idx) in enumerate(kf.split(df)):
    df.loc[valid_idx, "fold"] = fold

fold = 0
train = df.loc[df["fold"] != fold].reset_index(drop = True)
valid = df.loc[df["fold"] == fold].reset_index(drop = True)
train_ds = BostonData(train)
valid_ds = BostonData(valid)
train_dl = DataLoader(train_ds, batch_size = 16, shuffle = True)
valid_dl = DataLoader(valid_ds, batch_size = 16 * 2, shuffle = False)

model = BostonModel()
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters())

history = {"train": [], "valid": []}

for epoch in range(30):
    model.train()
    train_loss = 0
    for batch in train_dl:
        optimizer.zero_grad()
        logits = model(batch[0].float()).squeeze(-1)
        loss = criterion(logits, batch[1].float())
        loss.backward()
        optimizer.step()
        train_loss += loss.item()
    train_loss /= len(train_dl)

    model.eval()
    valid_loss = 0
    with torch.no_grad():
        for batch in valid_dl:
            logits = model(batch[0].float()).squeeze(-1)
            loss = criterion(logits, batch[1].float())
            valid_loss += loss.item()
    valid_loss /= len(valid_dl)
    
    history["train"].append(train_loss)
    history["valid"].append(valid_loss)

import matplotlib.pyplot as plt
plt.plot(history["train"], label = "train")
plt.plot(history["valid"], label = "valid")
plt.legend()
plt.show()

model.eval()
pred = []
truth = []
with torch.no_grad():
    for batch in valid_dl:
        logits = model(batch[0].float()).squeeze(-1)
        pred.extend(logits.tolist())
        truth.extend(batch[1].tolist())

plt.scatter(truth, pred)
plt.plot(range(50), range(50), "red")
plt.xlabel("truth")
plt.ylabel("pred")
plt.show()

from sklearn.metrics import mean_squared_error
rmse = np.sqrt(mean_squared_error(truth, pred))
print("RMSE:", rmse)