이번에는 훈련한 모델을 저장하는 방법과 불러오는 방법에 대해 알아보려고 한다. 1편에서 사용한 코드를 거의 그대로 사용하며, Trainer선언 부분만 바뀌는 정도이며, 전체 코드는 깃허브에 있으니 참고하기 바란다.

요약

• 모델을 특정경로에 체크포인트 저장하기
• 모델의 하이퍼파라미터 저장하기
• 체크포인트로 모델을 불러오기

모델을 특정경로에 체크포인트 저장하기

모델을 저장하는 것은 각 epoch마다 자동으로 저장된다. Jupyter 기준으로 설명하면, Jupyter 파일의 같은 디렉터리 내에 lightning_logs폴더에 각 validation_step마다 저장되는 것을 알 수 있다. 하지만, 나는 여기서 특정 경로에 저장하는 방법에 대해 소개하려고 한다.

# 이전 코드
trainer = pl.Trainer(max_epochs=2)

위의 코드는 1편에서 사용된 코드이다. 여기서 저장할 특정경로를 지정하지 않았기 때문에 실행파일과 동일한 디렉토리에 모델이 저장된다. 

trainer = pl.Trainer(
    max_epochs=2,
    # 저장 경로 지정
    default_root_dir='ckpt/'
    )

위의 코드는 수정된 코드이며, 'ckpt/' 경로에 모델이 저장되는 것이다. 그림으로 살펴보자.

ckpt 폴더 내에 무언가 저장된 것을 볼 수 있는데 각각 요소에 대해 설명하겠다.

• version_숫자 : n번째에 실행했을때 저장되는 것들이다. 
• hparams.yaml : hyperparameter들이 저장되어 있는 yaml 파일이다. 이에 대해서는 밑에서 다루도록 하겠다.
• metrics.csv : 우리가 측정하려는 metric들이 각 epoch마다 csv 형태로 저장되는 파일이다.
• checkpoints : 모델이 훈련하고 각 epoch가 끝날 때마다 얼마나 훈련되어 있는지 저장되어 있는 파일이다. 이 파일을 통해 어떠한 오류로 실험이 중단되면 저 ~~~.ckpt 파일을 읽어와서 실험을 진행할 수 있다. (그래서 사용한다.)

우리는 모델의 파라미터만 저장하고 하이퍼 파라미터는 저장하지 않았다. 그래서 하이퍼 파라미터를 저장하는 방법에 대해 다루려고 한다. 

모델의 하이퍼파라미터 저장하기

하이퍼 파라미터를 저장하는 방법은 간단하다. 코드 한줄을 추가하면 된다.

class LitAutoEncoder(pl.LightningModule):
    def __init__(self, encoder, decoder):
        super().__init__()
        # 코드 추가
        self.save_hyperparameters()
        
        
        self.encoder = encoder
        self.decoder = decoder

# 코드 중략

self.save_hyperparameters() 이 함수를 하나 추가하면 된다.  그리고 코드를 돌리면(훈련하면) 다음과 같이 나온다.

체크포인트로 모델을 불러오기

model = LitAutoEncoder(Encoder(), Decoder())
trainer = pl.Trainer()

# 수정 코드
trainer.fit(model,
            train_dataloaders=train_loader,
            val_dataloaders=valid_loader,
            # 체크포인트 저장한 경로
            ckpt_path='ckpt/lightning_logs/version_0/checkpoints/epoch=1-step=96000.ckpt')

여기서 Trainer에서는 건드릴 것이 없다. 그 이유는 ckpt파일에 정보가 저장되어 있다. 그래서 필요한 것은 다음과 같다.

• model : 훈련시킬 모델의 틀
• dataloader: 훈련시킬 데이터를 다시 가져와야 함
• ckpt_path = 훈련되어있는 parameter를 불러올 경로

다음 것들을 알면, 이제부터 훈련이 날아가서 처음부터 다시 돌리는 경우가 안 생긴다. 파이토치 라이트닝을 사용한다면, 꼭 이것을 알아주었으면 한다. 다음은 OverFitting을 방지하는 Early stopping 방법에 대해 소개하려고 한다. 

논문 링크
https://arxiv.org/abs/1910.13461

오늘은 바트에 대래 리뷰해보려고 한다. 나는 졸업작품에서 BART를 사용했다. 하지만 Architecture에 대해 모르고 그저 성능이 좋다고 사용했는데, 이번에 기회가 되어서 리뷰를 했었고 그에 대해 정리를 해서 블로그에 업로드해보려고 한다. 그리고 논문에 내용을 최대한 함축해서 핵심만 설명하려고 한다. 

Introduction

BART Architecture가 나오기 이전에 유명한 2가지 모델이 BERT(Bidirectional Encoder Representations from Transformer)와 GPT(Generative Pre-Trained Transformer)이었다. 이에 대해 간략하게 소개해본다.

 BERT

자세한 설명은 여기를 참고하기 바란다. 

특징은 다음과 같다.

• Transformer의 Encoder 부분만 이용
• 양방향으로 Pre-Training하도록 2가지 기법을 사용
  • MLM(Masked Language Model) : Token을 임의적으로 Masking 하여 Mask 토큰 예측
  • NSP(Next Sentence Prediction) : 2개의 문장이 다음 문장이 맞는지 다른지 예측
• Classification 관련 Task에 SOTA를 달성했음
• Generation Task를 수행할 수 없음(Encoder만 있는 Architecture의 한계)

GPT

특징은 BERT와 반대라고 생각하면 된다.

• Transformer의 Decoder 부분만 이용
• 단방향으로 Pre-training하도록 기법을 사용
  • Auto-Regressive : 입력과 출력이 동일하게 예측하도록 함
• Classification Task 보단, Generation Task(Summarization)에 특화

About BART

BART(Bidirectional and Auto-Regressive Transformers)는 BERT와 GPT를 합친 Seq-2-Seq Architecture이다. BART가 다른 모델과 다른 특징으로는 다음과 같다.

• Seq-to-Seq Architecture : Sequence 입력 시 Sequence 출력
• Denoising AutoEncoder : 입력에 Noise를 추가해서 Encoder에 입력하고 노이즈를 제거하여 Decoder가 출력
• BERT와 다르게 Encoder의 Final Hidden Layer에 FFNN(Feed Forward Neural Network)가 존재하지 않음
• BERT보다 10% 많은 Parameter의 수

BERT와 GPT와 다른게 크게 4가지 특징이 BART의 핵심이라고 볼 수 있다. 다음은 BART의 Pre-training 기법과 Fine-tuning 기법에 대해 살펴보겠다.

BART Pre-training

Pre-training 기법은 5가지가 있다.

원래 Text : ABC. DE.    

Mask 토큰은 언더바('_')로 표기한다.

Token Masking

BERT의 Pre-training 방식과 동일하다. 특정 Token을 Mask 토큰으로 대체해서 Mask Token을 맞추는 방식이다.

예시 : ABC.DC. → A_C. _E.

Token Deletion

특정 Token을 삭제하는 방식이다. 삭제한 Token을 맞추면서 훈련을 하는 방식이다.

예시 : A.C.E.

Token Infilling

Token Masking과 비슷하지만 다른 점이 있다. 일단 포아송 분포에 따른 난수를 정한다. 여기서 나온 수를 Span 범위로 정한 뒤, Masking 하는 방식이다. 만약 난수가 0인 경우는 <Mask> Token을 추가하는 방식이다.

예시 : Span이 각각 2와 0이라고 하면, BC를 _로 대체하며, D와 E 사이 <MASK> 토큰을 추가한다. → A_.D_E.

Sentence Permutation

말 그대로 문장을 회전하는 방식이다. 일단, 문장을 정하는 기준은 온점('.')을 기준으로 정한다. 그리고 문장을 임의로 섞는다.

예시 : ABC. 문장과 DE. 문장을 온점으로 나눈 뒤 섞으면 → DE.ABC. 이러한 방식으로 나온다.

Document Rotation

문서를 특정한 위치를 기준으로 회전시키는 방식이다. 

예시 : B와 C 사이를 기준으로 문서를 회전시킨 것이다.

BART Fine-tuning

방식은 총 4가지가 있다.

Sequence Classification Tasks

방식은 그림과 같으며 문장이 주어졌을 때, 문장의 문법이 합당했는지 아니면 문장의 감정을 분류하는 것이 있다. Decoder Final Hiddne Layer가 multi-class Linear Classifier에 입력으로 넣어서 분류하도록 한다. 그 이유는 Encoder, Decoder의 모든 정보를 참조하도록 하기 위함이다.

Encoder, Decoder 입력: 동일

Token Classification Tasks

방식은 그림과 같으며, Decoder Final Hidden State를 참고하는 것이다. Sequence Classification Task와 다른 것은 Token 단위로 Classification을 수행한다는 것이다. 예시로는 NER(Named Entity Recognition)등이 있다. 

Encoder, Decoder 입력: 동일

Sequence Generation Tasks

방식은 위의 그림이며, 노이즈가 추가되어 있는 입력을 복원하는 방식이다.

Encoder입력 : 노이즈 추가된 Text, Decoder 입력: 노이즈 제거된 Text 

Machine Translation

Encoder의 Embedding Layer만 매개변수가 초기화되어있는 Encoder로 교체 후 학습을 진행한다. 그 이유에 대해는 다음과 같다. Embedding Layer를 제외하고는 언어에 대해 이해를 하고 있다. 그래서 다른 Layer는 교체하지 않으며 Embedding Layer만 교체하는 이유는 특정 다른 언어와 영어를 매핑하기 위함이다. 한국어로 예시를 들어보자.

예시 : '나는 학생이다' 문장을 'I am student' 문장으로 매핑하기 위함이다. '나는' 이 단어와 'I'라는 단어 매핑, '학생'이란 단어를 'student'단어에 매핑하기 위함이다.

이러한 이유로 Embdding Layer를 교체한 뒤, self-attention input projection matrix를 학습시키는 데 이것도 특정 언어에 대해 이해를 시키기 위해 학습시킨다. 

2가지 과정을 거치면 특정언어(예시 : 한국어)가 영어로 단어별로 매핑이 되었으며, 그래서 적은 수의 반복으로 End-to-End로 학습시킨다.

Comparing Pre-training Objectives

여기에서는 noising 기법에 대해 알기 위해 여러 Architecture 간 비교를 진행한다. Architecture에 대해 간단하게 소개한다.

Model

• Language Model GPT와 유사한, Transformer의 Decoder 구조
• Permuted Language Model : XLnet구조와 매우 유사
• Masked Language Model : BERT구조와 매우 유사
• Multitask Masked Language Model : UniLM기반, Mask Token 예측 및 NER 동시에 예측하는 방식으로 훈련한 모델
• Masked Seq-to-Seq : 입력에서 Span Masking 후 Mask 토큰 예측

Task

1. SQuAD : Wikipedia 기반 extractive question answering task
2. MNLI : 한 문장이 다른 문장을 함축하는지에 대한 task
3. ELI5 : 긴 형식의 extractive question answering task
4. XSUM : 뉴스 요약 task
5. ConvAI2 : 대화의 답변에 대한 task
6. CNN/DM : 뉴스 요약 task
7.  

Results

결과는 몇 가지로 나온다.

Token masking은 필수적이다. 문서를 회전시키거나 문장을 순열하는 데 기반한 사전 학습 목표는 단독으로는 성능이 좋지 않으며, 좋은 방법은 token deletion, masking 그리고 self-attention mask를 사용한다.

Left-to-Right 방식의 Training은 성능이 좋다. Masked Language Model, Permuted Language Model은 Generation Task에서 성능이 떨어지기 대문이다.

양방향 Encoder는 SQuAD Task에서 필수적이다. 그 이유로는 미래의 Context도 중요하기 때문이다.

BART가 다른 모델과 달리 일관된 좋은 성능을 달성한다. 

Large-scale Pre-training Experiments

GPT 논문에서는 대규모의 Corpus를 사용하여 훈련을 하는 경우, 데이터가 많을수록 성능이 향상되는 것이 밝혀졌으며, 대용량의 데이터로 Training을 하고서 다른 모델과 비교를 진행한다.

Training Info

• Dataset : 160GB 크기의 뉴스, 책, 이야기, 웹 텍스트
• Batch size : 8,000
• Step : 5e6
• Dropout : 10%를 적용하며, 마지막 10%의 training step에서 dropout 적용 X

Results

Classification Task

다음은 Classification Task와 관련된 지표이다. 각 Task별로 SOTA(State-of-the-art)를 달성하는 것은 RoBERTa와 BART가 번갈아가며 성능을 달성하는 것을 통해 그게 성능이 차이는 없는 것을 알 수 있다.

Generation Task

다음은 Generation Task와 관련된 지표이다. 여기서는 BART가 모두 SOTA를 달성한 것을 볼 수 있다. 그 이유로는 BART의 Encoder의 경우 다른 모델보다 정보를 더욱 잘 갖고 있으며 다른 훈련이 Decoder의 성능을 다른 모델보다 이해를 잘하고 있는 것이다. 

Conclusion

논문에서 나온 결론이 아닌 내 결론은 이렇다. Classification과 같은 종류의 Task들은 BERT류를 사용할 것 같다. 아무래도 성능은 크게 차이 나지 않으며 Parameter의 수는 10% 정도 적으니 굳이 BART를 사용하지 않을 것 같다. 대신 Generation Task에서는 BART류를 무조건 사용할 것 같다.

오늘은 파이토치 라이트닝과 관련해 기본적인 자료가 부족해서 직접 작성해보기로 했다. 공식 홈페이지를 기반으로 했으니, 많이 도움이 될 것이라고 생각한다. 코드는 여기를 참고하면 된다.

대략적인 구조는 다음과 같다. 

1. 데이터를 모아서 DataLoader를 이용해 train, valid, test로 분할한다.
2. Model을 선언한다.
3. 훈련때마다 사용하는 방법을 선언한다. (e.g., train_step에서는 어떻게 진행하고 ...)
4. Model과 훈련 방법을 Pytorch Lightning Module에 선언한다.
5. Hyper Parameter + Lighitning Module을 Trainer에 추가한다.
6. Trainer를 통해 훈련된 모델(Trained Model)을 사용한다.

요약

• 라이브러리 설치
• 데이터셋 선언
• 모델 구조 만들기
• Lighitning Module 만들기
• Trainer를 이용해 훈련 및 테스트하기

5가지 단계로 구성되어 있다.

라이브러리 설치 및 선언

pip install torch==2.3.1
pip install torchvision==0.18.1
pip install pytorch_lightning==2.3.3

# 필요한 라이브러리 선언

import os
import torch
import numpy as np
import matplotlib.pyplot as plt
from torch import nn
import torch.nn.functional as F
from torchvision import transforms
from torchvision.datasets import MNIST
from torch.utils import data
import pytorch_lightning as pl

데이터셋 선언(분할) 및 확인

dataset은 손글씨로 유명헌 MNIST 데이터셋을 사용한다. 자세한 정보는 여기를 확인하면 된다.

데이터셋 선언 및 분할

# 데이터 셋 불러오기
train_set = MNIST(os.getcwd(), download=True, train=True, transform=transforms.ToTensor())
test_set = MNIST(os.getcwd(), download=True, train=False, transform=transforms.ToTensor())

# 데이터셋 분할
train_set_size = int(len(train_set) * 0.8)
valid_set_size = len(train_set) - train_set_size

# 8 : 2 비율로 train_set과 valid_set 분할
train_set, valid_set = data.random_split(train_set, [train_set_size, valid_set_size])

데이터 확인

image, label = train_set[0]
image = image.numpy()
image = image.squeeze()  # 차원 축소

# 이미지 시각화
plt.imshow(image, cmap='gray')
plt.title(f"answer = {label}")
plt.show()

데이터로더 변환

# DataLoader 변환
train_loader = data.DataLoader(train_set)
valid_loader = data.DataLoader(valid_set)
test_loader = data.DataLoader(test_set)

모델 만들기(Encoder, Decoder)

class Encoder(nn.Module):
    def __init__(self):
        super().__init__()
        self.l1 = nn.Sequential(nn.Linear(28 * 28, 64), nn.ReLU(), nn.Linear(64, 3))

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


class Decoder(nn.Module):
    def __init__(self):
        super().__init__()
        self.l1 = nn.Sequential(nn.Linear(3, 64), nn.ReLU(), nn.Linear(64, 28 * 28))

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

Encoder 

• 목적: 입력 데이터를 저차원의 표현으로 변환(인코딩).
• 구조:
  • nn.Linear(28 * 28, 64): 입력 데이터의 차원을 784(28x28)에서 64로 줄이는 선형 변환을 수행한다.
  • nn.ReLU(): 비선형 활성화 함수로, 선형 변환 후 적용되어 모델의 표현력을 증가시킨다.
  • nn.Linear(64, 3): 중간 표현의 차원을 64에서 3으로 더 줄여 저차원 특징을 추출한다.

Decoder 클래스

• 목적: 저차원 표현을 다시 원래의 차원으로 확장하여 데이터를 복원(디코딩).
• 구조:
  • nn.Linear(3, 64): 저차원 표현(3차원)을 중간 차원(64차원)으로 확장하는 선형 변환을 수행한다.
  • nn.ReLU(): 비선형 활성화 함수를 적용하여 데이터의 비선형적 특징을 유지 및 강화한다.
  • nn.Linear(64, 28 * 28): 최종적으로 중간 차원을 원래의 입력 데이터 차원(784차원)으로 확장하여 재구성한다.

Lighitning Module 만들기

class LitAutoEncoder(pl.LightningModule):
    def __init__(self, encoder, decoder):
        super().__init__()
        self.encoder = encoder
        self.decoder = decoder

    def training_step(self, batch, batch_idx):
        # training_step defines the train loop.
        x, _ = batch
        x = x.view(x.size(0), -1)
        z = self.encoder(x)
        x_hat = self.decoder(z)
        loss = F.mse_loss(x_hat, x)
        return loss

    def validation_step(self, batch, batch_idx):
        # this is the validation loop
        x, _ = batch
        x = x.view(x.size(0), -1)
        z = self.encoder(x)
        x_hat = self.decoder(z)
        val_loss = F.mse_loss(x_hat, x)
        self.log("val_loss", val_loss)
    
    def test_step(self, batch, batch_idx):
        # this is the test loop
        x, _ = batch
        x = x.view(x.size(0), -1)
        z = self.encoder(x)
        x_hat = self.decoder(z)
        test_loss = F.mse_loss(x_hat, x)
        self.log("test_loss", test_loss)

    def configure_optimizers(self):
        optimizer = torch.optim.Adam(self.parameters(), lr=1e-3)
        return optimizer

각각 함수에 대한 설명과 코드 해석을 하면 다음과 같다.

• __init__(self, encoder, decoder): 생성자에서는 인코더와 디코더 객체를 모듈의 속성으로 저장한다.
• training_step : 훈련 step마다 실행되며, x 입력에 대해 인코딩, 디코딩과정을 통해 정답과 예측에 대해 loss 계산한다.
• validation_step : 검증 step마다 실행되며, training_step과 동일하다.
• test_step : 테스트 step에서 실행되며, training_step과 동일하다.
• configure_optimizers : 학습 과정에서 사용할 옵티마이저를 설정한다. 필요하다면 학습률 스케줄러도 설정할 수 있지만, 여기서는 0.001로 설정한다.

모듈을 이용한 모델 선언

# model
model = LitAutoEncoder(Encoder(), Decoder())

Trainer로 훈련 및 테스트 하기

Trainer 선언

trainer = pl.Trainer(max_epochs=5)

Trainer 내부에 Hyper Parameters 선언하지만, 지금은 간단하게 epoch만 설정했다.

Trainer로  모델 훈련하기

trainer.fit(model, train_loader, valid_loader)

trainer의 fit 함수를 이용하여 train_loader와 valid_loader를 이용해 model 훈련한다.

• train_loader : training_step에서 사용된다.
• valid_loader : validation_step에서 사용된다.

실행하면 다음과 같이 모델 정보(파라미터 수, GPU 정보) 와 함께 실험 예상시간이 출력된다.

Trainer로 모델 테스트하기

trainer.test(model, dataloaders=test_loader)

실행하면 실행에 대한 결과가 출력된다. 여기서 코드 작성을 통해 원하는 metric을 추가할 수 있다.

다음은 훈련된 모델을 불러오고 사용하는 방법에 대해 알아보려고 한다.

논문 링크

https://aclanthology.org/W04-1013/

ROUGE

Rouge은 Recall-Oriented Understudy for Gisting Evaluation의 약자이다. ROUGE는 사람이 만든 참조 요약과 컴퓨터가 생성한 요약 사이의 일치 정도를 측정한다. 이 논문은 ROUGE가 포함된 네 가지 다른 ROUGE 메트릭, 즉 ROUGE-N, ROUGE-L, ROUGE-W, ROUGE-S를 소개합니다. 각 메트릭은 요약의 다른 측면을 평가한다.

• ROUGE-N: 이 메트릭은 N-gram 기반으로 평가한다. 이는 요약에서 사용된 단어의 연속적인 나열이 참조 요약과 얼마나 일치하는지를 측정하여, 텍스트의 어휘적 정확성을 평가한다. N이 클수록, 문맥적 일관성과 정확성을 더 자세히 분석한다.
• ROUGE-L: 가장 긴 공통 부분 문자열(LCS)을 기반으로 하여, 요약문과 참조 요약문 사이에 나타나는 최장 공통 부분문을 찾아냄으로써 요약의 문장 구조적 유사성을 평가한다. 
• ROUGE-W: 가중치가 있는 가장 긴 공통 부분 문자열을 사용한다. 이는 LCS의 기본 아이디어에 가중치를 추가하여, 더 긴 일치하는 구문을 더 높게 평가함으로써 요약의 문맥적 심도와 정보의 중요성을 평가한다.
• ROUGE-S: ‘skip-bigram’을 사용하여, 요약 내 단어들 사이의 임의의 간격에도 불구하고 겹치는 단어 쌍을 찾아내어 요약의 유연성과 단어 간 관계를 평가한다. 이 메트릭은 요약이 원문의 중요한 개념들을 얼마나 포괄적으로 포함하고 있는지를 측정한다.

Formula

ROUGE-N

여기서 $ Count_{match}(gram_n)$은 생성된 요약과 참조 요약에서 발견된 겹치는 n-gram의 수이며, $Count(gram_n)$은 참조 요약에서 n-gram의 수이다.

$$\text{ROUGE-N} = \frac{\sum_{s \in \{ \text{참조 요약} \}} \sum_{gram_n \in s} \text{Count}_{\text{match}}(gram_n)}{\sum_{s \in \{ \text{참조 요약} \}} \sum_{gram_n \in s} \text{Count}(gram_n)}$$
​

예시 1: (ROUGE-1 사용)

• 참조 요약: "그녀는 학교에 간다."
• 생성된 요약: "그녀는 집에 간다."
• 일치하는 unigram: "그녀는", "간다"
• 계산: $\frac{2}{3} ≈  0.67$ (3개 중 2개 일치)

예시 2: (ROUGE-2 사용)

• 참조 요약: "고양이는 소파에서 잔다."
• 생성된 요약: "고양이는 소파에서 놀고 있다."
• 일치하는 bigram: "고양이는 소파에서"
• 계산: $\frac{1}{2} = 0.5$(2개 중 1개 일치)

ROUGE-L

$$\text{ROUGE-L} = \frac{\text{LCS}(X, Y)}{\max(\text{len}(X), \text{len}(Y))}$$
​ 여기서 $X$와 $Y$는 각각 참조 요약과 생성된 요약의 단어 시퀀스를 나타내며, $LCS(X,Y)$는 두 시퀀스 간의 가장 긴 공통 부분 시퀀스의 길이이다.

예시 1:

• 참조 요약: "그는 그녀에게 책을 준다."
• 생성된 요약: "그는 책을 그녀에게 준다."
• LCS 길이: 4 (단어 순서 다름)
• 계산: $\frac{4}{5} = 0.8$

예시 2:

• 참조 요약: "아침에 커피를 마신다."
• 생성된 요약: "커피를 마신다."
• LCS 길이: 2
• 계산:$\frac{2}{3} ≈ 0.67 $

ROUGE-W

$$\text{ROUGE-W} = \frac{\text{WLCS}(X, Y)}{\max(\text{len}(X), \text{len}(Y))}$$
​ 여기서 $WLCS$는 가중치가 있는 $LCS$를 의미하며, 더 긴 일치하는 시퀀스에 더 큰 가중치를 준다.

예시 1:

• 참조 요약: "아침에 커피를 마시고 신문을 읽는다."
• 생성된 요약: "아침에 커피를 마시고 뉴스를 본다."
• WLCS 계산은 LCS에 가중치를 추가해야 하지만 간단히 LCS를 사용하고 장문 일치에 더 큰 가중치 부여
• 계산: 긴 일치 부분, 추정 가중치 추가 후 계산 필요

예시 2:

• 참조 요약: "그는 아침 조깅을 한 후에 샤워를 했다."
• 생성된 요약: "그는 샤워를 한 후에 조깅을 했다."
• WLCS: 가중치가 있는 LCS 계산, 여기서도 실제 가중치 계산을 위해서는 추가 데이터 필요

ROUGE-S

$$\text{ROUGE-S} = \frac{\sum_{\text{skip-bigram} \in \{ \text{참조 요약} \}} \text{Count}_{\text{match}}(\text{skip-bigram})}{\sum_{\text{skip-bigram} \in \{ \text{참조 요약} \}} \text{Count}(\text{skip-bigram})}$$
​ 여기서 $skip-bigram$은 두 단어가 연속적이지 않고, 문장 내에서 임의의 다른 단어들에 의해 분리될 수 있는 단어 쌍을 의미한다. $Count_{match}(skip-bigram)$은 생성된 요약과 참조 요약 모두에 나타나는 $skip-bigram$의 수이며, $Count(skip-bigram)$은 참조 요약 내의 $skip-bigram$ 수 이다

예시 1:

• 참조 요약: "그녀는 시장에서 과일을 샀다."
• 생성된 요약: "그녀는 과일을 시장에서 구입했다."
• 일치하는 skip-bigram: "그녀는-과일을", "과일을-시장에서", "그녀는-시장에서"
• 계산: $\frac{3}{3} = 1.0$

예시 2:

• 참조 요약: "아침에 커피를 마시고 신문을 읽는다."
• 생성된 요약: "신문을 읽으며 커피를 마신다."
• 일치하는 skip-bigram: "커피를-신문을", "아침에-커피를"
• 계산: $\frac{2}{4} = 0.5 $

이 예시를 통해 도움이 되시기를 바랍니다. 감사합니다

Introduction

Pre-trained Language Model(PLM) 장점

• 상당한 지식 습득
• 외부 메모리 접근 없이 지식 풍부

PLM 단점

• Hallucination
• 메모리의 확장 및 수정 불가능

PLM의 단점을 위해 parametric memory + non-parametric memory → Seq2Seq Model을 설계

• Parametric memory (e.g., Logistic Regression, Linear Regression)
  • Pre-trained seq2seq transformer
• Non-parametric memory(K-NN, Decision Tree, Random Forest)
  • dense vector index of Wikipedia

여기서 Retriever는 Top-K approximation 방법을 사용해 문서를 검색한다.

Methods

Model

• $x$ : Input sequence
• $y$ : target sequence
• $z$ : text document

RAG Model : $x$를 사용하여 $z$를 검색하고, 시퀀스 $y$를 생성할 때 사용하는 RAG 모델에 대한 탐구하며 모델은 두 가지 구성 요소를 활용

1. $x$에 대해 텍스트(Top-K)의 분포를 반환하는 매개변수 $η$를 가진 검색기 $p_η(z|x)$
2. 원래 입력 $x$와 검색된 구절 $z$, 이전 $i-1$ 토큰 $y_{1:i−1}$의 컨텍스트를 기반으로 현재 토큰을 생성하는 매개변수 $θ$로 매개화된 생성기 $p_θ(y_i|x, z, y_{1:i−1})$

Generator와 Retriever을 훈련(end-to-end)하기 위해 검색된 문서를 잠재 변수 취급하며, 생성된 텍스트에 대해 평가하기 위해 검색된 문서들을 두 가지 방식으로 사용하는 모델을 제안

RAG-Sequence Model

검색된 동일한 문서를 이용하여 전체 Sequence 생성. 검색된 문서를 단일 변수로 취급하고 나머지 문서에 대해 Sequence 예측

$$ p_{\text{RAG-Sequence}}(y|x) \approx \sum_{z \in \text{Top-K}(p(z|x))} p_{\eta}(z|x) p_{\theta}(y|z, x) = \sum_{z \in \text{Top-K}(p(z|x))} p_{\eta}(z|x) \prod_{i} p_{\theta}(y_i|z, x, y_{1:i-1}) $$

RAG-Token

생성되는 토큰마다 문서를 추출함. 이는 Generator가 토큰마다 문서를 다시 검색하고 검색된 문서를 바탕으로 다음 토큰을 예

$$ p_{\text{RAG-Token}}(y|x) \approx \prod_{i} \sum_{z \in \text{Top-K}(p(z|x))} p_{\eta}(z|x) p_{\theta}(y_i|z, x, y_{1:i-1}) $$

Retriever: DPR

Retrieval component $p_η(z|x)$ 는 DPR [26]을 기반으로 하며 bi-encoder 구조

$$ p_{\eta}(z|x) \propto \exp(d(z)^T q(x)) $$

$$d(z) = \text{BERT}{d}(z), \quad q(x) = \text{BERT}{q}(x) $$

• $d(z)$ : $BERT_{BASE}$ (Document Encoder)에 의해 생성된 문서의 Dense representation
• $q(x)$ : $BERT_{BASE}$(Query Encoder)에 의해 생성된 Quary Represenation

Query $x$ 에 대해 가장 높은 사전 확률 $p_η(z|x)$을 가진 $k$개의 문서 $z$ 목록인 top-k$(p_η(·|x))$를 계산하는 것은 최대 내적 검색(Maximum Inner Product Search, MIPS)을 이용(선형 시간내에 대략적 해결)

Generator: BART

Generator Component $p_\theta(y_i|x,z,y_{1:i-1})$는 Encoder-Decoder을 이용하여 모델링

논문에서는 400M 크기의 Pre-trained BART-large 사용

BART input = 입력 $x$와 생성된 문서 $z$를 단순히 연결

Training

• Retriever & Generator은 문서 검색에 대한 Direct supervision없이 Training을 진행.
• 입력/출력 쌍$(x_j, y_j)$으로 구성된 Training Corpus를 사용하여, 각 대상에 대해 $\sum_j - log p(y_j | x_j)$을 최소화하는 방식으로 훈련하며 최적화 기법은 Adam을 사용
• Document Encoder ($BERT_d$) 업데이트를 진행하지 않으며, Query Encoder($BERT_q$)와 Generator($BERT_{large}$)만 Fine-tunning을 진행

Experiments

1. $BERT_{BASE}$를 이용한 Document Encoder를 사용해 Wikipedia 기사를 임베딩
2. Facebook AI Similarity Search(FAISS)를 사용하여 single Maximum Inner Product Search(MIPS)를 통해 Top-K Document Search 함

Dataset

1. Open-Domain QA

1. Natural Question (NQ)
2. TriviaQA(TQA)
3. WebQuestion(WQ)
4. CuratedTrec(CT)

4개의 데이터셋을 이용하여 RAG를 훈련

2. Abstractive QA

https://huggingface.co/datasets/ms_marco

MSMARCO NLG Task v2.1을 사용함.

제공된 사이트를 사용하지 않고 질문과 답변을 할 수 없는 질문이 있으며, 일부 질문은 위키백과만을 사용하여 답변할 수 없습니다

→ RAG는 매개변수의 지식에 의존해 합리적인 응답 생성

3. Jeopardy QG

https://huggingface.co/datasets/search_qa

Jeopardy는 주어진 Entity에 대해 Entity를 추측하는 질문을 생성하는 독특한 데이터셋

e.g.,) “월드컵” ⇒ “1986년 멕시코가 처음으로 이 국제 스포츠 대회를 두 번째로 개최한 나라로 기록되었습니다.”

4. Fact Extraction and VERification(FEVER)

https://huggingface.co/datasets/fever

주장에 대해 반박할 충분한 정보가 있는지 없는지 분류 → 위키백과로 검증 불가능한지 분류

⇒ RAG이 생성이 아닌 분류를 처리할 수 있는 능력이 있는지 테스트

Results

1. Open-Domain QA

RAG = Closed Book + Open Book

→ REALM 및 T5 + SSM과 같이 비용이 많이 드는 Span masking 없이도 강력한 결과 산출

2. Abstractive QA

RAG는 최첨단 모델 성능에 다음과 같은 이유를 바탕으로 우수한 성능을 보여줌

1. RAG는 Gold access 없이 좋은 성능을 보여줌
2. 많은 질문은 Gold access 없이는 답변하기 어려움
3. 모든 질문이 Wikipedia로만 답변 가능한 것이 아님

RAG가 BART보다 Question Generation에서 Hallucination에 강하다.

3. Jeopardy QG

Jeopardy 질문에는 종종 두 개의 별개 정보가 포함되어 있는데, RAG-Token이 여러 문서의 내용을 결합한 응답을 생성할 수 있기 때문에 가장 잘 수행될 수 있다.

4. Fact Verification

3-way 분류에서, RAG 점수는 domain-specific Architecture 및 substantial engineering, trained using intermediate retrieval supervision 과 약 4.3% 범위 내의 최첨단 모델과 일치합니다.(RAG는 다음과 같은 복잡한 과정 X)

• 1편 : 프로젝트 요약 및 LightningModule 설계
• 2편 : Dataset 구축 및 DataLoader 구축
• 3편 : logger 작성, wandb 연동 및 확인
• 4편 : 모델 로딩 및 실행결과(wandb) 확인

이번에는 작성한 코드를 이용해서 모델 로딩 및 실행결과(wandb) 확인을 해보겠습니다.

환경설정 및 초기화

사용할 모델 및 토크나이저 선택

1편을 보시면 알겠지만, 저는 한국어로 입력했을 때, 영어로 출력하도록 Task를 만들었습니다. 그래서 저는 한국어-영어 쌍으로 훈련된 모델을 허깅페이스에서 가져와서 사용했습니다. 궁금하신분은 링크를 클릭해보시면 됩니다. 간단히 말하면 한국어를 입력했을 때, 영어로 출력하는 KoBART 모델입니다.

model_id = 'chunwoolee0/circulus-kobart-en-to-ko'

model = AutoModelForSeq2SeqLM.from_pretrained(args['model_path'], cache_dir='/home/hgjeong/hdd1/hub')
tokenizer = AutoTokenizer.from_pretrained(args['tokenizer'])

하이퍼 파라미터 설정

args = dict()
args['output_dir'] ='20240512'
args['model_path'] = 'chunwoolee0/circulus-kobart-en-to-ko'
args['tokenizer'] = 'chunwoolee0/circulus-kobart-en-to-ko'

args['gpus'] = '1'
args['epochs'] = 100
args['max_learning_rate'] = 2e-5
args['min_learning_rate'] = 1e-6
args['warmup_rate'] = 0.1
args['max_seq_len'] = 128
args['batch_size_train'] =64
args['batch_size_valid'] =8
args['logging_interval'] =100
args['evaluate_interval'] =1.0
args['seed'] =93
args['wandb_project'] ='blog'
args['accumulate_grad_batches'] = 1

• output_dir: 모델과 출력 파일들을 저장할 디렉토리 경로 ('20240512')
• model_path: 사용할 모델의 경로
• tokenizer: 사용할 토크나이저의 경로
• gpus: 훈련에 사용할 GPU 수
• epochs: 훈련할 총 에폭 수 (100)
• max_learning_rate: 최대 학습률 (2e-5)
• min_learning_rate: 최소 학습률 (1e-6)
• warmup_rate: 학습률 웜업 비율 (0.1), 학습 초기에 학습률을 점진적으로 증가시키는 비율
• max_seq_len: 입력 시퀀스의 최대 길이 (128), 너무 긴 입력은 잘라내기 처리
• batch_size_train: 훈련 데이터의 배치 크기 (64)
• batch_size_valid: 검증 데이터의 배치 크기 (8)
• logging_interval: 로깅 간격 (100), 이 간격으로 훈련 로그를 기록
• evaluate_interval: 평가 간격 (1.0), 각 에폭마다 모델 성능을 평가
• seed: 랜덤 시드 (93)
• wandb_project: Weights & Biases에서 사용할 프로젝트 이름 ('blog'), 실험 관리 및 모니터링을 위해 설정
• accumulate_grad_batches: 그래디언트 누적 배치 수

데이터 로딩

train_dataloader, valid_dataloader = OrderDataLoader("../test.jsonl", tokenizer, args['batch_size_train'], args['batch_size_valid'], args['max_seq_len'])

훈련 준비(pl.LightningModule 및 pl.Trainer 초기화)

model = AutoModelForSeq2SeqLM.from_pretrained(args['model_path'])

lightning_module = StoryModule(
    model,
    args['output_dir'],
    total_steps,
    args['max_learning_rate'],
    args['min_learning_rate'],
    args['warmup_rate'],
)

하이퍼 파라미터를 Module에 입력한다.

trainer = pl.Trainer(
    strategy="auto",
    accelerator="gpu",
    logger=train_loggers,
    max_epochs=args['epochs'],
    log_every_n_steps=args['logging_interval'],
    val_check_interval=args['evaluate_interval'],
    accumulate_grad_batches=args['accumulate_grad_batches'],
    callbacks=[LearningRateMonitor(logging_interval="step")],
    devices=args['gpus'],
)

하이퍼 파라미터를 Trainer에 입력한다.

훈련 시작

trainer.fit(lightning_module, train_dataloader, valid_dataloader)

train_dataloader와 valid_dataloader를 통해 훈련하고 검증을 진행한다.

실행 및 결과확인(wandb)

wandb를 통해 loss(train)와 loss(valid)가 둘 다 epoch가 늘어나면서 줄어드는 것을 확인할 수 있다.

전체 코드

https://github.com/Capstone-Amigo/OrderIt-AI/tree/main/ai

이곳에서 확인할 수 있다. BART 훈련에 도움이 되었으면 하는 마음에 다음과 같이 블로그 글을 적어보았다. 다들 이 글을 통해 시간을 아꼈으면 좋겠다. 감사합니다!

• 1편 : 프로젝트 요약 및 LightningModule 설계
• 2편 : Dataset 구축 및 DataLoader 구축
• 3편 : logger 작성, wandb 연동 및 확인
• 4편 : 모델 로딩 및 실행결과(wandb) 확인

이번에는 훈련이 잘되고 있는지 확인하기 위해 필요한 것들에 대해 알아보려고 합니다. 모델 인수들이 잘 입력되었는지 log에 확인하기 위한 logger 작성과 훈련 loss들이 잘 훈련되고 있는지에 대해 알려주는 wandb, tensorboard에 대해 알아보려고합니다.(tensorboard는 추후에 작성하도록 하겠습니다.)

logger 작성

import logging
import sys

def get_logger(name: str) -> logging.Logger:
    """Return logger for logging

    Args:
        name: logger name
    """
    logger = logging.getLogger(name)
    logger.propagate = False
    logger.setLevel(logging.DEBUG)
    if not logger.handlers:
        handler = logging.StreamHandler(sys.stdout)
        handler.setFormatter(logging.Formatter("[%(asctime)s] %(message)s"))
        logger.addHandler(handler)
    return logger

• logging.getLogger(name) : 이름이 'name'인 로거 객체를 생성하거나, 존재하는 경우 해당 로거를 반환한다.
• logger.setLevel(logging.DEBUG) : 로거의 로그 레벨을 'DEBUG'로 설정
• logger.handler : 로거에 핸들러(콘솔, 파일)가 없는 경우 설정
  • handler.setFormatter(logging.Formatter("[%(asctime)s] %(message)s"))는 로그 메시지의 형식을 설정
    • %(asctime)s는 로그가 기록된 시간, %(message)s는 로그 메시지를 나타낸다.
    • logger.addHandler(handler)는 생성된 핸들러를 로거에 추가

wandb 사용

wandb 로그인 방법은 여기를 클릭하세요.

import wandb
from pytorch_lightning.loggers import WandbLogger

wandb.login()

wandb.login()을 이용해 API KEY를 입력하면 wandb 연동 성공입니다.

이제 마지막으로 모델로딩, 훈련 및 실행결과를 확인해보도록 하겠습니다.

머신러닝 및 딥러닝 프로젝트에서 모델의 추적을 위해 강력한 도구인 wandb(Weights and Biases)를 사용하여 모델의 성능을 극대화하는 방법을 소개하려 한다.

wandb는 무엇이고 왜 사용하는가

wandb(Weights and Biases)는 머신러닝 및 딥러닝 프로젝트에서 모델의 파라미터에 대한 도구로, 실험 결과를 시각화하고, 하이퍼파라미터를 조정하며, 모델의 성능을 모니터링하는 도구이다.   특히, 하이퍼파라미터 최적화를 통해 최적의 모델 설정을 찾는 데 큰 도움이 된다.

wandb 가입 및 로그인

wandb를 사용하기 위해서는 wandb 공식 웹사이트에서 회원가입과 로그인을 해야 합니다. 이메일 주소와 비밀번호를 입력하거나, 구글 계정이나 GitHub 계정으로 손쉽게 가입하고 로그인할 수 있다.

공식사이트에 들어가서 가입버튼을 클릭한다.

나의 경우는 Github로 가입하고 개인정보를 입력한다.

다음과 같은 창이 나오게 되는데 여기서 API KEY를 복사해 두고 저장해 둔다.

wandb 로그인

pip install wandb

wandb 설치

import wandb
wandb.login()

다음과 같이 셀을 실행하면 밑에와 같은 창이 생긴다. 이때, 여기에 복사한 API KEY를 입력한다.

API KEY를 입력하면 다음과 같이 True가 출력된다.

혹시 키를 까먹었다면 여기를 클릭하면 된다.

wandb를 사용하는 방법

훈련코드

import pytorch_lightning as pl
import wandb
from pytorch_lightning.loggers import WandbLogger
import torch
from torch.nn import functional as F
from torch.utils.data import DataLoader, random_split
from torchvision import transforms
from torchvision.datasets import MNIST

# 1. wandb Logger 설정
wandb_logger = WandbLogger(project="pytorch-lightning-demo")

# 2. 데이터 준비
class MNISTDataModule(pl.LightningDataModule):
    def __init__(self, data_dir='./', batch_size=32):
        super().__init__()
        self.data_dir = data_dir
        self.batch_size = batch_size

    def prepare_data(self):
        MNIST(self.data_dir, train=True, download=True)
        MNIST(self.data_dir, train=False, download=True)

    def setup(self, stage=None):
        transform = transforms.Compose([transforms.ToTensor()])
        self.mnist_train = MNIST(self.data_dir, train=True, transform=transform)
        self.mnist_val = MNIST(self.data_dir, train=False, transform=transform)

    def train_dataloader(self):
        return DataLoader(self.mnist_train, batch_size=self.batch_size)

    def val_dataloader(self):
        return DataLoader(self.mnist_val, batch_size=self.batch_size)

# 3. 모델 정의
class LitMNIST(pl.LightningModule):
    def __init__(self):
        super().__init__()
        self.layer_1 = torch.nn.Linear(28 * 28, 128)
        self.layer_2 = torch.nn.Linear(128, 256)
        self.layer_3 = torch.nn.Linear(256, 10)

    def forward(self, x):
        x = x.view(x.size(0), -1)
        x = F.relu(self.layer_1(x))
        x = F.relu(self.layer_2(x))
        x = self.layer_3(x)
        return x

    def training_step(self, batch, batch_idx):
        x, y = batch
        y_hat = self.forward(x)
        loss = F.cross_entropy(y_hat, y)
        self.log('train_loss', loss)
        return loss

    def validation_step(self, batch, batch_idx):
        x, y = batch
        y_hat = self.forward(x)
        loss = F.cross_entropy(y_hat, y)
        self.log('val_loss', loss)
        return loss

    def configure_optimizers(self):
        return torch.optim.Adam(self.parameters(), lr=1e-3)

# 4. 학습
if __name__ == '__main__':
    dm = MNISTDataModule()
    model = LitMNIST()
    trainer = pl.Trainer(logger=wandb_logger, max_epochs=5)
    trainer.fit(model, dm)

위의 코드를 요약하면 다음과 같습니다. 

import wandb
from pytorch_lightning.loggers import WandbLogger

wandb_logger = WandbLogger(project="pytorch-lightning-demo")

trainer = pl.Trainer(logger=wandb_logger, max_epochs=5)

WandbLogger를 선언한 뒤, project 이름을 작성해 준 뒤 trainer 인수로 넣어주고 trinaing을 진행하면 wandb 사이트에 훈련하면서 loss값이 변하는 것을 확인할 수 있다. 

wandb를 통해 빠르게 모델 최적화를 하시기를 바랍니다!

• 1편 : 프로젝트 요약 및 LightningModule 설계
• 2편 : Dataset 구축 및 DataLoader 구축
• 3편 : logger 작성, wandb 연동 및 확인
• 4편 : 모델 로딩 및 실행결과(wandb) 확인

이번에는 Dataset 구축 및 DataLoader 구축에 대해 설명하려고 한다. 근데 생각을 해보았을 땐, 이 글을 보는 사람들은 데이터셋 구축은 관심이 없을 것이고 아무래도 어떻게 코드를 짜는지 궁금할 것이라 생각하기 때문에 데이터셋 구축에 대해서는 설명하지 않고 데이터셋이 어떻게 구성되어 있는지만 말하려고 한다.

Dataset 구축

# Text : Encoder 
# function : Decoder

# 예시
# 음료 개수, 메뉴 개수, 메뉴 개수 -> drink(quantity=1); menu(quantity=3); menu(quantity=2);
{"Text": "아이스 카페 라떼 4잔 주세요 그 외에 따듯한 그린티 라떼 줄래?", "function": "ICE_cafe_latte(quantity=4); HOT_greentea_latte(quantity=1); "}

데이터셋의 구성은 다음과 같다.

• Encoder에는 한국어로 되어있는 Text
• Decoder는 메뉴에 대해 요약된 Text

결론은 카페 메뉴에 대한 text가 들어갔을 때, 영어로 요약된 text가 출력되도록 하는 것이다.

DataLoader 구축

라이브러리

from datasets import Dataset
from torch.utils.data import DataLoader
from tokenizers.processors import TemplateProcessing

라이브러리는 다음과 같이 사용합니다.

데이터셋 로딩

# json을 dataset으로 변환
dataset = Dataset.from_json(fname)

토크나이저 설정

# cls token과 sep token이 설정되어있지 않으면 설정
if not tokenizer.cls_token:
    tokenizer.cls_token = tokenizer.eos_token
if not tokenizer.sep_token:
    tokenizer.sep_token = tokenizer.eos_token

토크나이저에 클래스 시작(cls_token)과 분리(sep_token) 토큰이 설정되어 있지 않다면, 각각을 문장 종료(eos_token) 토큰으로 설정

템플릿 프로세싱(포스트 프로세서 설정)

# TemplateProcessing
tokenizer._tokenizer.post_processor = TemplateProcessing(
    single=f"{tokenizer.cls_token} $0 {tokenizer.sep_token}",
    pair=f"{tokenizer.cls_token} $A {tokenizer.sep_token} $B:1 {tokenizer.sep_token}:1",
    special_tokens=[(tokenizer.cls_token, tokenizer.cls_token_id), (tokenizer.sep_token, tokenizer.sep_token_id)],
)

다음을 하는 이유는 다음과 같다. BART와 같이 모델이 요구하는 입력 형식에 맞추기 위함이다.

예를 들면 다음과 같다.

• 입력 : 아이스 카페 라테 4잔 주세요
• 출력 : [CLS] 아이스 카페 라뗴 4잔 주세요 [SEP]
• 입력 : 아이스 카페 라떼라테 4잔 주세요 그 외에 따듯한 그린티 라테 줘
• 출력 : [CLS] 아이스 카페 라뗴 4잔 주세요 [SEP] 그 외에 따듯한 그린티 라테 줘 [SEP]

다음과 같은 형식으로 처리한다.

전처리 함수

# 전처리 함수
def preprocess_function(examples):
    processed = {}
    inp = f'{examples["Text"]}'

    # 입력에 대한 토큰화
    tokenizer_input = tokenizer(
        inp,
        padding="max_length",
        max_length=max_length,
        truncation=True
    )
    processed["input_ids"] = tokenizer_input["input_ids"]
    processed["attention_mask"] = tokenizer_input["attention_mask"]

    # 훈련모드인 경우
    if mode == "train":
        #print(examples["code"])

        # 출력에 대한 토큰화
        tokenizer_output = tokenizer(
            examples["function"], 
            padding="max_length", 
            max_length=max_length, 
            truncation=True
        )
        processed["decoder_input_ids"] = tokenizer_output["input_ids"]
        processed["decoder_attention_mask"] = tokenizer_output["attention_mask"]

    # 토큰화된 배열 반환
    return processed

요약을 하면, 입력데이터를 BART의 출력에 맞도록 tokenizing 하는 것이라고 생각하면 된다.

• train mode : encoder, decoder 모두 tokenizing
• not train mode : encoder만 tokenizing

그 이유는 테스트 모드의 경우에는 decoder는 예측을 하는 것이기 때문에 정답이 없다.

데이터셋 전처리 적용

# dataset에 대해 전처리 함수로 매핑, columns제거 및 torch tensor로 변환
dataset = dataset.map(preprocess_function,remove_columns=dataset.column_names).with_format("torch")

dataset에 대해 전처리를 적용한 뒤, 이전에 필요 없는 column을 삭제한  뒤, pytorch 형태로 변환한다.

데이터셋 분할

dataset = dataset.train_test_split(0.2)

train_dataloader = DataLoader(dataset['train'], shuffle=True, batch_size=train_batch_size, num_workers=8, pin_memory=True)
valid_dataloader = DataLoader(dataset['test'], shuffle=True, batch_size=valid_batch_size, num_workers=8, pin_memory=True)

return train_dataloader, valid_dataloader

train과 valid를 8 : 2로 분할하는 코드이다. 이렇게 되면 dataset은 'train'과 'test'로 데이터를 8 : 2로 나눈다. 

• shuffle : data를 무작위로 섞는다.
• batch_size : batch_size를 조절한다. ram이 적다면, 낮게설정하고 ram이 많다면 높게 설정하는 것이 좋다.
• num_workers : 데이터 로딩을 위해 사용할 별도의 작업 프로세스 수를 지정, 8이면, 8개의 프로세스가 병렬로 데이터를 로딩한다.
• pin_memory : 데이터를 더 빠르게 GPU로 전송할 수 있게 하여, 특히 GPU를 사용할 때 데이터 전송 병목을 줄여준다.

다음과 같이 분할하여 train_dataloader과 valid_dataloader 형식으로 반환한다.

전체 코드

import json
import pandas as pd

from datasets import Dataset
from torch.utils.data import DataLoader
from tokenizers.processors import TemplateProcessing

def OrderDataLoader(fname, tokenizer, train_batch_size,valid_batch_size, max_length,split=False, mode="train"):
    """
    Build Data Loader

    """

    # json읽어와서 dataset으로 변환
    dataset = Dataset.from_json(fname)
    
    # cls token과 sep token이 설정되어있지 않으면 설정
    if not tokenizer.cls_token:
        tokenizer.cls_token = tokenizer.eos_token
    if not tokenizer.sep_token:
        tokenizer.sep_token = tokenizer.eos_token

    # TemplateProcessing
    tokenizer._tokenizer.post_processor = TemplateProcessing(
        single=f"{tokenizer.cls_token} $0 {tokenizer.sep_token}",
        pair=f"{tokenizer.cls_token} $A {tokenizer.sep_token} $B:1 {tokenizer.sep_token}:1",
        special_tokens=[(tokenizer.cls_token, tokenizer.cls_token_id), (tokenizer.sep_token, tokenizer.sep_token_id)],
    )

    # 전처리 함수
    def preprocess_function(examples):
        processed = {}
        inp = f'{examples["Text"]}'
       
        # 입력에 대한 토큰화
        tokenizer_input = tokenizer(
            inp,
            padding="max_length",
            max_length=max_length,
            truncation=True
        )
        processed["input_ids"] = tokenizer_input["input_ids"]
        processed["attention_mask"] = tokenizer_input["attention_mask"]
        
        # 훈련모드인 경우
        if mode == "train":
            #print(examples["code"])
             
            # 출력에 대한 토큰화
            tokenizer_output = tokenizer(
                examples["function"], 
                padding="max_length", 
                max_length=max_length, 
                truncation=True
            )
            processed["decoder_input_ids"] = tokenizer_output["input_ids"]
            processed["decoder_attention_mask"] = tokenizer_output["attention_mask"]
        
        # 토큰화된 배열 반환
        return processed
    

    # dataset에 대해 전처리 함수로 매핑, columns제거 및 torch tensor로 변환
    dataset = dataset.map(preprocess_function,remove_columns=dataset.column_names).with_format("torch")

    # dataset을 dataloader로 변환

    dataset = dataset.train_test_split(0.2)

    train_dataloader = DataLoader(dataset['train'], shuffle=True, batch_size=train_batch_size, num_workers=8, pin_memory=True)
    valid_dataloader = DataLoader(dataset['test'], shuffle=True, batch_size=valid_batch_size, num_workers=8, pin_memory=True)

    return train_dataloader, valid_dataloader

• 1편 : 프로젝트 요약 및 LightningModule 설계
• 2편 : Dataset 구축 및 DataLoader 구축
• 3편 : logger 작성, wandb 연동 및 확인
• 4편 : 모델 로딩 및 실행결과(wandb) 확인

오늘은 이전에 캡스톤디자인에서 BART를 이용해서 Text를 요약하는 Task를 진행했었는데, 이것에 대해 설명하면서 코드를 어떻게 작성했는지 설명하려고 한다. 

간단하게 요약하면 다음과 같다. 카페 메뉴에 대한 Text가 입력이 되면, BART를 이용해서 영어로 번역한 뒤 필요한 내용만 출력하는 것이다. 아이스 아메리카노 하나, 카페라떼 하나를 영어로 요약하는 것이다. 다음과 같은 Task를 위해 BART를 훈련시키는 코드에 대해 설명하겠다.

필요한 라이브러리 설치

pip install datasets==2.17.0
pip install pytorch_lightning==2.2.4
pip install torkenizers==0.15.2
pip install transformers==4.38.1
pip install pandas==2.0.3

위의 라이브러리는 제가 사용한 라이브러리들의 버전이다. 혹시 버전오류가 발생한다면 다음과 같이 라이브러리를 재설치하면 된다.

훈련 모듈 설계

요약

모델 초기화

def __init__(
        self,
        model,
        model_save_dir,
        total_steps,
        max_learning_rate: float = 2e-4,
        min_learning_rate: float = 2e-5,
        warmup_rate: float = 0.1,
    ):
    super().__init__()

    self.model = model
    self.total_steps = total_steps
    self.max_learning_rate = max_learning_rate
    self.min_learning_rate = min_learning_rate
    self.warmup_rate = warmup_rate
    self.model_save_dir = model_save_dir
    self.validation_step_loss = []

모델 설계에 필요한 인수들을 입력받습니다.

• model : BART 모델 인스턴스
• model_save_dir : 모델을 저장할 경로
• total_steps, max_learning_step, min_learning_rate, warmup_rate : 학습에 사용될 hyper parameter
• validation_step_loss : validation_epoch가 끝난 후 손실을 저장하기 위한 리스트

훈련 단계( training_step)

def training_step(self, batch, batch_idx):
    output = self.model(
        input_ids=batch["input_ids"],
        attention_mask=batch["attention_mask"],
        decoder_input_ids=batch["decoder_input_ids"],
        decoder_attention_mask=batch["decoder_attention_mask"],
        return_dict=True,
    )

    labels = batch["decoder_input_ids"][:, 1:].reshape(-1)
    logits = output["logits"][:, :-1].reshape([labels.shape[0], -1])

    loss = F.cross_entropy(logits, labels, ignore_index=self.model.config.pad_token_id)
    metrics = {"loss": loss}
    self.log_dict(metrics, prog_bar=True, logger=True, on_step=True)

    return metrics

모델에 특정 배치가 들어갔을 때 훈련을 하게 된다. 

• input_ids: 입력 시퀀스의 토큰 ID 배열.
• attention_mask: 입력 시퀀스의 어텐션 마스크 배열, 패딩 된 부분을 0으로, 유효한 토큰을 1로 표시.
• decoder_input_ids: 디코더에 입력되는 시퀀스의 토큰 ID 배열.
• decoder_attention_mask: 디코더 시퀀스의 어텐션 마스크 배열.
• labels : 맨 앞 토큰([BOS])을 제외하고 나머지 토큰을 사용
• logits : 모델 출력에서 마지막 토큰을 제외한 나머지
• cross_entropy : label과 logits의 손실을 계산하며 ignore_index를 통해 [PAD] 토큰 무시
• log_dict : 여러 메트릭을 로깅하는데 사용
  • metrics: 로깅하고자 하는 메트릭을 담고 있는 딕셔너리. 키는 메트릭의 이름, 값은 로깅하려는 수치.
  • prog_bar: 메트릭이 훈련 프로그레스 바에 표시. 
  • logger: 로깅하는 메트릭이 백엔드 로거(예: TensorBoard, CSVLogger 등)에 전송.
  • on_epoch: 로깅되는 메트릭이 에폭 단위로 기록. 각 에폭에서 메트릭의 평균값을 계산하여 로깅

cross_entropy함수를 이용해서 정답과 예측값의 손실을 계산한다. 또한 ignore_index인수를 통해 [PAD] 토큰들을 무시하도록 합니다. 마지막

검증 단계, 테스트 단계( validation_step, test_step)

def validation_step(self, batch, batch_idx):
    output = self.model(
        input_ids=batch["input_ids"],
        attention_mask=batch["attention_mask"],
        decoder_input_ids=batch["decoder_input_ids"],
        decoder_attention_mask=batch["decoder_attention_mask"],
        return_dict=True,
    )

    labels = batch["decoder_input_ids"][:, 1:].reshape(-1)
    logits = output["logits"][:, :-1].reshape([labels.shape[0], -1])

    loss = F.cross_entropy(logits, labels, ignore_index=self.model.config.pad_token_id)
    metrics = {"loss(v)": loss}
    self.validation_step_loss.append(loss)

    self.log_dict(metrics, prog_bar=True, logger=True, on_epoch=True)

    return metrics
    

def test_step(self, *args, **kwargs):
    return self.validation_step(*args, **kwargs)

train_step과 거의 동일하지만, 다른점은 validation_step_loss를 저장하며 test_step은 validation_step과 동일하다.

옵티마이저 설정( configure_optimizers)

def configure_optimizers(self):
    optimizer = torch.optim.AdamW(params=self.model.parameters(), lr=self.max_learning_rate)

    return {
        "optimizer": optimizer
    }

optimizer를 adamW로 설정하고 학습률에 따라서 model의 파라미터를 업데이트한 것을 optimizer를 반환하여 가중치를 업데이트한다.

검증 에폭 종료 시 ( on_validation_epoch_end)

def on_validation_epoch_end(self):
    if self.trainer.is_global_zero:
        losses = [output.mean() for output in self.validation_step_loss]
        loss_mean = sum(losses) / len(losses)

        self.model.save_pretrained(
            os.path.join(
                self.model_save_dir,
                f"model-{self.current_epoch:02d}epoch-{self.global_step}steps-{loss_mean:.4f}loss",
            ),
        )

    self.validation_step_loss.clear()  # free memory

• self.trainer.is_global_zero : 멀티 GPU 또는 분산 설정에서의 주 노드에서 실행되고 있는지를 검사
• self.model.save_pretrained : 모델의 파라미터를 저장하며, 제목은 다음과 같이 한다.
• self.validation_step_loss.clear : 리스트를 비우고 다음 에폭에서 새로운 손실 데이터를 깨끗하게 수집

전체코드

import os

import torch
import pytorch_lightning as pl
import torch.nn.functional as F


class StoryModule(pl.LightningModule):
    """
    Attributes:
        model: BART model
        total_steps: total training steps for lr scheduling
        max_learning_rate: Max LR
        min_learning_rate: Min LR
        warmup_rate: warmup step rate
        model_save_dir: path to save model
    """

    def __init__(
        self,
        model,
        model_save_dir,
        total_steps,
        max_learning_rate: float = 2e-4,
        min_learning_rate: float = 2e-5,
        warmup_rate: float = 0.1,
    ):
        super().__init__()

        self.model = model
        self.total_steps = total_steps
        self.max_learning_rate = max_learning_rate
        self.min_learning_rate = min_learning_rate
        self.warmup_rate = warmup_rate
        self.model_save_dir = model_save_dir
        self.validation_step_loss = []

        self.save_hyperparameters(
            {
                **model.config.to_dict(),
                "total_steps": total_steps,
                "max_learning_rate": self.max_learning_rate,
                "min_learning_rate": self.min_learning_rate,
                "warmup_rate": self.warmup_rate,
            }
        )

    def training_step(self, batch, batch_idx):
        output = self.model(
            input_ids=batch["input_ids"],
            attention_mask=batch["attention_mask"],
            decoder_input_ids=batch["decoder_input_ids"],
            decoder_attention_mask=batch["decoder_attention_mask"],
            return_dict=True,
        )

        labels = batch["decoder_input_ids"][:, 1:].reshape(-1)
        logits = output["logits"][:, :-1].reshape([labels.shape[0], -1])

        loss = F.cross_entropy(logits, labels, ignore_index=self.model.config.pad_token_id)
        metrics = {"loss": loss}
        self.log_dict(metrics, prog_bar=True, logger=True, on_step=True)

        return metrics
    
    def validation_step(self, batch, batch_idx):
        output = self.model(
            input_ids=batch["input_ids"],
            attention_mask=batch["attention_mask"],
            decoder_input_ids=batch["decoder_input_ids"],
            decoder_attention_mask=batch["decoder_attention_mask"],
            return_dict=True,
        )

        labels = batch["decoder_input_ids"][:, 1:].reshape(-1)
        logits = output["logits"][:, :-1].reshape([labels.shape[0], -1])

        loss = F.cross_entropy(logits, labels, ignore_index=self.model.config.pad_token_id)
        metrics = {"loss(v)": loss}
        self.validation_step_loss.append(loss)
        
        self.log_dict(metrics, prog_bar=True, logger=True, on_epoch=True)

        return metrics

    def test_step(self, *args, **kwargs):
        return self.validation_step(*args, **kwargs)

    def configure_optimizers(self):
        optimizer = torch.optim.AdamW(params=self.model.parameters(), lr=self.max_learning_rate)

        return {
            "optimizer": optimizer
        }

    def on_validation_epoch_end(self):
        if self.trainer.is_global_zero:
            losses = [output.mean() for output in self.validation_step_loss]
            loss_mean = sum(losses) / len(losses)

            self.model.save_pretrained(
                os.path.join(
                    self.model_save_dir,
                    f"model-{self.current_epoch:02d}epoch-{self.global_step}steps-{loss_mean:.4f}loss",
                ),
            )

        self.validation_step_loss.clear()  # free memory

다음은 dataset 및 DataLoader 구축에 대해 설명하겠습니다!