(2)-4-1 word2vec 속도 개선

앞서 구현한 간단한 CBOW 모델은 말뭉치에 포함된 어휘 수가 많아지면 계산량도 커져서 시간이 너무 오래 걸립니다.

이를 위해 두 가지 개선을 합니다.

1. Embedding이라는 새로운 계층을 도입합니다.

2. 네거티브 샘플링이라는 새로운 손실 함수를 도입합니다.

 

이 두 가지 개선으로 진정한 word2vec을 구현할 수 있습니다.

 

 

(1) word2vec 개선 1

앞에서 사용한 간단한 CBOW 모델을 이용하여 말뭉치에 어휘가 100만 개라고 가정한다면 입력층과 출력층에 각 100만 개의 뉴런들이 존재합니다.

이러한 많은 뉴런 때문에 중간 계산에 많은 시간이 소요됩니다.

구체적으로 다음 두 부분에서 병목현상이 발생합니다.

 

1. 입력층의 원핫 표현과 가중치 행렬 W(in)의 곱

2. 은닉층과 가중치 행렬 W(out)의 곱 및 Softmax 계층의 계산

 

이를 해결하기 위한 방법은 다음과 같습니다.

1번 문제를 해결하기 위해서 Embedding이라는 새로운 계층을 도입합니다.

2번 문제를 해결하기 위해서 네거티브 샘플링이라는 새로운 손실 함수를 도입해 해결합니다.

 

 

① Embedding 계층

앞장의 word2vec 구현을 100만 개의 어휘를 사용한다고 하면 다음과 같이 그려집니다.

단순히 보면 100만 크기의 행렬들을 곱해야 하므로 연산이 많이 걸리지만,

핵심만 바라본다면 결국 가중치 부분에서 특정 행을 추출하는 것뿐입니다.

따라서 굳이 행렬곱을 계산할 필요가 없습니다.
그러므로 이를 간단히 하게 하는 '단어 ID에 해당하는 가중치 행렬의 행'을 추출하는 Embedding 계층을 만들면 효율적으로 처리할 수 있습니다.

Embedding 계층에는 결과적으로 단어의 의미(분산 표현)가 저장됩니다.

 

(자연어 처리 분야에서 단어의 밀집 벡터 표현을 단어 임베딩 혹은 분산 표현이라고 합니다.

 통계 기반 기법으로 얻은 단어 벡터는 영어로 distributional representation이라고 하고,

 추론 기반 기법으로 얻은 단어 벡터는 영어로 distributed representation이라고 합니다.)

 

이를 코드로 표현하면 다음과 같습니다.

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

    def backward(self, dout):
        dW, = self.grads
        dW[...] = 0 # dW자체를 0으로 설정하는 것이 아니라, 
                    # dW의 형상을 유지한 채, 그 원소들을 0으로 덮어쓰는 것임
        np.add.at(dW, self.idx, dout)
        return None

변수에 반점을 붙인 것은 다음 예제를 통해 알 수 있습니다.

 

역전파 부분을 생각해보겠습니다.

Embedding의 순전파 부분은 단순하게 W의 행을 추출하는 것뿐입니다.

단순히 가중치 특정 행 뉴런만을 다음 층으로 흘러 보내는 것이므로

역전파에서도 앞 층으로부터 전해진 기울기를 다음 층으로 그대로 흘려주면 됩니다.

다만, 앞 층으로부터 전해진 기울기를 가중치 기울기 dW의 특정 행(idx번째 행)에 설정합니다.

하지만 역전파에서 그냥 흘리면 중복 문제가 발생합니다.

(corpus에는 한 단어가 중복되서 발생 가능하며 현 상황은 미니배치상황을 가정했을 때 인덱스가 여러 개 나오게 됨)

idx가 [0,2,0,4]인 경우를 생각해보면 다음과 같은 문제가 발생합니다.

0 인덱스가 두 개이므로 dW의 0자리에 중복되게 됩니다.

이를 해결하려면 '할당'이 아니라 '더하기'를 해야 합니다.

 

더하기를 하는 이유를 알기 위해서는 근본적인 행렬곱을 취하는 식 기준으로 역전파를 시행하면 알 수 있습니다.

주황색 : 역전파 과정

Embedding 코드 상에서는 가중치만 뽑아서 사용하지만 역전파를 생각할 때 더하는 이유를 알려면 위의 그림과 같이 본질적인 과정에서 더하는 이유를 알 수 있습니다.

 

이제 중복된 부분은 더해야 한다는 것을 알았으며,

중복된 부분은 더하게 하기 위해서 

np.add.at(dw, self.idx, dout)

코드를 사용하게 됩니다.

 

np.add.at(A, idx, B) : B를 A의 idx번째 행에 더해 줌

 

 

이로써 word2vec(CBOW 모델)의 구현은 입력 측 MatMul 계층을 Embedding 계층으로 전환함으로써 메모리 사용량을 줄이고 시간도 아낄 수 있습니다.

 

 

 

② word2vec 개선 2

이제 두 번째 문제 

은닉층과 가중치 행렬 W(out)의 곱 및 Softmax 계층의 계산

을 해결하기 위해서 네거티브 샘플링이라는 기법을 사용하면 됩니다.

 

우선 문제점을 다시 한번 짚으면,

행렬곱은 어휘가 많아지면 복잡하고 시간이 오래 걸린다는 것을 알 수 있고,

softmax의 경우 softmax 식을 보았을 때 100만 개에 대해 모두 고려하여 확률 형태로 내기 때문에 이 또한 시간이 오래 걸린다는 것을 알 수 있습니다.

 

③ 네거티브 샘플링 기법

 

이 기법의 핵심은,

다중 분류(Multi-class classification)를 이중분류(binary classification)로 근사하는 것입니다.

 

이전에 질문과 답은 ' 'you'와 'goodbye'를 주면 타깃 단어가 무엇인가'를 찾았다면,

이진 분류로 근사하기 위해서는 ' 'you'와 'goodbye'를 줬을 때 타깃 단어는 'say'인가'를 질문하면 됩니다.

그러면 출력층에는 'say'일 확률에 대한 점수만 출력하면 되므로 뉴런을 하나만 준비하면 됩니다.

 

따라서 그림처럼 은닉층과 출력 층의 가중치 행렬의 내적은 'say'에 해당하는 열만 추출하여 앞의 은닉층 결과와 내적을 계산하면 됩니다.

'say'에 해당하는 열만 추출하는 과정 역시 embedding 계층이 사용되어서 추출한 후 내적이 수행될 것입니다. 

 

 

이진 분류 문제를 신경망으로 풀려면 점수에 시그모이드 함수를 적용해 확률로 변환하고, 손실 함수를 구할 때 Cross entropy를 사용합니다.

(보통 다중 분류-softmax&cross-entropy, 이중분류-sigmoid&cross-entropy)

 

시그모이드 식은 다음과 같습니다.

또한 이진 분류의 Cross-entropy는 다음과 같이 나타낼 수 있습니다.

이 식은 t가 1이면  -logy가, t가 0이면 -log(1-y)가 나옴으로써 손실 함수 역할을 해냅니다.

 

sigmoid 계층과 Cross-Entropy Error 계층 그래프는 다음과 같습니다.

이를 통해 Sigmoid with Loss의 역전파는 y-t 임을 알 수 있습니다.

y는 모델이 예측한 확률, t는 정답 즉 0과 1을 의미합니다.

증명은 다음과 같습니다.

 

Sigmoid 역전파 자체의 증명은 아래에서 확인할 수 있습니다.

luminitworld.tistory.com/13?category=924576

5. 오차역전파법

 

이진 분류과정을 전체 그림을 나타내면 다음과 같습니다.

이러한 부분에서 Embed과정과 dot 부분을 합쳐서 생각할 수 있습니다.

Embed 과정과 dot 과정을 합친 것에 대한 코드는 다음과 같습니다.

class EmbeddingDot:
    def __init__(self, W):
        self.embed = Embedding(W)
        self.params = self.embed.params
        self.grads = self.embed.grads
        self.cache = None

    def forward(self, h, idx):
        target_W = self.embed.forward(idx)
        out = np.sum(target_W * h, axis=1) # *을 통해 각 원소끼리 곱하게 한 후 sum으로 더해서 
                                           # 내적을 실현시킴
        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 부분의 구체적 값 진행 예시는 다음과 같습니다.

  

 

 

지금까지 설명한 모델은 맥락이 주어졌을 때 특정 단어가 맞는지 틀린지 중에서 맞는 경우만 고려했습니다.

예시로 말하자면, 'You'와 'goodbye'를 주었을 때 타깃에 'say'가 맞을 때(긍정적) 1이라고 했습니다.

위의 모델은 'say'가 맞을 경우에만을 이진 분류를 했습니다.

왜냐하면 'say'가 맞을 확률이 y이었으며 그 y와 정답 t를 줄이는 방향으로 모델을 학습시키도록 유도했기 때문입니다.

 

그러나 부정적인 아니다('say' 이외의 단어)에 대해서는 어떤 정보도 얻지 못했습니다.

우리가 하고자 하는 것은 'say'에 대해서는 sigmoid 계층의 출력이 1에 가깝게 만들고,

'say' 이외의 단어에 대해서 sigmoid 계층의 출력이 0에 가깝게 만드는 것입니다.

 

이를 실현시키기 위해서, 모든 부정적 예를 대상으로 하여 이진 분류를 학습시킬 수 있지만,

이는 어휘 수가 늘어나면 감당할 수 없으므로,

부정적 예를 몇 개 선택합니다.

즉, 적은 수의 부정적 예를 샘플링해 사용합니다.

이것이 바로 '네거티브 샘플링' 기법입니다.

 

네거티브 샘플링 기법은 긍정적 예를 타깃으로 한 경우의 손실을 구하되,

그와 동시에 부정적 예를 몇 개 샘플링하여, 그 부정적 예에 대해서도 마찬가지로 손실을 구합니다.

그리고 각각의(긍정적, 부정적) 데이터의 손실을 더한 값으로 최종 손실로 합니다.

 

예시에서 'hello'와 'I' 두 개를 부정적 데이터로 샘플링한다고 했을 때 다음과 같이 그릴 수 있습니다.

 

 

 

④ 네거티브 샘플링의 샘플링 기법

앞서 부정적인 예를 모두 사용하면 비효율적이어서 일부만을 샘플링한다고 했는데,

부정적인 예를 어떻게 샘플링하는지 알아보겠습니다.

단순히 무작위로 샘플링(선별)하는 것보다,

말뭉치의 통계 데이터를 기초로 샘플링하는 것입니다.

한 말뭉치 안에서 자주 등장하는 단어를 많이 추출하고 드물게 등장하는 단어를 적게 추출하는 것입니다.

결국 네거티브 샘플링은 정답에 해당되지 않는 단어들을 대표해서 몇 개 선별하는 것인데 희소 단어들보다는 자주 나온 단어들이 부정적 예를 대표해서 학습시키는 것이 좋을 것입니다.

 

그러므로 말뭉치에서 단어별 출현 횟수를 바탕으로 확률분포를 구한 후, 그 확률분포에 따라서 샘플링을 수행하면 됩니다.

 

샘플링을 하는데 사용되는 핵심 함수는 np.random.choice() 함수입니다.

해당 함수의 기능은 다음 예시를 통해 알 수 있습니다.

 

이런 식으로 np.random.choice()를 이용하여 확률분포를 기반으로 랜덤 하게 샘플링할 수 있습니다.

 

그런데 여기서 확률분포에서 한 가지를 더 수정해줍니다.

바로 기존 기본 확률분포에 0.75를 제곱하는 것입니다.

 

p(wi)는 i번째 확률을 뜻합니다.

이렇게 제곱하는 이유는 원래 확률분포에서 낮은 확률을 살짝 높일 수 있기 때문입니다.

이로써 '희소 단어'를 버리지 않고 어느 정도 신경 써주겠다는 것입니다.

다음을 통해서 낮은 확률이 살짝 올라가는 것을 확인할 수 있습니다.

0.75라는 수치는 이론적인 의미는 없으니 다른 값으로 설정해도 됩니다.

 

이제 위의 내용들을 정리해서 만든 클래스의 코드 UnigramSampler class 코드는 다음과 같습니다.

import collection
GPU=False
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)  # 전체 합이 1이 되도록 확률 설정

    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() # 정답 부분의 확률을 0으로 만들어 버렸으니 남은 확률들의 합이 1이되도록 하기 위해서
                negative_sample[i, :] = np.random.choice(self.vocab_size, size=self.sample_size, replace=False, p=p)
        else:
            # GPU(cupy）로 계산할 때는 속도를 우선한다.
            # 부정적 예에 타깃이 포함될 수 있다.
            negative_sample = np.random.choice(self.vocab_size, size=(batch_size, self.sample_size),
                                               replace=True, p=self.word_p)

        return negative_sample

매개 변수로 corpus, 제곱 수, 샘플링할 수 이렇게 3가지를 받아서

corpus를 기반으로 말뭉치 안의 단어들의 발생 확률을 구해 확률분포를 구하고

이를 제곱 수만큼 제곱을 시켜서 확률로 다시 표현 후, get_negative_sample로 정답 단어의 확률은 0으로 설정하고 np.random.choice()로 샘플링할 수만큼 샘플링을 해서 샘플링된 값을 결과로 넘깁니다.

 

이 클래스는 다음과 같이 사용될 수 있습니다.

이를 통해 부정적인 샘플링을 할 수 있게 되었습니다.

 

 

이제 최종적인 네거티브 샘플링 코드는 다음과 같습니다.

클래스명은 NegativeSamplingLoss 이며, 부정적 샘플링을 한 후 Embedding Dot을 거쳐 SigmoidwithLoss까지 거치도록 설정한 코드입니다.

class NegativeSamplingLoss:
    def __init__(self, W, corpus, power=0.75, sample_size=5):
        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.grads

    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

생성자 매개변수로 W는 출력 측 가중치를 나타냅니다.

생정자에서는 UnigramSampler 클래스를 생성하여 인스턴스 변수인 sampler로 저장합니다.

부정적 예의 샘플링 횟수는 sample_size에 저장합니다.

 

loss_layers와 embed_dot_layers의 계층을 리스트로 보관하는데 

계층 수를 sample_size+1로 한 이유는 

부정적인 예를 다루는 계층 sample_size개가 있고, 여기에 추가로

긍정적인 예(정답)를 다루는 계층 1개가 필요하기 때문입니다.

정확히는 0번째 계층, 즉 loss_layers[0]과 embed_dot_layers[0]이 긍정적 예를 다루는 계층입니다.

 

순전파 부분은

우선 긍정적 예에 대한 계층들을 먼저 설계하고 

이후 부정적 예들에 대한 계층들을 설계한 후,

긍정적 예와 부정적 예 각각의 데이터에 대해서 순전파를 수행해 그 손실들을 더합니다.

여기에서 긍정적 예의 정답 레이블(correct_label)은 "1"이고,

부정적 예의 정답 레이블(negative_label)은 "0" 입니다.

 

역전파 부분은

각 계층의 backward로 구현하면 됩니다.

 

 

 

(위의 내용은 밑바닥부터 시작하는 딥러닝 2 (한빛미디어)를 바탕으로 글을 작성하였습니다)