NLP深度学习 DAY4：Word2Vec详解：两种模式（CBOW与Skip-gram）

5.2 训练模型

通过神经网络反向传播来计算更新，此时不光更新权重参数矩阵W，也会更新输入数据

训练完成后，我们就得到了比较准确的 Word Embeddings，从而得到了每个词的向量表示！！！ 

六、Python 代码实战

6.1 Model

from torch import nn


class DNN(nn.Module):
    def __init__(self, vocabulary_size, embedding_dim):
        super(DNN, self).__init__()

        self.embedding = nn.Linear(vocabulary_size, embedding_dim, bias=False)
        print("embedding_size:", list(self.embedding.weight.size()))

        self.layers = nn.Sequential(
            nn.Linear(vocabulary_size * embedding_dim, embedding_dim // 2),
            nn.LeakyReLU(),
            nn.Linear(embedding_dim // 2, 4),
            nn.LeakyReLU(),
            nn.Linear(4, 1),
        )

        # Mean squared error loss
        self.criterion = nn.MSELoss()
        # self.criterion = nn.CrossEntropyLoss()

    def forward(self, x):
        x = self.embedding(x)
        x = x.view(x.size()[0], -1)
        x = self.layers(x)
        x = x.squeeze(1)
        return x

    def cal_loss(self, pred, target):
        """ Calculate loss """
        return self.criterion(pred, target)

6.2 DataSet

import random

import numpy as np
from torch.utils.data import Dataset


class MyDataSet(Dataset):
    def __init__(self, features, labels):
        self.features = features
        self.labels = labels

    def __getitem__(self, index):
        return self.features[index], self.labels[index]

    def __len__(self):
        return len(self.features)


def get_data_set(data_path, window_width, window_step, negative_sample_num):
    with open(data_path, 'r', encoding='utf-8') as file:
        document = file.read()
        document = document.replace(",", "").replace("?", "").replace(".", "").replace('"', '')
        data = document.split(" ")
        print(f"数据中共有 {len(data)} 个单词")

        # 构造词典
        vocabulary = set()
        for word in data:
            vocabulary.add(word)
        vocabulary = list(vocabulary)
        print(f"词典大小为 {len(vocabulary)}")

        # index_dict
        index_dict = dict()
        for index, word in enumerate(vocabulary):
            index_dict[word] = index

        # 开始滑动窗口，构造数据
        features = []
        labels = []
        neighbor_dict = dict()

        for start_index in range(0, len(data), window_step):
            if start_index + window_width - 1 < len(data):
                mid_index = int((start_index + start_index + window_width - 1) / 2)
                for index in range(start_index, start_index + window_width):
                    if index != mid_index:
                        feature = np.zeros((len(vocabulary), len(vocabulary)))
                        feature[index_dict[data[index]]][index_dict[data[index]]] = 1
                        feature[index_dict[data[mid_index]]][index_dict[data[mid_index]]] = 1
                        features.append(feature)
                        labels.append(1)
                        if data[mid_index] in neighbor_dict.keys():
                            neighbor_dict[data[mid_index]].add(data[index])
                        else:
                            neighbor_dict[data[mid_index]] = {data[index]}
        # 负采样
        for _ in range(negative_sample_num):
            random_word = vocabulary[random.randint(0, len(vocabulary))]
            for word in vocabulary:
                if random_word not in neighbor_dict.keys() or word not in neighbor_dict[random_word]:
                    feature = np.zeros((len(vocabulary), len(vocabulary)))
                    feature[index_dict[random_word]][index_dict[random_word]] = 1
                    feature[index_dict[word]][index_dict[word]] = 1
                    features.append(feature)
                    labels.append(0)
                    break
        # 返回dataset和词典
        return MyDataSet(features, labels), vocabulary, index_dict

6.3 Main

import random
from math import sqrt

import numpy as np
import torch
from torch.utils.data import DataLoader

from Python.DataSet import get_data_set
from Python.Model import DNN


def same_seed(seed):
    """
    Fixes random number generator seeds for reproducibility
    固定时间种子。由于cuDNN会自动从几种算法中寻找最适合当前配置的算法，为了使选择的算法固定，所以固定时间种子
    :param seed: 时间种子
    :return: None
    """
    torch.backends.cudnn.deterministic = True  # 解决算法本身的不确定性，设置为True 保证每次结果是一致的
    torch.backends.cudnn.benchmark = False  # 解决了算法选择的不确定性，方便复现，提升训练速度
    np.random.seed(seed)  # 按顺序产生固定的数组，如果使用相同的seed，则生成的随机数相同， 注意每次生成都要调用一次
    torch.manual_seed(seed)  # 手动设置torch的随机种子，使每次运行的随机数都一致
    random.seed(seed)
    if torch.cuda.is_available():
        # 为GPU设置唯一的时间种子
        torch.cuda.manual_seed(seed)
        torch.cuda.manual_seed_all(seed)


def train(model, train_loader, config):
    # Setup optimizer
    optimizer = getattr(torch.optim, config['optimizer'])(
        model.parameters(), **config['optim_hyper_paras'])

    device = config['device']
    epoch = 0
    while epoch < config['n_epochs']:
        model.train()  # set model to training mode
        loss_arr = []
        for x, y in train_loader:  # iterate through the dataloader
            optimizer.zero_grad()  # set gradient to zero
            x, y = x.to(device).to(torch.float32), y.to(device).to(torch.float32)  # move data to device (cpu/cuda)
            pred = model(x)  # forward pass (compute output)
            mse_loss = model.cal_loss(pred, y)  # compute loss
            mse_loss.backward()  # compute gradient (backpropagation)
            optimizer.step()  # update model with optimizer
            loss_arr.append(mse_loss.item())
        print(f"epoch: {epoch}/{config['n_epochs']} , loss: {np.mean(loss_arr)}")
        epoch += 1

    print('Finished training after {} epochs'.format(epoch))


def find_min_distance_word_vector(cur_i, vector, embeddings, vocabulary):
    def calc_distance(v1, v2):
        # 计算欧式距离
        distance = 0
        for i in range(len(v1)):
            distance += sqrt(pow(v1[i] - v2[i], 2))
        return distance

    min_distance = None
    min_i = -1
    for i, word in enumerate(vocabulary):
        if cur_i != i:
            distance = calc_distance(vector, embeddings[i].tolist())
            if min_distance is None or min_distance > distance:
                min_distance = distance
                min_i = i
    return min_i


if __name__ == '__main__':
    data_path = './data/data.txt'
    config = {
        'seed': 3407,  # Your seed number, you can pick your lucky number. :)
        'device': 'cuda' if torch.cuda.is_available() else 'cpu',
        'n_epochs': 20,  # Number of epochs.
        'batch_size': 64,
        'optimizer': 'Adam',
        'optim_hyper_paras': {  # hyper-parameters for the optimizer (depends on which optimizer you are using)
            'lr': 0.001,  # learning rate of optimizer
        },
        'embedding_dim': 6,  # 词向量长度
        'window_width': 5,  # 窗口的宽度
        'window_step': 2,  # 窗口滑动的步长
        'negative_sample_num': 10  # 要增加的负样本个数
    }

    same_seed(config['seed'])

    data_set, vocabulary, index_dict = get_data_set(data_path, config['window_width'], config['window_step'],
                                                    config['negative_sample_num'])
    train_loader = DataLoader(data_set, config['batch_size'], shuffle=True, drop_last=False, pin_memory=True)

    model = DNN(len(vocabulary), config['embedding_dim']).to(config['device'])

    train(model, train_loader, config)

    # 训练完，看看embeddings，展示部分词的词向量，并找到离它最近的词的词向量
    embeddings = torch.t(model.embedding.weight)
    for i in range(10):
        print('%-50s%s' % (f"{vocabulary[i]} 的词向量为 :", str(embeddings[i].tolist())))
        min_i = find_min_distance_word_vector(i, embeddings[i].tolist(), embeddings, vocabulary)
        print('%-45s%s' % (
            f"离 {vocabulary[i]} 最近的词为 {vocabulary[min_i]} , 它的词向量为 :", str(embeddings[min_i].tolist())))
        print('-' * 200)

七、根据上面的代码举个例子 

一、数据处理流程（以示例文本说明）

假设输入文件 data.txt 内容为：

I love machine learning because it is interesting.

1. 文本预处理

• 清洗：移除标点（, ? . "），处理后得到：

  I love machine learning because it is interesting

• 分词：按空格切分为单词列表：

  data = ["I", "love", "machine", "learning", "because", "it", "is", "interesting"]

2. 构建词汇表

• 去重后得到词汇表（假设顺序固定）：

  vocabulary = ["I", "love", "machine", "learning", "because", "it", "is", "interesting"]

• 索引映射 index_dict：

  {"I":0, "love":1, "machine":2, "learning":3, "because":4, "it":5, "is":6, "interesting":7}

3. 滑动窗口生成正样本

假设窗口宽度 window_width=5，步长 window_step=2：

• 窗口划分：

  • 窗口1: ["I", "love", "machine", "learning", "because"]，中心词是第2个（machine，索引2）。

  • 窗口内上下文词：I（索引0）、love（索引1）、learning（索引3）、because（索引4）。

  • 每个上下文词与中心词生成一个正样本。

• 特征矩阵示例（中心词 machine 和上下文词 love）：

  # 特征矩阵形状 [8,8]（词汇表大小=8）
  feature = np.zeros((8,8))
  feature[1][1] = 1  # 上下文词 "love" 的对角线置1
  feature[2][2] = 1  # 中心词 "machine" 的对角线置1

  该样本标签为 1。

4. 负采样生成负样本

随机选择不共现的词对（例如 machine 和 it）：

feature = np.zeros((8,8))
feature[2][2] = 1  # "machine"
feature[5][5] = 1  # "it"

标签为 0。

二、模型架构详解（DNN 类）

1. Embedding 层

• 定义：nn.Linear(vocab_size, embed_dim, bias=False)

• 作用：将输入矩阵的每个词索引转换为嵌入向量。

• 输入形状：[batch_size, vocab_size, vocab_size]

• 输出形状：[batch_size, vocab_size, embed_dim]

  • 矩阵乘法等价于对每个词进行线性变换。

2. 全连接层

• 结构：

  nn.Sequential(
    nn.Linear(vocab_size * embed_dim, embed_dim//2),  # 展平后输入
    nn.LeakyReLU(),
    nn.Linear(embed_dim//2, 4),
    nn.LeakyReLU(),
    nn.Linear(4, 1)
  )

• 作用：将展平后的嵌入向量映射到标量输出。

3. 前向传播流程

以单个样本 [8,8]（词汇表大小=8）为例：

1. 输入矩阵：[1, 8, 8]（batch_size=1）

2. Embedding 层：与 [8, embed_dim] 权重矩阵相乘，得到 [1, 8, embed_dim]。

3. Reshape：展平为 [1, 8*embed_dim]。

4. 全连接层：逐步降维到标量输出。

三、训练过程（以示例说明）

1. 数据加载

• Dataset：包含特征矩阵和标签。

• DataLoader：按批次加载数据（batch_size=64）。

2. 损失函数与优化器

• 损失函数：MSELoss，优化目标使正样本输出接近1，负样本接近0。

• 优化器：Adam，学习率 0.001。

3. 训练循环

• 前向传播：输入特征矩阵，计算预测值。

• 反向传播：根据 MSE 损失更新参数。

四、词向量可视化与最近邻查找

1. 提取词向量

• 权重矩阵：embeddings = torch.t(model.embedding.weight)，形状 [vocab_size, embed_dim]。

• 示例输出：

  "machine" 的词向量为 : [0.12, -0.45, 0.78, ...]

2. 计算最近邻

• 欧式距离：遍历所有词向量，找到距离最小的非自身词。

• 示例结果：

  离 "machine" 最近的词为 "learning" ，距离 0.89