[밑바닥부터 시작하는 딥러닝 2]- 4. word2vec 속도 개선 I
word2vec의 속도 개선으로 새로운 계층을 도입하는 Embedding, 새로운 손실함수를 도입하는 negative sampling 이 두가지가 있다.
Embedding
word2vec 구현 시 단어를 one-hot vector 로바꿔 가중치 행렬을 곱한다. 만약 어휘수가 100만개일 경우, one-hot vector도 100만 차원이 된다. 도출되는 건 아래 사진 처럼 특정 행을 추출하는 것이다. 그러므로 원핫 표현 변환과 복잡한 행렬곱은 필요가 없다.
embedding 계층은 단어 ID에 해당하는 행을 추출하여 사용한다.
embedding 계층 구현
행 추출
행렬에서 W[2] W[5] 처럼 특정 행을 추출한다.
import numpy as np
W = np.arange(21).reshape(7, 3)
print(W)
#[[ 0 1 2]
# [ 3 4 5]
# [ 6 7 8]
# [ 9 10 11]
# [12 13 14]
# [15 16 17]
# [18 19 20]]
print(W[2])
#[6 7 8]
print(W[5])
#[15 16 17]원하는 행 번호들을 배열에 명시해서 추출할 수 있다. 이는 미니배치 처리를 가정하였을 때의 구현이다.
idx = np.array([1,0,3,0])
print(W[idx])
#[[ 3 4 5]
# [ 0 1 2]
# [ 9 10 11]
# [ 0 1 2]]
forward() 구현
가중치 특정 행 뉴런을 다음층으로 흘러보
class Embedding:
def __init__(self, W):
self.params = [W]
self.grads = [np.zeros_like(W)]
self.idx = None
def forward(self, idx):
W, = self.params
self.idx = idx
out = W[idx] #행의 인덱스를 배열로 저장
return out
backward() 구현
앞 층으로부터 전해진 기울기를 다음층으로 흘러보냄
def backward(self, dout):
dW, = self.grads #기울기 가져옴
dW[...] = 0 #dW 형상을 유지한 채, 그 원소를 0으로 덮어씀
dW[self.idx] = dout #앞층에서 전해진 기울기 dout를 idx번째 행에 할당
return Noneidx 가 중복될 수 있는 문제가 발생할 수 있다. 이러한 중복 문제를 해결 하기 위해 더하기를 해야한다.
dh의 각 행의 값을 dW의 해당 행에 더해준다.
def backward(self, dout):
dW, = self.grads
dW[...] = 0
for i, word_id in enumerate(self.idx):
dW[word_id] += dout[i] #해당 인덱스 기울기 더해줌
#혹은
#np.add.at(dW,self.idx,dout)
return NoneNegative Sampling
네거티브 샘플링을 이용하면 어휘가 아무리 많아져도 계산량을 낮은수준에서 일정하게 억제 할 수 있다.
Embedding 계층 도입을 통해 입력층 계산에서 낭비를 줄였다. 은닉층 이후로 은닉층 뉴런 x Wout(가중치 행렬), Softmax 계층에서 계산이 매우 오래걸린다. 그러므로 행렬곱을 가볍게 만들어야 한다.
어휘수를 100만개로 하면 exp 계산을 100만번 해야한다. 이러한 softmax 계산식을 대신할 방법이 필요하다
다중 분류에서 이진 분류
네거티브 샘플링을 이해하기 위해서는 다중분류 문제를 이진 분류 방식(Yes/No) 으로 해결하는 것이다.
CBOW 모델에서는 다음과 같은 기전이 일어난다.
출력층의 뉴런은 하나이다. 은닉층과 출력 측의 가중치 행렬의 내적은 say에 해당하는 열벡만을 추출, 추출된 열 벡터와 은닉층 뉴런과의 내적을 계산한다.
say에 해당하는 열 벡터가 (100*1) 이므로 앞의 은닉층 뉴런은 (1*100)(100*1) = (1*1)
추출되는 최종점수는 (1*1) 형상이다
시그모이드 함수와 교차 엔트로피 오차
시그모이드
입력값(x) 은 0에서 1사이의 실수로 변환.
시그모이드 함수에 사용되는 손실함수는 (이중분류 다중분류 둘다) cross entropy error ( 교차 엔트로피 오차) 임.
로그식은
L = -(tlogy + (1-t)log(1-y)) (t: 정답 레이블, 정답 레이블의 값은 0 or 1)
t=1이면 L= -logy, t=0이면, L= -log(1-y)
y는 신경망이 출력할 확률, t는 정답 레이블 , y가 1(100%)에 가까울수록 오차가 줄어든다.
오차가 크면 크게 학습하고, 오차가 작으면 작게 학습한다.
다중 분류에서 이진 분류로 (구현)
다중분류 수행 CBOW 모델
맥락이 you, goodbye, 정답 타겟 -> say 인 경우 이다. 입력층에서 단어 ID의 분산표현을 추출 하기 위해 embedding 계층을 사용함.
이진분류 수행 CBOW 모델
은닉층 뉴런 h와, 출력 측의 가중치 Wout에서 단어 say에 해당하는 단어 벡터와 내적을 계산함. 그리고 출력을 sigmoid with loss 계층에 입력해 최종 손실을 얻음. sigmoid with loss계층에 정답 레이블로 1을 입력한것은 정답이 yes를 의미함.
embedding dot계층은 embedding 계층과 dot 연산(내적)의 처리를 합친 계층이다. 은닉층 뉴런 h는 embedding dot 계층과 sigmoid with loss 계층을 통과한다.
embedding dot 계층 구현(순전파)
class EmbeddingDot:
def __init__(self, W):
self.embed = Embedding(W) #embedding 계층
self.params = self.embed.params #매개변수 저장
self.grads = self.embed.grads#기울기 저장
self.cache = None #cache 는 순전파 시의 계산 결과를 잠시 유지하기 위한 변수
def forward(self, h, idx):#은닉층 뉴런h, 단어id의 넘파이배열 idx (미니배치처리가정)
target_W = self.embed.forward(idx)
out = np.sum(target_W * h, axis=1)#내적 계산
self.cache = (h, target_W)
return out
def backward(self, dout):
h, target_W = self.cache
dout = dout.reshape(dout.shape[0], 1)
dtarget_W = dout * h
self.embed.backward(dtarget_W)
dh = dout * target_W
return dh
forward() 메서드 기전은 다음과 같다. idx[0,3,1]은 3개의 데이터를 미니배치로 한번에 처리하는 예임을 의미한다.
target_W는 W의 0번, 3번, 1번째 행을 추출한 결과이다. target_W*h는 원소별 곱을 수행하였고, 행별(axis=1) 합을 도출한다.
네거티브 샘플링
긍정적 예: 정답을 say라고 가정하면, say를 입력했을 때의 sigmoid 계층의 출력은 1에 가깝고, say 이외의 단어를 입력했을 때의 출력은 0에 가까워야 한다. 이렇게 만들어주는 가중치가 매우 중요하다.
근사적인 해법으로 부정적인 예 (5개든지 10개든지)를 선택한다. 적은 수의 부정적인 예를 샘플링하여 사용한다.
네거티브 샘플링 기법은 (긍정적인 예 타겟 + 부정적인 예 몇개 샘플링)의 각각 손실을 구한다.
각각 손실을 모두 더해서 최종 손실로 함.
긍정적 target은 say, 부정적 샘플이 hello와 I일때, 긍정적 타겟 say에 대해서는 sigmoid with loss 계층에 정답 레이블로 1을 입력함. 부정적 샘플에 대해서는 sigmoid with loss 계층에 0을 입력한다. 각 손실을 모두 더해 최종손실을 출력.
네거티브 샘플링의 샘플링 기법
부정적 예를 샘플링 하는 기법으로는 통계 데이터를 기초로 샘플링 한다.
말뭉치에서 자주 등장하는 단어를 많이 추출하고, 드물게 등장하는 단어를 적게 추출한다. 단어의 출현 횟수를 구해 확률 분포로 나타내고 확률분포대로 단어를 샘플링한다.
import numpy as np
# 0에서 9까지 숫자 중 하나를 무작위 샘플링
print(np.random.choice(10)) #3
# words에서 하나만 무작위로 샘플링
words = ['you', 'say', 'goodbye', 'I', 'hello', '.']
print(np.random.choice(words))#.
# 5개만 무작위로 샘플링 (중복 있음)
print(np.random.choice(words, size=5)) #['you' 'say' 'goodbye' 'goodbye' 'hello']
# 5개만 무작위로 샘플링 (중복 없음)
print(np.random.choice(words, size=5, replace=False)) #['you' '.' 'goodbye' 'say' 'hello']
# 확률분포에 따라 샘플링
p = [0.5, 0.1, 0.05, 0.2, 0.05, 0.1]
print(np.random.choice(words, p=p)) #I네거티브 샘플링에서는 확률이 낮은 단어의 확률을 살짝 높이기 위해 확률분포를 수정하는 것이 좋다
P(wi)는 i번째 단어의 확률을 뜻한다. 각 요소를 0.75제곱한 것이고, 확률의 총합은 1이 되어야 하므로 수정 후 확률분포의 총합이 필요하다.
p = [0.7,0.29,0.01]
new_p = np.power(p,0.75)
new_p /= np.sum(new_p)
print(new_p) #[0.64196878 0.33150408 0.02652714]확률이 0.01에서 0.0265로 근소한 상승이 있었다. 0.75라는 수치는 이론적인 의미는 없어서 다른 값으로 설정해도 괜찮다.
UnigramSampler는 다음과 같이 구현된다
import numpy as np
GPU = True
import collections
class UnigramSampler:
def __init__(self, corpus, power, sample_size):
self.sample_size = sample_size
self.vocab_size = None
self.word_p = None
counts = collections.Counter()
for word_id in corpus:
counts[word_id] += 1
vocab_size = len(counts)
self.vocab_size = vocab_size
self.word_p = np.zeros(vocab_size)
for i in range(vocab_size):
self.word_p[i] = counts[i]
self.word_p = np.power(self.word_p, power)
self.word_p /= np.sum(self.word_p)
def get_negative_sample(self, target):
batch_size = target.shape[0]
if not GPU:
negative_sample = np.zeros((batch_size, self.sample_size), dtype=np.int32)
for i in range(batch_size):
p = self.word_p.copy()
target_idx = target[i]
p[target_idx] = 0
p /= p.sum()
negative_sample[i, :] = np.random.choice(self.vocab_size, size=self.sample_size, replace=False, p=p)
else:
negative_sample = np.random.choice(self.vocab_size, size=(batch_size, self.sample_size),
replace=True, p=self.word_p)
return negative_sample위의 코드는 ssearch_while.py에 저장함
from ssearch_while import UnigramSampler
import numpy as np
corpus = np.array([0,1,2,3,4,1,2,3])
power = 0.75
sample_size=2 #부정적인 예 2개 사용
sampler = UnigramSampler(corpus,power,sample_size)
target = np.array([1,3,0]) #3개 데이터를 미니배치로 사용
negative_sample = sampler.get_negative_sample(target)
print(negative_sample)
#[[2 1]
# [2 0]
# [3 1]]네거티브 샘플링 구현
class NegativeSamplingLoss:
def __init__(self, W, corpus, power=0.75, sample_size=5):
#출력층 가중치 W, 말뭉치(단어ID), 확률분포에 제곱할 값, 부정적 예 샘플링 횟수
self.sample_size = sample_size
self.sampler = UnigramSampler(corpus, power, sample_size)
self.loss_layers = [SigmoidWithLoss() for _ in range(sample_size + 1)]
self.embed_dot_layers = [EmbeddingDot(W) for _ in range(sample_size + 1)]
self.params, self.grads = [], []
for layer in self.embed_dot_layers:
self.params += layer.params
self.grads += layer.gradsloss_layers와 embed_dot_layers는 원하는 계층을 리스트로 보관한다. sample_size+1을 하는 이유는 부정적 예를 다루는 계층이 sample_size 만큼이고, 긍정적 예를 다루는 계층이 하나 더 필요하기 때문이다. loss_layers[0]와 embed_dot_layers[0]이 긍정적 예를 다루는 계층이다. 각 계층에서 사용하는 매개변수와 기울기를 배열로 저장한다.
def forward(self, h, target): #순전파 구현
batch_size = target.shape[0]
negative_sample = self.sampler.get_negative_sample(target) #부정적 예 샘플링
# 긍정적 예 순전파
score = self.embed_dot_layers[0].forward(h, target)
correct_label = np.ones(batch_size, dtype=np.int32)
loss = self.loss_layers[0].forward(score, correct_label)
# 부정적 예 순전파
negative_label = np.zeros(batch_size, dtype=np.int32)
for i in range(self.sample_size):
negative_target = negative_sample[:, i]
score = self.embed_dot_layers[1 + i].forward(h, negative_target)
loss += self.loss_layers[1 + i].forward(score, negative_label) #긍정 부정 샘플 손실 더하기
return loss
def backward(self, dout=1):#역전파 구현
dh = 0
for l0, l1 in zip(self.loss_layers, self.embed_dot_layers):
dscore = l0.backward(dout)
dh += l1.backward(dscore)
return dh