고정 헤더 영역
상세 컨텐츠
본문
* 본 게시글은 http://nlp.seas.harvard.edu/annotated-transformer/ 의 코드를 바탕으로 작성되었습니다.
현재 Transformer의 구조를 바탕으로 설계된 모델들이 AI 모델의 주류로 자리 잡았다.
요즘 놀라운 성능을 보여주고 있는 LLM이나 Vision-Language Model도 기본 구조에
트랜스포머의 모듈을 활용한다.
그래서 이것들을 깊게 이해하고
또 적절한 방법으로 고쳐 쓸 수 있도록 하기 위해
가장 기본이 되는 트랜스포머의 동작을 잘 이해하는 것을 목표하고 있다.
(이 글을 작성하는 시점에 트랜스포머의 후계를 자처하는 논문이 나오긴 했다
: https://arxiv.org/abs/2307.08621)
마침 Annotated Transformer가 있어서 논문 내용과 그것을 구현하는 과정을 쉽게 따라갈 수는 있었지만
읽고 따라하는 것으로 누군가에게 쉽게 설명해줄 수 있을 정도의 이해를 얻지도 못했고
그나마 이해한 것도 또 금방 까먹는 일이 되풀이 되어서 이를 정리해두려 한다.
위 그림은 논문에서 트랜스포머의 전체 구조를 간략히 나타낸 도식(그림)이다.
Inputs(입력)을 받는 왼쪽 부분이 Encoder(인코더)
Outputs(출력)을 받고 Output Probabilities를 출력하는 오른쪽 부분이 Decoder(디코더)이다.
인코더는 입력을 (여기서 입력은 sequence 즉, 순서가 있는 정보) 벡터(Vector)로 압축하여 나타내는 역할을 한다.
여기서 벡터는 중등교육(중, 고등학교)에서 배우는 "크기"와 "방향"을 갖는 양만을 뜻하는 것은 아니고..
벡터 공간(Vector Space)의 원소라는 더 포괄적인 정의 하에서 벡터를 뜻하는 것이다.
단순하게 생각하면 v = [d1 d2 d3 ... dN] 과 같이 N차원의 정보를 갖는 형태로 나타내진다고 생각할 수 있다.
우리의 현실 세계를 생각하면, 예를 들어 v = [175 63 280] 이면
이 1, 2, 3의 정보는 3차원 공간에서 각 축(X Y Z)에 대응할 수 있는 것이므로, "크기"와 "방향"을 나타내는 익숙한 의미의 벡터로 생각할 수도 있다.
하지만 알고보니, 키 / 몸무게 / 발사이즈를 적어 놓은 것이라면? 크기는 그렇다고 쳐도, "방향"을 나타낸다고 보기는 어려울 것이다.
그렇지만 이것도 (벡터 공간의 정의를 따른다는 전제 하에) 벡터이다.
그러니까 벡터에 대해서 굳이 기하적인 의미로 상상할 필요는 없다.
"어떤" 정보를 담고 있는 공간에서 "어떤 값"으로 표현되는지를 보여주는 것이다.
인코더는 입력(일반적으로 우리에게 익숙한, 문장, 주가, 사진 등이 될 수 있다)을
다른 방식으로(일반적으로 정보를 최대한 손실없이 "압축"해서 보여줄 수 있도록) 표현해주는 것이다.
class Encoder(nn.Module):
# Core encoder is a stack of N layers
def __init__(self, layer, N):
super(Encoder, self).__init__()
self.layers = clones(layer, N)
self.norm = LayerNorm(layer.size)
def forward(self, x, mask):
# Pass the input (and mask) through each layer in turn.
for layer in self.layers:
x = layer(x, mask)
return self.norm(x)
더보기
# The encoder is composed of a stack of N = 6 identical layers.
def clones(module, N):
# Produce N identical layers.
return nn.ModuleList([copy.deepcopy(module) for _ in range(N)])
class LayerNorm(nn.Module):
# Construct a layernorm module (See citation for details).
def __init__(self, features, eps=1e-6):
super(LayerNorm, self).__init__()
self.a_2 = nn.Parameter(torch.ones(features))
self.b_2 = nn.Parameter(torch.zeros(features))
self.eps = eps
def forward(self, x):
mean = x.mean(-1, keepdim=True)
std = x.std(-1, keepdim=True)
return self.a_2 * (x - mean) / (std + self.eps) + self.b_2
코드 상에서 인코더의 구성은 위와 같이,
입력을 여러층의 (단층도 가능) 레이어로 통과시키는 형태로 되어 있다.
즉 여기서 구현된 부분은 아래 그림에 따르면 "Nx"에 해당한다.
원하는 모듈을 구현하면 복제하여 통과시키는 구조인 것이다.
LayerNorm은 정규화(Normalization)을 수행하는 것인데, 이는 딥러닝 모델에서 일반적으로 사용하는 Layer이며,
이를 사용하는 트랜스포머"만"의 특별한 이유가 있는 것은 아니다.
따라서 실제로 Encoder가 구현된 코드는 아래와 같이 "EncoderLayer"를 별도로 정의하여 계산을 한다.
Encoder(EncoderLayer(d_model, c(attn), c(ff), dropout), N)
# c = copy.deepcopy
class EncoderLayer(nn.Module):
# "Encoder is made up of self.attn and feed forward (defined below)"
def __init__(self, size, self_attn, feed_forward, dropout):
super(EncoderLayer, self).__init__()
self.self_attn = self_attn
self.feed_forward = feed_forward
self.sublayer = clones(SublayerConnection(size, dropout), 2)
self.size = size
def forward(self, x, mask):
# "Follow Figure 1 (left) for connections"
x = self.sublayer[0](x, lambda x: self.self_attn(x, x, x, mask))
return self.sublayer[1](x, self.feed_forward)
더보기
# LayerNorm(x + Sublayer(x))
class SublayerConnection(nn.Module):
"""
A residual connection followed by a layer norm.
Note for code simplicity the norm is first as opposed to last.
"""
def __init__(self, size, dropout):
super(SublayerConnection, self).__init__()
self.norm = LayerNorm(size)
self.dropout = nn.Dropout(dropout)
def forward(self, x, sublayer):
# "Apply residual connection to any sublayer with the same size."
return x + self.dropout(sublayer(self.norm(x)))
먼저 EncoderLayer의 Forward를 보면 두 개로 정해진 sublayer에 입력을 통과시키는 구조이다.
__init__으로 올라가서, sublayer의 구성을 보면 SublayerConnection이 두 개 있는 형태로 나타난다.
self.sublayer = clones(SublayerConnection(size, dropout), 2)
더보기의 SublayerConnection의 코드는, 아래 그림에서 분기하는 화살표와 Add & Norm을 구현한 것이다.
따라서 핵심적인 구조만 보면 "EncoderLayer"는 Multi-Head Attention → Feed Forward을 수행하는 것이고,
self.self_attn = self_attn
self.feed_forward = feed_forward
여기서도 핵심적인 레이어인 Multi-Head Attention과 Feed Forward는 별도로 구현한 것을 입력받아서 처리한 것임을 볼 수 있다.
x = self.sublayer[0](x, lambda x: self.self_attn(x, x, x, mask))
self_attn(x, x, x, mask)
다만, 그림은 동일한 입력(x)이 3개로 분기하여 Multi-Head Attention에 들어가는 것으로 그려져 있지만, 실제 구현체는 mask까지 총 4개의 입력을 받는다는 차이가 있다.
하지만 인코더에서 활용될 때는 Multi-Head Attention에도 소스(src)와 관련된 입력이
(정확히는 src가 embed layer를 거친 src_embed) 들어오는 것이고
src = torch.LongTensor([[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]])
max_len = src.shape[1]
src_mask = torch.ones(1, 1, max_len)
이와 같이 입력의 마스크(src_mask)는 src의 형태(shape)에 한 차원 더 올리고 1로 구성된 형태이다.
즉,
src → [[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]] / (1,10) - 가장 앞자리 1은 batch size에 해당
src_mask →[ [[1, 1, 1, 1, 1, 1, 1, 1, 1, 1]] ] / (1,1,10)
따라서 인코더에서 실제로 마스킹을 하는 것은 아니지만,
후술할 디코어와 입력형을 맞추기 위해 입력한다고 생각할 수 있다.
이제 어텐션(Attention)과 멀티헤드 어텐션(Multi-Head Attention)의 구현을 알아볼 차례이다.
어텐션이라는 개념은 트랜스포머 전에 제안된 개념이고, "주의"로 번역할 수 있는 것처럼
출력할 것을 예측하기 위해 전체 입력 중 어디에 더 주의를 집중해서 볼 것인가를 구현한 개념이다.
어텐션의 구현 자체는 위와 같은 계산 과정을 진행하는 것에 불과하지만,
왜 이렇게 구현한 것이 어텐션인가를 이해하는 것이 중요할 것이다.
먼저 Q, K, V는 각각
Query(쿼리-질문/질의)
Key(키)
Value(값) 를 뜻한다.
먼저 Key, Value는 파이썬의 Dictionary 자료형을 생각하면 간단하다.
실제 사전(dictionary)이 표제어 - 뜻으로 이루어진 것처럼
파이썬 Dictionary는 Key(표제어) - Value(뜻)으로 형태로 구성된다.
예를 들어 a = {0:"영", 1:"일"} 이라는 dictionary가 있으면 a[0] → "영" 이다.
이때 질의한 0이 "Query"이라고 볼 수 있고, 이와 매칭되는 "Key"를 찾아서 "Value" "영"을 반환한 것이다.
트랜스포머의 Q, K ,V 개념도 이와 같은 흐름과 비슷하다.
물어보고 싶은 (확인하고 싶은) 것들을 Q에 실어 넣으면
K 중, 즉 각 Key와 각 Query와의 유사한 정도를 계산하고 (Dictionary는 같은 것을 찾는다는 점에서 다름)
K와 엮인 V를 내보낼 때 Q와 K가 유사한 정도를 가중치로 곱해 내보내는 흐름이다.
(Dictionary는 Q와 같은 K와 엮인 V만 내보내는 점에서 다름)
그런데 다시 보면 인코더의 MHA(Multi-Head Attention)의 Q, K, V에 들어가는 입력이 모두 동일하다.
class EncoderLayer(nn.Module):
# "Encoder is made up of self.attn and feed forward (defined below)"
def __init__(self, size, self_attn, feed_forward, dropout):
super(EncoderLayer, self).__init__()
self.self_attn = self_attn
self.feed_forward = feed_forward
self.sublayer = clones(SublayerConnection(size, dropout), 2)
self.size = size
def forward(self, x, mask):
# "Follow Figure 1 (left) for connections"
x = self.sublayer[0](x, lambda x: self.self_attn(x, x, x, mask))
return self.sublayer[1](x, self.feed_forward)
self_attn(x, x, x, mask)
이것은 무엇을 뜻할까? 일단 이런 어텐션을 Self-Attention이라고 부른다.
Q, K, V의 입력이 모두 동일해도 되나 싶을 수도 있지만,
있는 그대로 받아들이면 된다.
아래와 같이 문장에서 문맥상 연관된 의미를 파악하는 상황이 Q, K, V가 동일한 상황이다.
문장 내 단어들간 유사도를 구하여 각 단어가 어떤 것과 관련이 있는지 파악하는 상황이다.
https://ai.googleblog.com/2017/08/transformer-novel-neural-network.html
Q가 "it"일 때 문장이 tired로 끝날 때와 wide로 끝날 때 유사도가 가장 높은 단어가 달라진다.
tired로 끝날 때는 K 중 animal과 유사도가 가장 높지만
wide로 끝날 때는 K 중 street와 유사도가 가장 높다.
V도 전부 문장 내의 단어이므로,
Q가 it 일 때 K 중 animal과 유사도가 가장 높다면,
최종적으로 V와 곱해졌을 때 animal이 두드러지는 경우에 대응하는 값이 나올 것이다.
다시 적으면
물어보고 싶은 (확인하고 싶은) 것들을 Q에 실어 넣으면
K 중, 즉 각 Key와 각 Query와의 유사한 정도를 계산하고
K와 엮인 V를 내보낼 때 Q와 K가 유사한 정도를 가중치로 곱해 내보내는 흐름이다.
이제 self_attn이 어떻게 구현되었는지 코드를 살펴본다.
class EncoderLayer(nn.Module):
# "Encoder is made up of self.attn and feed forward (defined below)"
def __init__(self, size, self_attn, feed_forward, dropout):
super(EncoderLayer, self).__init__()
self.self_attn = self_attn
self.feed_forward = feed_forward
self.sublayer = clones(SublayerConnection(size, dropout), 2)
self.size = size
def forward(self, x, mask):
# "Follow Figure 1 (left) for connections"
x = self.sublayer[0](x, lambda x: self.self_attn(x, x, x, mask))
return self.sublayer[1](x, self.feed_forward)
Encoder(EncoderLayer(d_model, c(attn), c(ff), dropout), N)
# d_model = 512
# h = 8
# c = copy.deepcopy
# attn = MultiHeadedAttention(h, d_model)
# ff = PositionwiseFeedForward(d_model, d_ff, dropout)
실제 코드는 전부 분산되어있지만, 이해를 위해 필요한 코드를 모아서 작성했다.
위와 같이 self_attn은 c(attn)이고, c는 객체를 복사하는 것이므로 attn 부분,
즉 MultiHeadedAttention(h, d_model)을 보면 된다.
여기서 h는 헤드의 개수이고, d_model은 모델의 dimension이다.
처음 제안된 어텐션은 한 번에 하는 것이었지만,
트랜스포머에서는 입력을 나누어 어텐션을 동시에 여러 개로 수행하도록 수정하여
멀티 헤드 어텐션이 된 것이다.
이렇게 나누면 전체적인 계산 비용이 감소하고,
여러 개의 헤드가 서로를 보완하는 효과가 있어 좋다고 한다.
class MultiHeadedAttention(nn.Module):
def __init__(self, h, d_model, dropout=0.1):
# Take in model size and number of heads.
super(MultiHeadedAttention, self).__init__()
assert d_model % h == 0
# We assume d_v always equals d_k
self.d_k = d_model // h
self.h = h
self.linears = clones(nn.Linear(d_model, d_model), 4)
self.attn = None
self.dropout = nn.Dropout(p=dropout)
먼저 Multi-Head Attention (이하 MHA) 객체의 생성자 부분을 보면,
assert d_model % h == 0 에서 h는 d_model을 나누어 떨어지게 만들어야 하는 것을 알 수 있다.
즉, 여러 개의 헤드가 모두 동일한 만큼 정보 나눠 갖게 될 것임을 알 수 있다.
self.d_k는 이름처럼 K의 차원(dimension)을 뜻하는데, d_model를 h로 나눈 값임을 알 수 있다.
그리고 V의 차원은 K의 차원과 같다. Key-Value로 엮이므로 다를 이유는 없다.
self.linears에서 입력, 출력의 차원이 d_model와 같은 Linear Layer "4개"를 만든다.
4라는 고정된 값이고, h나 d_model에 연관된 숫자가 아니므로,
여기서는 단지 구조상 그렇게 될 수 밖에 없다 정도로 넘어가도 될 것이다.
이제 본격적인 계산 부분이 시작된다.
class MultiHeadedAttention(nn.Module):
# def __init__
def forward(self, query, key, value, mask=None):
if mask is not None:
# Same mask applied to all h heads.
mask = mask.unsqueeze(1)
nbatches = query.size(0)
# ...
forward를 보면 self_attn(x, x, x, mask) 이었고, self_attn이 곧 MultiHeadAttention 객체였으므로,
입력 x가 각각 query, key, value로 mask는 mask로 들어가는 것임을 확인할 수 있다.
처음 처리하는 것은 mask에 unsqueeze하는 것인데,
위에서 src와 src_mask의 형태가 아래와 같음을 확인했다.
src → [[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]] / (1,10) - 가장 앞자리 1은 batch size에 해당
src_mask →[ [[1, 1, 1, 1, 1, 1, 1, 1, 1, 1]] ] / (1,1,10)
여기에 dim=1에 unsqueeze를 하면 mask는 (1,1,1,10) 이와 같은 형태로 차원이 확장된다.
이렇게 차원을 하나 더 늘리는 이유는
여러 개의 Head로 값을 분할해서 넣을 것이기 때문임을 추측할 수 있을 것이다.
그리고 query의 size의 Index 0의 값이 batch size에 해당하여 nbatches에 저장해둔다.
그런데 사실 여기까지 입력 x가 무엇인지 정확히 확인하지는 않았는데,
이제 query, key, value로 활용하는 입력 x가 무엇인지 정확히 확인할 필요가 있다.
우선 src_mask를 그대로 입력받는 mask와 다르게
query, key, value는 src를 그대로 받지 않는다.
여기까지 계속 인코더에서 인코딩하는 것(encode)를 하고 있는 과정임을 상기하고,
원래 모델 구조를 다시 보면 이렇다.
인코더 입력 전에 Embedding과 Positional Encoding이 먼저 수행되고
그것이 인코더에 입력으로 들어가는 것이다.
이 부분은 트랜스포머 동작을 이해하는데 핵심적인 부분은 아니다.
임베딩은 사실 중요한 개념인데,
트랜스포머를 떠나서 일반적으로 많이 사용하는 개념이라서 간단히만 언급한다.
Embedding은 단어를 벡터로 표현하는 것인데,
One-hot Vector로 표현했을 때 생기는 문제를 해결하고,
의미상 가까운 단어가 단어를 나타내는 공간에서도 가깝게 될 수 있도록 하여
실질적으로 의미를 반영하면서 단어를 벡터로 표현하는 방법이다.
Positional Encoding은 기존 순환신경망과 달리
트랜스포머가 입력을 동시에 처리하기 때문에
입력의 위치 정보가 훼손되는 문제를 해결하기 위해 도입된 것이다.
아래로 이어지는 어텐션 연산 과정을 보면 알겠지만,
입력 순서가 중요하지가 않다.
어차피 행렬(정확히는 텐서) 곱 연산이라서 순서만 바꾸면 같은 행렬이 된다.
문제는 이것들이 최종적으로 표현할 단어의 확률 분포로
Linear mapping되는데 (Linear layer를 통과하는 것 자체가 Linear mapping이다)
이때 예를 들어 입력을 [0,1]로 넣든 [1,0]으로 넣든
어떤 가중치랑 곱해져서 전부 더해서 값을 만들기 때문에
입력 순서는 변별력이 없게 된다.
그런데 트랜스포머가 문장을 처리한다는 점을 생각했을 때,
위치 정보가 없어지는 것은 큰 문제가 된다.
한국어는 그나마 기능어 때문에 어순에서 비교적 자유롭지만
영어는 I gave the cat the mouse. 와
I gave the mouse the cat. 은 완전히 다른 의미이다.
물론 다른 언어들도 다 마찬가지이다.
따라서 입력 순서를 반영할 추가 정보를 넣어줘야
위치 정보를 모델이 반영할 수 있다.
그것을 위해 벡터로 임베딩된 단어에 다시 Positional Encoding을 하는 것이고
다만, 위치 정보를 부여하기 위해 더한 값이 단어의 의미를 변화시켜서도 안 되기 때문에
(여기부터 단어는 이미 벡터 공간에 숫자로 표현되고 있음을 상기해야 한다)
적당한 조건을 만족하는 방법으로 Positional Encoding을 하는데,
본 논문에서는 heuristic 방법으로
Positional Encoding 방법을 정한 것으로 알려져 있고,
본 글은 트랜스포머에서 정보가 흐르는 과정을 살펴보는 것이 주 목적이므로,
이 부분은 따로 설명하지는 않는다.
(검색해보면 이것도 자세히 설명한 다른 좋은 글들이 많다)
그래서 정리하면 Embedding과 Positional Encoding은
단어의 의미를 포함하여 벡터로 나타내고,
위치 정보를 더해서 모델이 처리할 수 있게 처리해주는 과정으로 이해하면 된다.
query, key, value에는 src_embed가 입력되며
src → [[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]] / (1,10) - 가장 앞자리 1은 batch size에 해당
src_mask →[ [[1, 1, 1, 1, 1, 1, 1, 1, 1, 1]] ] / (1,1,10)
src_mask (unsqueeze(1) 후) →(1,1,1,10)
src_embed →(1,10,512)
src가 Embedding과 Positional Encoding을 거치면
d_model의 크기와 같은 Features 크기 (여기서는 512)를 갖도록 변환되는 것을 확인하고 넘어가도록 한다.
하지만 맨 앞이 batch size인 것은 src와 동일하다.
class MultiHeadedAttention(nn.Module):
# def __init__
def forward(self, query, key, value, mask=None):
# ...
# 1) Do all the linear projections in batch from d_model => h x d_k
query, key, value = [
lin(x).view(nbatches, -1, self.h, self.d_k).transpose(1, 2)
for lin, x in zip(self.linears, (query, key, value))
]
다음으로 query, key, value를 __init__에서 정의했던 Linear Layer에 통과시킨다.
lin에는 self.linears의 Linear layer가 하나씩 순서대로 대입된다.
x에는 (query, key, value)가 순서대로 대입되어,
총 3번 반복된다.
위에서 self.d_k = d_model // h 이었으므로 d_k = 64이다. (512 / 8)
Linear layer 통과 후 view를 이용하여 512(=d_model)를 h x d_k로 쪼개고 / (1,10,8,64)
transpose로 h와 입력의 길이(10)의 차원을 바꾼다. / (1,8,10,64)
각 헤드(여기서는 8개)에 나눠서 입력을 넣기 위한 것이다.
처음부터 view를 (nbatches, self.h, -1, self.d_k)로 하지 않는 것은
shape 자체는 (1,8,10,64)로 동일하겠지만, 내용의 구성이 다르기 때문이다.
512에 해당하는 부분을 쪼개야 하는데,
이렇게 하면 (1,10,8,64)에서 10에 해당하는 부분에서 쪼개지기 때문에 다르다.
class MultiHeadedAttention(nn.Module):
# def __init__
def forward(self, query, key, value, mask=None):
# ...
# 2) Apply attention on all the projected vectors in batch.
x, self.attn = attention(
query, key, value, mask=mask, dropout=self.dropout
)
이제 어텐션을 수행하는 차례이다.
헤드 개수에 맞게 입력을 쪼갰지만, 입력은 한 번에 넣는다.
그림 상으로는 각각 입력하는 것처럼 보이고,
실제로 그렇게 하는게 불가능한 것도 아니지만
효율적인 계산을 위해 실제로 각각 처리하는 것은 아니고
동일한 결과를 얻지만 행렬(Matrix) 계산은 한 번에 수행하기 때문이다.
def attention(query, key, value, mask=None, dropout=None):
# Compute 'Scaled Dot Product Attention'
d_k = query.size(-1)
scores = torch.matmul(query, key.transpose(-2, -1)) / math.sqrt(d_k)
if mask is not None:
scores = scores.masked_fill(mask == 0, -1e9)
p_attn = scores.softmax(dim=-1)
if dropout is not None:
p_attn = dropout(p_attn)
return torch.matmul(p_attn, value), p_attn
Attention의 구현체는 생각보다는 단순하다. 위의 식/그림을 코드로 구현한 것 뿐이기 때문이다.
트랜스포머의 Attention은 Scaled Dot Product Attention 인데,
Q와 K끼리 dot product를 수행하고, sqrt(d_k)으로 나누기(scale) 때문이다.
(어텐션을 구현하기 위해 사용하는 연산 방법은 여러 개 있고, 여기서는 이런 방법을 쓰는 것이다)
먼저 d_k는 query의 size의 마지막 부분으로 받는데,
위에서 query, key, value 모두 (1,8,10,64)
즉 (batch_size, head, input_length, d_k) 형태이기 때문이다.
그 다음으로
scores = torch.matmul(query, key.transpose(-2, -1)) / math.sqrt(d_k)
Q와 K의 유사도(정확히는 어텐션)을 계산한다.
query와 key 모두 (1,8,10,64)의 형태이므로 행렬 곱을 위해서 형태를 변환시켜야 한다.
여기서는 각 헤드별로 계산하는 것을 구현해야 하므로 헤드의 개수에 해당하는 8은 건들면 안 될 것이다.
다만 엄밀히 말하면 4차원이므로 행렬이 아니라 텐서(Tensor)이므로, 텐서끼리의 곱을 하게 되는 것이다.
key의 마지막에서 두 번째 차원과 첫 번째 차원끼리 바꾸므로,
결과적으로 (1,8,10,64) X (1,8,64,10) 형태이고, 계산 결과의 형태는 (1,8,10,10)이 된다.
이처럼 계산 자체는 한 번에 수행하지만,
각 헤드별로 독립적으로 계산한 것을 다시 붙인 것과 차이는 없다.
if mask is not None:
scores = scores.masked_fill(mask == 0, -1e9)
이어서 mask 연산을 수행한다.
현재 보고 있는 것은 인코더이고, 인코더에서는 mask는 있지만
실제로는 전부 1로 처리되어 masking을 하지는 않는다.
만약 mask에 0으로 처리된 부분이 있다면 -1e9로 scores 값이 바뀐다.
한편 여기서
scores의 형태는 (1,8,10,10) 이고,
mask의 형태는 (1,1, 1, 10) 이지만,
알아서 형태를 맞춰서 잘 처리해준다. (마지막 각 항목의 길이가 10으로 동일하기 때문)
마지막으로
p_attn = scores.softmax(dim=-1)
if dropout is not None:
p_attn = dropout(p_attn)
return torch.matmul(p_attn, value), p_attn
softmax를 취해준다.
dim=-1이어야 길이가 10(입력 길이)인 마지막 차원을 따라서 계산을 한다.
수식이나 그림에는 표현되어 있지 않지만, 일반화를 위해서 dropout도 포함하고
마지막으로 어텐션 결과와(p_attn)와 value를 곱해서 반환한다.
어텐션 결과는 예를 들면 다음과 같은 형태이다.
현재 예제는 입력의 길이가 10이지만, 아래 그림은 15인 것만 감안하면 된다.
p_attn의 형태는 (1,8,10,10)이지만,
맨 앞은 배치 사이즈이고, 그 다음은 헤드의 개수로, 따로 계산하는 것이므로
(10,10)이 한 문장(src)에서 각 헤드에서 어텐션을 계산한 결과를 보여주는 것이다.
왼쪽의 Y축을 Q, 아래의 X축을 K라고 생각하면
Q의 각 단어와 K의 각 단어간 어텐션을 표현하는 것이다.
이때 value의 형태는 (1,8,10,64) 이므로 p_attn과 value를 곱하면
(1,8,10,10) x (1,8,10,64)최종적으로 (1,8,10,64)의 형태로 결국 어텐션 모듈에 입력했을 때와 동일한 형태로 나온다.
즉, Q로 들어온 단어와 K에 있는 각 단어들간 어텐션을 계산한 것과
K에 엮인 V에 있는 단어(셀프 어텐션에서는 같은 단어임)를 곱한 형태가 된다.
위의 그림을 실제로 행렬 곱의 규칙에 따라 매칭해보면, 아래와 같다.
즉, 최종적으로 p_attn과 value를 곱해 어텐션을 끝낸 값(행렬)은
예를 들어 10번째 행의 의미는 Q로 10번째 단어가 들어왔을 때,
그 10번째 단어가 다른 단어들과 계산된 어텐션이 어느정도였는지를 반영한 수치이다.
쉽게 생각해서 인기를 얻는다고 생각해보면
10번째 단어가 다른 단어들을 순회하면서 어느정도 인기를 얻었는지 적어놓고,
인기 정도만큼 자기 자신을 높이는 것이다. (안타깝게도 작아질수도 있다)
(인지하기 쉽게 숫자로 쓰고 있지만
1→batch size
8→head 개수
10→input 길이
64→d_k라는 흐름을 기억해야 한다)
유사도를 계산한 값도 반환한다.
다시 attention 연산 모듈을 빠져나와 Multi-Head Attention을 마무리한다.
class MultiHeadedAttention(nn.Module):
def __init__(self, h, d_model, dropout=0.1):
# Take in model size and number of heads.
super(MultiHeadedAttention, self).__init__()
assert d_model % h == 0
# We assume d_v always equals d_k
self.d_k = d_model // h
self.h = h
self.linears = clones(nn.Linear(d_model, d_model), 4)
self.attn = None
self.dropout = nn.Dropout(p=dropout)
def forward(self, query, key, value, mask=None):
if mask is not None:
# Same mask applied to all h heads.
mask = mask.unsqueeze(1)
nbatches = query.size(0)
# 1) Do all the linear projections in batch from d_model => h x d_k
query, key, value = [
lin(x).view(nbatches, -1, self.h, self.d_k).transpose(1, 2)
for lin, x in zip(self.linears, (query, key, value))
]
# 2) Apply attention on all the projected vectors in batch.
x, self.attn = attention(
query, key, value, mask=mask, dropout=self.dropout
)
# 3) "Concat" using a view and apply a final linear.
x = (
x.transpose(1, 2)
.contiguous()
.view(nbatches, -1, self.h * self.d_k)
)
del query
del key
del value
return self.linears[-1](x)
3번의 과정은 Multi-Head로 입력하기 위해 쪼개놓은 것을 다시 붙이는 과정이다.
x 형태인 (1,8,10,64) / (bs, h, len, d_k)에서
transpose(1,2)로 (1,10,8,64)로 헤드 별로 계산하기 위해 변환한 형태를
다시 h x d_k 꼴로 바꿔준다.
contiguous()는 메모리 주소의 연속성을 위해 사용한 것이고
마지막으로 view를 통해서 h x d_k로 쪼개기 전 상태로 완전히 되돌린다.
최종적인 형태는 MHA 처음 입력한 형태와 동일한 (1,10,512) / (bs, len, d_model)이다.
그리고 어텐션 결과를 지금까지 사용하지 않았던 마지막 Linear layer에 통과시켜 내보낸다.
d_model = 512
h = 8
c = copy.deepcopy
attn = MultiHeadedAttention(h, d_model)
Encoder(EncoderLayer(d_model, c(attn), c(ff), dropout), N)
class EncoderLayer(nn.Module):
# Encoder is made up of self.attn and feed forward (defined below)
def __init__(self, size, self_attn, feed_forward, dropout):
super(EncoderLayer, self).__init__()
self.self_attn = self_attn
self.feed_forward = feed_forward
self.sublayer = clones(SublayerConnection(size, dropout), 2)
self.size = size
def forward(self, x, mask):
# Follow Figure 1 (left) for connections
x = self.sublayer[0](x, lambda x: self.self_attn(x, x, x, mask))
return self.sublayer[1](x, self.feed_forward)
여기까지가 forward에서
self.self_attn(x, x, x, mask)) 를 계산한 것이다.
사전에 정의해둔 sublayer 덕분에 아래 그림에서 보이는 Add & Norm 연산은 코드 상으로는 이어서 바로 수행된다.
이제 Feed Forward를 이해하면 인코더의 인코드 연산의 흐름은 모두 파악하는 것이다.
class PositionwiseFeedForward(nn.Module):
# Implements FFN equation.
def __init__(self, d_model, d_ff, dropout=0.1):
super(PositionwiseFeedForward, self).__init__()
self.w_1 = nn.Linear(d_model, d_ff)
self.w_2 = nn.Linear(d_ff, d_model)
self.dropout = nn.Dropout(dropout)
def forward(self, x):
return self.w_2(self.dropout(self.w_1(x).relu()))
PositionwiseFeedForward(d_model, d_ff, dropout)
# d_model = 512
# d_ff = 2048
# dropout = 0.1
Feed Forward Network는 Linear Transformation 두 번으로 이루어진다.
위에서 본 것처럼 FFN에 입력될 때
MHA의 출력의 차원이 모델의 차원과 같기 때문에 / (*,512)
Linear layer의 입력 차원이 d_model이고 출력 차원이 d_ff가 되었다가
다시 두 번째 Linear layer에서 나올 때는 d_model로 돌아온다.
수식에는 max(0, x)의 형식으로 나와 있지만,
Activation Function인 ReLU가 동일한 동작을 하므로 코드에는 relu로 구현되어 있다.
return self.sublayer[1](x, self.feed_forward)
최종적으로 sublayer에서 Add & Norm 연산이 실행되면서 인코더의 인코딩 과정이 마무리된다.
이 과정이 인코더 1개에서 수행되는 연산이며,
트랜스포머는 이런 인코더를 여러 개 쌓아서 구성될 수도 있다.
디코더(Decoder)는 기본적으로 인코딩과 반대로 압축된 정보를 다시 복원해주는 역할이다.
디코더는 트랜스포머 구조의 우측에 해당한다.
인코더처럼 처음에는 받은 입력("Outputs", target)에서 셀프 어텐션으로 결과를 내는데,
두 번째 어텐션에서는 인코딩된 값을 Q와 K로 쓰고 V를 디코더의 셀프 어텐션 값으로 받는다.
그리고 다음은 인코더와 동일하게 FFN을 지나고,
마지막으로 Linear layer를 통과해 Softmax를 통해 확률분포를 나타내는 형태이다.
# The decoder is also composed of a stack of N = 6 identical layers.
class Decoder(nn.Module):
# Generic N layer decoder with maksing
def __init__(self, layer, N):
super(Decoder, self).__init__()
self.layers = clones(layer, N)
self.norm = LayerNorm(layer.size)
def forward(self, x, memory, src_mask, tgt_mask):
for layer in self.layers:
x = layer(x, memory, src_mask, tgt_mask)
return self.norm(x)
더보기
class Encoder(nn.Module):
# Core encoder is a stack of N layers
def __init__(self, layer, N):
super(Encoder, self).__init__()
self.layers = clones(layer, N)
self.norm = LayerNorm(layer.size)
def forward(self, x, mask):
# Pass the input (and mask) through each layer in turn.
for layer in self.layers:
x = layer(x, mask)
return self.norm(x)
디코더 자체는 인코더와 마찬가지로 입력을 여러층의 (단층도 가능) 레이어로 통과시키는 형태로 되어 있다.
인코더와 마찬가지로, 실질적인 디코더 기능은 Decoder layer를 통해 구현하게 되고,
디코더는 입력이 여러 레이어를 통과할 수 있게 만들어진 구조이다.
하지만 인코더와 큰 차이점이 있는데,
인코더는 소스와 마스크만 입력으로 받지만,
디코더는 "Outputs" (그림), 메모리, 소스 마스크와 타겟 마스크를 입력으로 받는다.
다음으로 디코딩 과정을 알아보기 위해 Decoder layer를 본다.
"""
In addition to the two sub-layers in each encoder layer,
the decoder inserts a third sub-layer, which performs multi-head attention over the output of the encoder stack.
"""
class DecoderLayer(nn.Module):
# Decoder is made of self-attn, src-attn, and feed forward (defined below)
def __init__(self, size, self_attn, src_attn, feed_forward, dropout):
super(DecoderLayer, self).__init__()
self.size = size
self.self_attn = self_attn
self.src_attn = src_attn
self.feed_forward = feed_forward
self.sublayer = clones(SublayerConnection(size, dropout), 3)
def forward(self, x, memory, src_mask, tgt_mask):
# Follow Figure 1 (right) for connections.
m = memory
x = self.sublayer[0](x, lambda x: self.self_attn(x, x, x, tgt_mask))
x = self.sublayer[1](x, lambda x: self.src_attn(x, m, m, src_mask))
return self.sublayer[2](x, self.feed_forward)
그림에서 보이는 것처럼 처음에는 Q, K, V 모두에 같은 입력 넣는 셀프 어텐션을 수행한다.
이때 코드를 보면 mask에 tgt_mask가 들어가게 되는데,
그림에서 Masked Multi-Head Attention이라고 적혀 있는 것처럼
전부 1로 되어 있어 실질적으로 마스킹을 하지 않는 src_mask와 다르게
tgt_mask는 실제로 마스킹을 수행할 것임을 예상할 수 있다.
그리고 처음에는 self_attn으로 어텐션을 수행하지만,
다음에는 src_attn이라는 별도의 객체에서 어텐션을 수행한다.
셀프 어텐션은 위의 인코더에서 이미 보았지만,
tgt_mask가 있으므로 src_attn으로 넘어가기 전에
tgt_mask부터 살펴본다.
# We also modify the self-attention sub-layer in the decoder stack to prevent positions from attending to subsequent positions.
def subsequent_mask(size):
# Mask out subsequent positions.
attn_shape = (1, size, size)
subsequent_mask = torch.triu(torch.ones(attn_shape), diagonal=1).type(torch.uint8)
return subsequent_mask == 0
ys = torch.zeros(1, 1).type_as(src)
tgt_mask = subsequent_mask(ys.size(1)).type_as(src.data)
실제 코드를 편집해서 보기 편하게 재구성하였다.
tgt_mask는 decode()에 인자로 들어가는 형식이지만,
편의를 위해 명시적으로 분리해서 tgt_mask로 나타내었다.
tgt_mask는 subsequent_mask를 통해 생성된다.
subsequent_mask는
torch.ones로 1로 구성된 3차원 Tensor를 만들고,
triu, 즉 upper triangular를 만든다.
이때 diagonal = 1의 의미는 공식 예제를 참고하면 다음과 같다.
>>> a
tensor([[ 0.2309, 0.5207, 2.0049],
[ 0.2072, -1.0680, 0.6602],
[ 0.3480, -0.5211, -0.4573]])
>>> torch.triu(a)
tensor([[ 0.2309, 0.5207, 2.0049],
[ 0.0000, -1.0680, 0.6602],
[ 0.0000, 0.0000, -0.4573]])
>>> torch.triu(a, diagonal=1)
tensor([[ 0.0000, 0.5207, 2.0049],
[ 0.0000, 0.0000, 0.6602],
[ 0.0000, 0.0000, 0.0000]])
>>> torch.triu(a, diagonal=-1)
tensor([[ 0.2309, 0.5207, 2.0049],
[ 0.2072, -1.0680, 0.6602],
[ 0.0000, -0.5211, -0.4573]])
triu를 실행하면 대각선을 포함하여 위로 삼각형을 만드는데,
diagonal = 1 이면 대각선에서 위로 하나 올려 삼각형을 만든다.
반대로 -1이면 대각선에서 아래로 하나 내려 삼각형을 만든다.
위의 코드를 따르면 ys는 (1,1)의 형태를 가지므로,
subsequent_mask(1)이 실행되고,
subsequent_mask 내의 attn_shape은 (1,1,1)가 된다.
형태가 (1,1,1)인 upper triangular이면 [[ [1] ]]인데,
이때 diagonal = 1이므로 [[ [0] ]] 가 된다.
마지막으로
return subsequent_mask == 0
이므로, [[ [True] ]]가 반환될 것이다.
같은 방식으로 subsequent_mask(2)를 생각해보면
attn_shape = (1,2,2)
subsequent_mask = [[ [0,1],
[0,0] ]]
따라서 반환되는 값은 [[ [True,False]
[True,True] ]]
subsequent_mask(3)를 생각해보면
attn_shape = (1,3,3)
subsequent_mask = [[ [0,1,1],
[0,0,1],
[0,0,0] ]]
따라서 반환되는 값은 [[ [True,False,False],
[True,True,False],
[True,True,True] ]]
즉, 첫 번째 줄은 첫 번째만 True,
두 번째 줄은 두 번째까지 True,
세 번째 줄은 세 번 째까지 True,
네 번째 줄은 네 번째까지 True,
...
이렇게 mask가 만들어진다.
즉,
사이즈가 20이면 노란색이 True(1), 보라색이 False(0)로 위와 같은 그림처럼 된다.
subsequent라는 의미처럼 마스크를 하나씩 개방하는 것과 같다.
이것은 모델이 순서대로 다음 위치의 단어를 예측할 때,
예측할 단어 이후에 나오는 단어들을 참조하는 것을 막기 위한 것이다.
따라서 계산의 흐름은 위에서 살펴본 인코더와 동일하지만
def attention(query, key, value, mask=None, dropout=None):
# Compute 'Scaled Dot Product Attention'
d_k = query.size(-1)
scores = torch.matmul(query, key.transpose(-2, -1)) / math.sqrt(d_k)
if mask is not None:
scores = scores.masked_fill(mask == 0, -1e9)
p_attn = scores.softmax(dim=-1)
if dropout is not None:
p_attn = dropout(p_attn)
return torch.matmul(p_attn, value), p_attn
여기서
scores = scores.masked_fill(mask == 0, -1e9)
에 따라
보라색 부분이 -1e9 (~0)로 마스킹된다.
참고로 인코더와 달리 scores와 mask의 형태(shape)는 맨 앞 batch_size 부분을 제외하면 동일하다.
그 이유는
# We also modify the self-attention sub-layer in the decoder stack to prevent positions from attending to subsequent positions.
def subsequent_mask(size):
# Mask out subsequent positions.
attn_shape = (1, size, size)
subsequent_mask = torch.triu(torch.ones(attn_shape), diagonal=1).type(torch.uint8)
return subsequent_mask == 0
ys = torch.zeros(1, 1).type_as(src)
tgt_mask = subsequent_mask(ys.size(1)).type_as(src.data)
위와 같이 마스크를 만들 때 출력의 형태를 이용해서 만들기 때문이다.
이때, 구현 코드를 보면 ys가
ys = torch.zeros(1, 1).type_as(src)
out = test_model.decode(memory, src_mask, ys, subsequent_mask(ys.size(1)).type_as(src.data))
디코더에 입력으로 들어가는 "Outputs"에 해당하는데,
처음 들어가는 ys의 형태는 (1,1)에 불과하다.
for i in range(9):
out = test_model.decode(memory, src_mask, ys, subsequent_mask(ys.size(1)).type_as(src.data))
prob = test_model.generator(out[:, -1])
_, next_word = torch.max(prob, dim=1)
next_word = next_word.data[0]
ys = torch.cat([ys, torch.empty(1, 1).type_as(src.data).fill_(next_word)], dim=1)
하지만 위와 같이 인코더의 입력 길이(여기서는 10)까지 도달하도록 반복하고 → range(9)에 해당,
가장 마지막 출력 (out[:, -1])에서
next_word를 예측하여 다시 ys에 붙여서 갱신하는 방식으로
ys는 인코더의 입력 길이까지 도달하게 된다.
차이점은 ys는 모델의 결과를 붙여가면서 길이는 늘리는 것이다.
트랜스포머가 기계번역에 쓰일수 있도록 만들어진 것을 생각하면,
인코더에는 출발 언어(예를 들어 한국어)의 문장이 입력되고,
디코더는 도착 언어(예를 들어 영어)의 단어를 하나 출력하고,
그 전 단어들과 출발 언어의 문장을 다시 참고하여 그 다음 단어를 출력하는 식으로
문장을 만들어나가는 것이라고 생각할 수 있다.
다음으로는 셀프 어텐션을 지나서 인코더의 정보를 활용하는 어텐션을 진행하게 된다.
x = self.sublayer[1](x, lambda x: self.src_attn(x, m, m, src_mask))
# memory = test_model.encode(src, src_mask)
# m = memory
코드상의 어텐션은 순서대로 Query, Key, Value, Mask를 입력받으므로,
여기서 Query는 앞선 디코더의 셀프 어텐션 결과이고,
Key와 Value는 인코딩된 값을 이용한다.
여기서는 src_mask를 사용하므로 별도 마스킹이 존재하지 않는다.
이미 마스킹 처리된 값을 Query로 쓰기 때문이다.
인코더에서 나온 값은 위에서 언급한 것처럼
(1,10,512) / (bs, len, d_model)로
Embedding과 Positional Encoding을 거쳐
인코더/디코더에 입력되는 데이터 형태와 다르지 않다.
따라서 그대로 어텐션 연산을 수행할 수 있으며,
디코더에서 Self-Attention된 결과와
인코딩된 정보를 Attention 하여
디코더에서 질문하는 정보(Query)와 인코딩된 값(Key)의 어텐션을 보고
그 값에 따라 인코딩된 값(Value)에 곱해서 내보낸다.
이러한 방법으로 인코더에 입력한 값(트랜스포머의 목적에 따르면 출발 언어)과
디코더에서 출력한 값(트랜스포머의 목적에 따르면 도착 언어)을 결합시키는 것이다.
그리고 마지막으로 FFN에 통과시킨다.
return self.sublayer[2](x, self.feed_forward)
그리고 마지막으로 Linear layer와 Softmax를 통과시켜 확률분포로서 결과를 내보내는데,
트랜스포머의 본래 목적에 비추면 인코더에 입력한 문장과 이전 출력을 고려하여
다음에 올 가장 적절한 단어를 예측하는 것이다.
class Generator(nn.Module):
# Define standard linear + softmax generation step.
def __init__(self, d_model, vocab):
super(Generator, self).__init__()
self.proj = nn.Linear(d_model, vocab)
def forward(self, x):
return log_softmax(self.proj(x), dim=-1)
Linear layer와 Softmax는 위와 같이 Generator로 구현되어 있고,
보이는 것처럼 Linear와 log_softmax 하나씩 구현되어 있다.
여기까지 트랜스포머에서 정보를 처리하는 과정이 마무리된다.
def inference_test():
test_model = make_model(11, 11, 2)
test_model.eval()
src = torch.LongTensor([[1,2,3,4,5,6,7,8,9,10]])
src_mask = torch.ones(1, 1, 10)
memory = test_model.encode(src, src_mask)
ys = torch.zeros(1, 1).type_as(src)
for i in range(9):
out = test_model.decode(memory, src_mask, ys, subsequent_mask(ys.size(1)).type_as(src.data))
prob = test_model.generator(out[:, -1])
_, next_word = torch.max(prob, dim=1)
next_word = next_word.data[0]
ys = torch.cat([ys, torch.empty(1, 1).type_as(src.data).fill_(next_word)], dim=1)
print("Example Untrained Model Prediction:", ys)
정리를 위해 추론 코드를 살펴보면,
먼저 모델을 만들고
인코더의 입력(src)과 마스크 (실제로 마스킹을 하진 않음)을 정의하고,
인코더에 소스(src)와 소스 마스크(src_mask)를 통과시켜 인코딩된 정보를 얻는다.
그리고 맨 처음에는 출력(target, tgt)이 없으므로 더미로 빈 출력을 만들고,
(실제 자연어를 처리할 때는 시작을 뜻하는 토큰으로 처리한다)
디코더에 인코딩된 정보와 출력인 타겟,
그리고 타겟 마스크(미래에 해당하는 위치를 활용하지 않도록 마스킹 ~ 0으로 처리)를
디코더에 통과시켜 다음에 나올 단어(토큰)를 얻는다.
마지막으로 이 정보를 generator에 통과시켜 가능한 단어(토큰) 확률분포를 얻고,
그 중 가장 가능성이 높은 확률을 가진 단어(토큰)를 다음 단어(토큰)로 정한다.
여기까지 설명들은 트랜스포머 모델의 계산의 흐름에 어떤 의미가 있는지 설명한 것이다.
하지만 한편으로는 어쨌거나 그냥 순서에 따라 계산하는 것에 불과하기도 하다.
그래서 당연하지만 의도를 가진 설계와 달리 학습하지 않으면 무의미한 값만 나온다.
다만, 그러한 의도를 가지고 설계되었기 때문에
입력과 정답 쌍을 바탕으로 학습시킬 때
트랜스포머 구조를 이용하면 효과적으로 규칙/패턴을 찾아내는 것이다.
모델 구조 이외에도 실제로 사용하려면
데이터셋, 데이터로더, 학습, 성능 평가 등 필요한 내용이 많지만
이 부분은 트랜스포머에만 사용하는 내용들은 아니므로
본 글에서는 다루지 않는다.
따라서 실제적인 전체 구현은 맨 위에서도 밝힌
http://nlp.seas.harvard.edu/annotated-transformer/
The Annotated Transformer
nlp.seas.harvard.edu
해당 링크를 통해서 학습하면 된다.
댓글 영역