트랜스포머 바닥부터 구현하기2

WooSeongkyun·2023년 4월 4일
바닥부터 구현하기
0

바닥부터 구현하기

목록 보기
2/3
  • 참고사항
    - 이때 디코더 멀티헤드 어텐션에 들어가는 인코딩 아웃풋은 각각 키와 밸류의 역할을, 첫번째 멀티헤드 어텐션의 아웃풋은 두번째 멀티헤드 어텐션의 쿼리로 기능을 한다
    - 마스크드 멀티헤드 어텐션은 예측시 정답지를 보는 것을 방지하기 위하여, 아래삼각행렬 마스크를 사용한다

트랜스포머 아키텍쳐

  • 토큰 임베딩
    • 입력 텍스트를 토큰으로 바꿔주는 과정
  • 위치 임베딩
    • 위치정보를 뽑아내는 벡터
  • 인코더 디코더 구조
    • 인코더
      • Hidden state, context라 불리는 임베디 벡터의 시퀀스로 변환한다
    • 디코더
      • 인코더의 은닉상태를 활용하여 출력 토큰의 시퀀스를 출력한다
  • 인코더 구조
    -멀티헤드 셀프 어텐션
    • 완전 연결 피드포워드 층 FFNN

셀프어텐션

  • 어텐션 메커니즘은 신경망이 시퀀스의 각 토큰에 다른양의 가중치, 즉 어텐션을 할당한다
  • BERT에서 각 토큰은 768차원의 벡터로 표현된다. 셀프 어텐션의 셀프는 인코더의 모든 은닉 상태에 대하여 게산한다
  • 각 임베딩의 가중평균을 계산해주는 것이 셀프 어텐션의 기본개념이다
  • 다음과 같이 계산된다
    - xi′=∑j=1nwjixjx _{i}'=\displaystyle\sum\limits_{j=1}^{n}{w _{ji}x _{j}}
    - 이때 wjiw _{ji} 를 어텐션 가중치라고 부르며 ∑jwji=1\displaystyle\sum\limits_{j}^{}{w _{ji}}=1 이 되도록 정규화된다
  • 예
    - times flies like an arrow 와 같은 문장이 주어졌을 때 flies가 파리인지, 날라간다인지 해석하기 위해 주변 토큰들에게 어텐션값을 계산한다. 이때는 아마도 time과 arrow토큰 임베딩에 더 높은 가중치 값을 할당하게 될것이다
    - 이런식으로 생성된 임베딩을 문맥고려 임베딩contextualized embedding이라고 부른다
    - ![[Pasted image 20230331153534.png]]

셀프 어텐션 연산 순서

  1. Q,K,VQ,K,V 텐서 학습
  2. Q,KQ,K의 Tensor Contradiction으로 유사도 계산
  3. 스케일 조정
  4. (QtKQ ^{t}K)와 VV 의 Tensor contradiction
  5. 마스킹(디코더에서)
  6. 소프트맥스 함수
    ---- self_attention ----
  7. 헤드 concaterate
    ---- Multihead_attention----
    ![[Pasted image 20230319205238.png]]
## 어텐션 함수로 변형시키기
def self_attention(query,key,value):
    import torch 
    from math import sqrt
    import torch.nn.functional as F 

    """
    query,key,value 텐서는 
    쿼리 형태:    (N,query_len,embed_dim)
    키 형태:      (N,key_len,embed_dim)
    에너지 형태: (N,heads,query_len,key_len)
    """
    dim_k = key.size(-1)
    """
    책에서는 다음과 같은 방법이 제시되었지만 
    scores= torch.bmm(query, key.transpose(1,2))/sqrt(dim_k)
    einsum이 tensor-contradiction을 직관적으로 활용한다는 점에서 다음과 같이 수정하였다
    위와 아래의 scores를 비교하면 True가 나온다
    """
    scores= torch.einsum('ijk,ilk->ijl',[query,key])/sqrt(dim_k)

    #소프트 맥스 함수 적용하기
    weights= F.softmax(scores, dim=-1)
    weights.sum(dim=-1)
    
    #마지막 계산하기
    """기존 방법 : attn_outputs= torch.bmm(weights,value)
            attention shape: (N,query_len,key_len)
            value shape: (N,value_len,embed_dim)
            result: (N,query_len,embed_dim)
            여기선 key_len과 value_len이 같은것을 활용"""
    
    attn_outputs= torch.einsum('ijk,ijl->ikl',[weights,value])
    return attn_outputs

멀티헤드 어텐션

  • 왜 여러개의 어텐션 헤드를 사용하는가?
    - 한 헤드의 소프트맥스함수는 유사도의 한 측면에만 초점을 맞추는 경향이 있기 때문이다.
    - 여러개의 헤드가 있다면 모델은 동시에 여러 측면에 대해 초점을 맞춘다.
""" 
멀티헤드 어텐션을 구현하기 위해 먼저 어텐션헤드를 정의한다
입력받은 query,key,value값을 self_attention함수를 처리한 결과
를 반환한다
"""
class AttentionHead(nn.Module):
    def __init__(self,embed_dim, head_dim):
        super().__init__()
        self.q = nn.Linear(embed_dim, head_dim)
        self.k = nn.Linear(embed_dim, head_dim)
        self.v = nn.Linear(embed_dim, head_dim)

    def forward(self,hidden_states):
        attn_outputs= self_attention(
                self.q(hidden_states),
                self.k(hidden_states),
                self.v(hidden_states))
        return attn_outputs
    

""" 
멀티헤드 어텐션을 종합적으로 구현한다
"""
class MultiHeadAttention(nn.Module):
    """
    config= AutoConfig.from_pretrained(model_ckpt)
    로 기존 사전학습모델의 정보를 활용하는 것으로 보임
    """
    def __init__(self,config):
        super().__init__()
        embed_dim = config.hidden_size
        num_heads= config.num_attention_heads
        head_dim = embed_dim // num_heads 
        self.heads = nn.ModuleList(
            [AttentionHead(embed_dim,head_dim) for _ in range(num_heads)]
        )
        self.output_linear = nn.Linear(embed_dim,embed_dim)

    def forward(self,hidden_states):
    """torch.cat: 주어진 차원을 기준으로 텐서를 붙인다concatenate"""
        x= torch.cat([h(hidden_states) for h in self.heads],dim=-1)
        x= self.output_linear(x)
        return x
    
multihead_attn = MultiHeadAttention(config)
attn_output= multihead_attn(inputs_embeds)
attn_output.size()

어텐션 학습결과를 시각화하여 보기

from bertviz import head_view
from transformers import AutoModel 

model = AutoModel.from_pretrained(model_ckpt, 
                                  output_attentions=True)

sentence_a = "time flies like an arrow"
sentence_b = "fruit flies like a banana"

viz_inputs = tokenizer(sentence_a,sentence_b, return_tensors='pt')
attention= model(**viz_inputs).attentions 
sentence_b_start= (viz_inputs.token_type_ids ==0).sum(dim=1)
tokens =tokenizer.convert_ids_to_tokens(viz_inputs.input_ids[0])

head_view(attention,tokens,sentence_b_start,heads=[8])

"""
가중치 학습 결과 첫번째 문장의 'flies'에서 가장 중요하게 여긴 단어로
'arrow'가 꼽혔고, 두번째 문장의 학습 결과로 'fruit'와 'banana'를 선택하였다
모델이 문맥에 따라 'flies'가 어떤 의미인지, 명사인지 동사인지 구별한 것이다
"""

피드포워드 층

flowchart BT
	A[x] --> B;
	B[F_1=xW_1+b_1] --> C;
	C["F_2= max(0,F_1)"] --> D ;
	D["F_3= F_2W_2+b_2"];
  • 인코더와 디코더에 있는 피드 포워드 층은 간단한 두개의 층으로 구성된 완전 연결 신경망이다
  • 하지만 전체 임베딩 시퀀스를 하나의 벡터로 처리하지 않고, 각 임베딩을 독립적으로 처리한다 (N,query_len,embed_dim으로 이루어진 텐서를 N,query_len,intermediate_size > N,query_len,embed_dim순으로 변환시킴)
  • 그러한 이유로 이 층을 position-wise feed-forward layer라고 부르기도 하고, 1*1 convolutional network라고 부르기도 한다
  • 논문에선 첫번째 크기를 임베딩의 네배로 하고 GELU활성화 함수를 사용한다
class FeedForward(nn.Module):
    def __init__(self,config):
        """AutoCOnfig 클래스를 활용하여 bert-base-uncased
        체크포인트와 관련된 config.json 파일을 로드한다 
        트랜스포머에서의 모든 체크포인트는 vocab_size,hidden_size와
        같은 다양한 파라미터가 지정된 설정 파일이 할당된다

        이 예제 같은 경우 각 입력 단어의 정수ID가 nn.Embedding에 저장된
        30,522개의 임베딩 벡터중 하나에 매핑되고 이때, 벡터의 크기는
        768이다

        피드 포워드 층에 (batch_size,seq_len,hidden_len)의 텐서가
        입력되면 각 다른 (batch_size,seq_len)마다 진행함
        """
        self.linear_1 = nn.Linear(config.hidden_size,config.intermediate_size)
        self.linear_2 = nn.Linear(config.intermediate_size,config.hidden_size)
        self.gelu= nn.GELU()
        self.dropout = nn.Dropout(config.hidden_dropout_prob)

    def forward(self,x):
        x= self.linear_1(x)
        x= self.gelu(x)
        x= self.linear_2(x)
        x= self.dropout(x)
        return x

층 정규화하기

flowchart BT
	A[x] --> B;
	B["F(x)"] --> C;
	A[x] -->C;
	C["H(x)=x+F(x)"];
  • 사후 층 정규화
    • 트랜스포머 논문에서 사용한 방법. 멀티헤드 어텐션 및 FFNN에서 연산된 결과를 정규화한 방식
      • 그레디언트가 발산하는 경우가 생겨 처음부터 훈련시키기가 까다롭다
      • 이러한 이유로 훈련하는 동안은 작은 값에서 점점 최대값으로 증가시키는 learning rate warm-up 방식을 활용한다
  • 사전 층 정규화
    • 다른 논문에서 많이 활용하는 방법
    • 멀티헤드 어텐션 및 FFNN에서 연산하기 전에 정규화를 하는 방식
    • 여기선 두번째 방식을 채택한다
""" 
종합되어 작동하는 Transformer의 Encoder-layer
입력 텐서와 출력 텐서의 크기가 같다는 특징을 갖고 있다
단 아직 토큰의 위치정보를 활용하지 않았다
"""
class TransformerEncoderLayer(nn.Module):
    def __init__(self,config):
        super().__init__()
        """ 
        각각의 batch마다 전체 channel에 대한 normalization을 새행한다
        """
        self.layer_norm_1 = nn.LayerNorm(config.hidden_size)
        self.layer_norm_2= nn.LayerNorm(config.hidden_size)
        self.attention= MultiHeadAttention(config)
        self.feed_foward =FeedForward(config)

    def forward(self,x):
        #층 정규화를 적용하여 입력을 쿼리,키,값으로 복사한다
        hidden_states= self.layer_norm_1(x)
        #어텐션에 스킵 연결을 적용한다
        x= x + self.attention(hidden_states)
        #스킵 연결과 피드포워드 층을 적용한다
        x= x+ self.feed_foward(self.layer_norm_2(x))
        return x 
    

#입력 임베딩으로 테스트하기
encoder_layer= TransformerEncoderLayer(config)
inputs_embeds.shape ,encoder_layer(inputs_embeds).size()

위치정보 임베딩하기

"""위치정보를 임베딩 하기 위하여 사용하는 방법
"""
class Embeddings(nn.Module):
    def __init__(self,config):
        super().__init__()
        self.token_embeddings= nn.Embedding(
            config.vocab_size,
            config.hidden_size
        )
        self.positional_embeddings= nn.Embedding(
            config.max_position_embeddings,
            config.hidden_size
        )
        self.layer_norm = nn.LayerNorm( 
            config.hidden_size, eps=1e-12
        )
        self.dropout = nn.Dropout()

    def forward(self,input_ids):        
        """
        1.input_ids의 seq_length를 확인한다.
        2.해당 길이만큼의 1,2,...,seq_lenth-1을 담은 배열을 만든다
        3.토큰들을 임베딩 벡터로 변환한다
        4.config.max_positional_embeddings를 활용하여 임베딩을
        시행한다
        5.임베딩된 결과를 더하고 LN과 dropout을 시행하여 반환한다
        """
        seq_length= input_ids.size(1)
        positional_ids= torch.arange(seq_length,
                                     dtype=torch.long).unsqueeze(0)
        token_embeddings= self.token_embeddings(input_ids)
        positional_embeddings= self.positional_embeddings(positional_ids)
        embeddings= token_embeddings+ positional_embeddings
        embeddings= self.layer_norm(embeddings)
        embeddings= self.dropout(embeddings)
        return embeddings
    
"""여기서 inputs는 텍스트 시퀀스를 정수 인코딩한 값임"""
embedding_layer = Embeddings(config)
embedding_layer(inputs.input_ids).size()

![[PositionalEncoding.jpg]]

  • RNN은 단어 입력을 순차적으로 받는 방식으로서 위치 정보를 얻었다
  • 트랜스포머는 임베딩 벡터에 위치정보를 더하여 모델의 입력으로 활용하는데 이를 포지셔닝 인코딩이라 한다
  • 이때 포지셔널 인코딩은 다음과 같은 값으로 정의된다
    - PE(pos,2i)=sin(pos/100002i/dmodel)PE(pos,2i)=sin(pos/10000 ^{2i/d _{model}})
    - PE(pos,2i+1)=sin(pos/100002i/dmodel)PE(pos,2i+1)=sin(pos/10000 ^{2i/d _{model}})
    - pospos : 입력문장에서의 임베딩 위치
    - ii : 임베딩 백터 내의 차원의 인덱스
    - dmodeld _{model} : 임베딩 벡터의 크기
    - 문장 행렬은 j−thj-th row가 jj 번째 단어의 임베딩 벡터로 이루어진 행렬인데, 포지셔널 인코딩의 정의를 보면 행을 아래로 내려갈때마다 PEPE 의 입력값이 정수배 pospos 를 따라 변하고 있음을 알 수 있다

분류헤드 추가하기

  • 텍스트 분류에 특화된 모델을 만들고 싶다면 바디에 연결할 분류 헤드가 필요하다
  • 각 토큰에 대한 은닉 상태가 있어, 토큰마다 예측을 할 수도 있겠지만, 필요한 예측은 단 하나이다
  • 일반적으로 이런 모델의 첫번째 토큰을 예측에 사용하고, 드롭아웃층과 선형층을 추가하여 분류 예측을 만든다
class TransformerForSequenceClassification(nn.Module):
    def __init__(self,config):
        """
        Encoder를 거쳐 나온 결과값이 num_label개만큼만
        되도록 config.num_labels를 통하여 설정한다
        해당 결과값은 정규화되지 않은 logit이다
        """
        super().__init__()
        self.encoder =TransformerEncoder(config)
        self.dropout = nn.Dropout(config.hidden_dropout_prob)
        self.classifier = nn.Linear(config.hidden_size, config.num_labels)

    def forward(self,x):
        x =self.encoder(x)[:,0,:] #[CLS]토큰의 은닉상태를 선택한다
        x= self.dropout(x)
        x= self.classifier(x)
        return x 

config.num_labels= 3 
encoder_classifier =TransformerForSequenceClassification(config)
encoder_classifier(inputs.input_ids).size()

디코더

  • 마스크드 멀티헤드 셀프어텐션 층
    • 타임스텝마다 지난 출력과 동일한 현재 토큰만 사용하여 토큰을 생성한다.
    • 이렇게 해야 디코더를 훈련시키는 동안 단순히 타깃 번역을 복사하는 방식의 부정행위를 방지할 수 있다
  • 인코더-디코더 어텐션 층
    • 디코더의 중간표현을 쿼리처럼 사용하여 인코더 스택의 출력 키와 값 벡터에 멀티헤드 어텐션을 수행한다.
    • 이를 통해 인코더-디코더 어텐션 층은 두개의 다른 시퀀스에 있는 토큰을 연관짓는 방법을 학습하게 된다
""" 
torch.tril():하삼각행렬lower triangular matrix를 만드는 명령어
대각선을 포함하여 아래 성분들은 1이고, 대각선 위는 모두 0인 형태이다
이후 Tensor.maksed_fill()을 활용하여 0을 음의 무한대로 바꾸면
어텐션 헤드가 미래 토큰에 대한 정보를 참조할 수 없게 된다
"""
seq_len = inputs.input_ids.size(-1)
mask= torch.tril(torch.ones(seq_len,seq_len)).unsqueeze(0)
mask[0]


def scaled_dot_product_attention(query,key,value,mask=None):
   """ 
   쿼리 형태:    (N,query_len,embed_dim)
   키 형태:      (N,key_len,embed_dim)
   에너지 형태: (N,heads,query_len,key_len)
    """
   dim_k= query.size(-1) #임베딩 벡터 차원의 sqrt로 scaling해줌
   scores= torch.einsum('ijk,ilk->ijl',[query,key])/sqrt(dim_k)
   if mask is not None:
      scores= scores.masked_fill(mask==0, float('-1e+10'))
   weights= F.softmax(scores,dim=-1)

   """기존 방법 : attn_outputs= torch.bmm(weights,value)
        attention shape: (N,query_len,key_len)
        value shape: (N,value_len,embed_dim)
        result: (N,query_len,embed_dim)
        여기선 key_len과 value_len이 같은것을 활용"""
   attn_output= torch.einsum('ijk,ijl->ikl',[weights,value])
   return attn_output
profile
WooSeongkyun
안녕하세요!
이전 포스트

트랜스포머 바닥부터 구현하기

다음 포스트

파이토치 사용법 정리

0개의 댓글