개요

본 블로그 포스팅은 수도권 ICT 이노베이션 스퀘어에서 진행하는 AI 핵심 기술 집중 클래스의 자연어처리(NLP) 강좌 내용을 필자가 다시 복기한 내용에 관한 것입니다.

---

1. 실습 개요

(NER : Named Entity Recognition)는 위 사진처럼
원문에서 중요한 단어(키워드 혹은 분류 가능한 항목들) 에 대하여 미리 사전에 정의한 유형(태그 목록)에 따라
원문 내 단어들을 태깅(tagging)하는 작업을 말하며, 이런 작업을 하는 이유는

향후 컨텐츠 추천, 검색엔진, 기사 요약 과 같은 API 서비스를 만들고자 할 때 사용하는 텍스트 전처리 방법이라 보면 된다.

음.. 그러니까 NLD(자연어 데이터셋)을 사용하기 편리한 데이터셋으로 한번 전처리를 수행하는 작업

이렇게 이해하면 될 듯 하다.

물론 위 사진처럼 예전에는 중요 키워드에 대한 태깅작업이
약간 노가다성이 짙은 작업이었다면

이를 아래의 그림처럼 RNN : Many-to-Many 딥러닝 방법론으로 태깅 작업을 자동화 하는 것이 본 실습의 목표이다.

1.1 데이터셋 선정

NER 작업을 수행하기 위해서는 당연히 데이터셋이 필요하고 이 작업은 지도학습(Supervised Learning)에 속하는 작업이다.

그래서 누군가가 열심히
[원문, 원문에 대한 개체명 라벨데이터]
작업을 수행한 훈련 데이터셋을 구해와야 한다.

이를 위해 https://www.aihub.or.kr/
한국 지능정보사회 진흥원에서 운영하고 있는 AI Hub 에서 데이터셋을 구하고자 한다.

왠만한 AI작업에 사용되는 데이터는 여기서 구해 쓸 수 있으니 적극적으로 활용하는것이 좋다.

필자는 AI Hub에서 제공하는 데이터셋 중 실습을 위한 파일로

데이터셋은 약 100만건에 NER작업을 위해 사용 가능한 데이터셋이라는 설명 문구가 있으니
이를 활용하고자 한다.

데이터를 다운받으면 대략 위와 같이 폴더가 구성되어 있는데

필자는 이 중 '안전건설_93747.json' 데이터만을 활용하여 NER을 진행하고자 한다.

데이터셋 구조를 살펴본다면 *.json 확장자 명으로
데이터의 서두에 ['name'], ['create_data'], ['documents'] 항목이 있고

['documents'] 안에 민원 항목(93747 개)이 [id]로 라벨링 되어 있으며,
상세 항목으로 민원의 원문 데이터 [Q_refined]
그리고 원문 데이터 내 특별히 인식해야 할 키워드는
[entities]라는 항목으로 라벨링 처리되어 데이터셋이 구성되 있다.

이 데이터셋에 대한 파싱 작업을 수행 후
pandas - Dataframe 데이터 타입으로 변환하는 작업을 수행하고자 한다.

import json

file_path = './1.Training/라벨링데이터/안전건설/안전건설_93747.json'

#json파일 불러오기
with open(file_path, 'r', encoding='utf-8') as f:
    data = json.load(f)

parsed_data =[] # 파싱 데이터를 저장할 변수

for docu in data['documents']:
    # 문서 내 id, Q_refined 추출
    doc_id = docu['id']
    q_refiled = docu['Q_refined']

    # endity는 여러개 존재하니 [개채명 : 개체 분류] 형식으로 파싱
    entities = [[eneity['form'], eneity['label']] 
                for eneity in docu["labeling"]["entities"]]

    # 현재 document의 정보를 딕셔너리에 담고, 리스트에 추가
    raw_dict = {
        "문서ID": doc_id,
        "컨텐츠": q_refiled,
        "개체": entities
    }
    parsed_data.append(raw_dict)

import pandas as pd

# 파싱이 완료된 데이터를 Dataframe 형식으로 변환
raw_data = pd.DataFrame(parsed_data)

1.2 BIO Tagging

위 1.1 과정에서
원문, 원문에 대한 중요 키워드(개체)별 태그 정보
와 같은 데이터 전처리를 수행했으면

이를 BIO Tagging 방법론을 통해 라벨링 정보를 생성해야한다

위 사진처럼 원문, 태그정보 두개의 항목이 있을 때
두 정보 모두 토큰화를 수행하다 보면
가끔씩 위 사진의 붉은색 개체 처럼 개체가 토큰화로 인해 쪼개지는 상황이 발생한다.

이런 난감한 경우를 해결하는 방법론이 BIO Tagging으로

개체가 쪼개지면
맨 앞 개체 토큰 : 접두사 B-를 붙임
나머지 개체 토큰 : 접두사 I-를 붙임

으로 해결하는 것이다.

물론 쪼개지지 않더라도 편의성 추구를 위해 B-를 붙여버린다.

그리고 중요 단어(개체)가 아닌 토큰들은 모두 O 처리
-> O 처리된 토큰들은 나중에 단어장 생성 에도 참여하지 않고 모두 UNK토큰으로 날려버린다.

즉, 쓸모있는 단어(개체)만 y_label 처리하며, Entity이 토큰화로 분절되더라도 이를 상세하게 Re-tagging 하는것이 NER 과정에서 필수로 적용되는 BIO Tagging 방법론이다.

이전에 데이터 전처리 필수!!

# 데이터 및 텍스트 전처리 함수를 모듈화 시킨 파일
from NLP_pp import *
# 데이터셋읜 결측치 & 중복치 제거 함수 실행
raw_data = df_cleaning(raw_data, '컨텐츠')

따라서 이를 코드화를 수행한다면 아래와 같아진다.

# 데이터프레임의 항목을 분리 후 리스트 타입으로 변경
raw_x_data = raw_data['컨텐츠'].values.tolist()
# 개체 : 개체 라벨링 의 태그 정보는 따로 추출한다.
tag_data = raw_data['개체'].values.tolist()

먼저 컬럼의
'컨텐츠' 항목과 : raw_x_data
'개체:태그' 컬럼 : tag_data
두개를 리스트 처리한다

그 다음 정규표현식으로 특수문자를 삭제하는 코드를 실행한다.

import re

# 한글, 영어(소문자, 대문자), 숫자
p1 = re.compile(r'[^가-힣a-zA-Z0-9\s]')
# 개행문자 + 하나 이상의 공백문자
p2 = re.compile(r'\n|\s+')

def regex_sub(origin_sent):
    clean_text = p1.sub(repl=" ", string=origin_sent)
    clean_text = p2.sub(" ", clean_text)

    return clean_text

# [개체:태그]데이터는 리스트 형식이라서 위 함수를 콜백으로 사용함
def list_regex_sub(origin_list):
    return [[regex_sub(entity), label] for entity, label in origin_list]

# 설계한 정규표현식기반 특수문자 삭제 함수 적용
# apply함수는 inplace=True(덮어쓰기) 기능이 없음
raw_data['컨텐츠'] = raw_data['컨텐츠'].apply(regex_sub)
raw_data['개체'] = raw_data['개체'].apply(list_regex_sub)

이때 tag_data는 [개체:태그] 로 이뤄진 3중 리스트 데이터이기에 이중 '개체' 항목에 대해서만 정규표현식으로 문자열 클리닝 작업을 수행해야 하기에

regex_sub를 콜백함수로 받는 list_regex_sub로
[개체:태그] 데이터를 전처리 한다

추가로 꼭 실행해야 할 항목

[개체명 : 태그항목] 쌍에서 가끔 [태그]가 오염되어 있는 경우가 존재한다.

위 사진과 같은 경우인데.. 이런 데이터도 클리닝 처리해야한다.
이거 안하고 신나게 인코딩 하다 낭패봄

# 영어 대문자만 있는 경우를 감지하는 정규표현식 패턴
p999 = re.compile(r'^[A-Z]+$')

def clean_tags(entity_list):
    # 각 [개체명, 태그] 쌍을 검사하여 태그가 영어 대문자 외 단어가 포함된 경우
    # 이 태그는 오염된 태그이니 [개체명, 오염된 태그] 리스트 자체를 제거
    cleaned_list = [[entity, tag] for entity, tag in entity_list if p999.match(tag)]
    
    return cleaned_list

# 태그 항목에서 오염된 태그 항목을 제거하는 함수 구동
raw_data['개체'] = raw_data['개체'].apply(clean_tags)

토큰화 수행

다음으로 토큰화를 '원문', '개체' 두개를 진행한 뒤에
BIO Tagging 방법론으로 라벨 데이터(tokenized_y_label)를 생성하자

from mecab import MeCab #한글 단어 토크나이저
from tqdm import tqdm

#mecab 형태소 분석기 인스턴스화
word_tokenizer = MeCab()

# 데이터 및 텍스트 전처리 함수를 모듈화 시킨 파일
from NLP_pp import *

이때 이전 포스트 1. NLP-Text 전처리 마침 : 모듈화에서 설명한 Text Preprocessing(텍스트 전처리), Data Preprocessing(데이터 전처리)를 원활하게 수행하기 위한 함수모임 코드인

NLP_pp.py https://github.com/tbvjvsladla/ASH_NLP_lacture
를 사용하여 토큰화 함수를 구동하자

# 토큰화 수행
tokenized_x_data = tokenize(raw_x_data, word_tokenizer)

BIO Tagging 함수

BIO Tagging는 [개체:태그] 데이터에서
개체 -> 토큰화 한 개체
를 생성하고

토큰화 원문(token_x)와 토근화한 개체(token_entity)

두개 항목으로 y_label를 생성해야 하니 이를 코드화 한 함수를 사용한다

from tqdm import tqdm

def BIO_tagging(tokenized_x_data, tag_data, word_tokenizer):
    # BIO 태깅 규칙으로 라벨링 처리될 y_data 리스트 선언
    tagged_y_label = []
    # 태그 데이터도 토큰화가 잘 되었는지 확인이 팔요함...
    token_tag_data = []

    for token_x, tags in tqdm(zip(tokenized_x_data, tag_data), 
                              total=len(tokenized_x_data), 
                              desc="BIO 태깅 진행 중"):
        # 1) token_x(토큰화 처리된 x_data의 i번째 문서)와 매칭되는 y_label 생성
        # y_label은 token_x의 토큰 개수만큼 초기화 하는것을 의미함
        token_y = ['O'] * len(token_x)
        token_tags = [] # 태그가 잘 이뤄지는지 추적하는 데이터
        
        # 2) token_x에 대한 개체:태그 정보가 포함된 tag_data의 데이터 분해
        for entity, label in tags:
            # entity를 토큰화하고, 몇 개의 토큰으로 분해되었는지 확인
            entity_token = word_tokenizer.morphs(entity)
            entity_len = len(entity_token)

            # 토큰화된 개체명 정보를 붙이기
            token_tags.append(entity_token)
            # 3) token_x에서 entity_token과 일치하는 정보를 색인하여
            # 해당 정보를 올바르게 B-I labeling을 적용한다
            for i in range(len(token_x) - entity_len+1):
                # 정보를 찾아냈을 시, 시작은 'B-' 접두를 태그에 붙이고
                # 나머지 항목은 'I-' 접두를 태그에 붙이는 작업
                if token_x[i : i+entity_len] == entity_token:
                    token_y[i] = f'B-{label}'  # 'B-' 접두 추가 태깅
                    token_tags.append(token_y[i]) # 토큰태그가 잘 되엇는지 검증

                    for j in range(1, entity_len):
                        token_y[i+j] = f'I-{label}'  # 나머지는 'I-' 접두 추가 태깅
                        token_tags.append(token_y[i+j]) # 토큰태그가 잘 되엇는지 검증
                    break  # 첫 번째로 일치하는 위치만 태깅

        # 4) BIO 태깅 규칙으로 업데이트가 완료된 
        # i번째 문서에 대한 라벨 정보를 리스트에 추가
        tagged_y_label.append(token_y)
        # 검증하기 위한 태그 토큰 정보를 리턴
        token_tag_data.append(token_tags)

    return tagged_y_label, token_tag_data

# BIO 태깅 방법론으로 태깅처리된 y데이터와 태깅 검증을 위한 데이터(token_tag) 두개 반환
tagged_y_label, token_tag_data = BIO_tagging(tokenized_x_data, tag_data, word_tokenizer)

위 함수의 구동결과가 잘 되었는지 확인하는 코드까지 구동하면 어떻게 동작하는지 가늠될 것이다.

def val_BIO_tagging(raw_x, tok_x, tag, tok_tag, tok_y, sample_idx):
    print(f"원문 데이터 raw_x : {raw_x[sample_idx]}")
    print(f"토큰화 데이터(x)  : {tok_x[sample_idx]}")
    print(f"태그 데이터(tag)  : {tag[sample_idx]}")
    print(f"토큰 태그데이터   : {tok_tag[sample_idx]}")
    print(f"라벨링 데이터(y)  : {tok_y[sample_idx]}")

결과물을 확인한다면

두개의 태그 [마산합포구 : LOC], [친절요양병원 : ORG]
가 BIO Tagging방법론으로 각각 접두어 B-, I-가 붙으며,

그 외 토큰은 모두 O처리됨을 확인할 수 있다.

새로이 설계한 BIO_tagging, val_BIO_tagging 함수는 NLP_pp.py https://github.com/tbvjvsladla/ASH_NLP_lacture
에 추가하여 코드를 업데이트 했다.

1.3 나머지 텍스트 전처리

여기서부터는 통상적인 텍스트 전처리 과정이랑 거의 동일하다

데이터셋 분리

데이터셋은 위와 같이 80%, 15%, 5% 순으로 분리한다.

from sklearn.model_selection import train_test_split

# 훈련/검증/평가를 80%, 15%, 5%로 분할을 수행
# random_state -> 데이터셋을 내누는데 '재현성' 유지를 위해 넣음 -> 안넣어도 됨
# stratify -> y 클래스 비율을 알기 어렵기에 해당 항목은 없앰
x_train, x_etc, y_train, y_etc = train_test_split(
    tokenized_x_data, tagged_y_label, test_size=0.20
)

# 그 외 데이터셋을 반반으로 Val, Test로 나눔
x_val, x_test, y_val, y_test = train_test_split(
    x_etc, y_etc, test_size=0.25
)

단어장 만들기

단어장을 만들 때에는 기존의 텍스트 전처리 방법론인

bag of words 생성 후 희소단어 삭제 작업
을 수행하지 않고 '개체 태깅'된 단어만 bag of words로 생성한다

도식으로 본다면 O로 태깅된 단어들은 단어장 생성에 참여하지 않는다 이렇게 보면 된다.

위 과정에 대한 코드 작성은 아래와 같다

# NER 태깅용 단어장을 만드는 함수, 이때 토큰화된 태그 종류도 몇종인지 확인한다.
def set_vocab_label_forNER(tokenized_x_data, tagged_y_label, 
                      report=False):
    vocab = set() # 중복 회피를 위해 집합 변수로 선언

    for tokens, tag_labels in zip(tokenized_x_data, tagged_y_label):
        for token, tag_label in zip(tokens, tag_labels):
            # 토큰이 'O' 로 태깅이 된 경우가 아니면 vocab에 삽입
            if tag_label != 'O':
                vocab.add(token)

    vocab = list(vocab) #처리 완료 후 리스트로 변환

    # 태그종류(class)를 정렬하는 함수는 콜백함수로 뺀다.
    sorted_tags = sort_tags(tagged_y_label)

    if report:
        print(f"단어장에 포함된 단어는 {len(vocab)}")
        print(f"태깅된 항목 종류(class):")
        for idx, vel in enumerate(sorted_tags):
            print(vel, end=', ')
            if (idx+1) % 5 == 0:
                print()

    return vocab, sorted_tags


# 태그종류(class)를 정렬하는 함수
def sort_tags(tagged_y_label):
    # 태그로 라벨링된 데이터에서 중복 항목 제거
    unique_tags = set(label for labels in tagged_y_label for label in labels)
    # 중복을 제거한 태그 항목을 보기 좋게 정렬
    sorted_tags = ['O'] if 'O' in unique_tags else []
    other_tags = sorted(
        [label for label in unique_tags if label != 'O'],
        key=lambda x: (x[2:], x[0]) #두개의 조건으로 정렬
        # 여기서 두개의 조건은 태그 단어의 첫번째, 세번째 단어임
    )
    sorted_tags.extend(other_tags) # 정렬이 완료된 태그항목

    return sorted_tags

여기서 BIO Tagging 방법론으로 태깅 처리된 항목이 총 몇 종(class)인지 확인할 필요성이 있어 이에 대한 변수
sorted_tags도 함께 출력 및 연산을 수행하도록

set_vocab_label_forNER함수를 작성했다.

아무튼 수행 결과는 아래와 같다.

vocab, tags = set_vocab_label_forNER(tokenized_x_data, tagged_y_label,
                                     report=True)

중요 키워드(entity)항목만 단어장에 참가시키기에
양은 꽤 적은걸 확인할 수 있고
태깅 종류는.. 음 꽤 있는듯 하고
BIO Tagging 규칙은 잘 준수하는 것을 검증했다.

여기서 주의할 점은
훈련, 검증 데이터셋만 단어장에 참가시키고
평가 데이터셋은 단어장에서 빼기로 했으니

함수를 다시 구동하도록 하자

# 훈련/검증 데이터셋만 단어장 생성 및 클래스 종류 식별에 사용
vocab_1, tag_1 = set_vocab_label_forNER(x_train, y_train)
vocab_2, tag_2 = set_vocab_label_forNER(x_val, y_val)

vocab = list(set(vocab_1 + vocab_2))
tags = list(set(tag_1 + tag_2))

# 훈련 검증 데이터셋만 포함시킨 단어장
print(f"단어장에 포함된 단어는 : {len(vocab)}")
# 태깅된 항목 종류 출력
s_tags = sort_tags([tags])
print(s_tags)

수행 결과를 살펴보면

단어가 한 10개?... 정도 빠지고
클래스 종류는 변동이 없다.

정수인코딩

이제 범주형 데이터 $\rightarrow$ 정수형 데이터
변환인 정수인코딩을 수행해야 하는데
x_data, y_label두개 항목을 각각 진행해야 한다.

이때 당연히 {범주형 데이터 : 정수형 인덱스}쌍을 갖는
word_to_idx 딕셔너리를 만들어야 하는데
이걸 y_label용인 tag_to_idx 도 만들어야 하는 것이다.

이를 그림으로 표현하면 위와 같아진다.

이때 x_data의 Spec_token은 PAD, UNK 두개를 사용하는건 기존 방식과 동일하나
y_label은 O 항목이 사실상 UNK와 같은 기능을 하기에
PAD만 Spec_token으로 추가해주면 된다.

위 과정을 코드화 하면 아래와 같아진다

spec_x_token = ['<PAD>', '<UNK>'] # 스페셜 토큰(x)용 선언
spec_y_tag = ['<PAD>'] # 스페셜 태깅토큰(y) 선언

word_to_idx, idx_to_word = set_word_to_idx(spec_x_token, vocab, 
                                           report=True)
print()
tag_to_idx, idx_to_tag = set_word_to_idx(spec_y_tag, s_tags, 
                                           report=True, 
                                           content='태그')

# x_data(원문)의 정수 인코딩 수행
e_x_train = text_to_sequences(x_train, word_to_idx)
e_x_val = text_to_sequences(x_val, word_to_idx)
e_x_test = text_to_sequences(x_test, word_to_idx)

# y_label(태그)의 정수 인코딩 수행
e_y_train = text_to_sequences(y_train, tag_to_idx, spec_token='O')
e_y_val = text_to_sequences(y_val, tag_to_idx, spec_token='O')
e_y_test = text_to_sequences(y_test, tag_to_idx, spec_token='O')

정수인코딩 결과물을 본다면 x_data, y_label 각각 알맞게 수행됨을 확인할 수 있다.

문장 패딩

# 문서 당 시퀀스 길이를 정하는 하이퍼 파라미터
context_length = 50
set_sent_pad(e_x_train, context_length, report=True)

# x_data(원문)의 문장 패딩(정수인코딩의 완료)
padded_x_train = pad_seq_x(e_x_train, context_length)
padded_x_val = pad_seq_x(e_x_val, context_length)
padded_x_test = pad_seq_x(e_x_test, context_length)

# y_label(태그)의 문장 패딩(정수인코딩의 완료)
padded_y_train = pad_seq_x(e_y_train, context_length)
padded_y_val = pad_seq_x(e_y_val, context_length)
padded_y_test = pad_seq_x(e_y_test, context_length)

데이터 로더 생성

import torch
bs = 128 # Batch_size 하이퍼 파라미터

# 정수(원핫)인코딩 데이터를 데이터로더로 변환
trainloader = set_dataloader(padded_x_train, padded_y_train, bs, 
                             content='훈련', report=True)
valloader = set_dataloader(padded_x_val, padded_y_val, bs,
                           content='검증', report=True)
testloader = set_dataloader(padded_x_test, padded_y_test, bs, 
                            content='평가', report=True)

여기까지 수행했다면 NER(개체명 인식)을 위한
데이터의 전처리는 모두 끝난 것이다.

전체적으로 과정은 일반적인 NLP데이터 전처리 과정을 모두 수행하지만,

NER에서 수행하는 BIO Tagging으로 인해서 y_label 데이터가 살짝 복잡해진다

이 복잡해지는 것을 고려해 전처리 과정에 사용하는 함수를 조금씩 개선하면 된다.

위 NER과정을 고려하여 NLP 전처리 함수모음집 NLP_pp.py의 코드 업데이트를 수행했다.

https://github.com/tbvjvsladla/ASH_NLP_lacture 에 업로드 하였으니

1. NLP-Text 전처리 마침 : 모듈화 포스트에 기재한 NLP_pp.py 사용법을 함께 참조하면
쉽게 NER - Text Preprocessing(텍스트 전처리)를 수행할 수 있을 것이다.

---

2. Bi - RNN(LSTM, GRU) NER

NER 과정을 설명하면서 힘들게 Text Preprocessing(텍스트 전처리)를 완료했으니

이제 언어모델을 적용하여 학습/검증 그리고 평가를 진행해 보려 한다.

언어모델은 당연히 RNN, LSTM, GRU 를 사용하려 하는데

이번에는 양방향(Bidirectional) 옵션을 적용하여 학습/검증/평가를 진행하고자 한다.

그림으로 본다면 seq_data는 순서가 있으니
이 순서를 '정방향', 그리고 '역방향'

두 방향으로 각각 데이터를 읽어서 학습시킨다는게 주요 아이디어 이며,

cell의 개수가 x2배 되는 효과가 발생하고
각 셀의 데이터는 중간에 concat연산이 적용되서

각각 아래와 같이 변화한다.

출력 $y_{t}$ 차원 : (Batch_size, context_length, hid_dim x 2)

hidden 차원 : (num_layers x 2, Batch_size, hid_dim)

따라서 셀의 구조가 복잡해지기에 양방향(Bidirectional) RNN을 만드는 것은 어려울 것으로 생각되나

위 사진처럼 파이토치에서는 bidirectional=True로 옵션 하나만 딸깍 바꿔주면 되기에 코드화는 그리 어려운 편은 아니다.

아무튼 양방향 모델 설계에 대한 공부를 마쳣으니
코드를 작성해보자

주요 하이퍼 파라미터 정의

# 주요 하이퍼 파라미터 정리
VOCAB_SIZE = len(word_to_idx)
CONTEXT_LEN = context_length
EMB_DIM = 100 # 임베딩 차원은 100으로
NUM_Tags = len(tag_to_idx)
NUM_Layers = 2 #셀의 레이어는 1층이 아니라 2층으로
HIDE_DIM = 500

RNN계열 기반 NERTagger 모델 설계

import torch.nn as nn

class NERTagger_LSTM(nn.Module):
    def __init__(self, vocab_size, embed_dim, tag_dim,
                 hid_dim, num_layers, emb_matirx=None):
        super(NERTagger_LSTM, self).__init__()

        self.embed = nn.Embedding(vocab_size, embed_dim)
        # 사전훈련 임베딩 사용 유/무 함수
        if emb_matirx is not None:
            self.embed.weight = nn.Parameter(
                torch.tensor(emb_matirx, dtype=torch.float32))
            self.embed.weight.requires_grad = True

        self.lstm = nn.LSTM(input_size=embed_dim, # LSTM에 입력차원
                            hidden_size=hid_dim, # LSTM의 출력차원
                            num_layers=num_layers, # 내부 Cell 몇층?
                            bidirectional=True, #양방향 학습 옵션 On
                            batch_first=True) # 입력텐서 -> batch가 맨처음
        
        # 최종적으로 tag 종류를 맟추는 FC layer
        # 양방향 연산이기에 FC_layer에 입력되는 feature는 2배로 늘어난다
        self.classifier = nn.Sequential(
            nn.Linear(hid_dim*2, tag_dim),
        )

    def forward(self, x):
        # 입력 x : batch_size, seq_length
        emb = self.embed(x) # (batch_size, seq_length, embedding_dim)
        # 양방향 학습 On이기에
        # lstm_out : (bs, seq_len, hidden_dim * 2)
        # hidden = (num_layer * 2, bs, hidden_dim)
        lstm_out, (hidden, cell) = self.lstm(emb)

        # many-to-many 방식이니까 lstm_out을 사용한다.
        logits = self.classifier(lstm_out)
        # logits 출력 : (bs, seq_len, tag_dim)
        return logits

import torch.nn as nn

class NERTagger_GRU(nn.Module):
    def __init__(self, vocab_size, embed_dim, tag_dim,
                 hid_dim, num_layers, emb_matirx=None):
        super(NERTagger_GRU, self).__init__()

        self.embed = nn.Embedding(vocab_size, embed_dim)
        # 사전훈련 임베딩 사용 유/무 함수
        if emb_matirx is not None:
            self.embed.weight = nn.Parameter(
                torch.tensor(emb_matirx, dtype=torch.float32))
            self.embed.weight.requires_grad = True

        self.gru = nn.GRU(input_size=embed_dim, # GRU에 입력차원
                            hidden_size=hid_dim, # GRU의 출력차원
                            num_layers=num_layers, # 내부 Cell 몇층?
                            bidirectional=True, #양방향 학습 옵션 On
                            batch_first=True) # 입력텐서 -> batch가 맨처음
        
        # 최종적으로 tag 종류를 맟추는 FC layer
        # 양방향 연산이기에 FC_layer에 입력되는 feature는 2배로 늘어난다
        self.classifier = nn.Sequential(
            nn.Linear(hid_dim*2, tag_dim),
        )

    def forward(self, x):
        # 입력 x : batch_size, seq_length
        emb = self.embed(x) # (batch_size, seq_length, embedding_dim)
        # 양방향 학습 On이기에
        # gru_out : (bs, seq_len, hidden_dim * 2)
        # hidden = (num_layer * 2, bs, hidden_dim)
        gru_out, hidden = self.gru(emb)

        # many-to-many 방식이니까 gru_out을 사용한다.
        logits = self.classifier(gru_out)
        # logits 출력 : (bs, seq_len, tag_dim)
        return logits

코드를 본다면 LSTM, GRU버전 NER_tagger 모두 거의 동일한 구조를 띄고 있으며

many-to-many 옵션으로 모델을 사용하는 것이기에

lstm_out, gru_out 항목을 FC_layer에 넘겨주어 Tag를 맞추는 작업을 수행한다.

실험준비

# 학습 실험 조건을 구분하기 위한 키
model_key = ['LSTM', 'GRU']
metrics_key = ['Loss', '정확도']

# 모델 선언
Lstm_tagger = NERTagger_LSTM(VOCAB_SIZE, EMB_DIM, NUM_Tags,
                             HIDE_DIM, NUM_Layers)
Gru_tagger = NERTagger_GRU(VOCAB_SIZE, EMB_DIM, NUM_Tags,
                             HIDE_DIM, NUM_Layers)

# GPU사용 가능 유/무 확인
device = torch.device("cuda" if torch.cuda.is_available() else 'cpu')
models = {} # 딕셔너리

models[model_key[0]] = Lstm_tagger.to(device)
models[model_key[1]] = Gru_tagger.to(device)

손실함수 + 옵티마이저 설계

import torch.optim as optim
# 로스함수 및 옵티마이저 설계
# 로스함수에서 <PAD> 토큰의 정수인덱스 번호 -> 0번에 대해서는
# 틀리건 맞건 무시하겠다 : ignore_index에 해당 정수 인덱스 번호 기입
criterion = nn.CrossEntropyLoss(ignore_index=0)

LR = 0.001 # 러닝레이트는 통일
optimizers = {}

optimizers[model_key[0]] = optim.Adam(Lstm_tagger.parameters(), lr=LR)
optimizers[model_key[1]] = optim.Adam(Gru_tagger.parameters(), lr=LR)

위 손실함수를 보면

ignore_index=0 이라는 특이한 인자값이 하나 들어가 있다.

이것은 x_data $\rightarrow$ 모델 통과 $\rightarrow$ 예측값(y_pred)

정답지(y_label)과 예측값(y_pred)를 비교 $\rightarrow$ 손실함수

이 과정에서
정답지(y_label) : 0 : PAD 토큰인 항목에 관해서는
예측값(y_pred)값이 0으로 맞추던 못 맞추던 관심을 끄겠다는 옵션이다.

괜히 PAD 토큰까지 일일히 맞춰야 한다고 손실함수를 설계해 버리면 정작 의미있는 Tag를 맞추는 작업 성능이 하락하기에
의도적으로 PAD토큰은 정답을 맞추던 못맞추던 관여를 안하게끔 설계해야 한다.

학습 준비

이제 여기서 모델 학습/검증을 위한 코드를 구동하면 되는데

이전 포스트에서는 C_ModelTrainer.py 학습/검증을 위해 따로 작성한 코드

https://github.com/tbvjvsladla/ASH_NLP_lacture/blob/main/C_ModelTrainer.py

이 파일을 불러와서 간단하게 학습/검증을 수행했지만

NER(개체명 인식)의 모델 예측값($y_{pred}$)과 정답지($y_{label}$)의 차원이 살짝 달라지기에
이를 보정하는 과정이 필요하다

따라서 코드가 살짝 달라져셔 아에 새로이 코드작성을 진행했다

차원변환을 어떻게 진행해야하는지는 위 도식으로 표현할 수있다.
요약을 하자면 $y_{pred}$는 3차원, $y_{label}$는 2차원 데이터이기에 각각 view 메서드를 사용해서
CressEntropyLoss 메서드가 연산 가능한 2차원, 1차원으로 변환해야 한다

이에 대한 코드는 아래와 같다.

def data_reshape(y_pred, y_label):
    # 입력되는 y_pred : (Batch_size, seq_len, output_dim)
    # 입력되는 y_label : (Batch_size, seq_len)
    batch_size, seq_len, output_dim = y_pred.size()

    re_y_pred = y_pred.view(-1, output_dim) # (bs*seq, out)
    re_y_label = y_label.view(-1) #(bs*seq)

    return re_y_pred, re_y_label

다음으로 수행할 항목은 $y_{label}$의 현재 맞춰야 할 값이 PAD : 0값이면 이를 정답지에 카운트 하지 않는 함수를 설계해야 한다.

NER에서 문장패딩 토큰까지 정답을 맞추라는것은 너무 가혹하고 의미가 없는 행동이기에 위 과정을 코드화 해야한다

마치 segmentation에서 배경에 해당하는 class는 정답을 맞추던 말던 관심을 끄는 과정이랑 동일하다 보면 된다.

# 정답을 맞출 때 '무시'해야 할 클래스가 있을때 동작하는 함수
def cal_correct(y_pred, y_label, ignore_class):
    pred = y_pred.argmax(dim=1) #가장 높은 예측값 하나 추출

    if ignore_class is not None: #무시해야할 클래스가 있을 때
        mask = (y_label != ignore_class)
        correct = pred.eq(y_label).masked_select(mask).sum().item()
        total = mask.sum().item() # 전체 원소 개수중 마스크처리된것만
    else:
        correct = pred.eq(y_label).sum().item()
        total = y_label.numel() # 전체 원소 수 출력
    
    # 수치적 안정성을 보장하면서 연산을 수행하자
    iter_cor = correct / total if total > 0 else 0
    return iter_cor

위 두개의 함수를 callback함수로 받아서
전체적인 model_train / model_evaluate 함수를 개선해야 하다보니

기존의 모듈화된 코드를 더 업데이트 했다가는 스파게티 소스가 될것 같아서 아에 새로 설계했다...

def model_train(model, data_loader, loss_fn, optimizer_fn, 
                epoch, epoch_step, ignore_class=None):
    # 1개의 epoch내 batch단위(iter)로 연산되는 값이 저장되는 변수들
    iter_size, iter_loss, iter_correct = 0, 0, 0

    device = next(model.parameters()).device # 모델의 연산위치 확인
    model.train() # 모델을 훈련모드로 설정

    #특정 epoch_step 단위마다 tqdm 진행바가 생성되게 설정
    if (epoch+1) % epoch_step == 0 or epoch == 0:
        tqdm_loader = tqdm(data_loader)
    else:
        tqdm_loader = data_loader

    for x_data, y_label in tqdm_loader:
        x_data, y_label = x_data.to(device), y_label.to(device)
        y_pred = model(x_data) # Forward, 모델이 예측값을 만들게 함
        # 데이터셋의 구조 변형이 필요할 때 아래 함수 구동
        y_pred, y_label = data_reshape(y_pred, y_label)

        loss = loss_fn(y_pred, y_label) #로스함수로 예측갑~정답 차이계산

        #backward 과정 수행
        optimizer_fn.zero_grad()
        loss.backward()
        optimizer_fn.step() # 마지막에 스케줄러 있으면 업뎃코드넣기

        # 현재 batch 내 샘플 개수당 correct, loss, 수행 샘플 개수 구하기
        iter_correct += cal_correct(y_pred, y_label, 
                                    ignore_class) * x_data.size(0)
        iter_loss += loss.item() * x_data.size(0)
        iter_size += x_data.size(0)

        # tqdm에 현재 진행상태를 출력하기 위한 코드
        if (epoch+1) % epoch_step == 0 or epoch == 0:
            prograss_loss = iter_loss / iter_size
            prograss_acc = iter_correct / iter_size
            desc = (f"[훈련중]로스: {prograss_loss:.3f}, "
                    f"정확도: {prograss_acc:.3f}")
            tqdm_loader.set_description(desc)

    #현재 epoch에 대한 종합적인 정확도/로스 계산
    epoch_acc = iter_correct / iter_size
    epoch_loss = iter_loss / len(data_loader.dataset)
    return epoch_loss, epoch_acc

def model_evaluate(model, data_loader, loss_fn, 
                    epoch, epoch_step, ignore_class=None):
    # 1개의 epoch내 batch단위(iter)로 연산되는 값이 저장되는 변수들
    iter_size, iter_loss, iter_correct = 0, 0, 0

    device = next(model.parameters()).device # 모델의 연산위치 확인
    model.eval() # 모델을 평가 모드로 설정

    #특정 epoch_step 단위마다 tqdm 진행바가 생성되게 설정
    if (epoch+1) % epoch_step == 0 or epoch == 0:
        tqdm_loader = tqdm(data_loader)
    else:
        tqdm_loader = data_loader

    with torch.no_grad(): #평가모드에서는 그래디언트 계산 중단
        for x_data, y_label in tqdm_loader:
            x_data, y_label = x_data.to(device), y_label.to(device)
            y_pred = model(x_data) # Forward, 모델이 예측값을 만들게 함
            # 데이터셋의 구조 변형이 필요할 때 아래 함수 구동
            y_pred, y_label = data_reshape(y_pred, y_label)

            loss = loss_fn(y_pred, y_label) #로스함수로 예측갑~정답 차이계산

            # 현재 batch 내 샘플 개수당 correct, loss, 수행 샘플 개수 구하기
            iter_correct += cal_correct(y_pred, y_label, 
                                        ignore_class) * x_data.size(0)
            iter_loss += loss.item() * x_data.size(0)
            iter_size += x_data.size(0)

    #현재 epoch에 대한 종합적인 정확도/로스 계산
    epoch_acc = iter_correct / iter_size
    epoch_loss = iter_loss / len(data_loader.dataset)
    return epoch_loss, epoch_acc

학습 시작

num_epoch = 8 #총 훈련/검증 epoch값
ES = 2 # 디스플레이용 에포크 스텝

for key in model_key:
    print(f"\n--현재 훈련중인 조건: [{[key]}]--") # 조건에 맞는 실험시작
    for epoch in range(num_epoch): #에포크별 모델 훈련/검증
        # 모델 훈련
        train_loss, train_acc = model_train(
            models[key], trainloader, criterion,
            optimizers[key], epoch, ES, 
            ignore_class=ignore_class_idx #무시할 클래스 인덱스
        ) #모델 검증
        val_loss, val_acc = model_evaluate(
            models[key], valloader, criterion, 
            epoch, ES, ignore_class=ignore_class_idx #무시할 클래스 인덱스
        )

        # 손실 및 성과 지표를 history에 저장
        history[key]['Loss'].append((train_loss, val_loss))
        history[key]['정확도'].append((train_acc, val_acc))

        # Epoch_step(ES)일 때마다 print수행
        if (epoch+1) % ES == 0 or epoch == 0:
            print(f"epoch {epoch+1:03d}," + "\t" + 
                f"훈련 [Loss: {train_loss:.3f}, " +
                f"Acc: {train_acc*100:.2f}%]")
            print(f"epoch {epoch+1:03d}," + "\t" + 
                f"검증 [Loss: {val_loss:.3f}, " +
                f"Acc: {val_acc*100:.2f}%]")
    print(f"--조건[{[key]}] 훈련 종료--\n") # 조건에 맞는 실험종료

NER(개체명인식) 결과를 보니...
이거는 각 seq별로 예측을 태그/태그가아닌것 다 따로따로 예측을 해야해서 좀 정확도가 떨어질줄 알았지만

음.. 잘 맞춘다?

아무튼 LSTM, GRU로 NER을 Many-to-Many 방법론으로 수행하는 실습은 이것으로 마치겠다.