[PyTorch] MLP(Multi-Layer Perceptron) Regression(다층 퍼셉트론을 이용한 회귀)

선형 회귀식은 nn.Layer()가 하나 있는 모델을 의미한다. 선형식은 모든 데이터를 직선으로 예측하기 때문에 학습이 매우 빠르다는 장점이 있지만, 데이터 내의 변수들을 일반적으로 비선형 관계를 갖기 때문에 선형 모델을 이용한 예측에는 한계가 있다. 

 

nn.Layer()을 여러개 연결해 다층 신경망을 구성하여 집값 예측 회귀 모델을 구성해본다. 

 

import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split  # 데이터 나누기
import torch
from torch import nn, optim  # nn: 신경망 레이어, optim: 최적화 알고리즘
from torch.utils.data import DataLoader, Dataset  # 데이터셋, 미니배치 로더
import torch.nn.functional as F  # ReLU 등 다양한 함수 제공
from sklearn.metrics import mean_squared_error  # RMSE 계산에 사용
import matplotlib.pyplot as plt  # 그래프 시각화

# 1. GPU 장치 설정 (cuda 사용 가능하면 GPU, 아니면 CPU)
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

pandas는 표 형태의 데이터(스프레드 시트, 엑셀 등)를 다루기 위한 라이브러리이다. csv 파일을 처리하기 위해 사용한다.

sklearn은 scikit-learn 라이브러리로, 머신러닝에 사용되는 다양한 기능을 담고 있다. 

nn.functional은 ReLu, Sigmoid등의 함수들을 지원한다. 

 

# 2. 사용자 정의 Dataset 클래스
class TensorData(Dataset):
    def __init__(self, x_data, y_data):
        """
        x_data, y_data: numpy 배열 형태
        -> FloatTensor로 변환 후, device(GPU/CPU)에 올림
        """
        self.x_data = torch.FloatTensor(x_data).to(device)
        self.y_data = torch.FloatTensor(y_data).to(device)
        self.len = self.y_data.shape[0]  # 데이터 개수 (행 개수)

    def __getitem__(self, index):
        """
        인덱스(index)에 해당하는 (입력, 출력) 쌍을 반환
        """
        return self.x_data[index], self.y_data[index]

    def __len__(self):
        """
        전체 데이터셋의 샘플 개수 반환
        """
        return self.len

TensorData 클래스는 학습/평가 데이터를 다루기 위한 사용자 정의 클래스이다. (x_data, y_data)를 Tensor로 생성하여 GPU/CPU에 올린다. Dataset 클래스를 상속하였으므로 __getitem__ 함수는 DataLoader가 미니 배치를 생성할 때 자동으로 호출되어 데이터를 반환한다. 

 

# 3. 모델 정의: 회귀 모델 (nn.Module 상속)
class Regressor(nn.Module):
    def __init__(self):
        super().__init__()
        # fully-connected 레이어 정의
        # 입력 특징 차원: 13, 은닉층1: 50개, 은닉층2: 30개, 출력층: 1개
        self.fc1 = nn.Linear(13, 50, bias=True)
        self.fc2 = nn.Linear(50, 30, bias=True)
        self.fc3 = nn.Linear(30, 1, bias=True)
        self.dropout = nn.Dropout(0.5)  # 드롭아웃(50%)

    def forward(self, x):
        """
        순전파(Forward):
        입력 x를 차례대로 레이어에 통과시키며, 활성화 함수 ReLU 적용
        2번째 레이어 후에는 dropout으로 일부 노드를 무작위 삭제 (과적합 방지)
        """
        x = F.relu(self.fc1(x))
        x = self.dropout(F.relu(self.fc2(x)))
        x = F.relu(self.fc3(x))
        return x

nn.Module을 상속한 회귀 모델을 정의한다. 3개의 Linear 레이어로 구성되며, 입력층의 뉴런은 13 -> 50, 50 -> 30, 30 -> 1로 구성된다. drop out 수치는 50%로 설정한다. 

 

forward 메서드는 입력을 각 레이어에 순차적으로 통과시키며,두 번째 레이어에서 drop out을 적용한다. 활성화 함수로는 ReLu를 사용한다.

 

# 4. 평가 함수: RMSE 계산
def evaluation(dataloader):
    """
    dataloader를 사용해 미니배치로 데이터를 불러온 뒤,
    모델의 예측값(outputs)과 실제값(values)을 모두 모아 RMSE 계산.
    """
    predictions = torch.tensor([], dtype=torch.float).to(device)
    actual = torch.tensor([], dtype=torch.float).to(device)

    with torch.no_grad():  # 평가 시에는 역전파 비활성화
        model.eval()  # 드롭아웃 등 학습 전용 기능을 비활성화 (evaluation 모드)
        for data in dataloader:
            inputs, values = data
            outputs = model(inputs)  # 모델 예측
            # 예측값과 실제값을 축적
            predictions = torch.cat((predictions, outputs), 0)
            actual = torch.cat((actual, values), 0)

    # GPU 텐서를 CPU로 옮긴 뒤, numpy 배열로 변환
    predictions = predictions.cpu().numpy()
    actual = actual.cpu().numpy()

    # 평균 제곱 오차(MSE) -> RMSE
    rmse = np.sqrt(mean_squared_error(predictions, actual))
    return rmse

손실함수로 RMSE(Root Mean Square Error)를 사용한다. RMSE는 MSE의 제곱근이다. 

$$\text{RMSE} = \sqrt{\frac{1}{n} \sum_{i=1}^n (y_i - \hat{y}_i)^2}$$

예측값과 그에 대응하는 실제값을 저장하기 위한 빈 tensor predictions와 actual을 만들고 GPU에 올린다. torch.no_grad()와 model.eval()로 평가를 위한 설정을 한 후, dataloader로부터 배치 단위로 데이터를 받아 예측값과 실제값을 누적한다. 이때 torch.cat에서 0은 0번 째 차원(배치 차원)을 기준으로 누적한다는 의미이다. 

 

평가가 끝나면 GPU의 tensor를 CPU로 옮기고 numpy 배열로 변환한다. 그리고 rmse를 계산해 반환한다. 

 

# 5. 데이터 준비: CSV 파일 로드
df = pd.read_csv("./pytorch/data/reg.csv", index_col=[0])
# 예시로 'Price'라는 열이 종속변수(타겟), 나머지는 특징(입력)

# x: 입력 특징들, y: 타겟(Price)
x = df.drop("Price", axis=1).to_numpy()         # DataFrame -> numpy 배열
y = df["Price"].to_numpy().reshape(-1, 1)       # (N,) -> (N,1) 형태로 변환

학습을 위한 데이터를 준비한다. pands의 read_csv는 csv 파일을 읽는다. index_col[0]은 csv 파일의 0번째 열의 데이터를 인덱스로 하는 데이터 프레임을 생성한다. 이 데이터의 변수 개수는 13개로, 이에 맞춰 Regressor 클래스가 구축되었다. 

 

drop의 axis=1은 열을 기준으로 Price를 배제한다.

 

# 6. 학습/테스트 데이터 분할 (5:5 비율로)
x_train, x_test, y_train, y_test = train_test_split(x, y, test_size=0.5)

train_test_split은 데이터를 test_size의 비율로 학습 데이터와 테스트 데이터로 분할한다. 

 

# 7. Dataset, DataLoader 생성
trainsets = TensorData(x_train, y_train)
trainloader = torch.utils.data.DataLoader(trainsets, batch_size=32, shuffle=True)
# batch_size=32: 미니배치 크기 32, shuffle=True: 데이터를 섞어서 과적합 방지

testsets = TensorData(x_test, y_test)
testloader = torch.utils.data.DataLoader(testsets, batch_size=32, shuffle=False)
# 테스트셋은 shuffle=False (평가 시 데이터 순서 유지)

trainsets과 testsets를 각각 TensorData 객체로 생성한다. 그리고 DataLoader로 크기가 32인 배치로 묶는다. 과적합이란 모델이 학습 데이터에 대해서는 높은 정확도를 보이지만, 새로운 데이터에 대해서는 성능이 크게 떨어지는 현상을 말한다. 

 

# 8. 모델 객체 생성 & 손실 함수, 옵티마이저 설정
model = Regressor().to(device)     # Regressor 모델 객체를 GPU/CPU로 옮김
criterion = nn.MSELoss()          # 손실 함수로 MSE(평균 제곱 오차)
optimizer = optim.Adam(model.parameters(), lr=1e-3, weight_decay=1e-7)
# Adam 옵티마이저, weight_decay=1e-7: 가중치 감쇠(정규화)로 과적합 방지

모델 객체를 생성하고, 모델 내부에서 tensor를 device로 관리하므로 model도 동일한 device에 올려준다. 손실함수와 최적화 함수를 지정한다. 

# 9. 모델 학습
loss_ = []  # 각 epoch마다의 손실값을 저장할 리스트
n = len(trainsets)  # 학습 데이터셋의 전체 샘플 수

for epoch in range(400):  # 400번 반복 (Epoch)
    running_loss = 0.0
    # 미니배치 반복
    for data in trainloader:
        inputs, values = data       # (x_batch, y_batch)
        optimizer.zero_grad()       # 이전 배치에서 계산된 기울기 초기화
        outputs = model(inputs)     # 순전파로 예측값 계산
        loss = criterion(outputs, values)  # 예측값과 실제값으로 MSE 계산
        loss.backward()             # 역전파 -> 파라미터별 기울기 계산
        optimizer.step()            # 옵티마이저가 파라미터 업데이트 (가중치 최적화)
        running_loss += loss.item() # 손실값 누적

    # 한 epoch 끝날 때 평균 손실(= 누적 손실 / 전체 샘플 수)을 기록
    loss_.append(running_loss / n)
    print(f"Epoch {epoch}: {loss.item()}")

모델을 학습시킨다. 매 epoch마다 trainloader의 미니 배치를 하나씩 꺼내어 연산을 수행한다. 

 

# 10. 모델 평가: 학습 데이터 RMSE와 테스트 데이터 RMSE
train_rmse = evaluation(trainloader)
test_rmse = evaluation(testloader)
print("Train RMSE: ", train_rmse)
print("Test RMSE: ", test_rmse)

# 11. 학습 과정 시각화 (에폭별 훈련 손실)
plt.plot(loss_)
plt.title("Training Loss")
plt.xlabel("epoch")
plt.show()

모델을 평가하고 학습 데이터와 테스트 데이터의 손실을 비교한다. 

 

Epoch 392: 0.006565847434103489
Epoch 393: 0.005968224257230759
Epoch 394: 0.004061189014464617
Epoch 395: 0.008947031572461128
Epoch 396: 0.004307636991143227
Epoch 397: 0.005249027628451586
Epoch 398: 0.007499590050429106
Epoch 399: 0.006710496731102467
Train RMSE:  0.055454724
Test RMSE:  0.12850489

Training Loss는 0~50 epoch 사이에서 급격히 감소하고, 약 100 epoch이후에는 큰 변화 없이 미세하게 변동한다. 학습 데이터의 경우 rmse가 낮지만, 테스트 데이터의 rmse는 학습데이터에 비해 2배 이상 크다. 이는 학습 데이터에 대해서는 상당히 정확한 예측을 하고 있지만, 일반화 성능이 떨어짐(과적합)을 의미한다.