[NLP] Pytorch를 활용하여 CBOW 임베딩 학습하기 (2)모델 훈련

-- 본 포스팅은 파이토치로 배우는 자연어 처리 (한빛미디어) 책을 참고해서 작성된 글입니다.
-- 소스코드는 여기

 

<이전글>

[NLP] Pytorch를 활용하여 CBOW 임베딩 학습하기 (1)

 

1. 모델 생성

1. embedding층을 사용하여 문맥의 단어를 나타내는 인덱스를 각 단어에 대한 벡터로 만든다.
2. 전반적인 문맥을 감지하도록 벡터를 결합한다.
3. Linear 층에서 문맥 벡터를 사용하여 예측 벡터를 계산한다.
   • 이 예측벡터는 전체 어휘 사전에 대한 확률 분포이다.
   • 예측벡터에서 가장 큰 값이 타겟 단어에 대한 예측을 나타낸다.

class CBOWClassifier(nn.Module): # Simplified cbow Model
    def __init__(self, vocabulary_size, embedding_size, padding_idx=0):
        """
        매개변수3개로 제어된다:
            vocabulary_size (int): 어휘 사전 크기, 임베딩 개수와 예측 벡터 크기를 결정합니다
            embedding_size (int): 임베딩 크기
            padding_idx (int): 기본값 0; 임베딩은 이 인덱스를 사용하지 않습니다
        """
        super(CBOWClassifier, self).__init__()
        
        # padding idx는 기본값이 0 이지만, 데이터 포인트의 길이가 모두 같지 않을 때
        # embedding층에 패딩하는데 사용되는 매개변수이다.
        self.embedding =  nn.Embedding(num_embeddings=vocabulary_size, 
                                       embedding_dim=embedding_size,
                                       padding_idx=padding_idx)
        # linear층에서 문맥벡터를 사용하여 예측벡터를 계산한다
        # 예측벡터는 어휘사전에 대한 확률분포
        self.fc1 = nn.Linear(in_features=embedding_size,
                             out_features=vocabulary_size)

    def forward(self, x_in, apply_softmax=False):
        """분류기의 정방향 계산
        
        매개변수:
            x_in (torch.Tensor): 입력 데이터 텐서 
                x_in.shape는 (batch, input_dim)입니다.
            apply_softmax (bool): 소프트맥스 활성화 함수를 위한 플래그
                크로스-엔트로피 손실을 사용하려면 False로 지정합니다
        반환값:
            결과 텐서. tensor.shape은 (batch, output_dim)입니다.
        """
        x_embedded_sum = F.dropout(self.embedding(x_in).sum(dim=1), 0.3)
        y_out = self.fc1(x_embedded_sum)
        
        ## 출력: 소프트맥스함수는 선택적으로 계산할 예정 / 수치적 계산 낭비 발생, 불안정 발생 
        if apply_softmax:
            y_out = F.softmax(y_out, dim=1)
            
        return y_out

 

2. 모델 훈련

2.1 헬퍼함수 정의

• make_train_state : 현재 훈련 정도를 나타낼 수 있는 함수이다. 뒤에서 훈련하는 과정을 보면 state에 계속 훈련 상태를 업데이트 해주고, 모델을 훈련시켜주는 것을 확인할 수 있다.
• update_train_state : 최적의 모델을 찾기 위해서 해당 함수가 사용된다. 모델이 한번씩 돌아갈때마다 상태가 업데이트 되고, 업데이트된 값을 통해서 모델을 저장해준다. 손실을 좋게 만들어주고 최상의 모델을 찾을 때 사용된다.
• compute_accuracy : 정확도를 계산할때 사용되는 헬퍼함수이다.

# 현재 트레인 상태를 나타낼 수 있는 함수
# 뒤에 가면 모델을 훈련 도중 업데이트 하고 최상의 모델을 찾아내는데 사용된다.
# 손실 저장, 에포크 저장 등등 
def make_train_state(args):
    return {'stop_early': False,
            'early_stopping_step': 0,
            'early_stopping_best_val': 1e8,
            'learning_rate': args.learning_rate,
            'epoch_index': 0,
            'train_loss': [],
            'train_acc': [],
            'val_loss': [],
            'val_acc': [],
            'test_loss': -1,
            'test_acc': -1,
            'model_filename': args.model_state_file}


# 뒤에 가면 손실이 적게나오는 최적의 모델을 찾을 건데 해당 모델을 찾기위해
# 지속적으로 상태를 업데이트 해주기 위해 이 함수 사용
# 모델이 한번 돌아가면, 업데이트 해주는 작업을 할 건데, 그때 
# 이전에 저장되어 있는 train state와 성능을 비교하여 최적의 모델을 찾을 것
def update_train_state(args, model, train_state):
    """ 훈련 상태를 업데이트합니다.

    Components:
     - 조기 종료: 과대 적합 방지
     - 모델 체크포인트: 더 나은 모델을 저장합니다

    :param args: 메인 매개변수
    :param model: 훈련할 모델
    :param train_state: 훈련 상태를 담은 딕셔너리
    :returns:
        새로운 훈련 상태
    """

    # 적어도 한 번 모델을 저장합니다
    if train_state['epoch_index'] == 0:
        torch.save(model.state_dict(), train_state['model_filename'])
        train_state['stop_early'] = False

    # 성능이 향상되면 모델을 저장합니다
    elif train_state['epoch_index'] >= 1:
        loss_tm1, loss_t = train_state['val_loss'][-2:]

        # 손실이 나빠지면
        if loss_t >= train_state['early_stopping_best_val']:
            # 조기 종료 단계 업데이트
            train_state['early_stopping_step'] += 1
        # 손실이 감소하면
        else:
            # 최상의 모델 저장
            if loss_t < train_state['early_stopping_best_val']:
                torch.save(model.state_dict(), train_state['model_filename'])

            # 조기 종료 단계 재설정
            train_state['early_stopping_step'] = 0

        # 조기 종료 여부 확인
        train_state['stop_early'] = \
            train_state['early_stopping_step'] >= args.early_stopping_criteria

    return train_state


# 정확도를 계산해주는 함수를 정의한다.
def compute_accuracy(y_pred, y_target):
    _, y_pred_indices = y_pred.max(dim=1)
    n_correct = torch.eq(y_pred_indices, y_target).sum().item()
    return n_correct / len(y_pred_indices) * 100

 

2.2 일반 유틸리티 함수 정의

#추후 재현성을 위해서 랜덤시드를 다시 설정해줄 것인데 그때 사용하는 함수
# 어떤 연구를 똑같이 다시 반복함으로써 기존 원본에서 보고되었던 결과가 (거의) 똑같이 다시 나타나는지를 관찰하는 것.
def set_seed_everywhere(seed, cuda):
    np.random.seed(seed)
    torch.manual_seed(seed)
    if cuda:
        torch.cuda.manual_seed_all(seed)

# 디렉토리를 처리하기 위한 함수
def handle_dirs(dirpath):
    if not os.path.exists(dirpath):
        os.makedirs(dirpath)

 

2.3 설정과 전처리 작업 (환경설정)

args = Namespace(
    # 날짜와 경로 정보
    cbow_csv="data/books/frankenstein_with_splits.csv",
    vectorizer_file="vectorizer.json",
    model_state_file="model.pth",
    save_dir="model_storage/ch5/cbow",
    # 모델 하이퍼파라미터
    embedding_size=50,
    # 훈련 하이퍼파라미터
    seed=1337,
    num_epochs=100,
    learning_rate=0.0001,
    batch_size=32,
    early_stopping_criteria=5,
    # 실행 옵션
    cuda=True,
    catch_keyboard_interrupt=True,
    reload_from_files=False,
    expand_filepaths_to_save_dir=True
)

if args.expand_filepaths_to_save_dir:
    args.vectorizer_file = os.path.join(args.save_dir,
                                        args.vectorizer_file)

    args.model_state_file = os.path.join(args.save_dir,
                                         args.model_state_file)
    
    print("파일 경로: ")
    print("\t{}".format(args.vectorizer_file))
    print("\t{}".format(args.model_state_file))
    

# CUDA 체크
if not torch.cuda.is_available():
    args.cuda = False

args.device = torch.device("cuda" if args.cuda else "cpu")
    
print("CUDA 사용여부: {}".format(args.cuda))

# 재현성을 위해 시드 설정
set_seed_everywhere(args.seed, args.cuda)

# 디렉토리 처리
handle_dirs(args.save_dir)

 

2.4 데이터 초기화

• 데이터를 받아줄 것이다.
• 이미 전처리된 데이터를 사용한다.
• 이 책에서 라이트 버전의 전처리란, 크기를 좀 줄여서 전처리되어진 데이터셋을 의미한다.

# 만약 코랩에서 실행하는 경우 아래 코드를 실행하여 전처리된 라이트 버전의 데이터를 다운로드하세요.
!mkdir data
!wget https://git.io/JtX5A -O data/download.py
!wget https://git.io/JtX5F -O data/get-all-data.sh
!chmod 755 data/get-all-data.sh
%cd data
!./get-all-data.sh
%cd ..

 

2.5 데이터셋, vectorizer 생성

args에 저장된 정보를 바탕으로 데이터셋과 vectorizer를 만들어준다. 우리는 이미 사전에 생성된 vectorizer 객체가 없기 때문에 else문이 실행된다.

# 이미 생성되어있는 데이터셋과 vectorizer객체가 존재한다면 그 부분에서 데이터셋을 받아와서 다시 훈련 시작 
# args.reload_from_files = False 로 지정했음
# 아니면 데이터셋과 벡토라이저 만들기

''' args 설정 살펴보기!!
    # 모델 하이퍼파라미터
    embedding_size=50,
    # 훈련 하이퍼파라미터
    seed=1337,
    num_epochs=100,
    learning_rate=0.0001,
    batch_size=32,
    early_stopping_criteria=5,
'''
if args.reload_from_files:
    print("데이터셋과 Vectorizer를 로드합니다")
    dataset = CBOWDataset.load_dataset_and_load_vectorizer(args.cbow_csv,
                                                           args.vectorizer_file)
else:
    print("데이터셋을 로드하고 Vectorizer를 만듭니다")
    dataset = CBOWDataset.load_dataset_and_make_vectorizer(args.cbow_csv)
     #save vectorizer = 벡터를 디스크에 json 형태로 저장해주는 메서드
    dataset.save_vectorizer(args.vectorizer_file)
     
# get vectoirzer = 변환 벡터 객체를 반환해주는 메서드   
vectorizer = dataset.get_vectorizer()

# 모델 설정 (embedingsize = 50)
classifier = CBOWClassifier(vocabulary_size=len(vectorizer.cbow_vocab), 
                            embedding_size=args.embedding_size)

 

2.6 훈련 상태를 업데이트 하면서 훈련 반복

classifier = classifier.to(args.device)

# 다중클래스분류에 상요되는CrossEntropyLoss 사용
# 옵티마이저 adam    
loss_func = nn.CrossEntropyLoss()
optimizer = optim.Adam(classifier.parameters(), lr=args.learning_rate)
scheduler = optim.lr_scheduler.ReduceLROnPlateau(optimizer=optimizer,
                                                 mode='min', factor=0.5,

# 트레인상태 만들기 (헬퍼함수)                                           patience=1)
train_state = make_train_state(args)

# 에포크는 100
epoch_bar = tqdm.notebook.tqdm(desc='training routine', 
                               total=args.num_epochs,
                               position=0)

#set_split 메서드 : 데이터프레임에 있는 열을 사용해 분할 세트를 선택
dataset.set_split('train')

# get_num_batches 메서드 : 배치 크기가 주어지면 데이터셋으로 만들 수 있는 배치 개수를 반환해주는 메서드
train_bar = tqdm.notebook.tqdm(desc='split=train',
                               total=dataset.get_num_batches(args.batch_size), 
                               position=1, 
                               leave=True)
dataset.set_split('val')
val_bar = tqdm.notebook.tqdm(desc='split=val',
                             total=dataset.get_num_batches(args.batch_size), 
                             position=1, 
                             leave=True)

#외부 for문 : 에포크 횟수만큼 돌아감. (에포크는 100임)
#내부 for문 : 배치단위를 돌리는 반복문
try:
    for epoch_index in range(args.num_epochs):
        train_state['epoch_index'] = epoch_index

        # 훈련 세트에 대한 순회

        # 훈련 세트와 배치 제너레이터 준비, 손실과 정확도를 0으로 설정
        dataset.set_split('train')
        batch_generator = generate_batches(dataset, 
                                           batch_size=args.batch_size, 
                                           device=args.device)
        running_loss = 0.0
        running_acc = 0.0
        classifier.train()

        for batch_index, batch_dict in enumerate(batch_generator):
            # 훈련 과정은 5단계로 이루어집니다

            # --------------------------------------
            # 단계 1. 그레이디언트를 0으로 초기화합니다
            optimizer.zero_grad()

            # 단계 2. 출력을 계산합니다
            y_pred = classifier(x_in=batch_dict['x_data'])

            # 단계 3. 손실을 계산합니다
            loss = loss_func(y_pred, batch_dict['y_target'])
            loss_t = loss.item()
            running_loss += (loss_t - running_loss) / (batch_index + 1)

            # 단계 4. 손실을 사용해 그레이디언트를 계산합니다
            loss.backward()

            # 단계 5. 옵티마이저로 가중치를 업데이트합니다
            optimizer.step()
            # -----------------------------------------
            
            # 정확도를 계산합니다
            acc_t = compute_accuracy(y_pred, batch_dict['y_target'])
            running_acc += (acc_t - running_acc) / (batch_index + 1)

            # 진행 바 업데이트
            train_bar.set_postfix(loss=running_loss, acc=running_acc, 
                            epoch=epoch_index)
            train_bar.update()

        train_state['train_loss'].append(running_loss)
        train_state['train_acc'].append(running_acc)

        # 검증 세트에 대한 순회

        # 검증 세트와 배치 제너레이터 준비, 손실과 정확도를 0으로 설정
        dataset.set_split('val')
        batch_generator = generate_batches(dataset, 
                                           batch_size=args.batch_size, 
                                           device=args.device)
        running_loss = 0.
        running_acc = 0.
        classifier.eval()

        for batch_index, batch_dict in enumerate(batch_generator):

            # 단계 1. 출력을 계산합니다
            y_pred =  classifier(x_in=batch_dict['x_data'])

            # 단계 2. 손실을 계산합니다
            loss = loss_func(y_pred, batch_dict['y_target'])
            loss_t = loss.item()
            running_loss += (loss_t - running_loss) / (batch_index + 1)

            # 단계 3. 정확도를 계산합니다
            acc_t = compute_accuracy(y_pred, batch_dict['y_target'])
            running_acc += (acc_t - running_acc) / (batch_index + 1)
            val_bar.set_postfix(loss=running_loss, acc=running_acc, 
                            epoch=epoch_index)
            val_bar.update()

        train_state['val_loss'].append(running_loss)
        train_state['val_acc'].append(running_acc)

        train_state = update_train_state(args=args, model=classifier,
                                         train_state=train_state)

        scheduler.step(train_state['val_loss'][-1])

        if train_state['stop_early']:
            break

        train_bar.n = 0
        val_bar.n = 0
        epoch_bar.update()
except KeyboardInterrupt:
    print("Exiting loop")

 

2.7 가장 좋은 모델을 사용하여 정확도 계산

• test 데이터셋을 활용해서 정확도를 계산한다.

# 가장 좋은 모델을 사용해 테스트 세트의 손실과 정확도를 계산합니다
# train_state model filename에 가장 좋은 모델이 저장되어 있다.
classifier.load_state_dict(torch.load(train_state['model_filename']))
classifier = classifier.to(args.device)
loss_func = nn.CrossEntropyLoss()

# test 데이터셋을 선택
dataset.set_split('test')
batch_generator = generate_batches(dataset, 
                                   batch_size=args.batch_size, 
                                   device=args.device)
running_loss = 0.
running_acc = 0.
classifier.eval()

for batch_index, batch_dict in enumerate(batch_generator):
    # 출력을 계산합니다
    y_pred =  classifier(x_in=batch_dict['x_data'])
    
    # 손실을 계산합니다
    loss = loss_func(y_pred, batch_dict['y_target'])
    loss_t = loss.item()
    running_loss += (loss_t - running_loss) / (batch_index + 1)

    # 정확도를 계산합니다
    acc_t = compute_accuracy(y_pred, batch_dict['y_target'])
    running_acc += (acc_t - running_acc) / (batch_index + 1)

train_state['test_loss'] = running_loss
train_state['test_acc'] = running_acc

print("테스트 손실: {};".format(train_state['test_loss']))
print("테스트 정확도: {}".format(train_state['test_acc']))

(이게 말이 되는 정확도인가..?!)

• 결과값이 높지 않은 이유
  1. 이 예제에서는 범용적인 임베딩을 구성하는 방법을 설명하기 편하도록 여러가지 기능을 생략하여 CBOW 실습을 진행했다. (성능 최적화에 필수적인 여러가지 기능을 생략하고 단순 구현한 것이다.)
  2. 여기서 사용하는 데이터셋의 크기가 아주 작다. (단어의 수는 약 70000개 뿐이라서, zero base에서 규칙성을 감지하기에 충분하지 못하다. 

 

3. 임베딩 결과 출력해보기

임베딩 결과를 출력하기 위한 헬퍼 함수를 작성한다. 단지 이쁘게 임베딩된 결과를 출력하기 위함이다.
거리를 계산하여 입력된 단어와 가장 가까이 존재하는단어들을 출력할 것이다.

def pretty_print(results):
    """
    임베딩 결과를 출력합니다
    """
    for item in results:
        print ("...[%.2f] - %s"%(item[1], item[0]))
        
# n에 출력을원하는 개수 입력
def get_closest(target_word, word_to_idx, embeddings, n=5):
    """
    n개의 최근접 단어를 찾습니다.
    """

    # 다른 모든 단어까지 거리를 계산합니다
    word_embedding = embeddings[word_to_idx[target_word.lower()]]
    distances = []
    for word, index in word_to_idx.items():
        if word == "<MASK>" or word == target_word:
            continue
        distances.append((word, torch.dist(word_embedding, embeddings[index])))
    
    results = sorted(distances, key=lambda x: x[1])[1:n+2]
    return results

word = input('단어를 입력해 주세요: ')
embeddings = classifier.embedding.weight.data
word_to_idx = vectorizer.cbow_vocab._token_to_idx
pretty_print(get_closest(word, word_to_idx, embeddings, n=5))

target_words = ['frankenstein', 'monster', 'science', 'sickness', 'lonely', 'happy']

embeddings = classifier.embedding.weight.data
word_to_idx = vectorizer.cbow_vocab._token_to_idx

for target_word in target_words: 
    print(f"======={target_word}=======")
    if target_word not in word_to_idx:
        print("Not in vocabulary")
        continue
    pretty_print(get_closest(target_word, word_to_idx, embeddings, n=5))

 

 

이번 예제에서는 nn.Embedding 층을 사용하여 CBOW 분류 지도학습 작업으로 밑바닥부터 임베딩을 훈련시켜 보았다. 앞서 보았듯이 밑바닥부터 훈련시키기에 충분하지 못한 데이터의 크기를 가지고 있어서 정확도가 15%로 매우 낮다. 다음 공부에서는 말뭉치에 주어진 '사전 훈련된 임베딩'을 미세 조정해서 사용하는 '전이학습'방법을 살펴볼 것이다. 밑바닥 부터 훈련시키는 것은 매우 많은 데이터셋이 필요하기에 요즘에는 '전이학습'과 같이 훈련된 모델을 다른 작업의 초기 모델로 사용하는 방식이 많이 사용된다.