Day72_Ai(8)_Am

valid 라는 구역을 train 쪽에 하나 더 만듦

학습하면서 벨리 데이션을 loss 나 acc를 테스트로 쓴다.

 

1. 과적합이 안되는 모델을 사용 하던지 / /  dropout

2. 과적합 발생 시점이 됐을시 멈추던지  // early stop

 

dropout 몇퍼센트를 학습하는데 있어서 제외시킬것인지 => p 값 0.5는 절반을 랜덤하게 멈추겠다.

 

 

---

 

resultViewEx

import numpy as np
import matplotlib.pyplot as plt

plt.rc('font', family='Malgun Gothic')
plt.rcParams['axes.unicode_minus'] = False

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

torch.manual_seed(123)
from sklearn.datasets import load_iris

iris = load_iris()
x_org, y_org = iris.data, iris.target

x_data = iris.data[:100, :2]
y_data = iris.target[:100]

from sklearn.model_selection import train_test_split

x_train, x_valid, y_train, y_valid = train_test_split(x_data, y_data,
                                                     test_size=0.3, random_state=123)

class Net(nn.Module):
    def __init__(self, input_size, output_size):
        super().__init__()
        self.fc1 = nn.Linear(input_size, output_size)
        self.sigmoid = nn.Sigmoid()

        self.fc1.weight.data.fill_(1.0)      #1.0으로 채워넣는다
        self.fc1.bias.data.fill_(1.0)

    def forward(self, x):
        y = self.sigmoid(self.fc1(x))
        return y

model = Net(2,1)
loss_func = nn.BCELoss()
optimizer = optim.SGD(model.parameters(), lr=0.01)

x_train = torch.tensor(x_train).float()
x_valid = torch.tensor(x_valid).float()

y_train = torch.tensor(y_train).float()
y_valid = torch.tensor(y_valid).float()
y_train = y_train.view((-1,1))      #이건 왜해줌?
y_valid = y_valid.view((-1,1))

history = np.zeros((0,5))
print(history)

for epoch in range(10000):
    optimizer.zero_grad()
    hypothesis = model(x_train)
    loss = loss_func(hypothesis, y_train)
    loss.backward()
    optimizer.step()

    train_loss = loss.item()
    prediction = torch.where(hypothesis < 0.5, 0, 1)    #이 부분도 다시보기 먼지모름
    train_acc = (prediction == y_train).sum() / len(y_train)

    output_valid = model(x_valid)
    loss_valid = loss_func(output_valid, y_valid)
    val_loss = loss_valid.item()
    prediction_valid = torch.where(output_valid < 0.5, 0, 1)
    val_acc = (prediction_valid == y_valid).sum() / len(y_valid)

    if epoch % 10 == 0:
        print(f'epoch: {epoch+1} train_loss: {train_loss:.4f} train_acc: {train_acc:.4f} '+
              f'val_loss:{val_loss:.4f} val_acc:{val_loss:.4f}')
        history = np.vstack((history, np.array([epoch, train_loss, train_acc,
                                                val_loss, val_acc])))

plt.plot(history[:, 0], history[:, 1], label='train_loss')
plt.plot(history[:, 0], history[:, 3], label='valid_loss')
plt.xlabel('반복 횟수')
plt.ylabel('loss')
plt.legend(loc='best')
plt.ylim(0,1)
plt.show()

plt.plot(history[:, 0], history[:, 2], label='train_acc')
plt.plot(history[:, 0], history[:, 4], label='valid_acc')
plt.xlabel('반복 횟수')
plt.ylabel('acc')
plt.legend(loc='best')
plt.show()

 

---

droutEx1

import math, random, numpy as np, torch
from torch import nn
from torch.utils.data import TensorDataset, DataLoader
import matplotlib.pyplot as plt


SEED = 7
TRAIN_SIZE_PER_CLASS = 20     # 총 40개 (작을수록 오버피팅↑)
TRAIN_LABEL_NOISE = 0.20      # 훈련셋 라벨 뒤집기 비율(오버피팅 유도 핵심)
DATA_NOISE = 0.20             # 좌표에 가우시안 노이즈 (결정경계 어렵게)
EPOCHS = 300
LR = 3e-3
BATCH_TRAIN = 32
P_DROPOUT = 0.5               # 드롭아웃 확률
WEIGHT_DECAY_DROPOUT = 1e-3   # 드롭아웃 모델만 L2 함께 사용(일반화 도움)


random.seed(SEED); np.random.seed(SEED); torch.manual_seed(SEED)


def make_two_moons(n_samples=2000, noise=0.12):
    n = n_samples // 2
    t1 = np.random.rand(n) * math.pi
    x1 = np.stack([np.cos(t1), np.sin(t1)], axis=1)
    t2 = np.random.rand(n) * math.pi
    x2 = np.stack([1 - np.cos(t2), -np.sin(t2)], axis=1) + np.array([0.5, 0.25])
    X = np.concatenate([x1, x2], axis=0).astype(np.float32)
    y = np.concatenate([np.zeros(n, dtype=np.int64), np.ones(n, dtype=np.int64)])
    X += np.random.normal(scale=noise, size=X.shape).astype(np.float32)
    return X, y

X, y = make_two_moons(n_samples=2000, noise=DATA_NOISE)


c0_idx = np.where(y==0)[0]
c1_idx = np.where(y==1)[0]
np.random.shuffle(c0_idx); np.random.shuffle(c1_idx)

train_idx = np.concatenate([c0_idx[:TRAIN_SIZE_PER_CLASS], c1_idx[:TRAIN_SIZE_PER_CLASS]])
val_idx   = np.concatenate([c0_idx[TRAIN_SIZE_PER_CLASS:], c1_idx[TRAIN_SIZE_PER_CLASS:]])
np.random.shuffle(train_idx); np.random.shuffle(val_idx)

X_train, y_train = X[train_idx].copy(), y[train_idx].copy()
X_val,   y_val   = X[val_idx].copy(),   y[val_idx].copy()


def flip_labels(y_np, flip_ratio):
    y_noisy = y_np.copy()
    n_flip = int(len(y_noisy) * flip_ratio)
    if n_flip > 0:
        idx = np.random.choice(len(y_noisy), size=n_flip, replace=False)
        y_noisy[idx] = 1 - y_noisy[idx]  # 0<->1 뒤집기
    return y_noisy

y_train_noisy = flip_labels(y_train, TRAIN_LABEL_NOISE)

#======================================================
train_dataset = TensorDataset(torch.from_numpy(X_train),
                              torch.from_numpy(y_train_noisy))

valid_dataset = TensorDataset(torch.from_numpy(X_val),
                              torch.from_numpy(y_val))

train_loader = DataLoader(train_dataset, batch_size=BATCH_TRAIN, shuffle=True)
valid_loader = DataLoader(valid_dataset, batch_size=1024, shuffle=True)

#모델만드는 다른 방법
class MLP(nn.Module):
    def __init__(self, use_dropout=False, p=0.5):
        super().__init__()
        linears = []
        linears += [nn.Linear(2, 256), nn.ReLU()]
        if use_dropout: linears += [nn.Dropout(p)]
        linears += [nn.Linear(256, 256), nn.ReLU()]
        if use_dropout: linears += [nn.Dropout(p)]
        linears += [nn.Linear(256, 256), nn.ReLU()]
        if use_dropout: linears += [nn.Dropout(p)]
        linears += [nn.Linear(256, 2)]
        self.net = nn.Sequential(*linears)      # * 언패킹

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

def train_model(use_dropout=False, epoch=EPOCHS, lr=LR):
    model = MLP(use_dropout=use_dropout, p=P_DROPOUT)
    wd = WEIGHT_DECAY_DROPOUT if use_dropout else 0.0
    optimizer = torch.optim.Adam(model.parameters(), lr=lr, weight_decay=wd)
    # weight_decay = 드랍아웃을 사용한다 햇을때 웨이트 값이 커지지 않게 막아주는 역할
    loss_func = nn.CrossEntropyLoss()

    history = {'train_loss': [], 'val_loss':[], 'train_acc':[], 'val_acc':[]}
    for _ in range(epoch):
        model.train()
        #평가할때 이벨류에이션 모드를 사용할때
        tot = loss_sum = corr = 0
        for x_batch, y_batch in train_loader:
            optimizer.zero_grad()
            hypothesis = model(x_batch)
            loss = loss_func(hypothesis, y_batch)
            loss.backward()
            optimizer.step()
            loss_sum += loss.item() * x_batch.size(0)
            tot += x_batch.size(0)
            corr += (hypothesis.argmax(dim=1) == y_batch).sum().item()
        history['train_loss'].append(loss_sum / tot)
        history['train_acc'].append(corr / tot)

        model.eval()
        tot = loss_sum = corr = 0
        with torch.no_grad():
            for x_batch, y_batch in valid_loader:
                hypothesis = model(x_batch)
                loss = loss_func(hypothesis, y_batch)
                loss_sum += loss.item() * x_batch.size(0)
                tot += x_batch.size(0)
                corr += (hypothesis.argmax(dim=1) == y_batch).sum().item()
        history['val_loss'].append(loss_sum / tot)
        history['val_acc'].append(corr / tot)
    return model, history

model_over, h_over = train_model(use_dropout=False)
model_drop, h_drop = train_model(use_dropout=True)

plt.figure()
plt.plot(h_over['train_loss'], label='train loss')
plt.plot(h_over['val_loss'], label='val loss')
plt.title('No Dropout - loss')
plt.xlabel('epoch')
plt.ylabel('loss')
plt.legend(loc='best')
plt.show()


plt.figure()
plt.plot(h_over['train_acc'], label='train acc')
plt.plot(h_over['val_acc'], label='val acc')
plt.title('No Dropout - acc')
plt.xlabel('epoch')
plt.ylabel('acc')
plt.legend(loc='best')
plt.show()

No Dropout

Dropout

모든게 dropout으로 해결 되진 않음

 

---

 

earlyStppingEx

patient

best score 좋아지는 값 저장하는곳 

path

 

import torch
import torchvision.datasets as dset
import torchvision.transforms as transforms
from torch.utils.data import DataLoader
import numpy as np

torch.manual_seed(111)
from torch.utils.data.sampler import SubsetRandomSampler

def create_datasets(batch_size):
    train_data = dset.MNIST('MNIST_data/',
                            train=True,
                            download=True,
                            transform=transforms.ToTensor())    #하나만 있을때 직접연결가능
    test_data = dset.MNIST('MNIST_data/',
                            train=False,
                            download=True,
                            transform=transforms.ToTensor())

    num_train = len(train_data)
    indices = list(range(num_train))
    np.random.shuffle(indices)
    split = int(np.floor(0.2*num_train))
    train_idx, valid_idx = indices[split:],indices[:split]

    train_sampler = SubsetRandomSampler(train_idx)
    valid_sampler = SubsetRandomSampler(valid_idx)

    train_loader = DataLoader(train_data, sampler=train_sampler, batch_size=batch_size)
    valid_loader = DataLoader(train_data, sampler=valid_sampler, batch_size=batch_size)
    test_loader = DataLoader(test_data, batch_size=batch_size)

    return train_loader, test_loader, valid_loader

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

class ImageNN(nn.Module):
    def __init__(self, drop_p=0.5):
        super().__init__()
        self.fc1 = nn.Linear(784, 256)
        self.fc2 = nn.Linear(256, 128)
        self.fc3 = nn.Linear(128, 10)

        self.dropout_p = drop_p

    def forward(self, x):
        x = x.view(-1, 784)
        out = F.relu(self.fc1(x))
        out = F.dropout(out, p=self.dropout_p, training=self.training)
        out = F.relu(self.fc2(out))
        out = F.dropout(out, p=self.dropout_p, training=self.training)
        y = self.fc3(out)
        return y

model = ImageNN(drop_p=0.5)
loss_func = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters())

class EarlyStopping:
    def __init__(self, patience=7, verbose=False, delta=0, path='checkpoint.pt'):
        self.patience = patience
        self.verbose = verbose
        self.counter = 0
        self.best_score = None
        self.early_stop = False
        self.val_loss_min = np.inf      #inf 무한대
        self.delta = delta
        self.path = path
        
    def save_checkpoint(self, val_loss, model):
        if self.verbose:    #변화된 상태를 출력할껀지 말껀지
            print(f'validation loss:({self.val_loss_min:.6f}) -> ({val_loss:.6f} saving model!!!')
        torch.save(model.state_dict(), self.path)  #현재 모델안에 w,b 값을 딕셔너리로 가져와라
        self.val_loss_min = val_loss

    def __call__(self, val_loss, model):
        score = val_loss
        if self.best_score is None:
            self.best_score = score
            self.save_checkpoint(val_loss, model)
        elif score > self.best_score + self.delta:
            self.counter += 1

            if self.counter > self.patience:
                self.early_stop = True
            else:
                print(f'earlyStopping counter : {self.counter} / {self.patience}')

        else:
            self.best_score = score
            self.save_checkpoint(val_loss, model)
            self.counter= 0

def train_model(model, patience, n_epochs):
    train_losses = []
    valid_losses = []
    avg_train_losses = []
    avg_valid_losses = []
    early_stopping = EarlyStopping(patience=patience, verbose=True)

    for epoch in range(1, n_epochs+1):
        model.train()
        for x_train, y_train in train_loader:
            optimizer.zero_grad()
            hypothesis = model(x_train)
            loss = loss_func(hypothesis, y_train)
            loss.backward()
            optimizer.step()
            train_losses.append(loss.item())

        model.eval()
        for x_data, target in valid_loader:
            output = model(x_data)
            loss = loss_func(output, target)
            valid_losses.append(loss.item())
        #여기까지가 평가 한것

        train_loss = np.mean(train_losses)
        valid_loss = np.mean(valid_losses)
        avg_train_losses.append(train_loss)
        avg_valid_losses.append(valid_loss)
        epoch_len = len(str(n_epochs))

        print(f'[{epoch:<{epoch_len}} / {n_epochs:>{epoch_len}}]' +
              f'train_loss:{train_loss:.5f} valid_loss:{valid_loss:.5f}')

        train_losses = []
        valid_losses = []

        early_stopping(valid_loss, model)    #콜함수 호출한다.

        if early_stopping.early_stop:
            print('early stopping!!!!!')
            break

    model.load_state_dict(torch.load('checkpoint.pt'))

    return model, avg_train_losses, avg_valid_losses

batch_size = 256
n_epochs = 100

train_loader, test_loader, valid_loader = create_datasets(batch_size)
patience = 10
model, train_loss, valid_loss = train_model(model, patience, n_epochs)

---