はじめに

本記事は言語処理100本ノックの解説です。
100本のノックを全てこなした記録をQiitaに残します。
使用言語はPythonです。

今回は第9章: RNN, CNN(85~89)までの解答例をご紹介します。
9章前半(80~84)の解答例はこちらです。

85. 双方向RNN・多層化

コード

from torch.utils.data import TensorDataset, DataLoader
from torch.utils.tensorboard import SummaryWriter
import torch.nn as nn
import numpy as np
import torch

class RNN(nn.Module):
    def __init__(self, vocab_size, dw, dh, output):
        super().__init__()
        self.embed = nn.Embedding(vocab_size, dw, padding_idx=vocab_size-1)
        self.rnn = nn.RNN(dw, dh, batch_first=True, bidirectional=True)
        self.fc1 = nn.Linear(dh*2, output, bias=True)
        self.fc2 = nn.Softmax(dim=1)
        nn.init.xavier_normal_(self.rnn.weight_ih_l0)
        nn.init.xavier_normal_(self.rnn.weight_hh_l0)
        nn.init.xavier_normal_(self.rnn.weight_ih_l0_reverse)
        nn.init.xavier_normal_(self.rnn.weight_hh_l0_reverse)
        nn.init.xavier_normal_(self.fc1.weight)
    def forward(self, x):
        x = self.embed(x)
        _, x = self.rnn(x)
        rnn_out = torch.cat([x[-2,:,:], x[-1,:,:]], dim=1)
        x = self.fc1(rnn_out)
        x = self.fc2(x)
        return x

def calculate_loss_and_accuracy(model, dataset, device, criterion=None):
    dataloader = DataLoader(dataset, batch_size=1, shuffle=False)
    loss = 0.0
    total = 0
    correct = 0
    model = model.to(device)
    with torch.no_grad():
        for X, Y in dataloader:
            X = X.to(device)
            Y = Y.to(device)
            Y_pred = model(X)
            if criterion != None:
                loss += criterion(Y_pred, Y).item()
            pred = torch.argmax(Y_pred, dim=-1)
            total += len(Y)
            correct += (pred == Y).sum().item()

    return loss / len(dataset), correct / total

def train_model(X_train, y_train, X_test, y_test, batch_size, model, lr, num_epochs, device, collate_fn=None):
    dataset_train = TensorDataset(X_train, y_train)
    dataset_test = TensorDataset(X_test, y_test)
    model = model.to(device)
    dataloader_train = DataLoader(dataset_train, batch_size=batch_size, shuffle=True)
    dataloader_test = DataLoader(dataset_test, batch_size=batch_size, shuffle=True)
    criterion = nn.CrossEntropyLoss()
    criterion = criterion.to(device)
    for ep in range(num_epochs):
        optimizer = torch.optim.SGD(model.parameters(), lr=lr)
        if ep%30==0:
            lr = lr * 0.1
        model.train()
        for X, Y in dataloader_train:
            X = X.to(device)
            Y = Y.to(device)
            optimizer.zero_grad()
            Y_pred = model(X)
            loss = criterion(Y_pred, Y)
            loss.backward()
            optimizer.step()
        model.eval()

        loss_train, acc_train = calculate_loss_and_accuracy(model, dataset_train, device, criterion=criterion)
        loss_test, acc_test = calculate_loss_and_accuracy(model, dataset_test, device, criterion=criterion)

        print(f'epoch: {ep + 1}, loss_train: {loss_train:.4f}, accuracy_train: {acc_train:.4f}, loss_Test: {loss_test:.4f}, accuracy_Test: {acc_test:.4f}')
        TensorboardWriter(model, X_train, Y_train, ep, loss_train, "Train", device)
        TensorboardWriter(model, X_test, Y_test, ep, loss_test, "Test", device)

def TensorboardWriter(model, X, Y, epoch, loss, name, device):
    writer = SummaryWriter(log_dir="logs")
    model = model.to(device)
    X = X.to(device)
    Y_pred = model(X)
    result = torch.max(Y_pred.data, dim=1).indices
    result = result.cpu().data.numpy()
    Y_pred = np.array([np.argmax(y) for y in Y_pred.cpu().data.numpy()])
    Y = np.array([np.argmax(y) for y in Y.cpu().data.numpy()])
    result = torch.tensor(result)
    Y = torch.tensor(Y)
    Y_pred = torch.tensor(Y_pred)
    accuracy = result.eq(Y).sum().numpy()/len(Y_pred)
    writer.add_scalar("Loss/{}_Loss".format(name), loss, epoch)
    writer.add_scalar("Accuracy/{}_Accuracy".format(name), accuracy, epoch)
    writer.close()

def CountVocab(name):
    f = open("[PATH]/{}_code.txt".format(name), "r")
    lines = f.readlines()
    f.close()
    max_num = []
    for line in lines:
        line_t = line.split("\t")[2].replace("\n", "").split(" ")
        max_num.extend(map(int, line_t))
    vocab_max = max(max_num)+1
    return vocab_max

def GetCodeLow(name):
    f = open("[PATH]/{}_code.txt".format(name), "r")
    lines = f.readlines()
    f.close()
    num_list = []
    code_list = []
    pad_list = []
    for line in lines:
        line_s = line.split("\t")
        code_list.append(int(line_s[0]))
        num = line_s[2].replace("\n", "").split(" ")
        num = list(map(int, num))
        num_list.append(num)
        num_tensor = torch.tensor(num)
        pad_list.append(num_tensor)

    max_vocab = CountVocab("train")
    mlen = max([len(x) for x in num_list])
    pad_list = list(map(lambda x:x + [max_vocab]*(mlen-len(x)), num_list))
    pad_list = torch.tensor(pad_list)
    code_list = torch.tensor(code_list)
    return pad_list, code_list

X_train, Y_train = GetCodeLow("train")
X_test, Y_test = GetCodeLow("test")
BATCH_SIZE = 2
NUM_EPOCHS = 100
VOCAB_SIZE = CountVocab("train")+1
EMB_SIZE = 300
OUTPUT_SIZE = 4
HIDDEN_SIZE = 50
lr = 1e-2
device = "cuda:0"
model = RNN(VOCAB_SIZE, EMB_SIZE, HIDDEN_SIZE, OUTPUT_SIZE)
train_model(X_train, Y_train, X_test, Y_test, BATCH_SIZE, model, lr, NUM_EPOCHS, device)

出力結果

コメント
双方向RNNでやってみましたけど、精度はそれほど高くないですね。。。

86. 畳み込みニューラルネットワーク (CNN)

コード

import torch.nn.functional as F
from torch.utils.data import TensorDataset, DataLoader
from torch.utils.tensorboard import SummaryWriter
import torch.nn as nn
import numpy as np
import torch

class CNN(nn.Module):
  def __init__(self, vocab_size, dw, dh, output):
      super().__init__()
      self.embed = nn.Embedding(vocab_size, dw, padding_idx=vocab_size-1)
      self.conv1 = nn.Conv2d(1, 3, kernel_size=(3, 300))
      self.tanh = nn.Tanh()
      self.fc1 = nn.Linear(3, output, bias=True)
      self.fc2 = nn.Softmax(dim=1)
  def forward(self, x):
      x = self.embed(x)
      x = x.unsqueeze(1)
      x = self.conv1(x)
      x = self.tanh(x)
      x = F.max_pool2d(x, kernel_size=(x.size()[2], 1))
      x = x.view(-1, 3)
      x = self.fc1(x)
      x = self.fc2(x)
      return x

def CountVocab(name):
    f = open("[PATH]/{}_code.txt".format(name), "r")
    lines = f.readlines()
    f.close()
    max_num = []
    for line in lines:
        line_t = line.split("\t")[2].replace("\n", "").split(" ")
        max_num.extend(map(int, line_t))
    vocab_max = max(max_num)+1
    return vocab_max

def GetCodeLow(name):
    f = open("[PATH]/{}_code.txt".format(name), "r")
    lines = f.readlines()
    f.close()
    num_list = []
    code_list = []
    pad_list = []
    for line in lines:
        line_s = line.split("\t")
        code_list.append(int(line_s[0]))
        num = line_s[2].replace("\n", "").split(" ")
        num = list(map(int, num))
        num_list.append(num)
        num_tensor = torch.tensor(num)
        pad_list.append(num_tensor)
    max_vocab = CountVocab("train")
    mlen = max([len(x) for x in num_list])
    pad_list = list(map(lambda x:x + [max_vocab]*(mlen-len(x)), num_list))
    pad_list = torch.tensor(pad_list)
    code_list = torch.tensor(code_list)
    return pad_list, code_list

X_valid, Y_valid = GetCodeLow("valid")

VOCAB_SIZE = CountVocab("train")+1
EMB_SIZE = 300
OUTPUT_SIZE = 4
HIDDEN_SIZE = 50

model = CNN(VOCAB_SIZE, EMB_SIZE, HIDDEN_SIZE, OUTPUT_SIZE)
Y_pred = model(X_valid)
pred = torch.argmax(Y_pred, dim=-1)
print("accuracy: ", sum(1 for x,y in zip(Y_valid, pred) if x == y) / float(len(Y_valid)))

出力結果

コメント
今度はCNNをやっていきます。

87. 確率的勾配降下法によるCNNの学習

コード

%load_ext tensorboard
%tensorboard --logdir logs

import torch.nn.functional as F
from torch.utils.data import TensorDataset, DataLoader
from torch.utils.tensorboard import SummaryWriter
import torch.nn as nn
import numpy as np
import torch
import sys

class CNN(nn.Module):
  def __init__(self, vocab_size, dw, dh, output):
      super().__init__()
      self.embed = nn.Embedding(vocab_size, dw, padding_idx=vocab_size - 1)
      self.conv1 = nn.Conv2d(1, 3, kernel_size=(3, 300))
      self.tanh = nn.Tanh()
      self.conv2 = nn.Conv2d(1, 3, kernel_size=(5, 300))
      self.fc1 = nn.Linear(3, output, bias=True)
      self.fc2 = nn.Softmax(dim=1)
  def forward(self, x):
      x = self.embed(x)
      x = x.unsqueeze(1)
      x = self.conv1(x)
      x = self.tanh(x)
      x = F.max_pool2d(x, kernel_size=(x.size()[2], 1))
      x = x.view(-1, 3)
      x = self.fc1(x)
      x = self.fc2(x)
      return x

def calculate_loss_and_accuracy(model, dataset, device, criterion=None):
    dataloader = DataLoader(dataset, batch_size=1, shuffle=False)
    loss = 0.0
    total = 0
    correct = 0
    model = model.to(device)
    with torch.no_grad():
        for X, Y in dataloader:
            X = X.to(device)
            Y = Y.to(device)
            Y_pred = model(X)
            if criterion != None:
                loss += criterion(Y_pred, Y).item()
            pred = torch.argmax(Y_pred, dim=-1)
            total += len(Y)
            correct += (pred == Y).sum().item()

    return loss / len(dataset), correct / total


def train_model(X_train, y_train, X_test, y_test, batch_size, model, lr, num_epochs, device, collate_fn=None):
    dataset_train = TensorDataset(X_train, y_train)
    dataset_test = TensorDataset(X_test, y_test)
    model = model.to(device)
    dataloader_train = DataLoader(dataset_train, batch_size=batch_size, shuffle=True)
    dataloader_test = DataLoader(dataset_test, batch_size=batch_size, shuffle=True)
    criterion = nn.CrossEntropyLoss()
    criterion = criterion.to(device)
    for ep in range(num_epochs):
        optimizer = torch.optim.SGD(model.parameters(), lr=lr)
        model.train()
        for X, Y in dataloader_train:
            X = X.to(device)
            Y = Y.to(device)
            optimizer.zero_grad()
            Y_pred = model(X)
            loss = criterion(Y_pred, Y)
            loss.backward()
            optimizer.step()
        model.eval()

        loss_train, acc_train = calculate_loss_and_accuracy(model, dataset_train, device, criterion=criterion)
        loss_test, acc_test = calculate_loss_and_accuracy(model, dataset_test, device, criterion=criterion)

        print(f'epoch: {ep + 1}, loss_train: {loss_train:.4f}, accuracy_train: {acc_train:.4f}, loss_Test: {loss_test:.4f}, accuracy_Test: {acc_test:.4f}')
        TensorboardWriter(model, X_train, Y_train, ep, loss_train, "Train", device)
        TensorboardWriter(model, X_test, Y_test, ep, loss_test, "Test", device)

def TensorboardWriter(model, X, Y, epoch, loss, name, device):
    writer = SummaryWriter(log_dir="logs")
    model = model.to(device)
    X = X.to(device)
    Y_pred = model(X)
    result = torch.max(Y_pred.data, dim=1).indices
    result = result.cpu().data.numpy()
    Y_pred = np.array([np.argmax(y) for y in Y_pred.cpu().data.numpy()])
    Y = np.array([np.argmax(y) for y in Y.cpu().data.numpy()])
    result = torch.tensor(result)
    Y = torch.tensor(Y)
    Y_pred = torch.tensor(Y_pred)
    accuracy = result.eq(Y).sum().numpy()/len(Y_pred)
    writer.add_scalar("Loss/{}_Loss".format(name), loss, epoch)
    writer.add_scalar("Accuracy/{}_Accuracy".format(name), accuracy, epoch)
    writer.close()

def CountVocab(name):
    f = open("[PATH]/{}_code.txt".format(name), "r")
    lines = f.readlines()
    f.close()
    max_num = []
    for line in lines:
        line_t = line.split("\t")[2].replace("\n", "").split(" ")
        max_num.extend(map(int, line_t))
    vocab_max = max(max_num)+1
    return vocab_max

def GetCodeLow(name):
    f = open("[PATH]/{}_code.txt".format(name), "r")
    lines = f.readlines()
    f.close()
    num_list = []
    code_list = []
    pad_list = []
    for line in lines:
        line_s = line.split("\t")
        code_list.append(int(line_s[0]))
        num = line_s[2].replace("\n", "").split(" ")
        num = list(map(int, num))
        num_list.append(num)
        num_tensor = torch.tensor(num)
        pad_list.append(num_tensor)

    max_vocab = CountVocab("train")
    mlen = max([len(x) for x in num_list])
    pad_list = list(map(lambda x:x + [max_vocab]*(mlen-len(x)), num_list))
    pad_list = torch.tensor(pad_list)
    code_list = torch.tensor(code_list)
    return pad_list, code_list

X_train, Y_train = GetCodeLow("train")
X_test, Y_test = GetCodeLow("test")
BATCH_SIZE = 2
NUM_EPOCHS = 100
VOCAB_SIZE = CountVocab("train")+1
EMB_SIZE = 300
OUTPUT_SIZE = 4
HIDDEN_SIZE = 50
lr = 1e-3
device = "cuda:0"
model = CNN(VOCAB_SIZE, EMB_SIZE, HIDDEN_SIZE, OUTPUT_SIZE)
train_model(X_train, Y_train, X_test, Y_test, BATCH_SIZE, model, lr, NUM_EPOCHS, device)

出力結果

コメント
CNNでも精度はそれほど高くなかったですね。。。学習データの作り方がいけなかったのか。。

88. パラメータチューニング

準備
パラメータチューニングのためoptunaを入れます。

コマンド

$ pip install optuna

コード

コード

import optuna
import torch.nn.functional as F
from torch.utils.data import TensorDataset, DataLoader
from torch.utils.tensorboard import SummaryWriter
import torch.nn as nn
import numpy as np
import torch

class RNN(nn.Module):
    def __init__(self, vocab_size, dw, dh, output):
        super().__init__()
        self.embed = nn.Embedding(vocab_size, dw, padding_idx=vocab_size-1)
        self.rnn = nn.RNN(dw, dh, batch_first=True, bidirectional=True)
        self.fc1 = nn.Linear(dh*2, output, bias=True)
        self.fc2 = nn.Softmax(dim=1)
    def forward(self, x):
        x = self.embed(x)
        _, x = self.rnn(x)
        rnn_out = torch.cat([x[-2,:,:], x[-1,:,:]], dim=1)
        x = self.fc1(rnn_out)
        x = self.fc2(x)
        return x

class LSTM(nn.Module):
    def __init__(self, vocab_size, dw, dh, output):
        super().__init__()
        self.embed = nn.Embedding(vocab_size, dw, padding_idx=vocab_size-1)
        self.lstm = nn.LSTM(dw, dh, batch_first=True,bidirectional=True)
        self.fc1 = nn.Linear(dh*2, output, bias=True)
        self.fc2 = nn.Softmax(dim=1)
    def forward(self, x):
        x = self.embed(x)
        _, x = self.lstm(x)
        x = torch.cat([x[0][0],x[0][1]], dim=1)
        x = self.fc1(x)
        x = self.fc2(x)
        return x

class CNN(nn.Module):
  def __init__(self, vocab_size, dw, output, layer, unit, activation):
      super().__init__()
      self.layer = layer
      self.embed = nn.Embedding(vocab_size, dw, padding_idx = vocab_size-1)
      if unit == 6:
          units = [6, 4, 2]
      elif unit == 4:
          units = [4, 3, 2]
      elif unit == 2:
          units = [2, 2, 2]
      self.conv1 = nn.Conv2d(1, units[0], kernel_size=(units[0], 300))
      linearoutput = units[0]
      if layer > 1:
          self.conv2 = nn.Conv2d(units[0], units[1], kernel_size=(units[1],1))
          linearoutput = units[1]
      if layer > 2:
          self.conv3 = nn.Conv2d(units[1], units[2], kernel_size=(units[2],1))
          linearoutput = units[2]
      self.fc1 = nn.Linear(linearoutput, output, bias=True)
      self.fc2 = nn.Softmax(dim=1)

      if activation == "Tanh":
          self.active = nn.Tanh()
      elif activation == "ReLU":
          self.active = nn.ReLU()
      elif activation == "Sigmoid":
          self.active = nn.Sigmoid()
  def forward(self, x):
      x = self.embed(x)
      x = x.unsqueeze(1)
      x = self.conv1(x)
      x = self.active(x)
      if self.layer > 1:
          x = self.conv2(x)
          x = self.active(x)
      if self.layer > 2:
          x = self.conv3(x)
          x = self.active(x)
      x = F.max_pool2d(x, kernel_size=(x.size()[2], 1))
      x = x.view(x.size()[0], -1)
      x = self.fc1(x)
      x = self.fc2(x)
      return x

def calculate_loss_and_accuracy(model, dataset, device, criterion=None):
    dataloader = DataLoader(dataset, batch_size=1, shuffle=False)
    loss = 0.0
    total = 0
    correct = 0
    model = model.to(device)
    with torch.no_grad():
        for X, Y in dataloader:
            X = X.to(device)
            Y = Y.to(device)
            Y_pred = model(X)
            if criterion != None:
                loss += criterion(Y_pred, Y).item()
            pred = torch.argmax(Y_pred, dim=-1)
            total += len(Y)
            correct += (pred == Y).sum().item()
    return loss / len(dataset), correct / total


def train_model(X_train, y_train, X_test, y_test, batch_size, model, lr, num_epochs, device, collate_fn=None, optimizer_select="SGD"):
    dataset_train = TensorDataset(X_train, y_train)
    dataset_test = TensorDataset(X_test, y_test)
    model = model.to(device)
    dataloader_train = DataLoader(dataset_train, batch_size=batch_size, shuffle=True)
    dataloader_test = DataLoader(dataset_test, batch_size=batch_size, shuffle=True)
    criterion = nn.CrossEntropyLoss()
    criterion = criterion.to(device)
    for ep in range(num_epochs):
        if ep%20==0:
            lr = lr * 0.1
        if optimizer_select == "SGD":
            optimizer = torch.optim.SGD(model.parameters(), lr=lr)
        elif optimizer_select == "Adam":
            optimizer = torch.optim.Adam(model.parameters(), lr=lr)
        elif optimizer_select == "RMSprop":
          optimizer = torch.optim.RMSprop(model.parameters(), lr=lr)
        model.train()
        for X, Y in dataloader_train:
            X = X.to(device)
            Y = Y.to(device)
            optimizer.zero_grad()
            Y_pred = model(X)
            loss = criterion(Y_pred, Y)
            loss.backward()
            optimizer.step()
    model.eval()
    _, acc_test = calculate_loss_and_accuracy(model, dataset_test, device, criterion=criterion)

    return acc_test


def CountVocab(name):
    f = open("[PATH]/{}_code.txt".format(name), "r")
    lines = f.readlines()
    f.close()
    max_num = []
    for line in lines:
        line_t = line.split("\t")[2].replace("\n", "").split(" ")
        max_num.extend(map(int, line_t))
    vocab_max = max(max_num)+1
    return vocab_max

def GetCodeLow(name):
    f = open("[PATH]/{}_code.txt".format(name), "r")
    lines = f.readlines()
    f.close()
    num_list = []
    code_list = []
    pad_list = []
    for line in lines:
        line_s = line.split("\t")
        code_list.append(int(line_s[0]))
        num = line_s[2].replace("\n", "").split(" ")
        num = list(map(int, num))
        num_list.append(num)
        num_tensor = torch.tensor(num)
        pad_list.append(num_tensor)
    max_vocab = CountVocab("train")
    mlen = max([len(x) for x in num_list])
    pad_list = list(map(lambda x:x + [max_vocab]*(mlen-len(x)), num_list))
    pad_list = torch.tensor(pad_list)
    code_list = torch.tensor(code_list)
    return pad_list, code_list

def objective_RNN(trial):
    X_train, Y_train = GetCodeLow("train")
    X_test, Y_test = GetCodeLow("test")
    BATCH_SIZE = 2
    NUM_EPOCHS = 10
    VOCAB_SIZE = CountVocab("train")+1
    EMB_SIZE = 300
    OUTPUT_SIZE = 4
    lr = 1e-2
    device = "cuda:0"
    model_name_display_only = trial.suggest_categorical("model_name_RNN", ["RNN"])
    HIDDEN_SIZE = trial.suggest_categorical("HIDDEN_SIZE", [10, 50, 100, 500, 1000])
    optimizer_select = trial.suggest_categorical("optimizer_select", ["SGD", "Adam", "RMSprop"])
    model = RNN(VOCAB_SIZE, EMB_SIZE, HIDDEN_SIZE, OUTPUT_SIZE)
    score = train_model(X_train, Y_train, X_test, Y_test, BATCH_SIZE, model, lr, NUM_EPOCHS, device, optimizer_select)
    return score

def objective_LSTM(trial):
    X_train, Y_train = GetCodeLow("train")
    X_test, Y_test = GetCodeLow("test")
    BATCH_SIZE = 2
    NUM_EPOCHS = 10
    VOCAB_SIZE = CountVocab("train")+1
    EMB_SIZE = 300
    OUTPUT_SIZE = 4
    lr = 1e-2
    device = "cuda:0"
    model_name_display_only = trial.suggest_categorical("model_name_LSTM", ["LSTM"])
    HIDDEN_SIZE = trial.suggest_categorical("HIDDEN_SIZE", [10, 50, 100, 500, 1000])
    optimizer_select = trial.suggest_categorical("optimizer_select", ["SGD", "Adam", "RMSprop"])
    model = LSTM(VOCAB_SIZE, EMB_SIZE, HIDDEN_SIZE, OUTPUT_SIZE)
    score = train_model(X_train, Y_train, X_test, Y_test, BATCH_SIZE, model, lr, NUM_EPOCHS, device, optimizer_select)
    return score

def objective_CNN(trial):
    X_train, Y_train = GetCodeLow("train")
    X_test, Y_test = GetCodeLow("test")
    BATCH_SIZE = 2
    NUM_EPOCHS = 10
    VOCAB_SIZE = CountVocab("train")+1
    EMB_SIZE = 300
    OUTPUT_SIZE = 4
    lr = 1e-2
    device = "cuda:0"
    model_name_display_only = trial.suggest_categorical("model_name_CNN", ["CNN"])
    layer = trial.suggest_categorical("layer", [1,2,3])
    unit = trial.suggest_categorical("unit", [2,4,6])
    activation = trial.suggest_categorical("activation", ["Tanh", "Sigmoid", "ReLU"])
    optimizer_select = trial.suggest_categorical("optimizer_select", ["SGD", "Adam", "RMSprop"])
    model = CNN(VOCAB_SIZE, EMB_SIZE, OUTPUT_SIZE, layer, unit, activation)
    score = train_model(X_train, Y_train, X_test, Y_test, BATCH_SIZE, model, lr, NUM_EPOCHS, device, optimizer_select)
    return score

study = optuna.create_study(direction='maximize')
study.optimize(objective_CNN, n_trials=81)
study.optimize(objective_RNN, n_trials=15)
study.optimize(objective_LSTM, n_trials=15)
print(study.best_params)
print(study.best_value)

出力結果

{'model_name_CNN': 'CNN', 'layer': 1, 'unit': 6, 'activation': 'Tanh', 'optimizer_select': 'RMSprop'}
0.7241379310344828

コメント
各モデル10epochでの評価値で比較しました。他のモデルとはどんぐりの背比べでしたので、試行回数によっては違う結果になりますね。

89. 事前学習済み言語モデルからの転移学習

準備
Bertを使えるようにします。

コマンド

$ pip install transformers -q

コード

コード

%load_ext tensorboard
%tensorboard --logdir logs
from transformers import AutoTokenizer, AutoModel
from torch.utils.data import TensorDataset, DataLoader, Dataset
from torch.utils.tensorboard import SummaryWriter
import torch.nn as nn
import numpy as np
import torch

class Bert(nn.Module):
    def __init__(self):
        super().__init__()
        self.bert = AutoModel.from_pretrained("bert-base-uncased")
        self.classifier = nn.Linear(in_features = 768, out_features = 4)
    def forward(self, input_ids, attention_mask, token_type_ids):
        outputs = self.bert(input_ids = input_ids, attention_mask = attention_mask, token_type_ids = token_type_ids)
        pooler_output = outputs.pooler_output
        logits = self.classifier(pooler_output).squeeze(-1)
        return logits

class BertDataset(Dataset):
    def __init__(self, data, label):
        super().__init__()
        self.data_length = len(data["input_ids"])
        self.x_input_ids = data["input_ids"]
        self.x_token_type_ids = data["token_type_ids"]
        self.x_attention_mask = data["attention_mask"]
        self.y = label
    def __len__(self):
        return self.data_length
    def __getitem__(self, idx):
        x_input_ids = torch.tensor(self.x_input_ids[idx])
        x_token_type_ids = torch.tensor(self.x_token_type_ids[idx])
        x_attention_mask = torch.tensor(self.x_attention_mask[idx])
        return {"input_ids":x_input_ids, "token_type_ids":x_token_type_ids, "x_attention_mask":x_attention_mask}, torch.tensor(self.y[idx])

def calculate_loss_and_accuracy(model, dataset, device, criterion=None):
    dataloader = DataLoader(dataset, batch_size=256, shuffle=False)
    loss = 0.0
    total = 0
    correct = 0
    model = model.to(device)
    with torch.no_grad():
        for X, Y in dataloader:
            input_ids = X["input_ids"].to(device)
            attention_mask = X["x_attention_mask"].to(device)
            token_type_ids = X["token_type_ids"].to(device)
            Y = Y.to(device)
            Y_pred =  model(input_ids, attention_mask, token_type_ids)
            if criterion != None:
                loss += criterion(Y_pred, Y).item()
            pred = torch.argmax(Y_pred, dim=-1)
            total += len(Y)
            correct += (pred == Y).sum().item()
    return loss / len(dataset), correct / total


def train_model(X_train, y_train, X_test, y_test, batch_size, model, lr, num_epochs, device, collate_fn=None):
    dataset_train = BertDataset(X_train, y_train)
    dataset_test = BertDataset(X_test, y_test)
    dataloader_train = DataLoader(dataset_train, batch_size=batch_size, shuffle=True)
    dataloader_test = DataLoader(dataset_test, batch_size=batch_size, shuffle=True)
    model = model.to(device)
    criterion = nn.CrossEntropyLoss()
    criterion = criterion.to(device)
    for ep in range(num_epochs):
        optimizer = torch.optim.SGD(model.parameters(), lr=lr)
        model.train()
        if ep%30==0:
            lr = lr * 0.1
        for X, Y in dataloader_train:
            input_ids = X["input_ids"].to(device)
            attention_mask = X["x_attention_mask"].to(device)
            token_type_ids = X["token_type_ids"].to(device)

            Y = Y.to(device)
            optimizer.zero_grad()
            Y_pred = model(input_ids, attention_mask, token_type_ids)
            loss = criterion(Y_pred, Y)
            loss.backward()
            optimizer.step()
        model.eval()

        loss_train, acc_train = calculate_loss_and_accuracy(model, dataset_train, device, criterion=criterion)
        loss_test, acc_test = calculate_loss_and_accuracy(model, dataset_test, device, criterion=criterion)

        print(f'epoch: {ep + 1}, loss_train: {loss_train:.4f}, accuracy_train: {acc_train:.4f}, loss_Test: {loss_test:.4f}, accuracy_Test: {acc_test:.4f}')
        TensorboardWriter(ep, loss_train, acc_train, "Train")
        TensorboardWriter(ep, loss_test, acc_test, "Test")

def TensorboardWriter(epoch, loss, accuracy, name):
    writer = SummaryWriter(log_dir="logs")
    writer.add_scalar("Loss/{}_Loss".format(name), loss, epoch)
    writer.add_scalar("Accuracy/{}_Accuracy".format(name), accuracy, epoch)
    writer.close()

def CountVocab(name):
    f = open("[PATH]/{}_code.txt".format(name), "r")
    lines = f.readlines()
    f.close()
    max_num = []
    for line in lines:
        line_t = line.split("\t")[2].replace("\n", "").split(" ")
        max_num.extend(map(int, line_t))
    vocab_max = max(max_num)+1
    return vocab_max

def GetStrLow(name):
    f = open("[PATH]/RNN_CNN/{}_code.txt".format(name), "r")
    lines = f.readlines()
    f.close()
    sent_list = []
    code_list = []

    for line in lines:
        line_s = line.split("\t")
        code_list.append(int(line_s[0]))
        sent = line_s[1].replace("\n", "")
        sent_list.append(sent)
    code_list = torch.tensor(code_list)
    return sent_list, code_list

X_train, Y_train = GetStrLow("train")
X_test, Y_test = GetStrLow("test")

MAX_LENGTH = 32
tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased")
X_train_tokenizer = tokenizer.batch_encode_plus(X_train, padding = "max_length", max_length = MAX_LENGTH, truncation=True)
X_test_tokenizer = tokenizer.batch_encode_plus(X_test, padding = "max_length", max_length = MAX_LENGTH, truncation=True)

BATCH_SIZE = 8
NUM_EPOCHS = 100
lr = 1e-3
device = "cuda:0"
model = Bert()
train_model(X_train_tokenizer, Y_train, X_test_tokenizer, Y_test, BATCH_SIZE, model, lr, NUM_EPOCHS, device)

出力結果

コメント
分類精度93.7%を達成しました。やはり大規模言語モデルはすごいですね。

他章の解答例