[총정리] Pytorch로 AI 모델링을 하는데 필요한 모든 것

1. Dataset

Dataset은 단순히 csv나 Json을 이용하는 거 아니야? 라고 생각하지만

 

엄밀히 말하면 파일을 가져와서 아래와 같은 방식으로 변환해 줘야 한다(대부분)

DataFrame -> numpy -> Tensor

 

즉 Tensor 형식으로 Data를 불러와야 하고 방법은 torch.utils.data를 이용하는 것이다

 

쉽게 아래만 기억하자

1. DataFrame 가져오기
2., DataFrame 에서 x 와 y 를 나누기
3. x, y를 각각 to_numpy() 를 이용해서 numpy로 바꾸기
4, numpy를 torch.FloatTensor(numpy).to(device) 를 통해 Tensor로 변환하고 장치에 올리기

 

Tensor 종류는 이 링크 참고

https://pytorch.org/docs/stable/tensors.html

torch.Tensor — PyTorch 2.4 documentation

 

 

#라이브러리 가져오기
import torch
from torch.utils.data import Dataset

class CustomDataset(Dataset):
    def __init__(self,x_data_df,y_data_df,device):
        x_data = x_data_df.to_numpy() #DataFrame to numpy
        y_data = y_data_df.to_numpy() #DataFrame to numpy

		#GPU를 사용할거면 device에 올려야하고 학습을 위해 Tensor로 바꿔야 한다
        self.data = torch.FloatTensor(x_data).to(device)
        self.label = torch.FloatTensor(y_data).to(device)

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

    def __getitem__(self, idx):
        #self.x_data와 self.y_data가 이미 GPU(CUDA)로 이동되어 있기 때문에, __getitem__ 메서드에서 다시 .to(device)를 사용하지 않아야 합니다. 
        x = self.data[idx]
        y = self.label[idx]
        return x, y

 

여기까지는 엄밀히 Data만 준비된 거지 분리를 하지는 않았다

엄밀히 말하면 틀만 준비한 것이다

이제 train data 와 test data 그리고 validation data로 나눠야 하는데

random_split , train_test_split 등 다양하게 사용할 수 있다

import json
import pandas as pd
from sklearn.model_selection import train_test_split
import torch

def make_train_val_test_dataset(data_path,test_size=0.2,train_val_rate=0.8, device =  "cuda" if torch.cuda.is_available() else "cpu") -> list[Dataset,Dataset,Dataset]:
    with open(data_path, 'r') as f:
        data = json.load(f)

    total_data = pd.DataFrame(data)
 	
    '''
	Data 전처리 파트
	'''
    # 위에서 나온 결과가 prepare_data_df 라 가정
    df_x_data = prepare_data_df[['x1', 'x2', 'x3']]
    df_y_data = prepare_data_df[['y2', 'y2', 'y3']]

    df_x_train, df_x_test, df_y_train, df_y_test = train_test_split(df_x_data, df_y_data, test_size=test_size, random_state=777)

    dataset = CustomDataset(df_x_train,df_y_train,device)

    dataset_size = len(dataset)
    train_size = int(dataset_size * train_val_rate)
    validation_size = dataset_size - train_size

    train_dataset, validation_dataset = random_split(dataset, [train_size, validation_size])
    test_dataset = CustomDataset(df_x_test,df_y_test,device)
    print("Finished to Prepared DataSet")
    print(f"Training Data Size : {len(train_dataset)}")
    print(f"Validation Data Size : {len(validation_dataset)}")
    print(f"Testing Data Size : {len(test_dataset)}")

    return train_dataset, validation_dataset, test_dataset

 

이렇게 만들 걸 Data loader에 넣어야 한다

그 이유는 mini-batch 가 가장 크다 (이거 아님 슈퍼 하드 코딩임)\

아래와 같이 만들어 주자

from torch.utils.data import DataLoader

train_dataloader = DataLoader(
    train_dataset, 
    batch_size=args.train_batch_size, 
    shuffle=True, 
    drop_last=True)

validation_dataloader = DataLoader(
    validation_dataset, 
    batch_size=args.valid_batch_size, 
    shuffle=True, 
    drop_last=True)

test_dataloader = DataLoader(
    test_dataset, 
    batch_size=1, 
    shuffle=False, #순서대로 이용하기 위해서 
    drop_last=False)

 

이제 모델을 만들 차례인데

Regression 모델을 만들어 보자

 

model은 nn.Module 을 통해서 만들 수 있는데 class 형식이다

 

가장 기본 형태는

class CustomRegressor(nn.Module):
    
    def __init__(self,input_features,output_features,device):
        super().__init__()

        self.device = device
        #self.input_features = input_features
        #self.output_features =output_features

        self.input_layer = nn.Linear(
            in_features = input_features,
            out_features = 32,
            device=device)
        
        self.fc1 = nn.Linear(
            in_features = 32,
            out_features = 128,
            device=device)

        self.fc2 = nn.Linear(
            in_features = 128,
            out_features = 64,
            device=device)


        self.output_layer = nn.Linear(
            in_features = 64,
            out_features = output_features,
            device=device)
        )
        
     def forward(self, input_data):
        x = torch.relu(self.input_layer(input_data))
        x = torch.relu(self.fc1(x))
        x = torch.relu(self.fc2(x))
        output = self.output_layer(x)
        return output

 

이런 식으로 만들어 줄 수 있고 (__str__ (self) 등을 활용해서 더 많이 꾸밀 수도 있음)

nn,Sequential을 이용해서 아래와 같이

class CustomRegressor(nn.Module):
    
    def __init__(self,input_features,output_features,device):
        super().__init__()

        self.device = device
        #self.input_features = input_features
        #self.output_features =output_features

        self.input_layer = nn.Linear(in_features = input_features,out_features = 32,device=device)
        
        self.fc1 = 

        self.fc2 = nn.Linear(
            in_features = 128,
            out_features = 64,
            device=device)


        self.output_layer = nn.Linear(in_features = 64,out_features = output_features,device=device)
        
        self.model = nn.Sequential(
            nn.Linear(in_features = input_features,out_features = 32,device=device)
            nn.ReLU(),
            nn.Linear(in_features = 32,out_features = 128,device=device)
            nn.ReLU(),
            nn.Linear(in_features = 128,out_features = 64,device=device)
            nn.ReLU(),
            nn.Linear(in_features = 64,out_features = output_features,device=device)
        )
        
    def forward(self, input_data):
        output = self.model(input_data)
        return output

 

만들 수도 있다.

 

이러면

Data 준비 -> Model 준비 까지 끝났다

그러면 뭐가 남았냐!

Training 이 남았다

 

Train, Validation, Test 방식 먼저 정해보자

 

Test는 별개로 구성을 해야 하지만

 

Train과 Validation은 약간만 다르고 복붙이다

 

어떤 게 다르냐면

Train

#code related to training
model.train()




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

 

Validation / Test

#code related to inference
model.eval()
with torch.no_grad():

딱 이것만 다르고 나머지는 똑같다 예시를 보자

def train_loop(model,opt,train_dataloader,device):
    model.train() #train mode
    total_loss = 0
    total_accuracy = 0
    total_cnt = 0

    for X,y in train_dataloader:
        X, y = X.to(device), y.to(device)  # Move data to GPU

        pred = model(X).to(device)

        loss_func = torch.nn.MSELoss()
        loss = loss_func(pred, y)
        
        opt.zero_grad()
        loss.backward()
        opt.step()

        total_loss += loss.detach().item()
    
    train_loss = total_loss / len(y)
    #train_acc 구현
    train_acc = ''' '''
    return train_loss , train_acc
    
    
def validation_loop(model, val_dataloader,device):
    model.eval()
    total_loss = 0
    with torch.no_grad(): # 이부분만 다르고 나머지는 training과 동일
        for X,y in val_dataloader:
            X, y = X.to(device), y.to(device)  # Move data to GPU
            
            pred = model(X).to(device)
            loss_func = torch.nn.MSELoss()
            loss = loss_func(pred, y)
            total_loss += loss.detach().item()

    val_loss = total_loss / len(y)
    #val_accuracy 구현
    val_accuracy = ''' '''
    return val_loss, val_accuracy

 

참고 링크 : https://rabo0313.tistory.com/entry/Pytorch-modeltrain-modeleval-%EC%9D%98%EB%AF%B8

[Pytorch] model.train(), model.eval() 의미

그러면 Test는 ? 

Validation을 참고하면 된다 (근데 loss를 볼 필요도 없음)

def test(model,test_dataloader,device = torch.device('cuda') if torch.cuda.is_available() else torch.device('cpu')):
    model.eval()

    for X,y in test_dataloader:
        X, y = X.to(device), y.to(device)  # Move data to GPU
        pred = model(X).to(device)

	'''
    accuracy 구현하기
    '''

    return accuracy

 

자 이제 train loop와 validation loop를 준비했으니

진짜 fit() 함수인 train()함수를 만들어 보자

 

아래의 코드는 복잡해 보이지만

현재 시간을 참고해서 저장할 파일들의 이름을 만들고

train / validation 의 loss 와 accuracy 또한 list -> txt 로 저장하여서

나중에 loss accuracy 그래프를 확인 하고자 하였다

 

또한 모델을 저장해야 하는데 다음 3가지를 저장하였다

1. Training 기준 acc 가 제일 높은 것

2, Validation 기준 acc 가 제일 높은 것

3. 가장 마지막 모델

 

모델 저장은 state_dict()으로 하였다

 

코드 먼저 보면

 

from copy import deepcopy 
RECORD_TMP = 10

def train(model, opt, train_dataloader, val_dataloader, epochs, save_log_file,device):
    best_val_acc = 0
    best_train_acc = 0
    best_val_best_acc_model = None
    best_train_best_acc_model = None
    file = open(save_log_file, 'a')
    current_time = datetime.now()
    file.write("\n\n"+"="*64+"\n")
    file.write("                     Recording The Training                     "+"\n")
    file.write("                training Time : "+current_time.strftime('%Y-%m-%d %H:%M:%S')+"\n")
    file.write("="*64+"\n\n")
    # plotting하기 위한 리스트 생성 

    train_loss_list, validation_loss_list = [], []
    train_acc_list, validation_acc_list = [],[]
 
    print("Training and validating model")
    for epoch in trange(epochs):
        if epoch % RECORD_TMP == 0 :
            file.write("-"*25 +" Epoch {"+str(epoch + 1)+"} " +"-"*25+"\n")

        train_loss, train_accuracy = train_loop(model,opt,train_dataloader,device)
        train_loss_list += [train_loss]
        train_acc_list += [train_accuracy]


        if train_accuracy > best_train_acc:
            best_train_acc = train_accuracy
            best_train_best_acc_model = deepcopy(model.state_dict())         

        validation_loss,validation_acc= validation_loop(model, val_dataloader,device)
        validation_loss_list += [validation_loss]
        validation_acc_list += [validation_acc]

        if validation_acc > best_val_acc:
            best_val_acc = validation_acc
            best_val_best_acc_model = deepcopy(model.state_dict())            

        if epoch % RECORD_TMP == 0 :
            file.write(f"Training loss: {train_loss:.4f}"+"\n")
            file.write(f"Validation loss: {validation_loss:.4f}"+"\n")
            file.write(f"Training accuracy: {train_accuracy:.4f}"+"\n")
            file.write(f"Validation accuracy: {validation_acc:.4f}"+"\n")
            #print(f"Training loss: {train_loss:.4f}")
            #print(f"Validation loss: {validation_loss:.4f}")
        
    file.close()
    return best_train_acc,best_val_acc,best_train_best_acc_model,best_val_best_acc_model,train_loss_list, train_acc_list, validation_loss_list,validation_acc_list

 

인데 필요한 것들을 쪼개서 보면

 

먼저 현재 시간을 가져오는 코드

from datetime import datetime
current_time = datetime.now()

 

txt파일을 써주는 코드

file = open(save_log_file, 'a')
file.write(f"Training loss: {train_loss:.4f}"+"\n")
file.write("\n\n"+"="*64+"\n")
file.close()

여기서 save_log_file 은 파일 명을 포함한 (~~~.txt) 저장할 파일의 경로이고

옆에 옵션은

이거 이다

만약 뭘 적었는지 출력하고 싶다면 그대로 return 받아서

file = open(FILE_PATH, 'a')
output = file.write("Hello World")
print(output)
file.close()

 

이렇게 해주면 된다

읽을 때는

file = open('test.txt', 'r')
text = file.read() 
file.close()

d할 수도 있고

 

이렇게도 쓴다

#저장하기
with open('text.txt', 'a') as f:
    f.write('python')
    
#불러오기    
with open('text.txt', 'r') as f:
    print(f.read())

 

 

또 코드에서 재밌는 건 이 부분인데

if train_accuracy > best_train_acc:
    best_train_acc = train_accuracy
    best_train_best_acc_model = deepcopy(model.state_dict())

 

여기에 굳이 deepcopy를 써야하나 라는 생각을 할 수도 있지만

모델 저장할 때 deepcopy를 해줘야 한다 아니면 에러남

관련링크 : https://tutorials.pytorch.kr/beginner/saving_loading_models.html

모델 저장하기 & 불러오기

 

그러면 이제 모든 준비가 끝났으니

train.py 를 만들어 보자 (다른 폴더의 py 에서 가져온 함수들은 일단 무시하자)

from torch.utils.data import DataLoader
import torch
import torch.optim as optim
import pickle
from datetime import datetime

#다른 폴더에 있는 py의 함수 가져오는 부분
from src.Data_Loader import make_train_val_test_dataset
from src.func import train,createDirectory,test, make_acc_under
from configs.train_setting import parse_args

device = torch.device('cuda') if torch.cuda.is_available() else torch.device('cpu')
data_path = '/home/virtual_display_task/data/2024_BR214_2D_Plus.json'
SAVE_LOG_DIR = "/home/virtual_display_task/result/train_log/"
SAVE_LOSS_ACC_DIR = "/home/virtual_display_task/result/visualization/"

date = datetime.today().strftime("%Y%m%d%H")

def make_acc_under(accuarcy):
    formatted_accuarcy = f"{accuarcy:.2f}"
    formatted_accuarcy = formatted_accuarcy.replace('.','_')
    return formatted_accuarcy

if __name__ == "__main__":

    print("** Start to Training : ",date," **")
    args = parse_args()

    # build_dataset
    train_dataset, validation_dataset, test_dataset = make_train_val_test_dataset(data_path,test_size=0.2,train_val_rate=0.8)

    train_dataloader = DataLoader(
        train_dataset, 
        batch_size=args.train_batch_size, 
        shuffle=True, 
        drop_last=True)

    validation_dataloader = DataLoader(
        validation_dataset, 
        batch_size=args.valid_batch_size, 
        shuffle=True, 
        drop_last=True)

    test_dataloader = DataLoader(
        test_dataset, 
        batch_size=1, 
        shuffle=False, #순서대로 이용하기 위해서 
        drop_last=False)



    # build model
    createDirectory(SAVE_LOG_DIR)
    #input과 output Dimension이 모두 3으로 동일 했음
    model = args.model(input_features=3,output_features=3,device=device)
    optimizer = optim.Adam(model.parameters(), lr=args.lr)
    # 아래의 list 들은 나중에 loss, acc 그래프를 사용하기 위함
    best_train_acc,best_val_acc,\
        best_train_best_acc_model,best_val_best_acc_model,\
            train_loss_list, train_acc_list,\
                valiadation_loss_list,validation_acc_list = train(model, optimizer, 
                                                                train_dataloader,
                                                                validation_dataloader,
                                                                args.epochs,
                                                                SAVE_LOG_DIR+ args.log_file_name,
                                                                device)


	#최종 test 에 대한 정확도도 보고 싶어서
    accuarcy = test(model,test_dataloader,device=device)
    print(accuarcy)
	
    #정확도는 소숫점이므로 . 을 _으로 바꿔주는 코드
    tuned_train_acc = make_acc_under(best_train_acc)
    tuned_val_acc = make_acc_under(best_val_acc)
    formatted_accuarcy = make_acc_under(accuarcy)

    save_dir_path = '/home/virtual_display_task/result/model_result/'+str((args.model).__name__ )+'_'+str(formatted_accuarcy)+"_"+str(date)+'/'
    
    # save model (train acc 가 높은 거 val acc가 높은거 제일 마지막 꺼)
    createDirectory(save_dir_path)
    torch.save(best_train_best_acc_model,
               save_dir_path+"train_"+str(tuned_train_acc)+'_'+str(date)+'.pt')
    torch.save(best_val_best_acc_model,
            save_dir_path+"val_"+str(tuned_val_acc)+'_'+str(date)+'.pt')
    torch.save(model.state_dict(),
            save_dir_path+"latest"+'.pt') 
    


    # save for visualizaiton -> loss , validation 그래프 그리는 데에 이용
    save_vis_dir_path = save_dir_path + 'visualization/'
    createDirectory(save_vis_dir_path)
    with open(save_vis_dir_path+str(formatted_accuarcy)+'_'+str(args.loss_acc_file_name),"wb") as f:
        pickle.dump(train_loss_list, f)
        pickle.dump(valiadation_loss_list, f)
        pickle.dump(train_acc_list, f)
        pickle.dump(validation_acc_list, f)   
    
    print("** Finish to Training **")