[코드리뷰] Attention is All You Need-Pytorch [2] Model Architecture

Attention is all you need논문의 pytorch Implementation 코드를 리뷰하는 포스팅이다. 

https://github.com/jadore801120/attention-is-all-you-need-pytorch

GitHub - jadore801120/attention-is-all-you-need-pytorch: A PyTorch implementation of the Transformer model in "Attention is All

Transformer 코드 리뷰는 세 개의 포스팅으로 나누어 진행을 할 것이고 본 포스팅은 두 번째로 모델 구조 부분이다.

다만 자세한 모델 설명은 이전의 포스팅에서 설명했으므로 생략하겠다.

1. Word Embedding

지난 전처리에서 vocab size는 10077이였다. 이 vocab 내 모든 단어들을 고유 벡터로 만들어서 512차원으로 embedding 시켜주기 위해 nn.Embedding을 사용한다.

import torch.nn as nn
vacab_len = 10077
import torch.nn as nn
embedding_layer = nn.Embedding(num_embeddings=vacab_len, 
                               embedding_dim=512,
                               padding_idx=1)

print('vocab 내 모든 단어들을 고유 벡터로 만들어서 512차원으로 embedding 시켜줌')
print(f'LOOK UP TABLE SIZE:  {embedding_layer.weight.shape}')
print(embedding_layer.weight)

vocab 내 모든 단어들을 고유 벡터로 만들어서 512차원으로 embedding 시켜줌
LOOK UP TABLE SIZE:  torch.Size([10077, 512])
Parameter containing:
tensor([[ 0.9914, -0.4633, -1.2087,  ..., -2.0206, -2.9503, -0.4875],
        [ 0.0000,  0.0000,  0.0000,  ...,  0.0000,  0.0000,  0.0000],
        [-0.5413, -0.2046, -0.0866,  ..., -0.1290,  0.0060,  0.8837],
        ...,
        [-0.6048, -0.5161,  0.6198,  ..., -0.0604, -0.0853,  0.4896],
        [ 0.2955,  0.1371,  0.4321,  ..., -0.3920,  0.7107,  0.5773],
        [-0.6375, -0.2004,  0.9800,  ...,  0.6815,  0.3131,  1.8058]],
       requires_grad=True)

 

2. Positional Encoding

transformer에서 문장 내 각 token의 위치정보를 넣어주기 위해 positional encoding을 해준다. Positional Encoding은 embedding과 차원이 동일하며, embedding vector와 더하여 위치정보를 추가하게 된다.

 

class PositionalEncoding(nn.Module):
    def __init__(self, d_hid, n_position=200):
        super(PositionalEncoding, self).__init__()
        # Not a parameter
        self.register_buffer('pos_table', self._get_sinusoid_encoding_table(n_position, d_hid)) # ①
        
    def _get_sinusoid_encoding_table(self, n_position, d_hid):
        '''Sinusoid position encoding table'''
        
        def get_position_angle_vec(position):
            return [position / np.power(10000, 2 * (hid_j // 2) / d_hid) for hid_j in range(d_hid)]
        
        sinusoid_table = np.array([get_position_angle_vec(pos_i) for pos_i in range(n_position)])
        sinusoid_table[:, 0::2] = np.sin(sinusoid_table[:, 0::2]) # dim 2i
        sinusoid_table[:, 1::2] = np.cos(sinusoid_table[:, 1::2]) # dim 2i+1
        
        return torch.FloatTensor(sinusoid_table).unsqueeze(0)
    
    def forward(self, x):
        return x + self.pos_table[:, :x.size(1)].clone().detach() # x : torch.size([328, 36, 512])

1. torch.nn.Module.register_buffer는 parameter가 아니라 말 그대로 buffer를 수행하기 위한 목적으로 활용한다. 자세한 설명은 다음의 포스팅을 참고하면 된다.

3. Encoder

3-1. Multi head Attention

class ScaledDotProductAttention(nn.Module):
    ''' Scaled Dot-Product Attention '''

    def __init__(self, temperature, attn_dropout=0.1):
        super().__init__()
        self.temperature = temperature
        self.dropout = nn.Dropout(attn_dropout)

    def forward(self, q, k, v, mask=None):
        
        # ① attention score 계산 
        attn = torch.matmul(q / self.temperature, k.transpose(2, 3))
        
        # ② attention을 계산하지 않는 위치를 매우작은값으로 치환
        # - 매우작은 값은 softmax를 거칠때 0과 가까운 값이 되므로 무시한다.
        if mask is not None:
            attn = attn.masked_fill(mask == 0, -1e9)

        # ③ soft max를 이용하여 attention weight 계산
        attn = self.dropout(F.softmax(attn, dim=-1))
        
        # ④ 해당 분포값에 v를 곱하여 attention value를 구한다
        output = torch.matmul(attn, v)
        
        return output, attn
        # attention values, attention weight

class MultiHeadAttention(nn.Module):
    ''' Multi-Head Attention module '''

    def __init__(self, n_head, d_model, d_k, d_v, dropout=0.1):
        super().__init__()

        self.n_head = n_head
        self.d_k = d_k
        self.d_v = d_v

        self.w_qs = nn.Linear(d_model, n_head * d_k, bias=False)
        self.w_ks = nn.Linear(d_model, n_head * d_k, bias=False)
        self.w_vs = nn.Linear(d_model, n_head * d_v, bias=False)
        self.fc = nn.Linear(n_head * d_v, d_model, bias=False)

        self.attention = ScaledDotProductAttention(temperature=d_k ** 0.5)

        self.dropout = nn.Dropout(dropout)
        self.layer_norm = nn.LayerNorm(d_model, eps=1e-6)


    def forward(self, q, k, v, mask=None):

        d_k, d_v, n_head = self.d_k, self.d_v, self.n_head
        sz_b, len_q, len_k, len_v = q.size(0), q.size(1), k.size(1), v.size(1)

        residual = q

        # Pass through the pre-attention projection: b x lq x (n*dv)
        # Separate different heads: b x lq x n x dv
        q = self.w_qs(q).view(sz_b, len_q, n_head, d_k) # [256, len, 8, 64]
        k = self.w_ks(k).view(sz_b, len_k, n_head, d_k) # [256, len, 8, 64]
        v = self.w_vs(v).view(sz_b, len_v, n_head, d_v) # [256, len, 8, 64]

        # Transpose for attention dot product: b x n x lq x dv
        q, k, v = q.transpose(1, 2), k.transpose(1, 2), v.transpose(1, 2) # [256, 8, len, 64]

        if mask is not None:
            mask = mask.unsqueeze(1)   # For head axis broadcasting.

        q, attn = self.attention(q, k, v, mask=mask) # [256, 8, len, 64], [256, 8, len, 36]

        # Transpose to move the head dimension back: b x lq x n x dv
        # Combine the last two dimensions to concatenate all the heads together: b x lq x (n*dv)
        q = q.transpose(1, 2).contiguous().view(sz_b, len_q, -1) # [256, 36, 512]
        q = self.dropout(self.fc(q))
        q += residual

        q = self.layer_norm(q)

        return q, attn

Multi Head Attention은 세가지 요소를 입력으로 받는다. 

• query
• key
• value

3-2. Position wise Feed Forward

class PositionwiseFeedForward(nn.Module):
    ''' A two-feed-forward-layer module '''

    def __init__(self, d_in, d_hid, dropout=0.1):
        super().__init__()
        self.w_1 = nn.Linear(d_in, d_hid) # position-wise 512 -> 2048
        self.w_2 = nn.Linear(d_hid, d_in) # position-wise 2048 -> 512
        self.layer_norm = nn.LayerNorm(d_in, eps=1e-6)
        self.dropout = nn.Dropout(dropout)

    def forward(self, x):

        residual = x

        x = self.w_2(F.relu(self.w_1(x)))
        x = self.dropout(x)
        x += residual

        x = self.layer_norm(x)

        return x

3-3. Encoder Layer

다음은 하나의 인코더 레이어를 정의한다. 인코더 레이어는 입력과 출력의 차원이 같다. 이러한 특징을 이용해 인코더 레이어를 여러 번 중첩해 사용할 수 있다.

class EncoderLayer(nn.Module):
    ''' Compose with two layers '''

    def __init__(self, d_model, d_inner, n_head, d_k, d_v, dropout=0.1):
        super(EncoderLayer, self).__init__()
        self.slf_attn = MultiHeadAttention(n_head, d_model, d_k, d_v, dropout=dropout)
        self.pos_ffn = PositionwiseFeedForward(d_model, d_inner, dropout=dropout)

    def forward(self, enc_input, slf_attn_mask=None):
        enc_output, enc_slf_attn = self.slf_attn(
            enc_input, enc_input, enc_input, mask=slf_attn_mask)
        enc_output = self.pos_ffn(enc_output)
        return enc_output, enc_slf_attn

여기서 slf_attn_mask는 pad_idx에 대하여 mask값을 0으로 처리하는 다음과 같은 함수이다. 이는 필요 없는 패딩 인덱스에 대해 attention 연산을 안 하겠다는 의미이다.

def get_pad_mask(seq, pad_idx):
    return (seq != pad_idx).unsqueeze(-2)

3-4. Encoder Architechture

이 클래스에서 전체 인코더 아키텍처를 정의한다.

class Encoder(nn.Module):
    ''' A encoder model with self attention mechanism. '''

    def __init__(
            self, n_src_vocab, d_word_vec, n_layers, n_head, d_k, d_v,
            d_model, d_inner, pad_idx, dropout=0.1, n_position=200, scale_emb=False):
        

        super().__init__()
        
        # 모든 단어들을 embedding (고유 백터를 가진 차원으로 변경)
        # 현재 코드에서는10077의 vocab 내 단어들을 512 차원으로 embedding 시킴
        self.src_word_emb = nn.Embedding(n_src_vocab, d_word_vec, padding_idx=pad_idx)
        
        # 문장의 최대길이까지 positional encoding을함
        # positional encoding : rnn이 아니므로 순서에 대한 정보를 반영하기위한 방법
        self.position_enc = PositionalEncoding(d_word_vec, n_position=n_position)
        self.dropout = nn.Dropout(p=dropout)
        
        # multiple encoder
        # encoder layer stack으로 encoder 층을 6개로 쌓는다.
        # encoder layer : multihead attentions, feedforward로 구성
        self.layer_stack = nn.ModuleList([
            EncoderLayer(d_model, d_inner, n_head, d_k, d_v, dropout=dropout)
            for _ in range(n_layers)])
        
        # layer_norm 
        self.layer_norm = nn.LayerNorm(d_model, eps=1e-6)
        self.scale_emb = scale_emb
        self.d_model = d_model

    def forward(self, src_seq, src_mask, return_attns=True):

        enc_slf_attn_list = []

        # -- Forward
        # ① word embedding
        enc_output = self.src_word_emb(src_seq)
        if self.scale_emb:
            enc_output *= self.d_model ** 0.5
            
        # ② positional encoding
        enc_output = self.dropout(self.position_enc(enc_output))
        
        # normalization
        enc_output = self.layer_norm(enc_output)
        
        # ③ stacked encoder layers
        for enc_layer in self.layer_stack:
            enc_output, enc_slf_attn = enc_layer(enc_output, slf_attn_mask=src_mask)
            enc_slf_attn_list += [enc_slf_attn] if return_attns else []
            
        if return_attns:
            return enc_output, enc_slf_attn_list
        return enc_output,

 

 

4. Decoder

4-1. Decoder Layer

decoder layer도 encoder layer와 같이 입력과 출력의 차원이 같다. 이러한 특징을 이용해 트랜스포머의 decoder는 decoder layer를 여러 번 중첩해 사용한다.

또한 decoder layer는 encoder layer와 다르게 두 개의 multi head attention이 사용된다. 

Masked Multi-head Attention에서는 뒤쪽 단어의 점수를 참고하지 않도록 대각행렬 위쪽을 마스킹 처리를 해준다.

# masked multi-head attention을 하기 위한 mask 과정
def get_subsequent_mask(seq):
    ''' For masking out the subsequent info. '''
    sz_b, len_s = seq.size()
    subsequent_mask = (1 - torch.triu(
        torch.ones((1, len_s, len_s), device=seq.device), diagonal=1)).bool() # 대각행렬 윗부분을 False로 치환
    return subsequent_mask

class DecoderLayer(nn.Module):
    ''' Compose with three layers '''

    def __init__(self, d_model, d_inner, n_head, d_k, d_v, dropout=0.1):
        super(DecoderLayer, self).__init__()
        
        # 
        self.slf_attn = MultiHeadAttention(n_head, d_model, d_k, d_v, dropout=dropout)
        self.enc_attn = MultiHeadAttention(n_head, d_model, d_k, d_v, dropout=dropout)
        self.pos_ffn = PositionwiseFeedForward(d_model, d_inner, dropout=dropout)

    def forward(
            self, dec_input, enc_output,
            slf_attn_mask=None, dec_enc_attn_mask=None):
        
        # ① sublayer 1: decoder input 값에 대한 self attention
        dec_output, dec_slf_attn = self.slf_attn(dec_input, dec_input, dec_input, mask=slf_attn_mask)
        
        # ② sublayer 2 : ①의 결과값을 query로, key,value는 encoder의 output 값으로 attention
        dec_output, dec_enc_attn = self.enc_attn(dec_output, enc_output, enc_output, mask=dec_enc_attn_mask)
        
        # ③ sublayer 3 : position wise feed forward를 통과
        dec_output = self.pos_ffn(dec_output)
        
        return dec_output, dec_slf_attn, dec_enc_attn

4-2. Decoder

class Decoder(nn.Module):
    ''' A decoder model with self attention mechanism. '''
    def __init__(
            self, n_trg_vocab, d_word_vec, n_layers, n_head, d_k, d_v,
            d_model, d_inner, pad_idx, n_position=200, dropout=0.1, scale_emb=False):

        super().__init__()
        
        # target word embedding
        self.trg_word_emb = nn.Embedding(n_trg_vocab, d_word_vec, padding_idx=pad_idx)
        
        # positional encoding
        self.position_enc = PositionalEncoding(d_word_vec, n_position=n_position)
        self.dropout = nn.Dropout(p=dropout)
        
        # stacked decoder layers
        self.layer_stack = nn.ModuleList([
            DecoderLayer(d_model, d_inner, n_head, d_k, d_v, dropout=dropout)
            for _ in range(n_layers)])
        
        # layer_normalization
        self.layer_norm = nn.LayerNorm(d_model, eps=1e-6)
        self.scale_emb = scale_emb
        self.d_model = d_model

    def forward(self, trg_seq, trg_mask, enc_output, src_mask, return_attns=True):

        dec_slf_attn_list, dec_enc_attn_list = [], []

        # -- Forward
        
        # ① target words embedding
        dec_output = self.trg_word_emb(trg_seq)
        if self.scale_emb:
            dec_output *= self.d_model ** 0.5
            
        # ② positional encoding
        dec_output = self.dropout(self.position_enc(dec_output))
        dec_output = self.layer_norm(dec_output)

        # ③ decoder_layer stacked
        for dec_layer in self.layer_stack:
            dec_output, dec_slf_attn, dec_enc_attn = dec_layer(
                dec_output, enc_output, slf_attn_mask=trg_mask, dec_enc_attn_mask=src_mask)
            
            dec_slf_attn_list += [dec_slf_attn] if return_attns else []
            dec_enc_attn_list += [dec_enc_attn] if return_attns else []

        if return_attns:
            return dec_output, dec_slf_attn_list, dec_enc_attn_list
        return dec_output,

5. Transformer Model

마지막으로 transformer model을 구현해보자. transformer 구조는 크게보면 다음과 같다.

1. source, target 데이터에 대한 mask 생성
2. encoder 층 통과
3. decoder 층 통과
4. 최종 가중치 층을 통과

''' Define the Transformer model '''
import torch
import torch.nn as nn
import numpy as np
#from transformer.Layers import EncoderLayer, DecoderLayer

class Transformer(nn.Module):
    ''' A sequence to sequence model with attention mechanism. '''

    def __init__(
            self, n_src_vocab, n_trg_vocab, src_pad_idx, trg_pad_idx,
            d_word_vec=512, d_model=512, d_inner=2048,
            n_layers=6, n_head=8, d_k=64, d_v=64, dropout=0.1, n_position=200,
            trg_emb_prj_weight_sharing=True, emb_src_trg_weight_sharing=True,
            scale_emb_or_prj='prj'):

        super().__init__()
        
        # padding index 저장
        self.src_pad_idx, self.trg_pad_idx = src_pad_idx, trg_pad_idx

        assert scale_emb_or_prj in ['emb', 'prj', 'none']
        scale_emb = (scale_emb_or_prj == 'emb') if trg_emb_prj_weight_sharing else False
        self.scale_prj = (scale_emb_or_prj == 'prj') if trg_emb_prj_weight_sharing else False
        self.d_model = d_model

        # encoder 정의
        self.encoder = Encoder(
            n_src_vocab=n_src_vocab, n_position=n_position,
            d_word_vec=d_word_vec, d_model=d_model, d_inner=d_inner,
            n_layers=n_layers, n_head=n_head, d_k=d_k, d_v=d_v,
            pad_idx=src_pad_idx, dropout=dropout, scale_emb=scale_emb)

        # decoder 정의
        self.decoder = Decoder(
            n_trg_vocab=n_trg_vocab, n_position=n_position,
            d_word_vec=d_word_vec, d_model=d_model, d_inner=d_inner,
            n_layers=n_layers, n_head=n_head, d_k=d_k, d_v=d_v,
            pad_idx=trg_pad_idx, dropout=dropout, scale_emb=scale_emb)

        # 최종 output layers 정의
        self.trg_word_prj = nn.Linear(d_model, n_trg_vocab, bias=False)

        for p in self.parameters():
            if p.dim() > 1:
                nn.init.xavier_uniform_(p) 

        assert d_model == d_word_vec, \
        'To facilitate the residual connections, \
         the dimensions of all module outputs shall be the same.'

        if trg_emb_prj_weight_sharing:
            # Share the weight between target word embedding & last dense layer
            self.trg_word_prj.weight = self.decoder.trg_word_emb.weight

        if emb_src_trg_weight_sharing:
            self.encoder.src_word_emb.weight = self.decoder.trg_word_emb.weight

    def forward(self, src_seq, trg_seq):
        
        # ① source, target 데이터에 대한 mask 생성
        src_mask = get_pad_mask(src_seq, self.src_pad_idx)
        trg_mask = get_pad_mask(trg_seq, self.trg_pad_idx) & get_subsequent_mask(trg_seq)

        # ② encoder 층을 통과
        #enc_output, *_ = self.encoder(src_seq, src_mask)
        enc_output, *_ = self.encoder(src_seq, src_mask)
        
        # ③ decoder 층을 통과
        #dec_output, *_ = self.decoder(trg_seq, trg_mask, enc_output, src_mask)
        dec_output, attention1, attention2 = self.decoder(trg_seq, trg_mask, enc_output, src_mask)
        
        # ④ 최종 weight 층을 통과
        seq_logit = self.trg_word_prj(dec_output)
        if self.scale_prj:
            seq_logit *= self.d_model ** -0.5

        return seq_logit.view(-1, seq_logit.size(2))

 

여기까지 transformer를 구성하는 모듈들에 대해 알아보았다. 다음 포스팅에는 전체적인 학습과 옵티마이저, 예측 함수에 대해 알아보도록 하겠다.