在线问答5

这个屌丝很懒，什么也没留下！

热门标签

word2vec改进之hierarchical softmax

作者：在线问答5 | 2024-07-16 13:06:43

踩

word2vec改进

概要

本篇博客主要是详细讲述使用了hierarchical softmax（层级softmax）改进的word2vec的具体流程和代码实现，关于word2vec的基础实现可以看我上一篇博客：

【学习笔记】手写神经网络之word2vec_一无是处le的博客-CSDN博客

整体流程

整体流程基本跟基础的word2vec相似，只是在使用softmax计算损失的部分进行改进，看到这可能你们会想为什么要进行改进呢？原因很简单，因为softmax计算量非常大，一旦embedding的维度稍微大一点，窗口大一点，训练的时间就会大大拉长，这是因为softmax需要进行大量的指数运算，速度十分慢，而这里采用的hierarchical softmax其本质就是构建霍夫曼树代替使用softmax多分类计算损失。如下图：

我们需要做的就是将softmax这块使用霍夫曼树来代替，霍夫曼树如下图：

使用霍夫曼树代替softmax的主要问题在于如何使用霍夫曼树完美的插入到word2vec当中来代替使用多分类求交叉熵的方式来计算损失。这一步我认为是最重要的，其次才是如何构建霍夫曼树。（我看了很多文章，它们大多都没有通俗易懂的讲解如何将霍夫曼树融入word2vec而是花大量篇幅去讲解如何构建霍夫曼树以及hierarchical softmax在word2vec中的计算公式，梯度公式等等，我认为本末倒置了，而我也是花了比较长的时间来理解）其关键就是使用sigmoid函数对每个非叶子节点以及词向量取值判断是霍夫曼树是向左走还是向右走，用这个值与每个真实路径计算损失。下图我使用手绘方式大致解析一下其整体流程（这里以CBOW为例）：

         和之前的神经网络语言模型相比，我们的霍夫曼树的所有内部节点就类似之前神经网络隐藏层的神经元,其中，根节点的词向量对应我们的投影后的词向量，而所有叶子节点就类似于之前神经网络softmax输出层的神经元，叶子节点的个数就是词汇表的大小。在霍夫曼树中，隐藏层到输出层的softmax映射不是一下子完成的，而是沿着霍夫曼树一步步完成的，因此这种softmax取名为"Hierarchical Softmax"。

        如何“沿着霍夫曼树一步步完成”呢？在word2vec中，我们采用了二元逻辑回归的方法，即规定沿着左子树走，那么就是负类(霍夫曼树编码1)，沿着右子树走，那么就是正类(霍夫曼树编码0)。判别正类和负类的方法是使用sigmoid函数，即：

                                 $P(+)=\sigma (x_{w}\theta^{T})=\frac{1}{1+e^{x_{w}\theta^{T}}}$

其中 $x_{w}$ 是当前内部节点的词向量，而 θ 则是我们需要从训练样本求出的逻辑回归的模型参数。

　　使用霍夫曼树有什么好处呢？首先，由于是二叉树，之V,现在变成了 $log_{2}V$ 。第二，由于使用霍夫曼树是高频的词靠近树根，这样高频词需要更少的时间会被找到，这符合我们的贪心优化思想。

　　容易理解，被划分为左子树而成为负类的概率为P(−)=1−P(+)。在某一个内部节点，要判断是沿左子树还是右子树走的标准就是看P(−),P(+)谁的概率值大。而控制P(−),P(+)谁的概率值大的因素一个是当前节点的词向量，另一个是当前节点的模型参数 θ。

　　对于上图中的 jumps，如果它是一个训练样本的输出，那么我们期望对于里面的隐藏节点的n(w,1)P(+)概率大，n(w,3), n(w, 5)的P(+)概率大，即最终的目标路径为:011

　　回到基于Hierarchical Softmax的word2vec本身，我们的目标就是找到合适的所有节点的词向量和所有内部节点θ, 使训练样本达到最大似然。

具体流程：

1.构建霍夫曼树

构建霍夫曼树其本质就是使用二叉树求解最优路径，我们这里直接以词频为标准构建二叉树，词出现的频率越高其越靠近根节点，这样需要整个词汇表需要遍历的节点就会最少，代码及解析如下：


import heapq
# heapq 库是Python标准库之一，提供了构建小顶堆的方法和一些对小顶堆的基本操作方法(如入堆，出堆等)，可以用于实现堆排序算法。
from collections import defaultdict
# defaultdict主要是collections中的一个dict的子类
 
 
class HuffmanNode:
# 这个类用于构建霍夫曼树节点
    def __init__(self, word, freq):
        self.word = word
        self.freq = freq
        self.left = None
        self.right = None
        self.parent = None
 
    def __lt__(self, other):
    # 这个魔法方法其实就是当两个HuffmanNode实例对象比较时会自动触发
        return self.freq < other.freq
 
 
# 这个类用于构建霍夫曼树
def build_huffman_tree(vocabulary):
    # defaultdict本质就是构建一个dict然后默认赋值为0，之后可以直接对空dict赋值，会自动生成key
    word_freq = defaultdict(int)
    # 计算每个词的词频
    for word in vocabulary:
        word_freq[word] += 1
 
    # 将词频字典中所有值取出来构建Huffman node放入min_heap，min_heap的意思就是小根堆，
    # 但是这里的列表没有排序
    min_heap = [HuffmanNode(word, freq) for word, freq in word_freq.items()]
    # 这里调用heapq.heapify才进行排序构建堆，默认是构建小根堆
    # 因为这里的heapq默认直接调用数组中的值比较大小，而min_heap中都是Huffman Node，
    # 直接比较会报错，因此才需要编写__lt__这个魔法方法，否则会报错
    heapq.heapify(min_heap)
 
    # 下面这个循环语句就是真正开始构建霍夫曼树
    while len(min_heap) > 1:
        # heappop是弹出最小值，即小根堆的根节点，每次调用heapq时都会重新根据当前值构建小根堆
        left = heapq.heappop(min_heap)
        right = heapq.heappop(min_heap)
        # 这个new_node就是构建的节点，这个节点在霍夫曼树中是从下往上构建（因为这里用的是小根堆）
        # 而在霍夫曼树中节点没有word属性，词频属性为其左右子树之和
        new_node = HuffmanNode(None, left.freq + right.freq)
        new_node.left = left
        new_node.right = right
        # 这里的None是一个临时值，霍夫曼树中的节点都有其对应的地址（之后会被new_node的地址覆盖）
        new_node.parent = None
        left.parent = new_node  # 给左子节点赋值其父母节点
        right.parent = new_node  # 给右子节点赋值其父母节点
        # 这里的heappushu就是在min_heap中加入new_node，再重新构建小根堆
        heapq.heappush(min_heap, new_node)
    
    # 这个就是min_heap剩下最后的一个最大的节点，即霍夫曼树的根节点
    huffman_tree = min_heap[0]
 
    return huffman_tree
 
 
# 这个方法是构建每个词在霍夫曼树中的路径
def calculate_probabilities(huffman_tree):
    parents = []
 
    def traverse(node, path, probs):
        # 存储叶子节点的路径
        if node.word is not None:
            probs[node.word] = path
        # 将所有非叶子节点取出来，之后构建θ矩阵需要
        if node.parent is not None:
            parents.append(node.parent)
        if node.left is not None:
            traverse(node.left, path + "1", probs)
        if node.right is not None:
            traverse(node.right, path + "0", probs)
 
    probs = {}
    traverse(huffman_tree, "", probs)
    return probs, parents
 
 
# 展示构建的霍夫曼树，None表示非叶子节点
def display_huffman_tree(node, prefix='', is_left=True):
    if node is not None:
        print(f"{'L' if is_left else 'R'}{prefix}{node.word}:{node.freq}")
        display_huffman_tree(node.left, prefix + '|-- ', True)
        display_huffman_tree(node.right, prefix + '|-- ', False)

2.将CBOW和霍夫曼树结合起来

这部分就是上面说的关键部分，通过计算经过霍夫曼树某个节点的最大似然估计值与真实路径直接的熵差来计算损失，并计算梯度进行反向传播，代码如下：


class CBOWWithHuffman:
    # 实例化各个值       词典大小     词映射的维度  霍夫曼树根节点     学习率
    def __init__(self, vocab_size, embedding_dim, huffman_tree, learning_rate=0.01):
        self.vocab_size = vocab_size
        self.embedding_dim = embedding_dim
        self.huffman_tree = huffman_tree
        self.learning_rate = learning_rate
    
    # 这里的各个传入值分别表示‘窗口中所有词的加权求和平均值’，‘窗口中除中心词外的所有词’
    # ‘霍夫曼树非叶子节点对应的theta值字典’，‘目标路径（霍夫曼树走的方向）’，‘当前词要走的深度’
    def train(self, x_w, embedding_bag, parents_theta, direction, l_w):
        # 因为要计算平均损失，因此每次训练损失归零
        self.loss = 0
        # count用来监测l_w
        self.count = 0
        # e用来更新窗口中各个词的embediing值，每次训练都需要归零
        self.e = 0
 
        # 下面这段递归计算是计算公式的具象化，具体计算公式下面会讲到
        def travel(node, d, x_w):
            d = int(d)
            if node is None:
                return
            if node.parent is not None:
                theta = parents_theta[node.parent]
                f = self.softmax(np.dot(x_w, theta.T))
                # l_f = np.dot(np.power(f, 1 - d), np.power(1 - f, d))
                g = (1 - d - f) * self.learning_rate
                self.e += np.dot(theta, g)
                parents_theta[node.parent] += np.dot(x_w, g)
                l = -(1-d)*np.log(f) - d*np.log(1-f)
                self.loss += l
 
            if d == 0:
                self.count += 1
                if self.count == l_w:
                    pass
                else:
                    travel(node.left, direction[self.count], x_w)
            if d == 1:
                self.count += 1
                if self.count == l_w:
                    pass
                else:
                    travel(node.left, direction[self.count], x_w)
 
 
        travel(self.huffman_tree, direction[self.count], x_w)
        # 更新参数
        embedding_bag += self.e
        return self.loss, embedding_bag
 
    def softmax(self, x):
        return 1 / (1 + np.exp(-x))

这里我们讲讲上面写到的公式。

        我们使用最大似然法来寻找所有节点的词向量和所有内部节点θ。先拿上面的jumps例子来看，我们期望最大化下面的似然函数：

                 $\prod_{i=1}^{3} P(n(w_{i}), i)=\frac{1}{1+e^{-x_{w}\theta_{1}^{T}}}(1-\frac{1}{1+e^{-x_{w}\theta_{2}^{T}}})(1-\frac{1}{1+e^{-x_{w}\theta_{3}^{T}}})$

        对于所有的训练样本，我们期望最大化所有样本的似然函数乘积。

　　为了便于我们后面一般化的描述，我们定义输入的词为w，其从输入层词向量求和平均后的霍夫曼树根节点词向量为 $x_{w}$ ，从根节点到w所在的叶子节点，包含的节点总数为 $l_{w}$ ，w在霍夫曼树中从根节点开始，经过的第i个节点表示为 $p_{i}^{w}$ ，对应的霍夫曼编码为 $d_{i}^{w}$ ∈{0,1}，其中i=2,3,... $l_{w}$ 。而该节点对应的模型参数表示为 $\theta _{i}^{w}$ , 其中i=1,2,... $l_{w}-1$ ，没有 $i=l_{w}$ 是因为模型参数仅仅针对于霍夫曼树的内部节点。

        定义w经过的霍夫曼树某一个节点j的逻辑回归概率为 $P(d_{j}^{w}|x_{w},\theta_{j-1}^{w})$ ，其表达式为：

                        $P(d_{j}^{w}|x_{w},\theta_{j-1}^{w})=\left\{\begin{matrix}\sigma (x_{w}(\theta_{j}^{w})^{T}) & & d_{j}^{w}=0\\ 1-\sigma (x_{w}(\theta_{j}^{w})^{T}) & & d_{j}^{w}=1 \end{matrix}\right.$

        那么对于某一个目标输出词w，其最大似然为：

         $\prod_{j=2}^{l_{w}} P(d_{j}^{w}|x_{w}, \theta _{j-1}^{w})=\prod_{j=2}^{l_{w}}[\sigma(x_{w}(\theta_{j-1}^{w})^{T})]^{1-d_{j}^{w}}[1-\sigma(x_{w}(\theta_{j-1}^{w})^{T})]^{d_{j}^{w}}$

为了简化运算，我们通常会对最大似然估计进行log处理，这样同时也满足了交叉熵的定义，得到：

                 $L=log\prod_{j=2}^{l_{w}} P(d_{j}^{w}|x_{w}, \theta _{j-1}^{w})=\sum_{j=2}^{l_{w}}(1-d_{j}^{w})log[\sigma(x_{w}(\theta_{j-1}^{w})^{T})]+d_{j}^{w}log[1-\sigma(x_{w}(\theta_{j-1}^{w})^{T})]$

得到了损失函数之后，我们就可以进行梯度计算，以便于反向传播：

                 $\frac{\partial L}{\partial \theta_{j-1}^{w}}=\frac{\partial L}{\partial f(X)}\frac{\partial f(X)}{\partial X}\frac{\partial X}{\partial \theta_{j-1}^{w}}$

                 $f(X)=\frac{1}{1+e^{-X}}$

                 $X=x_{w}(\theta_{j-1}^{w})^T$

则带入公式并化简得：

                 $\frac{\partial L}{\partial \theta_{j-1}^{w}}=(1-d_{j}^{w}-\sigma (x_{w}(\theta_{j-1}^{w})^{T}))x_w$

                 $\frac{\partial L}{\partial x_{w}}=(1-d_{j}^{w}-\sigma (x_{w}(\theta_{j-1}^{w})^{T}))\theta_{j-1}^{w}$

有公式之后我们就可以带入计算了，正如上面代码所示，g代表梯度，e代表x_w对应的梯度，θ就对应θ的梯度。

3.构建word2vec模型(CBOW)

构建词袋模型，代码如下：


def word2vec(vocabulary, windows_size):
    length = len(vocabulary)
    n = windows_size * 2 + 1
    # 词袋one_hot编码
    bag = np.eye(n)
    # 计算总损失
    sum_loss = 0
    # 储存所有词向量
    word_dict = {}
    epochs = 10000
    for i, word in enumerate(vocabulary):
        target_path = list(tree_path[word])
        theta = np.random.randn(len(parents), 20)
        parents_theta = {}
        for i, parent in enumerate(parents):
            parents_theta[parent] = theta[i]
 
        # 初始化x_w权重
        w_linear = [np.random.randn(n, 20) if (j + i) >= windows_size and (j + i - 1) <= length else np.zeros((n, 20))
                    for j in range(windows_size * 2)]
        # x_i
        temp_bag = [bag[x] for x in range(n) if x != windows_size]
        # context_embedding
        embedding_bag = [np.dot(temp_bag[j], w_linear[j]) for j in range(windows_size * 2)]
        # x_w
        x_w = sum(embedding_bag) / (windows_size*2)
        temp_loss = 0
        for epoch in range(epochs):
            loss, embedding_bag = cbow.train(x_w, embedding_bag, parents_theta, target_path, len(target_path))
            temp_loss += loss
 
        sum_loss += temp_loss
        print("当前单词： {}  的损失为： {}".format(word, temp_loss/epochs))
        vector = sum(embedding_bag) / (windows_size * 2)
        # 将vector放入字典
        if word in word_dict:
            word_dict[word] = (word_dict[word] + vector) / 2
        else:
            word_dict[word] = vector
 
    return sum_loss / length / epochs, word_dict

这一步就不做过多的讲解，如有不懂的可以看我上一篇博客。

小结

本篇博客是我在学习NLP过程中自己实现的，代码什么的都是我自己手写的，如果有错误，欢迎指出。

本篇博客参考：

word2vec原理(二) 基于Hierarchical Softmax的模型 - 刘建平Pinard - 博客园 (cnblogs.com)

完整代码


import heapq
import jieba
import numpy as np
from collections import defaultdict
 
 
class HuffmanNode:
    def __init__(self, word, freq):
        self.word = word
        self.freq = freq
        self.left = None
        self.right = None
        self.parent = None
 
    def __lt__(self, other):
        return self.freq < other.freq
 
 
def build_huffman_tree(vocabulary):
    word_freq = defaultdict(int)
    for word in vocabulary:
        word_freq[word] += 1
 
    min_heap = [HuffmanNode(word, freq) for word, freq in word_freq.items()]
    heapq.heapify(min_heap)
 
    while len(min_heap) > 1:
        left = heapq.heappop(min_heap)
        right = heapq.heappop(min_heap)
        new_node = HuffmanNode(None, left.freq + right.freq)
        new_node.left = left
        new_node.right = right
        new_node.parent = None
        left.parent = new_node  # 给左子节点赋值其父母节点
        right.parent = new_node  # 给右子节点赋值其父母节点
        heapq.heappush(min_heap, new_node)
 
    huffman_tree = min_heap[0]
 
    return huffman_tree
 
 
def calculate_probabilities(huffman_tree):
    parents = []
 
    def traverse(node, path, probs):
        if node.word is not None:
            probs[node.word] = path
        if node.parent is not None:
            parents.append(node.parent)
        if node.left is not None:
            traverse(node.left, path + "1", probs)
        if node.right is not None:
            traverse(node.right, path + "0", probs)
 
    probs = {}
    traverse(huffman_tree, "", probs)
    return probs, parents
 
 
def display_huffman_tree(node, prefix='', is_left=True):
    if node is not None:
        print(f"{'L' if is_left else 'R'}{prefix}{node.word}:{node.freq}")
        display_huffman_tree(node.left, prefix + '|-- ', True)
        display_huffman_tree(node.right, prefix + '|-- ', False)
 
 
class CBOWWithHuffman:
    def __init__(self, vocab_size, embedding_dim, huffman_tree, learning_rate=0.01):
        self.vocab_size = vocab_size
        self.embedding_dim = embedding_dim
        self.huffman_tree = huffman_tree
        self.learning_rate = learning_rate
 
    def train(self, x_w, embedding_bag, parents_theta, direction, l_w):
        self.loss = 0
        self.count = 0
        self.e = 0
 
        def travel(node, d, x_w):
            d = int(d)
            if node is None:
                return
            if node.parent is not None:
                theta = parents_theta[node.parent]
                f = self.softmax(np.dot(x_w, theta.T))
                # l_f = np.dot(np.power(f, 1 - d), np.power(1 - f, d))
                g = (1 - d - f) * self.learning_rate
                self.e += np.dot(theta, g)
                parents_theta[node.parent] += np.dot(x_w, g)
                l = -(1-d)*np.log(f) - d*np.log(1-f)
                self.loss += l
 
            if d == 0:
                self.count += 1
                if self.count == l_w:
                    pass
                else:
                    travel(node.left, direction[self.count], x_w)
            if d == 1:
                self.count += 1
                if self.count == l_w:
                    pass
                else:
                    travel(node.left, direction[self.count], x_w)
 
 
        travel(self.huffman_tree, direction[self.count], x_w)
        # 更新参数
        embedding_bag += self.e
        return self.loss, embedding_bag
 
    def softmax(self, x):
        return 1 / (1 + np.exp(-x))
 
 
def word2vec(vocabulary, windows_size):
    length = len(vocabulary)
    n = windows_size * 2 + 1
    # 词袋one_hot编码
    bag = np.eye(n)
    sum_loss = 0
    # 储存所有词向量
    word_dict = {}
    epochs = 10000
    for i, word in enumerate(vocabulary):
        target_path = list(tree_path[word])
        theta = np.random.randn(len(parents), 20)
        parents_theta = {}
        for i, parent in enumerate(parents):
            parents_theta[parent] = theta[i]
 
        # 初始化x_w权重
        w_linear = [np.random.randn(n, 20) if (j + i) >= windows_size and (j + i - 1) <= length else np.zeros((n, 20))
                    for j in range(windows_size * 2)]
        # x_i
        temp_bag = [bag[x] for x in range(n) if x != windows_size]
        # context_embedding
        embedding_bag = [np.dot(temp_bag[j], w_linear[j]) for j in range(windows_size * 2)]
        # x_w
        x_w = sum(embedding_bag) / (windows_size*2)
        temp_loss = 0
        for epoch in range(epochs):
            loss, embedding_bag = cbow.train(x_w, embedding_bag, parents_theta, target_path, len(target_path))
            temp_loss += loss
 
        sum_loss += temp_loss
        print("当前单词： {}  的损失为： {}".format(word, temp_loss/epochs))
        vector = sum(embedding_bag) / (windows_size * 2)
        # 将vector放入字典
        if word in word_dict:
            word_dict[word] = (word_dict[word] + vector) / 2
        else:
            word_dict[word] = vector
 
    return sum_loss / length / epochs, word_dict
 
 
if __name__ == '__main__':
    sentence = '这篇文章探讨了 AI 是否能够替代工程经理的角色。作者的一个朋友提出了一个用 AI 智能体 EMAI 来管理软件工程团队的想法，' \
               '认为这样可以提高效率和客观性。作者则反驳了这个想法，指出人类经理有着 AI 无法比拟的同理心和直觉，以及能够塑造组织文化，' \
               '激励人员，建立友谊，和适应变化的能力。作者最后表明了自己的立场，选择信任以咖啡（或茶）为动力的工程经理，而不是电力驱动的 AI。' \
               '作者希望能够自动化那些妨碍有意义工作的任务，从而投入到同情心，关注和真正服务人们的工作中。'
 
    vocabulary = jieba.lcut(sentence)
    huffman_tree = build_huffman_tree(vocabulary)
 
    cbow = CBOWWithHuffman(len(vocabulary), 20, huffman_tree)
 
    tree_path, temp = calculate_probabilities(huffman_tree)
    parents = np.unique(temp)
    sum_loss, word_dict = word2vec(vocabulary, 2)
    print()
    print('loss = ', sum_loss)
    print('dict: ')
    for i in set(vocabulary):
        print("word: {}   vector: {}".format(i, word_dict[i]))

本文内容由网友自发贡献，转载请注明出处：【wpsshop博客】