Deeplearning - chap 11-3

트랜스포머 아키텍처

셀프 어텐션 이해하기

일반화된 셀프 어텐션 : 쿼리 -키 -값 모델
멀티 헤드 어텐션

트렌스포머 인코더

< self- attention >

import torch
import torch.nn.functional as F

def self_attention(inputs):
    batch_size, seq_len, input_dim = inputs.size()
    
    # Query, Key, Value를 위한 선형 변환 파라미터
    W_q = torch.nn.Linear(input_dim, input_dim)
    W_k = torch.nn.Linear(input_dim, input_dim)
    W_v = torch.nn.Linear(input_dim, input_dim)
    
    # Query, Key, Value 계산
    queries = W_q(inputs)
    keys = W_k(inputs)
    values = W_v(inputs)
    
    # Scaled Dot-Product Attention 수행
    scaled_dot_products = torch.bmm(queries, keys.transpose(1, 2)) / torch.sqrt(input_dim)
    attention_scores = F.softmax(scaled_dot_products, dim=2)
    attention_output = torch.bmm(attention_scores, values)
    
    return attention_output

• import torch를 사용하여 PyTorch를 임포트합니다.

• import torch.nn.functional as F를 사용하여 PyTorch의 함수형 인터페이스를 임포트합니다.

• batch_size, seq_len, input_dim = inputs.size()를 사용하여 입력의 크기 정보를 가져옵니다.

• inputs는 크기가 (batch_size, seq_len, input_dim)인 텐서

• W_q, W_k, W_v는 입력을 선형 변환하는 세 개의 선형 레이어입니다. 이들 레이어는 각각 입력 차원에서 동일한 차원으로 변환

• queries = W_q(inputs), keys = W_k(inputs), values = W_v(inputs)를 사용하여 입력을 각각의 선형 레이어에 통과시킵니다. 이를 통해 쿼리, 키, 값에 해당하는 텐서들을 얻습니다.

• scaled_dot_products = torch.bmm(queries, keys.transpose(1, 2)) / torch.sqrt(input_dim)는 쿼리와 키의 점곱을 계산하고, 행렬 곱셈 후 스케일링을 수행합니다. 이를 통해 어텐션 스코어 행렬을 얻습니다
  -> 쿼리와 키의 차원이 일치해야 한다.
  -> 쿼리의 마지막 차원 크기와 key의 두번쨰 차원 크기가 동일 해야 한다.
  -> 이를 위해 transpose(1,2)를 사용하여 두번쨰와 세번째를 바꿔준다.

-> torch.sqrt(input_dim)을 사용하여 제곱근 값을 구한다. -> 이는 스케일링 요소로 사용

• torch.bmm(attention_scores, values)
  -> PyTorch의 배치 행렬 곱셈(batch matrix multiplication) 연산이다

• 최종 결과인 scaled_dot_products

-> (batch_size, seq_len, seq_len)인 어텐션 스코어 텐서
-> 이는 각 입력 위치의 다른 위치들과의 어텐션 가중치를 나타냅니다.

• attention_scores = F.softmax(scaled_dot_products, dim=2)는 어텐션 스코어 행렬을 소프트맥스 함수를 사용하여 정규화합니다.
  -> dim=2: 소프트맥스 함수가 적용될 차원을 지정합니다. 여기서는 마지막 차원에 대해 소프트맥스 함수를 적용하도록 지정
  -> 이렇게 함으로써 각 위치의 어텐션 가중치를 확률로 표현

-> 소프트맥스 함수는 각 요소를 입력의 합으로 나눠주는 작업을 수행

• 최종 결과인 scaled_dot_products

-> (batch_size, seq_len, seq_len)인 어텐션 스코어 텐서
-> 이는 각 입력 위치의 다른 위치들과의 어텐션 가중치를 나타냅니다.

Layers 층을 상속하여 구현한 트랜스포머 인코더

class TransformerEncoder(layers.Layer):
    def __init__(self, embed_dim, dense_dim, num_heads, **kwargs):
        super().__init__(**kwargs)
        self.embed_dim = embed_dim
        self.dense_dim = dense_dim
        self.num_heads = num_heads
        self.attention = layers.MultiHeadAttention(   
            num_heads=num_heads, key_dim=embed_dim) 
        self.dense_proj = keras.Sequential(
            [layers.Dense(dense_dim, activation="relu"),
             layers.Dense(embed_dim),]
        )
        self.layernorm_1 = layers.LayerNormalization() 
        self.layernorm_2 = layers.LayerNormalization()

• embed_dim : 임베딩 차원의 크기를 나타내는 매개변수

• dense_dim: 밀집 레이어(dense layer)의 출력 차원의 크기를 나타내는 매개변수

• num_heads: 멀티헤드 어텐션(multi-head attention)에서 사용될 어텐션 헤드의 수를 나타내는 매개변수

• **kwargs: 추가적인 매개변수를 받기 위한 가변 키워드 매개변수

• self.attention = layers.MultiHeadAttention(num_heads=num_heads, key_dim=embed_dim)

<layers.MultiHeadAttention>

멀티헤드 어텐션 레이어를 생성합니다. 이 레이어는 입력 시퀀스에 대한 멀티헤드 어텐션 메커니즘을 적용

num_heads: 멀티헤드 어텐션에서 사용될 어텐션 헤드의 수를 설정

key_dim: 어텐션 키(key)의 차원 크기를 설정

• self.dense_proj = keras.Sequential([layers.Dense(dense_dim, activation="relu"), layers.Dense(embed_dim),])

-> keras.Sequential: 여러 개의 레이어를 연속적으로 적용하는 시퀀셜 모델을 생성

• self.layernorm_1 = layers.LayerNormalization()
  -> layers.LayerNormalization: 레이어 정규화(layer normalization) 레이어를 생성

    def call(self, inputs, mask=None):  
        if mask is not None:
            mask = mask[:, tf.newaxis, :]
        attention_output = self.attention(
            inputs, inputs, attention_mask=mask)
        proj_input = self.layernorm_1(inputs + attention_output)
        proj_output = self.dense_proj(proj_input)
        return self.layernorm_2(proj_input + proj_output)

• def call(self, inputs, mask=None):
  inputs: 입력 텐서입니다.
  mask: 입력에 대한 마스크 텐서로, 선택적으로 제공

• if mask is not None: mask = mask[:, tf.newaxis, :]:

mask가 제공되었을 경우, 마스크의 차원을 확장합니다. 이를 위해 tf.newaxis를 사용하여 마스크 텐서에 차원을 추가합니다.
이렇게 하면 마스크 텐서의 크기는 (batch_size, 1, seq_len)이 됩니다

• attention_output = self.attention(inputs, inputs, attention_mask=mask):

• attention_mask=mask: 어텐션 마스크를 적용합니다. 이를 통해 패딩 토큰에 대한 어텐션을 제한할 수 있습니다.

• proj_input = self.layernorm_1(inputs + attention_output):
  -> inputs와 attention_output을 더한 후, 첫 번째 레이어 정규화를 적용합니다. 이를 통해 입력과 어텐션 출력을 합친 값을 정규화

• proj_output = self.dense_proj(proj_input):
  -> proj_input을 밀집 레이어를 통과시킵니다. 이를 통해 입력을 밀집된 차원으로 변환

• return self.layernorm_2(proj_input + proj_output)
  -> proj_input + proj_output을 반환

이를 통해 인코더 층의 동작을 구현한다

get_config() 메서드는 사용자 정의 층(class)의 구성(configuration)을 반환하는 역할

def get_config(self):   # get_

        config = super().get_config()
        config.update({
            "embed_dim": self.embed_dim,
            "num_heads": self.num_heads,
            "dense_dim": self.dense_dim,
        })
        return config 

• config = super().get_config()
  -> 상위 클래스의 get_config() 메서드를 호출하여 기본 구성을 가져옵니다
  -> get_config() 메서드는 기본 층(configuration)을 반환

• config.update({...})
  -> 기본 구성에 사용자 정의 층의 추가 구성을 업데이트

트랜스포머 인코더를 사용하여 텍스트 분류하기

vocab_size = 20000
embed_dim = 256
num_heads = 2
dense_dim = 32

inputs = keras.Input(shape=(None,), dtype="int64")
x = layers.Embedding(vocab_size, embed_dim)(inputs)  # attention다 들어있는 것
x = TransformerEncoder(embed_dim, dense_dim, num_heads)(x)
x = layers.GlobalMaxPooling1D()(x)  # 차원을 줄이기 위해서 ->하나의 벡터를 만든다. ,seqquential 
x = layers.Dropout(0.5)(x)
outputs = layers.Dense(1, activation="sigmoid")(x)
model = keras.Model(inputs, outputs)
model.compile(optimizer="rmsprop",
              loss="binary_crossentropy",
              metrics=["accuracy"])
model.summary()

• x = layers.Embedding(vocab_size, embed_dim)(inputs)
  임베딩 층은 단어의 정수 인덱스를 해당 단어 벡터로 변환

• x = TransformerEncoder(embed_dim, dense_dim, num_heads)(x)
  멀티헤드 어텐션, 레이어 정규화, 밀집 레이어의 연산을 수행

• x = layers.GlobalMaxPooling1D()(x)
  시퀀스의 가장 큰 값만 추출한다-> 이를 통해 시퀀스의 중요한 특징 강조

서브 클래싱으로 위치 임베딩 구현하기

class PositionalEmbedding(layers.Layer):
    def __init__(self, sequence_length, input_dim, output_dim, **kwargs):
        super().__init__(**kwargs)
        self.token_embeddings = layers.Embedding(
            input_dim=input_dim, output_dim=output_dim)
        self.position_embeddings = layers.Embedding(
            input_dim=sequence_length, output_dim=output_dim)
        self.sequence_length = sequence_length
        self.input_dim = input_dim
        self.output_dim = output_dim

:positionalEmbedding은 입력 시퀀스의 토큰 임베딩과 위치 임베딩을 결합하는 역할을 수행

• self.token_embeddings = layers.Embedding(
  input_dim=input_dim, output_dim=output_dim)

        -> 입력 시퀀스의 토큰 임베딩을 위한 임베딩 층
        

• self.position_embeddings = layers.Embedding(
  input_dim=sequence_length, output_dim=output_dim)
  

-> 위치 임베딩을 위한 임베딩 층
-> input_dim 은 시퀀스의 길이

 def call(self, inputs):
        length = tf.shape(inputs)[-1]
        positions = tf.range(start=0, limit=length, delta=1)
        embedded_tokens = self.token_embeddings(inputs)
        embedded_positions = self.position_embeddings(positions)  
        return embedded_tokens + embedded_positions

• length = tf.shape(inputs)[-1]:
  입력 텐서 inputs의 마지막 차원의 크기를 가져옵니다. 이는 입력 시퀀스의 길이를 나타냅니다.

• positions = tf.range(start=0, limit=length, delta=1)
  0부터 length-1까지의 값을 갖는다. 이는 위치 임베딩을 위해 사용된다

• embedded_tokens = self.token_embeddings(inputs): 입력 토큰 시퀀스 inputs를 토큰 임베딩에 적용합니다. 즉, 각 토큰에 대한 임베딩 벡터를 얻습니다

• embedded_positions = self.position_embeddings(positions): 위치 임베딩에 입력 시퀀스의 위치 정보를 적용합니다. positions에 해당하는 위치에 대한 임베딩 벡터를 얻습니다.

• return embedded_tokens + embedded_positions: 토큰 임베딩과 위치 임베딩을 더하여 결합된 임베딩 벡터를 반환합니다. 이를 통해 각 토큰에 대한 위치 정보가 포함된 임베딩이 생성됩니다.

    def compute_mask(self, inputs, mask=None):
        return tf.math.not_equal(inputs, 0)

• inputs: 입력 텐서로, 여기서는 텍스트 데이터를 나타내는 정수 시퀀스

• compute_mask 메서드는 입력 텐서의 값이 0이 아닌 위치를 True로 표시하는 마스크를 계산합니다. 즉, 입력 텐서에서 0이 아닌 위치는 True로, 0인 위치는 False로 이루어진 마스크를 반환

트랜스포머 인코더와 위치 임베딩 합치기

inputs = keras.Input(shape=(None,), dtype="int64")
x = PositionalEmbedding(sequence_length, vocab_size, embed_dim)(inputs)
x = TransformerEncoder(embed_dim, dense_dim, num_heads)(x)
x = layers.GlobalMaxPooling1D()(x)
x = layers.Dropout(0.5)(x)
outputs = layers.Dense(1, activation="sigmoid")(x)
model = keras.Model(inputs, outputs)

• inputs = keras.Input(shape=(None,), dtype="int64")
  : shape -> none은 길이가 가변적이라는 것을 나타내고 , , 을 사용하여 튜플을 생성하고 있다

• x = PositionalEmbedding(sequence_length, vocab_size, embed_dim)(inputs)
  : sequence_length - 문장의 최대 길이
  input_dim - vocab_size -> 입력차원을 나타내는 값
  output_dim - embed_dim -> 출력차원을 나타내는 값

• x = layers.GlobalMaxPooling1D()(x)
  :1D 글로벌 맥스 풀링은 시계열 데이터, 텍스트 데이터 등과 같이 1차원 구조를 가진 입력에서 주요 특성을 추출하는 데 사용

model = keras.models.load_model(
    "full_transformer_encoder.keras",
    custom_objects={"TransformerEncoder": TransformerEncoder,
                    "PositionalEmbedding": PositionalEmbedding})

-> 모델을 저장할 때 TransformerEncoder와 PositionalEmbedding 클래스를 사용자 정의 객체로 지정했다면, 모델을 로드할 때도 custom_objects에 해당 클래스들을 제공해야 합니다. 이렇게 하면 Keras가 모델을 로드하는 동안 해당 클래스를 인식하고 올바르게 재구성할 수 있다.