파이토치(PyTorch) 딥러닝 모델링 전략 (2) - 파이토치 딥러닝 모델링 단계별 기본 셋팅
파이토치를 활용한 딥러닝 모델링이 익숙하지 않아서
중요하다고 생각되는 부분들을 정리하고자 포스팅합니다.
파이토치 딥러닝 모델링 단계별 기본 셋팅
※ 앞서 포스팅한 '파이토치 딥러닝 모델링 단계'를 기준으로 포스팅하였습니다.
4) 베이스라인 모델 셋팅/모델 훈련 및 예측 수행
(1) 환경 설정
: Seed 고정
# 시드값 고정
import torch # 파이토치
import random
import numpy as np
import os
# 시드값 고정
seed = 50
os.environ['PYTHONHASHSEED'] = str(seed)
random.seed(seed) # 파이썬 난수 생성기 seed
np.random.seed(seed) # 넘파이 난수 생성기 seed
torch.manual_seed(seed) # 파이토치 난수 생성기 seed(CPU)
torch.cuda.manual_seed(seed) # 파이토치 난수 생성기 seed(GPU)
torch.cuda.manual_seed_all(seed) # 파이토치 난수 생성기 seed(멀티GPU)
torch.backends.cudnn.deterministic = True # Deterministic(확정적) 연산 사용
torch.backends.cudnn.benchmark = False # 벤치마크 기능 해제
torch.backends.cudnn.enabled = False # cudnn 사용 해제*만약 결과 재현이 필요 없을 경우는 seed 고정을 생략(※ seed 고정시 속도 및 예측성능이 저하될 수 있음.)
: GPU 자원 활용 Device 설정
# GPU 장비 설정
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
device
# 출력결과 >>> device(type='cuda')
(2) 데이터 준비
: 데이터 경로 지정
import pandas as pd
from sklearn.model_selection import train_test_split
# 데이터 경로 지정
data_path = './data/'
train = pd.read_csv(data_path + 'train.csv')
test = pd.read_csv(data_path + 'test.csv')
: 훈련 데이터, 검증 데이터 분리
# 훈련 데이터, 검증 데이터 분리(이진분류)
train, valid = train_test_split(train,
test_size=0.1,
stratify=labels['pos'],
random_state=50)
# 훈련 데이터, 검증 데이터 분리(다중분류)
train, valid = train_test_split(train,
test_size=0.1,
stratify=train[['A', 'B', 'C', 'D']],
random_state=50)
# 출력
print("훈련 데이터 개수: ', len(train))
print("검증 데이터 개수: ', len(valid))
: 데이터셋 클래스 정의 및 생성 (ex. 이미지 데이터일 경우를 예시로 함)
# 데이터셋 클래스 정의
import cv2
from torch.utils.data import Dataset # 데이터 생성을 위한 클래스
import numpy as np
class ImageDataset(Dataset):
# 초기화 메서드(생성자)
def __init__(self, df, img_dir='./', transform=None, is_test=False):
super().__init__() # 상속받은 Dataset의 생성자 __init__() 메서드 호출
# 전달받은 인수 저장
self.df = df
self.img_dir = img_dir
self.transform = transform
self.is_test = is_test
# 데이터셋 크기 반환 메서드
def __len__(self):
return len(self.df)
# 인덱스(idx)에 해당하는 데이터 반환 메서드
def __getitem__(self, idx):
img_id = self.df.iloc[idx, 0] # 이미지 ID
img_path = self.img_dir + img_id + '.jpg' # 이미지 파일 경로
image = cv2.imread(img_path) # 이미지 파일 읽기
image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB) # 이미지 색상 보정
# label = self.df.iloc[idx, 1] ← 이진 분류일 경우 이미지 레이블(타깃값)
# 이미지 변환 (변환기가 있을 경우)
if self.transform is not None:
image = self.transform(image=image)['image']
# 테스트 or 훈련/검증용일때 반환값 조정
if self.is_test: # 테스트용일 때
return image
else: # 훈련/검증용일 때
label = np.argmax(self.df.iloc[idx, 1:5]) # 타깃값 4개 중 가장 큰 값의 인덱스
return image, label
: 이미지 변환기 셋팅(transforms 사용시)
from torchvision import transforms # 이미지 변환을 위한 모듈
# 훈련 데이터용 변환기
transform_train = transforms.Compose([transforms.ToTensor(),
transforms.Pad(32, padding_mode='symmetric'),
transforms.RandomHorizontalFlip(),
transforms.RandomVerticalFlip(),
transforms.RandomRotation(10),
transforms.Normalize((0.485, 0.456, 0.406),
(0.229, 0.224, 0.225))])
# 검증 및 테스트 데이터용 변환기
transform_test= transforms.Compose([transforms.ToTensor(),
transforms.Pad(32, padding_mode='symmetric'),
transforms.Normalize((0.485, 0.456, 0.406),
(0.229, 0.224, 0.225))])
: 이미지 변환기 셋팅(Albumentations 사용시)
# 이미지 변환기 정의
# 이미지 변환을 위한 모듈
import albumentations as A
from albumentations.pytorch import ToTensorV2
# 훈련 데이터용 변환기
transform_train = A.Compose([
A.Resize(450, 650), # 이미지 크기 조절
A.RandomBrightnessContrast(brightness_limit=0.2, # 밝기 대비 조절
contrast_limit=0.2, p=0.3),
A.VerticalFlip(p=0.2), # 상하 대칭 변환
A.HorizontalFlip(p=0.5), # 좌우 대칭 변환
A.ShiftScaleRotate( # 이동, 스케일링, 회전 변환
shift_limit=0.1,
scale_limit=0.2,
rotate_limit=30, p=0.3),
A.OneOf([A.Emboss(p=1), # 양각화, 날카로움, 블러 효과
A.Sharpen(p=1),
A.Blur(p=1)], p=0.3),
A.PiecewiseAffine(p=0.3), # 어파인 변환
A.Normalize(), # 정규화 변환
ToTensorV2() # 텐서로 변환
])
# 검증 및 테스트 데이터용 변환기
transform_test = A.Compose([
A.Resize(450, 650), # 이미지 크기 조절
A.Normalize(), # 정규화 변환
ToTensorV2() # 텐서로 변환
])
: 데이터셋 생성
# 데이터셋 생성(1)
dataset_train = ImageDataset(df=train, img_dir='train/', transform=transform_train)
dataset_valid = ImageDataset(df=valid, img_dir='train/', transform=transform_test)
# 데이터셋 생성(2)
img_dir = './data/images/'
dataset_train = ImageDataset(train, img_dir=img_dir, transform=transform_train)
dataset_valid = ImageDataset(valid, img_dir=img_dir, transform=transform_test)
: 데이터 로더 멀티 프로세싱
# 멀티 프로세서 설정
def seed_worker(worker_id): # 데이터 로더 시드값 고정 함수
worker_seed = torch.initial_seed() % 2**32
np.random.seed(worker_seed)
random.seed(worker_seed)
g = torch.Generator() # 제너레이터 생성
g.manual_seed(0) # 제너레이터 시드값 고정
: 데이터 로더 생성
from torch.utils.data import DataLoader # 데이터 로더 클래스
batch_size = 4
loader_train = DataLoader(dataset_train, batch_size=batch_size, shuffle=True,
worker_init_fn=seed_worker, generator=g, num_workers=2)
loader_valid = DataLoader(dataset_valid, batch_size=batch_size, shuffle=False,
worker_init_fn=seed_worker,generator=g, num_workers=2)
# 멀티프로세싱일 경우 "worker_init_fn=seed_worker,generator=g, num_workers=2" 부분을 추가함
(3) 모델 훈련, 성능 검증
: 합성곱(CNN) 모델 생성(예시)
import torch.nn as nn # 신경망 모듈
import torch.nn.functional as F # 신경망 모듈에서 자주 사용되는 함수
class Model(nn.Module):
# 신경망 계층 정의
def __init__(self):
super().__init__() # 상속받은 nn.Module의 __init__() 메서드 호출
# 첫 번째 합성곱 계층
self.conv1 = nn.Conv2d(in_channels=3, out_channels=32,
kernel_size=3, padding=2)
# 두 번째 합성곱 계층
self.conv2 = nn.Conv2d(in_channels=32, out_channels=64,
kernel_size=3, padding=2)
# 최대 풀링 계층
self.max_pool = nn.MaxPool2d(kernel_size=2)
# 평균 풀링 계층
self.avg_pool = nn.AvgPool2d(kernel_size=2)
# 전결합 계층
self.fc = nn.Linear(in_features=64 * 4 * 4, out_features=2)
# 순전파 출력 정의
def forward(self, x):
x = self.max_pool(F.relu(self.conv1(x)))
x = self.max_pool(F.relu(self.conv2(x)))
x = self.avg_pool(x)
x = x.view(-1, 64 * 4 * 4) # 평탄화
x = self.fc(x)
return x
: 합성곱(CNN) 모델 생성(예시2)
class Model(nn.Module):
def __init__(self):
super().__init__()
# 첫 번째 합성곱, 최대 풀링 계층
self.layer1 = nn.Sequential(nn.Conv2d(in_channels=3,
out_channels=32,
kernel_size=3,
padding=2),
nn.ReLU(),
nn.MaxPool2d(kernel_size=2))
# 두 번째 합성곱, 최대 풀링 계층
self.layer2 = nn.Sequential(nn.Conv2d(in_channels=32,
out_channels=64,
kernel_size=3,
padding=2),
nn.ReLU(),
nn.MaxPool2d(kernel_size=2))
# 평균 풀링 계층
self.avg_pool = nn.AvgPool2d(kernel_size=2)
# 전결합 계층
self.fc = nn.Linear(in_features=64 * 4 * 4, out_features=2)
# 순전파 출력 정의
def forward(self, x):
x = self.layer1(x)
x = self.layer2(x)
x = self.avg_pool(x)
x = x.view(-1, 64 * 4 * 4) # 평탄화
x = self.fc(x)
return x
: 합성곱(CNN) 모델 개선(예시3)
import torch.nn as nn # 신경망 모듈
import torch.nn.functional as F # 신경망 모듈에서 자주 사용되는 함수
class Model(nn.Module):
# 신경망 계층 정의
def __init__(self):
super().__init__() # 상속받은 nn.Module의 __init__() 메서드 호출
# 1 ~ 5번째 {합성곱, 배치 정규화, 최대 풀링} 계층
self.layer1 = nn.Sequential(nn.Conv2d(in_channels=3, out_channels=32,
kernel_size=3, padding=2),
nn.BatchNorm2d(32), # 배치 정규화
nn.LeakyReLU(), # LeakyReLU 활성화 함수
nn.MaxPool2d(kernel_size=2))
self.layer2 = nn.Sequential(nn.Conv2d(in_channels=32, out_channels=64,
kernel_size=3, padding=2),
nn.BatchNorm2d(64),
nn.LeakyReLU(),
nn.MaxPool2d(kernel_size=2))
self.layer3 = nn.Sequential(nn.Conv2d(in_channels=64, out_channels=128,
kernel_size=3, padding=2),
nn.BatchNorm2d(128),
nn.LeakyReLU(),
nn.MaxPool2d(kernel_size=2))
self.layer4 = nn.Sequential(nn.Conv2d(in_channels=128, out_channels=256,
kernel_size=3, padding=2),
nn.BatchNorm2d(256),
nn.LeakyReLU(),
nn.MaxPool2d(kernel_size=2))
self.layer5 = nn.Sequential(nn.Conv2d(in_channels=256, out_channels=512,
kernel_size=3, padding=2),
nn.BatchNorm2d(512),
nn.LeakyReLU(),
nn.MaxPool2d(kernel_size=2))
# 평균 풀링 계층
self.avg_pool = nn.AvgPool2d(kernel_size=4)
# 전결합 계층
self.fc1 = nn.Linear(in_features=512 * 1 * 1, out_features=64)
self.fc2 = nn.Linear(in_features=64, out_features=2)
# 순전파 출력 정의
def forward(self, x):
x = self.layer1(x)
x = self.layer2(x)
x = self.layer3(x)
x = self.layer4(x)
x = self.layer5(x)
x = self.avg_pool(x)
x = x.view(-1, 512 * 1 * 1) # 평탄화
x = self.fc1(x)
x = self.fc2(x)
return x
: Pretrained Mode 사용시(ex. EfficientNet)
!pip install efficientnet-pytorch==0.7.1
from efficientnet_pytorch import EfficientNet # EfficientNet 모델
# 사전 훈련된 efficientnet-b7 모델 불러오기
model = EfficientNet.from_pretrained('efficientnet-b7', num_classes=4)
model = model.to(device) # 장비 할당
: 디바이스에 모델 할당
model = Model().to(device)
model
: 손실함수 설정
# 손실함수
criterion = nn.CrossEntropyLoss()
: 옵티마이저 설정
# 옵티마이저
optimizer = torch.optim.SGD(model.parameters(), lr=0.01)
'''
다른 옵티마이저 사용도 가능하다.
optimizer = torch.optim.Adamax(model.parameters(), lr=0.00006)
'''
: 성능개선 (1) (스케줄러 추가)
*스케쥴러 : 훈련 과정에서 학습률 조정(훈련 초반에는 학습률을 크게 - 빠른 가중치 갱신/후반에는 학습률을 작게 - 미세조정)
from transformers import get_cosine_schedule_with_warmup
epochs = 39 # 총 에폭
# 스케줄러 생성
scheduler = get_cosine_schedule_with_warmup(optimizer,
num_warmup_steps=len(loader_train)*3,
num_training_steps=len(loader_train)*epochs)
: 모델 훈련(1 - 이진분류)
epochs = 10 # 총 에폭
# 총 에폭만큼 반복
for epoch in range(epochs):
epoch_loss = 0 # 에폭별 손실값 초기화
# '반복 횟수'만큼 반복
for images, labels in loader_train:
# 이미지, 레이블 데이터 미니배치를 장비에 할당
images = images.to(device)
labels = labels.to(device)
# 옵티마이저 내 기울기 초기화
optimizer.zero_grad()
# 순전파 : 이미지 데이터를 신경망 모델의 입력값으로 사용해 출력값 계산
outputs = model(images)
# 손실 함수를 활용해 outputs와 labels의 손실값 계산
loss = criterion(outputs, labels)
# 현재 배치에서의 손실 추가
epoch_loss += loss.item()
# 역전파 수행
loss.backward()
# 가중치 갱신
optimizer.step()
# 훈련 데이터 손실값 출력
print(f'에폭 [{epoch+1}/{epochs}] - 손실값: {epoch_loss/len(loader_train):.4f}')
: 모델 훈련(2 - 다중분류, pretrained model 사용)
from sklearn.metrics import roc_auc_score # ROC AUC 점수 계산 함수
from tqdm.notebook import tqdm # 진행률 표시 막대
epochs = 5
# 총 에폭만큼 반복
for epoch in range(epochs):
# == [ 훈련 ] ==============================================
model.train() # 모델을 훈련 상태로 설정
epoch_train_loss = 0 # 에폭별 손실값 초기화 (훈련 데이터용)
# '반복 횟수'만큼 반복
for images, labels in tqdm(loader_train):
# 이미지, 레이블(타깃값) 데이터 미니배치를 장비에 할당
images = images.to(device)
labels = labels.to(device)
# 옵티마이저 내 기울기 초기화
optimizer.zero_grad()
# 순전파 : 이미지 데이터를 신경망 모델의 입력값으로 사용해 출력값 계산
outputs = model(images)
# 손실 함수를 활용해 outputs와 labels의 손실값 계산
loss = criterion(outputs, labels)
# 현재 배치에서의 손실 추가 (훈련 데이터용)
epoch_train_loss += loss.item()
loss.backward() # 역전파 수행
optimizer.step() # 가중치 갱신
# scheduler.step() # 스케쥴러 학습률 갱신시
# 훈련 데이터 손실값 출력
print(f'에폭 [{epoch+1}/{epochs}] - 훈련 데이터 손실값 : {epoch_train_loss/len(loader_train):.4f}')
# == [ 검증 ] ==============================================
model.eval() # 모델을 평가 상태로 설정
epoch_valid_loss = 0 # 에폭별 손실값 초기화 (검증 데이터용)
preds_list = [] # 예측 확률값 저장용 리스트 초기화
true_onehot_list = [] # 실제 타깃값 저장용 리스트 초기화
with torch.no_grad(): # 기울기 계산 비활성화
# 미니배치 단위로 검증
for images, labels in loader_valid:
images = images.to(device)
labels = labels.to(device)
outputs = model(images)
loss = criterion(outputs, labels)
epoch_valid_loss += loss.item()
preds = torch.softmax(outputs.cpu(), dim=1).numpy() # 예측 확률값
# 실제값 (원-핫 인코딩 형식)
true_onehot = torch.eye(4)[labels].cpu().numpy()
# 예측 확률값과 실제값 저장
preds_list.extend(preds)
true_onehot_list.extend(true_onehot)
# 검증 데이터 손실값 및 ROC AUC 점수 출력
print(f'에폭 [{epoch+1}/{epochs}] - 검증 데이터 손실값 : {epoch_valid_loss/len(loader_valid):.4f} / 검증 데이터 ROC AUC : {roc_auc_score(true_onehot_list, preds_list):.4f}')
: 성능 검증
from sklearn.metrics import roc_auc_score # ROC AUC 점수 계산 함수 임포트
# 실제값과 예측 확률값을 담을 리스트 초기화
true_list = []
preds_list = []
model.eval() # 모델을 평가 상태로 설정
with torch.no_grad(): # 기울기 계산 비활성화
for images, labels in loader_valid:
# 이미지, 레이블 데이터 미니배치를 장비에 할당
images = images.to(device)
labels = labels.to(device)
# 순전파 : 이미지 데이터를 신경망 모델의 입력값으로 사용해 출력값 계산
outputs = model(images)
preds = torch.softmax(outputs.cpu(), dim=1)[:, 1] # 예측 확률
true = labels.cpu() # 실제값
# 예측 확률과 실제값을 리스트에 추가
preds_list.extend(preds)
true_list.extend(true)
# 검증 데이터 ROC AUC 점수 계산
print(f'검증 데이터 ROC AUC : {roc_auc_score(true_list, preds_list):.4f}')
: 예측(1 - 이진분류)
dataset_test = ImageDataset(df=submission, img_dir='test/', transform=transform)
loader_test = DataLoader(dataset=dataset_test, batch_size=32, shuffle=False)
model.eval() # 모델을 평가 상태로 설정
preds = [] # 타깃 예측값 저장용 리스트 초기화
with torch.no_grad(): # 기울기 계산 비활성화
for images, _ in loader_test:
# 이미지 데이터 미니배치를 장비에 할당
images = images.to(device)
# 순전파 : 이미지 데이터를 신경망 모델의 입력값으로 사용해 출력값 계산
outputs = model(images)
# 타깃값이 1일 확률(예측값)
preds_part = torch.softmax(outputs.cpu(), dim=1)[:, 1].tolist()
# preds에 preds_part 이어붙이기
preds.extend(preds_part)
: 예측(2 - 다중분류)
dataset_test = ImageDataset(test, img_dir=img_dir,
transform=transform_test, is_test=True)
loader_test = DataLoader(dataset_test, batch_size=batch_size,
shuffle=False, worker_init_fn=seed_worker,
generator=g, num_workers=2)
model.eval() # 모델을 평가 상태로 설정
preds = np.zeros((len(test), 4)) # 예측값 저장용 배열 초기화
with torch.no_grad():
for i, images in enumerate(loader_test):
images = images.to(device)
outputs = model(images)
# 타깃 예측 확률
preds_part = torch.softmax(outputs.cpu(), dim=1).squeeze().numpy()
preds[i*batch_size:(i+1)*batch_size] += preds_part
: 성능개선 (1) (TTA, Test 단계 데이터 증강)
*테스트 단계에서도 데이터 증강기법을 이용하여 예측성능을 끌어올리는 방법
# 테스트 데이터 원본 데이터셋 및 데이터 로더
dataset_test = ImageDataset(test, img_dir=img_dir,
transform=transform_test, is_test=True)
loader_test = DataLoader(dataset_test, batch_size=batch_size,
shuffle=False, worker_init_fn=seed_worker,
generator=g, num_workers=2)
# TTA용 데이터셋 및 데이터 로더
dataset_TTA = ImageDataset(test, img_dir=img_dir,
transform=transform_train, is_test=True)
loader_TTA = DataLoader(dataset_TTA, batch_size=batch_size,
shuffle=False, worker_init_fn=seed_worker,
generator=g, num_workers=2)
원본을 이용한 예측 후 TTA 적용하여 예측
model.eval() # 모델을 평가 상태로 설정
preds_test = np.zeros((len(test), 4)) # 예측값 저장용 배열 초기화
with torch.no_grad():
for i, images in enumerate(loader_test):
images = images.to(device)
outputs = model(images)
# 타깃 예측 확률
preds_part = torch.softmax(outputs.cpu(), dim=1).squeeze().numpy()
preds_test[i*batch_size:(i+1)*batch_size] += preds_part
submission_test = submission.copy() # 샘플 파일 복사
submission_test[['healthy', 'multiple_diseases', 'rust', 'scab']] = preds_test
num_TTA = 7 # TTA 횟수
preds_tta = np.zeros((len(test), 4)) # 예측값 저장용 배열 초기화 (TTA용)
# TTA를 적용해 예측
for i in range(num_TTA):
with torch.no_grad():
for i, images in enumerate(loader_TTA):
images = images.to(device)
outputs = model(images)
# 타깃 예측 확률
preds_part = torch.softmax(outputs.cpu(), dim=1).squeeze().numpy()
preds_tta[i*batch_size:(i+1)*batch_size] += preds_part
preds_tta /= num_TTA
submission_tta = submission.copy()
submission_tta[['healthy', 'multiple_diseases', 'rust', 'scab']] = preds_tta
: 성능개선 (2) (레이블 스무딩)
* 특정 타깃값의 확률을 과잉평가(1에 가깝게)하는 것을 방지 위한 예측값 보정 방법
def apply_label_smoothing(df, target, alpha, threshold):
# 타깃값 복사
df_target = df[target].copy()
k = len(target) # 타깃값 개수
for idx, row in df_target.iterrows():
if (row > threshold).any(): # 임계값을 넘는 타깃값인지 여부 판단
row = (1 - alpha)*row + alpha/k # 레이블 스무딩 적용
df_target.iloc[idx] = row # 레이블 스무딩을 적용한 값으로 변환
return df_target # 레이블 스무딩을 적용한 타깃값 반환
alpha = 0.001 # 레이블 스무딩 강도
threshold = 0.999 # 레이블 스무딩을 적용할 임계값
# 레이블 스무딩을 적용하기 위해 DataFrame 복사
submission_test_ls = submission_test.copy()
submission_tta_ls = submission_tta.copy()
target = ['healthy', 'multiple_diseases', 'rust', 'scab'] # 타깃값 열 이름
# 레이블 스무딩 적용
submission_test_ls[target] = apply_label_smoothing(submission_test_ls, target,
alpha, threshold)
submission_tta_ls[target] = apply_label_smoothing(submission_tta_ls, target,
alpha, threshold)
submission_test_ls.to_csv('submission_test_ls.csv', index=False)
submission_tta_ls.to_csv('submission_tta_ls.csv', index=False)
반응형
'DataScience > 딥러닝' 카테고리의 다른 글
| 파이토치(PyTorch) 딥러닝 모델링 전략 (2) - 파이토치 딥러닝 모델링 단계 (0) | 2024.06.30 |
|---|---|
| 파이토치(PyTorch) 딥러닝 모델링 전략 (1) - 모델링 할 때 지녀야 할 마인드셋 (0) | 2024.06.30 |
| 딥러닝 :: 밑바닥부터 시작하는 딥러닝 Chap5. 오차역전파법 (0) | 2023.05.06 |
| 딥러닝 :: 밑바닥부터 시작하는 딥러닝 Chap4. 신경망 학습 (0) | 2023.05.05 |
| 딥러닝 :: 밑바닥부터 시작하는 딥러닝 Chap3. 신경망 (0) | 2023.05.03 |
댓글