NLP Transformer

Attention is all you need (2017)

기계 번역을 목적으로 만들어 졌을 것으로 예상

• 인코더: 
• skip 커넥션(resnet), Batch Norm(yolov2) 을 갖고 있다.
  
  
• 디코더:
• 포지셔널 인코딩
• 병렬적으로 처리한다. 순차적으로 처리하지 않는다.
• 순차적이지 않기 때문에 위치적 고유 값을 부여한다.

Attention(Q, K, V)는 순차적일 필요가 없다.

• Q: query, t시점(t시퀀스)의 디코더 LTSTM의 은닉값 (벡터 형태의 hidden)
• K: key, 모든 시점의 인코더 셀의 은닉값 (벡터 형태의 hidden)
• V: =K

Self-Attention

• 인코더 안에서 전체 단어와 각 단어 간에 유사도를 구한다.
• skip 커넥션(resnet), Batch Norm(yolov2) 을 사용한다.

논문 살펴보기

1) 모델의 구조, 2) 학습 방식, 3) 성능개선 전략

RNN에 Attention을 사용했던 복잡한 네트워크 구조에서 Attention만 사용하는 단순한 구조를 사용했다. 과연, 어떻게 순서 정보 없이 병렬 처리로 가능할까? (어텐션 & 포지셔널인코딩)

병렬 처리의 장점은 무엇인가? 

한 번에 입력 받는다 -> 학습 속도가 빠르다.

순서 정보가 학습되어 있으면, 순서 정보 없이 입력이 들어와도 문맥 파악이 가능하다.

Model Architecture

Encode: [two sub-layers with skip-connection] 를 6개, 총 12층

1. 쿼리, 키, 밸류가 인코더에서 모두 처리된다.

• Queries: 현재 처리 단어, 기존에 디코더에서 하나씩 입력하던 것
• keys: 각 토큰과의 상관관계, 예를 들어 유사도
• values: 단어의 의미

2. Scaled Dot-Product Attention

> MatMul

> Scale(크기 조절)
> Mask(특정 토큰을 유사도 계산 비활성화. 아주 큰 음수를 곱함으로써, y값을 0에 가깝게 만들어 비활성화 시킨다. ex:패딩)
> SoftMax (여기까지가 가중치를 만드는 단계다.)
> MatMul(SoftMax에 Value를 곱한다.)

한 번의 어텐션을 하는 것보다 여러 번의 어텐션을 병렬로 사용하는 것이 효과적이라고 판단.

heads를 8개를 둔다. Multi-Head는 8개의 묶음이다.

3. Feed Forward

소규모 완전연결층

FFX(x) = max(0, xW1 + b1)W2 + b2

512 > 2048 > 512

Decoder: [multi-head attention * 2 + FC] 가 6개, 총 18층

Seq2Seq의 모델과 달리 Attention만 있는 트랜스포머의 경우,

디코더를 시작하는 첫번째 Attention에 Context Vector를 사용하지 않는다.
인코더의 결과값(문맥파악)을 참고하지 않고 먼저 Attention을 구하고
그 다음에 인코더 값을 참조한다.

또한, 첫번째 Attention은 마스킹 처리가 있다.
(인코더는 양방향, 디코더는 단방향)
(입력은 병렬로 하더라도, 출력은 순차적으로 해야 한다.)

자기 앞에 있는 애들까지만 유사도 계산을 하고,
뒤에 나올 토큰은 유사도 계산을 하지 않는다. (우측 하단 녹색 계단 참고)

두번째 Attention - 인코더 디코더 어텐션: 

Query = 디코더 행렬

key, Value = 인코더 행렬

어텐션 (셀프 어텐션, 단어의 문맥 파악)

포지셔널 인코딩

ex) 

1. 임베딩: 나는 오늘 학교에 간다 -> 각 단어를 벡터화

2. 위치적 고유값, PE (동일한 n차원)

• 어떤 특정 토큰을 다른 토큰으로 설명할 수 있다.

• pos는 위치 값: 0, 1, 2, 3
• i는 단어 하나 임베딩 벡터의 차원 값
  
• 왜 sin, cos 인가? 겹치지지만 않으면 되지 않을까? 랜덤 돌려도 될텐데?
• 결과 값이 선형(단순한) 관계가 될 수 있도록,
  선형으로 예측 가능한 비선형을 주입한다. 학습시 명료함을 주기 위해.
• sin(짝수) 과 cos(홀수) 두 개를 사용하면 겹칠 일이 상쇄한다.
• 0(2i): sin(... i=0)
• 1(2i+1): cos(...i=0)
• 2: sin(....i=1이므로 2)
• 3: cos(... i=1이므로 2)
  
  
• 10000은 무엇인가? 단순 실험 값.
• 지수로 처리한 이유는?
• 인간이 연구한 값 중, 익숙한 값은 로그 값. 그래서 로그의 역함수인 지수를 사용

3. 단어 벡터화(n차원) + 위치값(동일한 n차원) = 인풋(n차원)

BERT & GPT

BERT의 핵심은 분류, GPT는 생성형 챗봇이 목적이었다.

분류 레이블 없이 문서만 가지고 학습한다.

GPT

• 디코더만 사용하는 단방향

BERT

• 인코더만 사용하는 양방향
• 문장의 특정 단어에 마스크 처리를 해서, 단어를 맞추는 학습을 한다.
• 서브워드 토크나이저 사용
• NSP: 다음 문장 예측. 두 개의 문장을 준 후에 이 문장이 이어지는 문장인지 아닌지 맞추는 방식

실습

# 토큰화를 위한 토크나이저

from transformers import BertTokenizer

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

tokenizer = BertTokenizer.from_pretrained("bert-base-multilingual-cased")

전처리

질문과 답이 쌍을 이루는 학습데이터

토큰 길이

패딩 및 인코딩

# 질문 패딩 및 인코딩

def src_encoding(tokens):

    # (src_length - 토큰 수) 만큼 패딩토큰 추가 - 패딩토큰을 추가하여 최대길이로 맞춤

    tokens = tokens + ['<PAD>'] *  (src_length - len(tokens))

    # 인코딩 된 숫자를 담아 둘 리스트

    index_sequences = []

    # 문장에서 토큰을 꺼내오며

    for word in tokens:

      try: # 토큰 인코딩 (단어 사전에 없는 토큰이 들어오면 except로)

          index_sequences.append(vocab[word])

      except KeyError: # 단어 사전에 없는 토큰이 들어오면 '<UNK>' 토큰의 숫자로 변환

          index_sequences.append(vocab['<UNK>'])

    return index_sequences

# 답변문 패딩 및 인코딩

def trg_encoding(tokens):

    # (src_length - 토큰 수) 만큼 패딩토큰 추가

    tokens = tokens + ['<PAD>'] *  (trg_length - len(tokens))

    # 인코딩 된 숫자를 담아 둘 리스트

    index_sequences = []

    # 문장에서 토큰을 꺼내오며

    for word in tokens:

      try: # 토큰 인코딩 (단어 사전에 없는 토큰이 들어오면 except로)

          index_sequences.append(vocab[word])

      except KeyError: # 단어 사전에 없는 토큰이 들어오면 '<UNK>' 토큰의 숫자로 변환

          index_sequences.append(vocab['<UNK>'])

    return index_sequences

데이터셋 클래스 생성 -> 데이터프레임에서 전처리된 데이터를 가져와서 텐서로 변환

트랜스포머 모델의 구축 순서

1. 포지셔널 인코딩 클래스

2. 마스킹 함수

3. 모델 설계

# 수학 연산을 위한 math 라이브러리

import math

class PositionalEncoding(nn.Module):

    # 클래스 생성 시 emb_size, dropout, maxlen을 입력으로 받음

    def __init__(self, emb_size, dropout, maxlen=5000):

        super(PositionalEncoding, self).__init__()

        # (maxlen * emb_size)크기의 PostionalEncoding 행렬 생성 후 원본과 타겟입력값에 더해줌

        # den :10000^(2i/d_model) 구현

        den = 10000 ** (torch.arange(0, emb_size, 2) / emb_size)

        # position

        pos = torch.arange(0, maxlen).reshape(maxlen, 1)

        # 영 행렬 생성(maxlen, emb_size)

        pos_embedding = torch.zeros((maxlen, emb_size))

        # 영 행렬의 짝수 열은 사인함수 적용(0,2,4,6 .....)

        pos_embedding[:, 0::2] = torch.sin(pos * den)

        # 영 행렬의 홀수 열은 코사인 함수 적용(1,3,5,7)

        pos_embedding[:, 1::2] = torch.cos(pos * den)

        # pistional embedding에 배치 차원 추가(1, maxlen, embsize)

        pos_embedding = pos_embedding.unsqueeze(0)

        # 논문에서는 포지셔널인코딩을 거친 후 드롭아웃을 적용

        self.dropout = nn.Dropout(dropout)

        # PyTorch의 nn.Module에서 제공하는 기능으로, 모델의 상태에 포함되지만 학습되지 않는 텐서를 등록할 때 사용

        # PositionalEncoding 벡터는 변하지 않고 고정

        self.register_buffer('pos_embedding', pos_embedding)

    def forward(self, token_embedding):

        # 임베딩 벡터와(batch, seq_length, embsize) PE벡터(1, maxlen, embsize) 더함

        # 배치의 각 데이터들에대해 각각 PE 더함

        # 이후 드롭아웃 적용

        return self.dropout(token_embedding + self.pos_embedding[:, :token_embedding.size(1), :])

class Seq2SeqTransformer(nn.Module):

    def __init__(self, num_encoder_layers, num_decoder_layers, emb_size,

                 nhead, src_vocab_size, tgt_vocab_size, dim_feedforward=512,

                 dropout=0.1):

        super(Seq2SeqTransformer, self).__init__()

        # 모델에서 사용할 레이어 정의

        # 임베딩 레이어

        self.emb_size = emb_size

        self.embedding = nn.Embedding(vocab_size, emb_size)

        # 포지셔널 인코딩 레이어

        self.positional_encoding = PositionalEncoding(emb_size, dropout=dropout)

        # 트랜스포머 레이어(인코더+디코더)

        # d_model : 입출력 차원 수, num_encoder(decoder)_layers : 인코더(디코더)의 층 수)

        # dim_feedforward : FFNN의 차원 수(attention 거친 후 FFNN에서 dmodel -> dim_feedforward -> dmodel)

        self.transformer = nn.Transformer(d_model=emb_size,

                                       nhead=nhead,

                                       num_encoder_layers=num_encoder_layers,

                                       num_decoder_layers=num_decoder_layers,

                                       dim_feedforward=dim_feedforward,

                                       dropout=dropout,

                                       batch_first=True)

        # 출력 레이어(단어사전에 있는 단어로 출력될 수 있도록)

        self.fc = nn.Linear(emb_size, tgt_vocab_size)

    def forward(self, src, tgt, src_mask, tgt_mask, src_padding_mask, tgt_padding_mask, memory_key_padding_mask):

        # 질문, 답변문 임베딩

        # 논문에서는 임베딩 벡터의 스케일을 조정하여 초기화 과정에서의 불안정성을 줄이기 위해 임베딩 크기의 제곱근을 곱함

        src, tgt = src.long(), tgt.long()

        src = self.embedding(src) * math.sqrt(self.emb_size)

        tgt = self.embedding(tgt) * math.sqrt(self.emb_size)

        # 포지셔널 인코딩 적용

        src_emb = self.positional_encoding(src)

        tgt_emb = self.positional_encoding(tgt)

        # 트랜스포머 모델 통과

        # src_emb : 질문 데이터, tgt_emb : 답변문 데이터, src_mask : 질문 마스크, tgt_mask : 답변문 마스크(어텐션 시 미래 시점 못보도록)

        # src(tgt)_padding_mask : 질문(답변문)이 어텐션 시 패딩토큰은 적용되지 않도록 패딩 토큰에 마스킹-> self-atteniton에 적용

        # memory_key_padding_mask : 디코더가 인코더의 출력 정보 활용 시 패딩된 부분 활용하지 않도록 마스 -> encoder-decoder attention에 적용

        outs = self.transformer(src_emb, tgt_emb, src_mask, tgt_mask, None, # None 부분은 memory 마스크로 보통 None으로 둠

                                src_padding_mask, tgt_padding_mask, memory_key_padding_mask)

        outs = self.fc(outs)

        return outs # (배치사이즈, 출력시퀀스 길이, 단어 사전 수)

    # 예측 시 적용할 인코딩 함수

    def encode(self, src, src_mask):

        # 예측 시 사용할 인코더 통과(질문 -> 임베딩 -> 포지셔널인코딩 -> 트랜스포머 인코더 통과)

        src = src.long()

        src = self.embedding(src) * math.sqrt(self.emb_size)

        src_emb = self.positional_encoding(src)

        # 실제 예측 시 패딩을 넣지 않기에 패딩 마스크는 적용 X

        outs = self.transformer.encoder(src_emb, src_mask)

        return outs

    # 예측 시 적용할 디코딩 함수

    def decode(self, tgt, memory, tgt_mask):

        # 에측 시 디코더 통과(답변문 -> 임베딩 -> 포지셔널인코딩 -> 트랜스포머 디코더 통과)

        # 예측 시 tgt는 sos 토큰, memory는 인코더의 출력(매 층에서 인코더 디코더 어텐션에 활용)

        tgt = tgt.long()

        tgt = self.embedding(tgt) * math.sqrt(self.emb_size)

        tgt_emb = self.positional_encoding(tgt)

        outs = self.transformer.decoder(tgt_emb, memory, tgt_mask)

        return outs

# 미래 시점 마스크를 위한 함수 생성

def generate_square_subsequent_mask(sz):

    # torch.ones((sz, sz) : (sz, sz)크기의 1로 채워진 행렬 생성

    # torch.triu() : 대각선 기준으로 아래의 원소를 0으로 변경

    # 1 1 1

    # 0 1 1

    # 0 0 1

    # 전치

    # 1 0 0

    # 1 1 0

    # 1 1 1

    mask = torch.triu(torch.ones((sz, sz), device=device)).transpose(0, 1)

    # 0인 부분을 음의 무한 값으로, 1인 부분을 0.0으로 변경

    # 타켓문의 시퀀스 길이 만큼의 마스크 행렬을 생성하여 self-attention 시 미래 시점의 값들은 음의 무한대로 갈 수 있도록

    # 음의 무한대 값은 softmax 통과 시 0으로 수렴

    mask = mask.float().masked_fill(mask == 0, float('-inf')).masked_fill(mask == 1, float(0.0))

    return mask

# 전체 마스크(패딩, 미래시점)를 생성하기 위한 함수

def create_mask(src, tgt):

    # 질문과 답변문의 시퀀스 길이 확인

    src_seq_len = src.shape[1]

    tgt_seq_len = tgt.shape[1]

    # 미래시점 마스크

    tgt_mask = generate_square_subsequent_mask(tgt_seq_len)

    # 질문은 마스킹을 적용하지 않기에 False로 이루어진 행렬 생성

    src_mask = torch.zeros((src_seq_len, src_seq_len), device=device).type(torch.bool)

    # 패딩마스크 생성(입력 데이터 행렬에서 입력 토큰이 0(<PAD>)인 부분을 True로 나머지는 False인 형태로 반환)

    src_padding_mask = (src == 0)

    tgt_padding_mask = (tgt == 0)

    return src_mask, tgt_mask, src_padding_mask, tgt_padding_mask