STEP 05 / 18

단어를 벡터로, 순서를 신호로

원본: 임베딩과 소프트맥스 · 위치 인코딩

모델에 들어가는 것은 단어 id(정수)다. 이것을 512차원 벡터로 바꾸는 부품 두 개를 만든다.

Embeddings: nn.Embedding 표에서 id에 해당하는 행을 꺼낸 뒤 $\sqrt{d_{model}}$을 곱한다. 바로 뒤에 더할 위치 인코딩 값(-1~1)과 크기를 맞추기 위해서다.

PositionalEncoding: 어텐션은 순서를 모른다. 단어 순서를 섞어 넣어도 결과가 그대로 섞여 나올 뿐이다. 그래서 위치마다 고유한 사인·코사인 무늬를 임베딩에 더한다.

$$PE_{(pos,2i)} = \sin(pos / 10000^{2i/d_{model}}),\quad PE_{(pos,2i+1)} = \cos(pos / 10000^{2i/d_{model}})$$

차원마다 파장이 다르다. 앞쪽 차원은 빠르게, 뒤쪽 차원은 느리게 진동한다. 코드는 div_term을 로그 공간에서 계산하고, 짝수 열에 sin, 홀수 열에 cos을 채운다. 학습하는 값이 아니라서 nn.Parameter가 아닌 register_buffer로 등록한다.

실행 결과에서 볼 것

× √512를 하면 임베딩 표준편차가 약 1에서 약 22로 커진다.
위치 0은 [0, 1, 0, 1, ...](sin 0, cos 0)로 시작한다.
같은 단어(id 42)가 두 위치에 있으면 임베딩은 같지만 PE를 더한 뒤에는 달라진다.
차트는 원본의 example_positional이다. 차원 4~7의 파형인데, 차원 번호가 클수록 주기가 길다.

코드 변화

+33 −0 · 파일 전체 119줄

이전 단계의 transformer.py와 비교했다. 초록 줄이 새로 생긴 줄, 빨간 줄이 사라진 줄이다.

GitHub에서 보기 ↗

실행 결과

실행한 코드 demo.py 보기

1	import math
2
3	import altair as alt
4	import pandas as pd
5	import torch
6
7	from show import show
8	from transformer import Embeddings, PositionalEncoding
9
10	torch.manual_seed(0)
11
12	emb = Embeddings(d_model=512, vocab=1000)
13	tokens = torch.tensor([[17, 42, 7]])
14	e = emb(tokens)
15	print("토큰 id", tokens.tolist(), "→ 임베딩", tuple(e.shape))
16	print(f"nn.Embedding 값의 표준편차 : {emb.lut(tokens).std():.3f}")
17	print(f"× √512 = × {math.sqrt(512):.1f} 한 뒤 : {e.std():.3f}")
18	print()
19
20	pe = PositionalEncoding(d_model=512, dropout=0.0)
21	print("PE 버퍼:", tuple(pe.pe.shape), " (1, max_len, d_model)")
22	print("학습되는 파라미터 수:", sum(p.numel() for p in pe.parameters()), " ← register_buffer라 학습하지 않는다")
23	print()
24	y = pe(torch.zeros(1, 4, 512))
25	print("위치 0~3의 앞 6개 차원 (0차원=sin, 1차원=cos, ...)")
26	print(y[0, :, :6].round(decimals=3))
27	print()
28	print("같은 단어라도 위치가 다르면 다른 벡터가 된다")
29	same = emb(torch.tensor([[42, 42]]))
30	z = pe(same)
31	print(f" 임베딩만: 두 벡터 차이 {(same[0, 0] - same[0, 1]).norm():.2f}")
32	print(f" PE 더한 뒤: 두 벡터 차이 {(z[0, 0] - z[0, 1]).norm():.2f}")
33
34
35	# 원본 노트북의 example_positional
36	def example_positional():
37	pe = PositionalEncoding(20, 0)
38	y = pe.forward(torch.zeros(1, 100, 20))
39
40	data = pd.concat(
41	[
42	pd.DataFrame(
43	{
44	"embedding": y[0, :, dim],
45	"dimension": dim,
46	"position": list(range(100)),
47	}
48	)
49	for dim in [4, 5, 6, 7]
50	]
51	)
52
53	return (
54	alt.Chart(data)
55	.mark_line()
56	.properties(width=800)
57	.encode(x="position", y="embedding", color="dimension:N")
58	.interactive()
59	)
60
61
62	show(example_positional(), "positional")

토큰 id [[17, 42, 7]] → 임베딩 (1, 3, 512)
nn.Embedding 값의 표준편차   : 0.980
× √512 = × 22.6 한 뒤 : 22.181

PE 버퍼: (1, 5000, 512)  (1, max_len, d_model)
학습되는 파라미터 수: 0  ← register_buffer라 학습하지 않는다

위치 0~3의 앞 6개 차원 (0차원=sin, 1차원=cos, ...)
tensor([[ 0.0000,  1.0000,  0.0000,  1.0000,  0.0000,  1.0000],
        [ 0.8410,  0.5400,  0.8220,  0.5700,  0.8020,  0.5970],
        [ 0.9090, -0.4160,  0.9360, -0.3510,  0.9580, -0.2860],
        [ 0.1410, -0.9900,  0.2450, -0.9700,  0.3430, -0.9390]])

같은 단어라도 위치가 다르면 다른 벡터가 된다
  임베딩만: 두 벡터 차이 0.00
  PE 더한 뒤: 두 벡터 차이 3.71

차트 불러오는 중…


1	1		import copy
2	2		import math
3	3
4	4		import torch
5	5		import torch.nn as nn
6	6
7	7
8	8		def attention(query, key, value, mask=None, dropout=None):
9	9		"Compute 'Scaled Dot Product Attention'"
10	10		d_k = query.size(-1)
11	11		scores = torch.matmul(query, key.transpose(-2, -1)) / math.sqrt(d_k)
12	12		if mask is not None:
13	13		scores = scores.masked_fill(mask == 0, -1e9)
14	14		p_attn = scores.softmax(dim=-1)
15	15		if dropout is not None:
16	16		p_attn = dropout(p_attn)
17	17		return torch.matmul(p_attn, value), p_attn
18	18
19	19
20	20		def subsequent_mask(size):
21	21		"Mask out subsequent positions."
22	22		attn_shape = (1, size, size)
23	23		subsequent_mask = torch.triu(torch.ones(attn_shape), diagonal=1).type(
24	24		torch.uint8
25	25		)
26	26		return subsequent_mask == 0
27	27
28	28
29	29		def clones(module, N):
30	30		"Produce N identical layers."
31	31		return nn.ModuleList([copy.deepcopy(module) for _ in range(N)])
32	32
33	33
34	34		class MultiHeadedAttention(nn.Module):
35	35		def __init__(self, h, d_model, dropout=0.1):
36	36		"Take in model size and number of heads."
37	37		super(MultiHeadedAttention, self).__init__()
38	38		assert d_model % h == 0
39	39		# We assume d_v always equals d_k
40	40		self.d_k = d_model // h
41	41		self.h = h
42	42		self.linears = clones(nn.Linear(d_model, d_model), 4)
43	43		self.attn = None
44	44		self.dropout = nn.Dropout(p=dropout)
45	45
46	46		def forward(self, query, key, value, mask=None):
47	47		"Implements Figure 2"
48	48		if mask is not None:
49	49		# Same mask applied to all h heads.
50	50		mask = mask.unsqueeze(1)
51	51		nbatches = query.size(0)
52	52
53	53		# 1) Do all the linear projections in batch from d_model => h x d_k
54	54		query, key, value = [
55	55		lin(x).view(nbatches, -1, self.h, self.d_k).transpose(1, 2)
56	56		for lin, x in zip(self.linears, (query, key, value))
57	57		]
58	58
59	59		# 2) Apply attention on all the projected vectors in batch.
60	60		x, self.attn = attention(
61	61		query, key, value, mask=mask, dropout=self.dropout
62	62		)
63	63
64	64		# 3) "Concat" using a view and apply a final linear.
65	65		x = (
66	66		x.transpose(1, 2)
67	67		.contiguous()
68	68		.view(nbatches, -1, self.h * self.d_k)
69	69		)
70	70		del query
71	71		del key
72	72		del value
73	73		return self.linears[-1](x)
74	74
75	75
76	76		class PositionwiseFeedForward(nn.Module):
77	77		"Implements FFN equation."
78	78
79	79		def __init__(self, d_model, d_ff, dropout=0.1):
80	80		super(PositionwiseFeedForward, self).__init__()
81	81		self.w_1 = nn.Linear(d_model, d_ff)
82	82		self.w_2 = nn.Linear(d_ff, d_model)
83	83		self.dropout = nn.Dropout(dropout)
84	84
85	85		def forward(self, x):
86	86		return self.w_2(self.dropout(self.w_1(x).relu()))
	87	+
	88	+
	89	+	class Embeddings(nn.Module):
	90	+	def __init__(self, d_model, vocab):
	91	+	super(Embeddings, self).__init__()
	92	+	self.lut = nn.Embedding(vocab, d_model)
	93	+	self.d_model = d_model
	94	+
	95	+	def forward(self, x):
	96	+	return self.lut(x) * math.sqrt(self.d_model)
	97	+
	98	+
	99	+	class PositionalEncoding(nn.Module):
	100	+	"Implement the PE function."
	101	+
	102	+	def __init__(self, d_model, dropout, max_len=5000):
	103	+	super(PositionalEncoding, self).__init__()
	104	+	self.dropout = nn.Dropout(p=dropout)
	105	+
	106	+	# Compute the positional encodings once in log space.
	107	+	pe = torch.zeros(max_len, d_model)
	108	+	position = torch.arange(0, max_len).unsqueeze(1)
	109	+	div_term = torch.exp(
	110	+	torch.arange(0, d_model, 2) * -(math.log(10000.0) / d_model)
	111	+	)
	112	+	pe[:, 0::2] = torch.sin(position * div_term)
	113	+	pe[:, 1::2] = torch.cos(position * div_term)
	114	+	pe = pe.unsqueeze(0)
	115	+	self.register_buffer("pe", pe)
	116	+
	117	+	def forward(self, x):
	118	+	x = x + self.pe[:, : x.size(1)].requires_grad_(False)
	119	+	return self.dropout(x)

1	import copy
2	import math
3
4	import torch
5	import torch.nn as nn
6
7
8	def attention(query, key, value, mask=None, dropout=None):
9	"Compute 'Scaled Dot Product Attention'"
10	d_k = query.size(-1)
11	scores = torch.matmul(query, key.transpose(-2, -1)) / math.sqrt(d_k)
12	if mask is not None:
13	scores = scores.masked_fill(mask == 0, -1e9)
14	p_attn = scores.softmax(dim=-1)
15	if dropout is not None:
16	p_attn = dropout(p_attn)
17	return torch.matmul(p_attn, value), p_attn
18
19
20	def subsequent_mask(size):
21	"Mask out subsequent positions."
22	attn_shape = (1, size, size)
23	subsequent_mask = torch.triu(torch.ones(attn_shape), diagonal=1).type(
24	torch.uint8
25	)
26	return subsequent_mask == 0
27
28
29	def clones(module, N):
30	"Produce N identical layers."
31	return nn.ModuleList([copy.deepcopy(module) for _ in range(N)])
32
33
34	class MultiHeadedAttention(nn.Module):
35	def __init__(self, h, d_model, dropout=0.1):
36	"Take in model size and number of heads."
37	super(MultiHeadedAttention, self).__init__()
38	assert d_model % h == 0
39	# We assume d_v always equals d_k
40	self.d_k = d_model // h
41	self.h = h
42	self.linears = clones(nn.Linear(d_model, d_model), 4)
43	self.attn = None
44	self.dropout = nn.Dropout(p=dropout)
45
46	def forward(self, query, key, value, mask=None):
47	"Implements Figure 2"
48	if mask is not None:
49	# Same mask applied to all h heads.
50	mask = mask.unsqueeze(1)
51	nbatches = query.size(0)
52
53	# 1) Do all the linear projections in batch from d_model => h x d_k
54	query, key, value = [
55	lin(x).view(nbatches, -1, self.h, self.d_k).transpose(1, 2)
56	for lin, x in zip(self.linears, (query, key, value))
57	]
58
59	# 2) Apply attention on all the projected vectors in batch.
60	x, self.attn = attention(
61	query, key, value, mask=mask, dropout=self.dropout
62	)
63
64	# 3) "Concat" using a view and apply a final linear.
65	x = (
66	x.transpose(1, 2)
67	.contiguous()
68	.view(nbatches, -1, self.h * self.d_k)
69	)
70	del query
71	del key
72	del value
73	return self.linears[-1](x)
74
75
76	class PositionwiseFeedForward(nn.Module):
77	"Implements FFN equation."
78
79	def __init__(self, d_model, d_ff, dropout=0.1):
80	super(PositionwiseFeedForward, self).__init__()
81	self.w_1 = nn.Linear(d_model, d_ff)
82	self.w_2 = nn.Linear(d_ff, d_model)
83	self.dropout = nn.Dropout(dropout)
84
85	def forward(self, x):
86	return self.w_2(self.dropout(self.w_1(x).relu()))
87
88
89	class Embeddings(nn.Module):
90	def __init__(self, d_model, vocab):
91	super(Embeddings, self).__init__()
92	self.lut = nn.Embedding(vocab, d_model)
93	self.d_model = d_model
94
95	def forward(self, x):
96	return self.lut(x) * math.sqrt(self.d_model)
97
98
99	class PositionalEncoding(nn.Module):
100	"Implement the PE function."
101
102	def __init__(self, d_model, dropout, max_len=5000):
103	super(PositionalEncoding, self).__init__()
104	self.dropout = nn.Dropout(p=dropout)
105
106	# Compute the positional encodings once in log space.
107	pe = torch.zeros(max_len, d_model)
108	position = torch.arange(0, max_len).unsqueeze(1)
109	div_term = torch.exp(
110	torch.arange(0, d_model, 2) * -(math.log(10000.0) / d_model)
111	)
112	pe[:, 0::2] = torch.sin(position * div_term)
113	pe[:, 1::2] = torch.cos(position * div_term)
114	pe = pe.unsqueeze(0)
115	self.register_buffer("pe", pe)
116
117	def forward(self, x):
118	x = x + self.pe[:, : x.size(1)].requires_grad_(False)
119	return self.dropout(x)