파이토치 큰그림 정리

회사에서 해커톤 대회에 출전할 때 텐서플로우로 이미지 분류를 써본적이 있다.

 

 

이번에도 다시 ai를 쓸 일이 있어서 이번에는 파이토치로 도전해보게 되었다.

 

매번 까먹고 다시공부하고 해서 이번에는 간단히 정리하려 한다

 

 

 

 

1. 큰그림에서 돌아가는 구조

 

우리가 일반적으로 아는 인공지능 트레이닝은 흔히 생각하는 신경망 구조를 만든뒤 데이터 넣어주는 작업이다.

여기서 막막한 부분은 신경망을 어떻게 구성하고 짜야하는가 인데, 결국 trial and error다. 아무도 모르다는거임

 

이것만 기억하고 가면 된다.

 

 

 

2. 키워드

 

한줄요약으로 외우자

 

텐서 : 인공지능의 넘파이배열 같은거임. 데이터 단위

CNN : 이미지 분류같은 2차원에 사용되는 신경망

RNN : 1차원에 사용되는 신경망

히든레이어 : 신경망 구성할때 세로로 한줄이라 생각하면 됨

Weight와 Bias : 신경망 하나의 동그라미 노드에 들어가는 계수와 가중치. 예를들어 ax+b라 하면 a는 weight이고 b는 bias임

경사하강법 : weight과 bias를 계산하는 방법중 하나임.

옵티마이저 : 경사하강법에 쓰이는 함수

역전파 : 대충 신경망 한번 돌리고, 정답과 비교해서 오차를 계산하고 그걸 신경망 뒤쪽방향으로 데이터를 날려서 weight와 bias 계산하도록 함

 

 

 

활성함수 : weight와 bias가 형태가 ax+b 즉 선형적이라고 하자. 대충 신경망 다 통과하면 a1x+b1 + a2x+b2 ... 뭐 이런식으로 정리가 될텐데 결국 이것도 정리해보면 ax+b형태로 귀결된다. weight과 bias여러개 계산한게 쓸모없다는 얘기. 이를 위해 선형적인걸 한번 꼬아주는 함수라고 하면 된다.

 

 

 

 

3. 신경망은 어떻게 짜야함?

class SimpleNN(nn.Module):
    def __init__(self):
        super(SimpleNN, self).__init__()
        # Sequential을 사용하여 네트워크 정의
        self.network = nn.Sequential(
            nn.Linear(5, 10),  # 입력 5 -> 은닉층 10
            nn.ReLU(),         # ReLU 활성화 함수
            nn.Linear(10, 10), # 은닉층 10 -> 은닉층 10
            nn.ReLU(),         # ReLU 활성화 함수
            nn.Linear(10, 1)   # 은닉층 10 -> 출력 1
        )
        
    def forward(self, x):
        return self.network(x)

위에 대충 예제임 여기서 nn.Sequential이 뉴럴네트워크 만들어주는 과정이다.

대충 보면 히든레이어 사이사이에 ReLu라는 활성함수를 꼽사리 껴주면 된다.

 

 

 

4. 파이토치 돌아가는 큰그림

 

내가짜려는건 hrv데이터를 읽어와서 이사람이 스트레스 받나 안받나를 학습시키는 과정임.

즉 인풋7개 가지고 정답 1개를 도출하는 과정

 

사전준비

 

1. 먼저 csv같은 데이터 파일을 pandas로 로드한다

2. 데이터를 로드할때 인풋과 정답을 구분해서 로드해야함. 보통 정답을 label이라고 표현하는거 같음

3. 로드한 pandas 데이를 .value를 통해서 넘파이 배열로 바꿔줌

4. 이를 torch.tensor로 텐서라는 자료형으로 바꿔준다.

 

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

train = pd.read_csv('data/final/train.csv')
test = pd.read_csv('data/final/train.csv') 


train_inputs = train[['MEAN_RR','RMSSD', 'pNN50','VLF','LF','HF','LF_HF']].values
train_labels = train['condition'].values

test_inputs = train[['MEAN_RR','RMSSD', 'pNN50','VLF','LF','HF','LF_HF']].values
test_labels = train['condition'].values

train_input_tensor = torch.tensor(train_inputs,dtype=torch.float32)
train_labels_tensor = torch.tensor(train_labels,dtype=torch.long)

test_input_tensor = torch.tensor(test_inputs,dtype=torch.float32)
test_labels_tensor = torch.tensor(test_labels,dtype=torch.long)

 

 

 

다음으로 Dataset과 DataLoader라는 클래스를 알아야함

 

Dataset: 데이터를 로드하는 클래스. 사실 위 코드 작업을 Dataset의 생성자에서 처리해도 됨.

DataLoader : Dataset을 이용해 배치사이즈, 셔플, 워커등을 정한뒤 데이터를 읽어와 토치에서 사용함

 

 

배치사이즈 : 데이터를 읽어올 크기.

셔플 : 데이터를 섞어서 로드할지 결정하는거

워커 : 대충 쓰레드나 프로세스 갯수 같은거라 생각하면 편함

 

 

Dataset 클래스는 반드시 __init__, __len__, __getitem__ 함수를 정의해야 한다.

여기서 중요한건 getitem인데, 리턴값을 인풋, 정답 형식으로 리턴하면 된다.

 

 

이 Dataset과 DataLoader가 파이토치를 메모리 관리 어쩌구 해서 특별하게 어쩌구 저쩌구 하게 해준다 하니 대충 이정도만 알자.

class CustomDataset(Dataset):
    def __init__(self, inputs, labels):
        self.inputs = inputs
        self.labels = labels

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

    def __getitem__(self, idx):
        # 인덱스에 해당하는 입력 데이터와 정답(레이블)을 반환
        return self.inputs[idx], self.labels[idx]


train_dataset = CustomDataset(train_input_tensor,train_labels_tensor)
test_dataset = CustomDataset(test_input_tensor,test_labels_tensor)

train_loader = DataLoader(train_dataset,batch_size=32,shuffle=true,num_workers=2)
test_loader = DataLoader(test_dataset,batch_size=32,shuffle=true,num_workers=2)

 

 

 

 

 

다음으로 모델을 정의해야함. 모델=신경망 이라고 생각하면 편함.

 

1. nn.Module이라는 걸 상속하고 생성자에 nn.Sequential을 이용해 뉴럴네트워크를 만들어주자.

2. nn.Linear(인풋수,아웃풋수) 로 히든레이어를 만들어준다

3. nn.ReLu()로 히든레이어 사이마다 활성함수를 꼽사리 껴준다.

4. 마지막 히든레이어는 활성함수가 필요없다.

 

여기서 중요한건 forward라는 함수를 정의해주는거 인데, 데이터가 들어오면 뉴럴네트워크를 한번 통과시켜주도록 짜면 된다.

 

그리고 device라는걸 정의해줘서 뉴럴네트워크를 생성할때 넘겨줘야 한다. 이는 gpu를 사용할지 cpu를 사용할지 정의해주는거라 생각하면된다.

 

cuda의 경우 엔비디아의 gpu를 사용하고, mps는 애플꺼, 그마저도 없으면 cpu를 사용하도록 짜준다.

class NeuralNetwork(nn.Module):
    def __init__(self):
        super().__init__()
        self.linear_relu_stack = nn.Sequential(
            nn.Linear(7, 24),  #히든레이어 인풋7 -> 아웃풋24
            nn.ReLU(),         #활성함수
            nn.Linear(24, 24), #히든레이어 인풋24 -> 아웃풋24
            nn.ReLU(),         #활성함수
            nn.Linear(24, 1)   #히든레이어 인풋24 -> 아웃풋1
        )

    def forward(self, x):
        logits = self.linear_relu_stack(x) #뉴럴네트워크에 들어온 데이터 x를 굴려줌
        return logits # 굴려준 결과를 리턴



# 하드웨어 가속기를 무엇을 사용할지 정한다
device = ("cuda" if torch.cuda.is_available() else "mps" if torch.backends.mps.is_available() else "cpu")
model = NeuralNetwork().to(device) # 그리고 뉴럴네트워크를 만들때 넘겨준다

 

 

 

 

다음으로 손실함수와 옵티마이저를 만들어준다.

 

손실함수 : 뉴럴네트워크에 인풋 굴린뒤 나온 결과가 얼마나 틀렸는지 계산해주는 함수

옵티마이저 : 경사하강법에 사용될 함수

경사하강법 : 대충 손실을 최소화 해주는 작업에 사용되는 방법

 

옵티마이저는 Adam이 무난하다고 한다. SGD는 커스터마이징이 강점이라하는듯

loss_fn = nn.CrossEntropyLoss()
# optimizer = torch.optim.SGD(model.parameters(), lr=0.01)
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)

 

 

 

 

자 이제 거의 다왔다. 마지막에는 학습함수와, 테스트 함수를 작성해주도록 하자!

def train(dataloader, model, loss_fn, optimizer):
    size = len(dataloader.dataset)
    model.train() # 뉴럴 네트워크를 학습모드로 설정
    for batch, (X, y) in enumerate(dataloader):
        X, y = X.to(device), y.to(device) # x, y는 인풋과 정답이라 생각하면 됨

        # Compute prediction error
        pred = model(X)
        loss = loss_fn(pred, y)

        # Backpropagation
        loss.backward()
        optimizer.step()
        optimizer.zero_grad()

        if batch % 100 == 0:
            loss, current = loss.item(), (batch + 1) * len(X)
            print(f"loss: {loss:>7f}  [{current:>5d}/{size:>5d}]")
            
            
def test(dataloader, model, loss_fn):
    size = len(dataloader.dataset)
    num_batches = len(dataloader)
    model.eval() # 뉴럴 네트워크를 테스트모드로 설정
    test_loss, correct = 0, 0
    with torch.no_grad():
        for X, y in dataloader:
            X, y = X.to(device), y.to(device)
            pred = model(X)
            test_loss += loss_fn(pred, y).item()
            correct += (pred.argmax(1) == y).type(torch.float).sum().item()
    test_loss /= num_batches
    correct /= size
    print(f"Test Error: \n Accuracy: {(100*correct):>0.1f}%, Avg loss: {test_loss:>8f} \n")

 

 

 

 

마지막으로 epochs (수행횟수)를 설정하고 학습과 테스트를 반복하면 된다

 

 

epochs = 40
for t in range(epochs):
    print(f"Epoch {t+1}\n-------------------------------")
    train(train_loader, model, loss_fn, optimizer)
    test(test_loader, model, loss_fn)
print("Done!")