9. 파이토치(PyTorch) 튜토리얼 - Torch.NN

본 포스팅은 파이토치(PYTORCH) 한국어 튜토리얼을 참고하여 공부하고 정리한 글임을 밝힙니다.

---

PyTorch에서 신경망을 생성하고 학습시키는 것을 도와주기 위해 torch.nn, torch.optim, Dataset, DataLoader와 같은 모듈과 클래스들 제공

Torch.nn

MNIST 데이터 준비

• 경로 설정 담당하는 pathlib
• requests를 이용한 데이터셋 다운로드

from pathlib import Path
import requests

DATA_PATH = Path("data")
PATH = DATA_PATH / "mnist"

PATH.mkdir(parents=True, exist_ok=True)

URL = "https://github.com/pytorch/tutorials/raw/master/_static/"
FILENAME = "mnist.pkl.gz"

if not (PATH / FILENAME).exists():
    content = requests.get(URL + FILENAME).content
    (PATH / FILENAME).open("wb").write(content)

• numpy 배열 포맷인 데이터셋을 직렬화하기 위한 pickle

import pickle
import gzip

with gzip.open((PATH / FILENAME).as_posix(), "rb") as f:
		((x_train, y_train), (x_valid, y_valid), _) = pickle.load(f, encoding="latin-1")

from matplotlib import pyplot
import numpy as np

pyplot.imshow(x_train[0].reshape((28,28)), cmap="gray")
print(x_train.shape)

Out:

(50000, 784)

• PyTorch에서 torch.tensor를 사용하므로 데이터를 변환

import torch

x_train, y_train, x_valid, y_valid = map(torch.tensor, (x_train, y_train, x_valid, y_valid))

n,c = x_train.shape
print(x_train, y_train)
print(x_train.shape)
print(y_train.min(), y_train.max())

Out:

tensor([[0., 0., 0.,  ..., 0., 0., 0.],
        [0., 0., 0.,  ..., 0., 0., 0.],
        [0., 0., 0.,  ..., 0., 0., 0.],
        ...,
        [0., 0., 0.,  ..., 0., 0., 0.],
        [0., 0., 0.,  ..., 0., 0., 0.],
        [0., 0., 0.,  ..., 0., 0., 0.]]) tensor([5, 0, 4,  ..., 8, 4, 8])
torch.Size([50000, 784])
tensor(0) tensor(9)

torch.nn.functional 사용하기

• PyTorch의 nn 클래스의 장점을 활용하여 코드를 더 간결하고 유연하게 만들 수 있음

• 활성화, 손실 함수를 torch.nn.functional의 함수로 대체

• torch.nn 라이브러리 (다른 부분에는 클래스가 포함되어 있음)에는 다양한 손실 및 활성화 함수뿐만 아니라, 풀링(pooling) 함수와 같이 신경망을 만드는데 편리한 함수들도 존재함
  (컨볼루션 연산, 선형 레이어 등을 수행하는 함수도 있지만 일반적으로 라이브러리의 다른 부분을 사용하여 더 잘 처리 가능)

• 음의 로그 우도 손실(Negative Log Likelihood)와 로그 소프트맥스 (log softamx) 활성화 함수를 사용하는 경우 ➡️ Pytorch는 이 둘을 결합하는 단일 함수인 F.cross_entropy 제공

import math

weights = torch.randn(784, 10) / math.sqrt(784)
weights.requires_grad_()
bias = torch.zeros(10, requires_grad=True)

bs = 64  # 배치 크기

xb = x_train[0:bs]  # x로부터 미니배치(mini-batch) 추출
yb = y_train[0:bs]

def accuracy(out, yb):
    preds = torch.argmax(out, dim=1)
    return (preds == yb).float().mean()
    

import torch.nn.functional as F

loss_func = F.cross_entropy

def model(xb):
    return xb @ weights + bias

print(loss_func(model(xb), yb), accuracy(model(xb), yb))

Out:

tensor(0.0796, grad_fn=<NllLossBackward0>) tensor(1.)

nn.Module 을 이용하여 리팩토링 하기

• 더 명확하고 간결한 훈련 루프를 위해 nn.Module 및 nn.Parameter 사용
• nn.Module 하위 클래스(subclass)를 만들어 forward 단계에 대한 가중치, 절편, 메소드 등을 유지하는 클래스 만들어보자
• nn.Module은 속성(attribute)과 메소드 (.parameters(), .zero_grad()와 같은)를 가지고 있음

from torch import nn

class Mnist_Logistic(nn.Module):
    def __init__(self):
        super().__init__()
        self.weights = nn.Parameter(torch.randn(784, 10) / math.sqrt(784))
        self.bias = nn.Parameter(torch.zeros(10))

    def forward(self, xb):
        return xb @ self.weights + self.bias

• 함수 사용하는 대신 오브젝트(Object)를 사용하기 때문에 먼저 모델을 인스턴스화 한다

model = Mnist_Logistic()

• nn.Module 오브젝트들은 마치 함수처럼 사용됨(즉, 호출가능)

print(loss_func(model(xb), yb))

Out:

tensor(2.3468, grad_fn=<NllLossBackward0>)

• nn.Module에 의해 model.parameters() 및 model.zero_grad()를 활용하여 각 매개변수의 값을 업데이터하고, 각 매개 변수에 대한 기울기들을 자동으로 0으로 정의해줌

# 예시 코드
with torch.no_grad():
	for p in model.parameters(): p -= p.grad * lr
    model.zero_grad()

lr = 0.5  # 학습률(learning rate)
epochs = 2  # 훈련에 사용할 에폭(epoch) 수

def fit():
    for epoch in range(epochs):
        for i in range((n - 1) // bs + 1):
            start_i = i * bs
            end_i = start_i + bs
            xb = x_train[start_i:end_i]
            yb = y_train[start_i:end_i]
            pred = model(xb)
            loss = loss_func(pred, yb)

            loss.backward()
            with torch.no_grad():
                for p in model.parameters():
                    p -= p.grad * lr
                model.zero_grad()

fit()

print(loss_func(model(xb), yb))

Out:

tensor(0.0827, grad_fn=<NllLossBackward0>)

nn.Linear 이용하여 리팩토링 하기

• nn.Linear를 선형 레이어로 사용하여 self.weights 및 self.bias를 수동으로 정의 및 초기화하고 xb @ self.weights + welf.bias를 계산하는 것을 대신해서 모두 해줌

class Mnist_Logistic(nn.Module):
    def __init__(self):
        super().__init__()
        self.lin = nn.Linear(784, 10)

    def forward(self, xb):
        return self.lin(xb)

model = Mnist_Logistic()
print(loss_func(model(xb), yb))

Out:

tensor(2.2836, grad_fn=<NllLossBackward0>)

fit()

print(loss_func(model(xb), yb))

Out:

tensor(0.1940, grad_fn=<NllLossBackward0>)

optim 이용하여 리팩토링 하기

• torch.optim: 다양한 최적화 알고리즘을 가진 패키지

• 각 매개변수를 수동으로 업데이트 하는 대신, 옵티마이저의 step 메소드 사용하여 업데이트 진행

• 이렇게 코드를 작성하면 수동으로 코딩한 최적화 단계를 손쉽게 대체할 수 있음

• optim.zero_grad(): 기울기를 0으로 재설정 해줌, 다음 미니 배치에 대한 기울기를 계산하기 전에 호출해야 함

opt.step()
opt.zero_grad()

• 나중에 다시 사용할 수 있도록 모델과 옵티마이저를 만드는 함수 정의

from torch import optim

def get_model():
    model = Mnist_Logistic()
    return model, optim.SGD(model.parameters(), lr=lr)

model, opt = get_model()
print(loss_func(model(xb), yb))

for epoch in range(epochs):
    for i in range((n - 1) // bs + 1):
        start_i = i * bs
        end_i = start_i + bs
        xb = x_train[start_i:end_i]
        yb = y_train[start_i:end_i]
        pred = model(xb)
        loss = loss_func(pred, yb)

        loss.backward()
        opt.step()
        opt.zero_grad()

print(loss_func(model(xb), yb))

Out:

tensor(2.3160, grad_fn=<NllLossBackward0>)
tensor(0.0822, grad_fn=<NllLossBackward0>)

Dataset 이용하여 리팩토링하기

• Dataset은 __len__ 함수 및 __getitem__ 함수를 가진 어떤 것이라도 될 수 있음

• 이 예제에선 Dataset의 하위 클래스로써 사용자 지정 FacialLandmarkDataset 클래스를 만드는 예시 제시

• PyTorch의 TensorDataset은 텐서를 감싸는 Dataset ➡️ 길이와 인덱식 방식 정의함으로써 텐서의 첫 번째 차원을 따라 반복, 인덱싱 및 슬라이스하는 방법도 제공

from torch.utils.data import TensorDataset

• x_train 및 y_train 모두 하나의 TensorDataset에 합쳐질 수 있고, 따라서 반복시키고 슬라이스 하기 편리함

train_ds = TensorDataset(x_train, y_train)

• 이전에는 아래 첫 번째 코드와 같이 x 및 y 값의 미니 배치를 별도로 반복해야 했다면, 이제는 이 두 단계를 두 번째 코드와 같이 함께 수행 가능함

# 예전 방식
xb = x_train[start_i:end_i]
yb = y_train[start_i:end_i]

# 현재 방식
xb, yb = train_ds[i*bs : i*bs+bs]

model, opt = get_model()

for epoch in range(epochs):
    for i in range((n - 1) // bs + 1):
        xb, yb = train_ds[i * bs: i * bs + bs]
        pred = model(xb)
        loss = loss_func(pred, yb)

        loss.backward()
        opt.step()
        opt.zero_grad()

print(loss_func(model(xb), yb))

Out:

tensor(0.0837, grad_fn=<NllLossBackward0>)

DataLoader 이용하여 리팩토링하기

• DataLoader: 배치 관리 ➡️ 매 미니배치를 자동적으로 제공

from torch.utils.data import DataLoader

train_ds = TensorDataset(x_train, y_train)
train_dl = DataLoader(train_ds, batch_size=bs)

• 이제 (xb, yb)가 DataLoader에서 자동으로 로드되므로 루프가 훨씬 깨끗해짐

for xb,yb in train_dl:
    pred = model(xb)

model, opt = get_model()

for epoch in range(epochs):
    for xb, yb in train_dl:
        pred = model(xb)
        loss = loss_func(pred, yb)

        loss.backward()
        opt.step()
        opt.zero_grad()

print(loss_func(model(xb), yb))

Out:

tensor(0.0833, grad_fn=<NllLossBackward0>)

Pytorch의 nn.Module, nn.Parameter, Dataset 및 DataLoader 덕분에 이제 훈련 루프가 훨씬 더 작아지고 이해하기 쉬워졌음
➡️ 이제 실제로 효과적인 모델을 만드는 데 필요한 기본 기능을 추가해보자

검증(validation) 추가하기

• 과적합 확인하기 위해 항상 검증 데이터셋이 있어야 함
• 훈련 데이터를 섞는 것은 배치와 과적합 사이의 상관관계를 방지하기 위해 매우 중요하지만, 반면 검증 손실은 검증 데이터셋을 섞든 안섞든 동일함
• 검증 데이터셋에 대한 배치 크기는 학습 데이터셋 배치 크기의 2배를 사용 ➡️ 역전파가 필요하지 않으므로 메모리를 덜 사용하기 때문(기울기 저장할 필요 X)
• 더 큰 배치 크기를 사용하여 손실을 더 빨리 계산하기 위해 이렇게 함

train_ds = TensorDataset(x_train, y_train)
train_dl = DataLoader(train_ds, batch_size=bs, shuffle=True)

valid_ds = TensorDataset(x_valid, y_valid)
valid_dl = DataLoader(valid_ds, batch_size=bs * 2)

• 각 epoch이 끝날 때 검증 손실을 게산하고 출력함
• 훈련 전에 항상 model.train() 호출하고, 추론 전에 model.eval()을 호출함
  • 이는 nn.BatchNorm2d 및 nn.Dropout과 같은 레이어에서 이러한 다른 단계(훈련, 추론)에 대한 적절한 동작이 일어나게 하기 위함

model, opt = get_model()

for epoch in range(epochs):
    model.train()
    for xb, yb in train_dl:
        pred = model(xb)
        loss = loss_func(pred, yb)

        loss.backward()
        opt.step()
        opt.zero_grad()

    model.eval()
    with torch.no_grad():
        valid_loss = sum(loss_func(model(xb), yb) for xb, yb in valid_dl)

    print(epoch, valid_loss / len(valid_dl))

Out:

0 tensor(0.3308)
1 tensor(0.3219)

fit()과 get_data() 생성하기

• 훈련 데이터셋과 검증 데이터셋 모두에 대한 손실을 계산하는 유사한 프로세스를 두 번 거침 ➡️ 이를 하나의 배치에 대한 손실을 계산하는 자체 함수 loss_batch 로 만들어보자

• 훈련 데이터셋에 대한 옵티마이저를 전달하고 이를 사용하여 역전파를 수행 (검증 데이터셋의 경우 옵티마이저를 전달하지 않으므로 메소드가 역전파를 수행 X)

def loss_batch(model, loss_func, xb, yb, opt=None):
    loss = loss_func(model(xb), yb)

    # train
    if opt is not None:
        loss.backward()
        opt.step()
        opt.zero_grad()

    return loss.item(), len(xb)
	

• fit: 모델 훈련하고 각 epoch에 대한 훈련 및 검증 손실 계산하는 작업```python

import numpy as np

def fit(epochs, model, loss_func, opt, train_dl, valid_dl):
    for epoch in range(epochs):
        model.train()
        for xb, yb in train_dl:
            loss_batch(model, loss_func, xb, yb, opt)

        model.eval()
        with torch.no_grad():
            losses, nums = zip(
                *[loss_batch(model, loss_func, xb, yb) for xb, yb in valid_dl]
            )
        val_loss = np.sum(np.multiply(losses, nums)) / np.sum(nums)

        print(epoch, val_loss)

• get_data: 학습 및 검증 데이터셋에 대한 dataloader를 출력

def get_data(train_ds, valid_ds, bs):
    return(
        DataLoader(train_ds, batch_size=bs, shuffle=True),
        DataLoader(valid_ds, batch_size=bs*2)
    )

• 이제 dataloader를 가져오고 모델을 훈련하는 전체 프로세스를 3 줄의 코드로 실행 가능

trian_dl, valid_dl = get_data(train_ds, valid_ds, bs)
model, opt = get_model()
fit(epochs, model, loss_func, opt, train_dl, valid_dl)

Out:

0 0.3312061833024025
1 0.3091698085784912

CNN 으로 넘어가기

• 3개의 컨볼루션 레이어로 신경망을 구축 ➡️ 별도의 수정없이 이전 섹션의 함수를 CNN 학습하는 데 사용할 수 있음
• PyTorch에 사전 정의된 Conv2d 클래스를 컨볼루션 레이어로 사용
• avg_pool2d을 이용하여 평균 풀링(average pooling) 수행
• view: PyTorch의 numpy reshape 버전

class Mnist_CNN(nn.Module):
    def __init__(self):
        super().__init__()
        self.conv1 = nn.Conv2d(1, 16, kernel_size=3, stride=2, padding=1)
        self.conv2 = nn.Conv2d(16, 16, kernel_size=3, stride=2, padding=1)
        self.conv3 = nn.Conv2d(16, 10, kernel_size=3, stride=2, padding=1)

    def forward(self, xb):
        xb = xb.view(-1, 1, 28, 28)
        xb = F.relu(self.conv1(xb))
        xb = F.relu(self.conv2(xb))
        xb = F.relu(self.conv3(xb))
        xb = F.avg_pool2d(xb, 4)
        return xb.view(-1, xb.size(1))

lr = 0.1

• 모멘텀(Momentum): 이전 업데이트도 고려하고 일반적으로 더 빠른 훈련으로 이어지는 확률적 경사하강법(SGD)의 변형

model = Mnist_CNN()
opt = optim.SGD(model.parameters(), lr=lr, momentum=0.9)

fit(epochs, model, loss_func, opt, train_dl, valid_dl)

Out:

0 0.3451383782863617
1 0.24721240463256836

nn.Sequential

• torch.nn의 Sequential 클래스: 안에 포함된 각 모듈을 순차적으로 실행 ➡️ 신경망을 더 간단하게 작성하는 방법
• 이를 활용하려면 주어진 함수에서 사용자정의 레이어(custom layer)를 쉽게 정의할 수 있어야 함
• ex) PyTorch에는 view 레이어가 없으므로 신경망 용으로 직접 만들어야 함
• Lambda는 Sequential로 신경망 정의할 때 사용할 수 있는 레이어 생성함

class Lambda(nn.Module):
    def __init__(self, func):
        super().__init__()
        self.func = func

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


def preprocess(x):
    return x.view(-1, 1, 28, 28)

model = nn.Sequential(
    Lambda(preprocess),
    nn.Conv2d(1, 16, kernel_size=3, stride=2, padding=1),
    nn.ReLU(),
    nn.Conv2d(16, 16, kernel_size=3, stride=2, padding=1),
    nn.ReLU(),
    nn.Conv2d(16, 10, kernel_size=3, stride=2, padding=1),
    nn.ReLU(),
    nn.AvgPool2d(4),
    Lambda(lambda x: x.view(x.size(0), -1)),
)

opt = optim.SGD(model.parameters(), lr=lr, momentum=0.9)

fit(epochs, model, loss_func, opt, train_dl, valid_dl)

Out:

0 0.3579630592823029
1 0.22187287817001342

DataLoader 감싸기

• 우리의 CNN은 상당히 간결하지만, MNIST에서만 작동함
  • 입력이 28x28의 긴 벡터라고 가정
  • 최종적으로 CNN 그리드 크기는 4x4라고 가정 (평균 풀링 커널 크기 때문)
• 이러한 두 가지 가정을 제거하여 모델이 만든 2d 단일 채널 이미지에서 작동하도록 해보자
• 초기 Lambda 레이어를 제거하고 데이터 전처리를 제네레이터(generator)로 이동시킬 수 있음

def preprocess(x, y):
    return x.view(-1, 1, 28, 28), y

class WrappedDataLoader:
    def __init__(self, dl, func):
        self.dl = dl
        self.func = func

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

    def __iter__(self):
        batches = iter(self.dl)
        for b in batches:
            yield (self.func(*b))

train_dl, valid_dl = get_data(train_ds, valid_ds, bs)
train_dl = WrappedDataLoader(train_dl, preprocess)
valid_dl = WrappedDataLoader(valid_dl, preprocess)

• nn.AvgPool2d를 nn.AdaptiveAvgPool2d로 대체하여 우리가 가진 입력 텐서가 아니라 원하는 출력 텐서의 크기를 정의 가능
• 결과적으로 모든 크기의 입력과 함께 작동 가능

model = nn.Sequential(
    nn.Conv2d(1, 16, kernel_size=3, stride=2, padding=1),
    nn.ReLU(),
    nn.Conv2d(16, 16, kernel_size=3, stride=2, padding=1),
    nn.ReLU(),
    nn.Conv2d(16, 10, kernel_size=3, stride=2, padding=1),
    nn.ReLU(),
    nn.AdaptiveAvgPool2d(1),
    Lambda(lambda x: x.view(x.size(0), -1)),
)

opt = optim.SGD(model.parameters(), lr=lr, momentum=0.9)

fit(epochs, model, loss_func, opt, train_dl, valid_dl)

Out:

0 0.3100432321071625
1 0.2719866509437561

GPU 사용하기

print(torch.cuda.is_available())

Out:

True

• 디바이스 오브젝트 생성

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

• GPU로 배치 옮기도록 preprocess 업데이트

def preprocess(x, y):
    return x.view(-1, 1, 28, 28).to(dev), y.to(dev)


train_dl, valid_dl = get_data(train_ds, valid_ds, bs)
train_dl = WrappedDataLoader(train_dl, preprocess)
valid_dl = WrappedDataLoader(valid_dl, preprocess)

• 모델도 GPU로 이동시킴

model.to(dev)
opt = optim.SGD(model.parameters(), lr=lr, momentum=0.9)

• CPU보다 더 빨리 실행됨

fit(epochs, model, loss_func, opt, train_dl, valid_dl)

---

Recap

• torch.nn

  • Module: 함수처럼 동작하지만, 또한 상태(state) (예를 들어, 신경망의 레이어 가중치)를 포함할 수 있는 호출 가능한 오브젝트를 생성합니다. 이는 포함된 Parameter가 어떤 것인지 알고, 모든 기울기를 0으로 설정하고 가중치 업데이트 등을 위해 반복할 수 있습니다.
  • Parameter: Module 에 역전파 동안 업데이트가 필요한 가중치가 있음을 알려주는 텐서용 래퍼입니다. requires_grad 속성이 설정된 텐서만 업데이트 됩니다.
  • functional: 활성화 함수, 손실 함수 등을 포함하는 모듈 (관례에 따라 일반적으로 F 네임스페이스로 임포트 됩니다) 이고, 물론 컨볼루션 및 선형 레이어 등에 대해서 상태를 저장하지않는(non-stateful) 버전의 레이어를 포함합니다.

• torch.optim: 역전파 단계에서 Parameter 의 가중치를 업데이트하는, SGD 와 같은 옵티마이저를 포함합니다.

• Dataset: TensorDataset 과 같이 Pytorch와 함께 제공되는 클래스를 포함하여 __len__ 및 __getitem__ 이 있는 객체의 추상 인터페이스

• DataLoader: 모든 종류의 Dataset 을 기반으로 데이터의 배치들을 출력하는 반복자(iterator)를 생성합니다