词向量进阶

|
          根(θ₁)
         /     \
      (θ₂)    (θ₃)
      /  \      /  \
    w₁   w₂   w₃   w₄

|
算法：GloVe训练
输入：共现矩阵 X, 向量维度 d, 迭代次数 epochs
输出：词向量 W, W̃
 
1. 初始化 W, W̃, b, b̃ 为小随机值
2. for epoch = 1 to epochs:
3.     for (i, j) where X_ij > 0:
4.         计算权重 w_ij = f(X_ij)
5.         计算梯度：
6.             diff = w_i^T w̃_j + b_i + b̃_j - log(X_ij)
7.             ∇J_w_i = w_ij * diff * w̃_j
8.             ∇J_w̃_j = w_ij * diff * w_i
9.             ∇J_b_i = w_ij * diff
10.            ∇J_b̃_j = w_ij * diff
11.        更新参数（使用AdaGrad等优化器）
12. 返回 W 和 W̃ 的平均作为最终词向量

|
import torch
import torch.nn as nn
import numpy as np
from collections import Counter
 
class SkipGramNegSampling(nn.Module):
    """带负采样的Skip-gram模型"""
    
    def __init__(self, vocab_size, embedding_dim):
        super(SkipGramNegSampling, self).__init__()
        
        self.vocab_size = vocab_size
        self.embedding_dim = embedding_dim
        
        # 中心词嵌入
        self.center_embeddings = nn.Embedding(vocab_size, embedding_dim)
        
        # 上下文词嵌入
        self.context_embeddings = nn.Embedding(vocab_size, embedding_dim)
        
        # 初始化
        self._init_weights()
    
    def _init_weights(self):
        """初始化权重"""
        init_range = 0.5 / self.embedding_dim
        self.center_embeddings.weight.data.uniform_(-init_range, init_range)
        self.context_embeddings.weight.data.uniform_(-init_range, init_range)
    
    def forward(self, center_words, pos_contexts, neg_contexts):
        """
        Args:
            center_words: (batch_size,) 中心词索引
            pos_contexts: (batch_size,) 正样本上下文词索引
            neg_contexts: (batch_size, k) k个负样本索引
        
        Returns:
            loss: 负采样损失
        """
        batch_size, k = neg_contexts.shape
        
        # 获取词向量
        # (batch_size, embedding_dim)
        center_embeds = self.center_embeddings(center_words)
        # (batch_size, embedding_dim)
        pos_embeds = self.context_embeddings(pos_contexts)
        # (batch_size, k, embedding_dim)
        neg_embeds = self.context_embeddings(neg_contexts)
        
        # 计算正样本得分
        # (batch_size,)
        pos_score = torch.sum(center_embeds * pos_embeds, dim=1)
        pos_loss = -torch.log(torch.sigmoid(pos_score))
        
        # 计算负样本得分
        # (batch_size, k)
        neg_score = torch.bmm(neg_embeds, center_embeds.unsqueeze(2)).squeeze(2)
        neg_loss = -torch.sum(torch.log(torch.sigmoid(-neg_score)), dim=1)
        
        # 总损失
        loss = torch.mean(pos_loss + neg_loss)
        
        return loss
    
    def get_embedding(self, word_idx):
        """获取词向量（中心词+上下文词的平均）"""
        center_vec = self.center_embeddings.weight[word_idx].detach().cpu().numpy()
        context_vec = self.context_embeddings.weight[word_idx].detach().cpu().numpy()
        return (center_vec + context_vec) / 2
 
 
class NegativeSampler:
    """负采样器"""
    
    def __init__(self, word_counts, power=0.75):
        """
        Args:
            word_counts: 词频字典 {word_idx: count}
            power: 词频的幂次，默认0.75
        """
        # 计算采样概率
        vocab_size = len(word_counts)
        self.vocab_size = vocab_size
        
        # 构建概率分布
        word_freqs = np.array([word_counts.get(i, 0) for i in range(vocab_size)])
        word_freqs = np.power(word_freqs, power)
        self.probs = word_freqs / np.sum(word_freqs)
    
    def sample(self, batch_size, k):
        """
        采样负样本
        
        Args:
            batch_size: 批次大小
            k: 每个样本的负样本数量
        
        Returns:
            (batch_size, k) 负样本索引
        """
        neg_samples = np.random.choice(
            self.vocab_size,
            size=(batch_size, k),
            p=self.probs,
            replace=True
        )
        return torch.LongTensor(neg_samples)
 
 
# 训练示例
def train_skipgram_neg_sampling(model, data_loader, neg_sampler, epochs=10, lr=0.01, k=5):
    """
    训练带负采样的Skip-gram
    
    Args:
        model: SkipGramNegSampling模型
        data_loader: 数据加载器，返回(center_words, context_words)
        neg_sampler: 负采样器
        epochs: 训练轮数
        lr: 学习率
        k: 负样本数量
    """
    optimizer = torch.optim.Adam(model.parameters(), lr=lr)
    
    for epoch in range(epochs):
        total_loss = 0
        num_batches = 0
        
        for center_words, pos_contexts in data_loader:
            batch_size = center_words.size(0)
            
            # 采样负样本
            neg_contexts = neg_sampler.sample(batch_size, k)
            
            # 前向传播
            optimizer.zero_grad()
            loss = model(center_words, pos_contexts, neg_contexts)
            
            # 反向传播
            loss.backward()
            optimizer.step()
            
            total_loss += loss.item()
            num_batches += 1
        
        avg_loss = total_loss / num_batches
        print(f"Epoch {epoch + 1}/{epochs}, Loss: {avg_loss:.4f}")
 
 
# 使用示例
if __name__ == "__main__":
    # 假设已经预处理好数据
    vocab_size = 10000
    embedding_dim = 100
    k_negative = 5
    
    # 创建模型
    model = SkipGramNegSampling(vocab_size, embedding_dim)
    
    # 假设word_counts是词频统计
    word_counts = {i: np.random.randint(1, 1000) for i in range(vocab_size)}
    neg_sampler = NegativeSampler(word_counts, power=0.75)
    
    print("模型创建成功！")
    print(f"中心词嵌入维度: {model.center_embeddings.weight.shape}")
    print(f"上下文词嵌入维度: {model.context_embeddings.weight.shape}")

|
import torch
import torch.nn as nn
import numpy as np
from scipy import sparse
 
class GloVeModel(nn.Module):
    """GloVe模型实现"""
    
    def __init__(self, vocab_size, embedding_dim):
        super(GloVeModel, self).__init__()
        
        self.vocab_size = vocab_size
        self.embedding_dim = embedding_dim
        
        # 主词向量
        self.w_embeddings = nn.Embedding(vocab_size, embedding_dim)
        # 上下文词向量
        self.w_tilde_embeddings = nn.Embedding(vocab_size, embedding_dim)
        
        # 偏置项
        self.w_biases = nn.Embedding(vocab_size, 1)
        self.w_tilde_biases = nn.Embedding(vocab_size, 1)
        
        # 初始化
        self._init_weights()
    
    def _init_weights(self):
        """随机初始化参数"""
        init_range = 0.5 / self.embedding_dim
        self.w_embeddings.weight.data.uniform_(-init_range, init_range)
        self.w_tilde_embeddings.weight.data.uniform_(-init_range, init_range)
        self.w_biases.weight.data.zero_()
        self.w_tilde_biases.weight.data.zero_()
    
    def forward(self, i_indices, j_indices, co_occurrences, weights):
        """
        Args:
            i_indices: (batch_size,) 词i的索引
            j_indices: (batch_size,) 词j的索引
            co_occurrences: (batch_size,) log(X_ij)
            weights: (batch_size,) f(X_ij)权重
        
        Returns:
            loss: 加权平方损失
        """
        # 获取词向量和偏置
        w_i = self.w_embeddings(i_indices)
        w_tilde_j = self.w_tilde_embeddings(j_indices)
        b_i = self.w_biases(i_indices).squeeze()
        b_tilde_j = self.w_tilde_biases(j_indices).squeeze()
        
        # 计算内积
        inner_product = torch.sum(w_i * w_tilde_j, dim=1)
        
        # 计算差异
        diff = inner_product + b_i + b_tilde_j - co_occurrences
        
        # 加权平方损失
        loss = weights * (diff ** 2)
        
        return torch.mean(loss)
    
    def get_embedding(self, word_idx):
        """获取最终词向量（W和W̃的平均）"""
        w = self.w_embeddings.weight[word_idx].detach().cpu().numpy()
        w_tilde = self.w_tilde_embeddings.weight[word_idx].detach().cpu().numpy()
        return (w + w_tilde) / 2
 
 
def compute_weight(x, x_max=100, alpha=0.75):
    """
    计算GloVe权重函数
    
    Args:
        x: 共现次数
        x_max: 最大共现次数阈值
        alpha: 幂次参数
    
    Returns:
        权重值
    """
    return np.where(x < x_max, (x / x_max) ** alpha, 1.0)
 
 
def build_cooccurrence_matrix(corpus, vocab, window_size=5):
    """
    构建共现矩阵
    
    Args:
        corpus: 词索引序列列表
        vocab: 词汇表字典
        window_size: 窗口大小
    
    Returns:
        稀疏共现矩阵
    """
    vocab_size = len(vocab)
    cooccur_matrix = sparse.lil_matrix((vocab_size, vocab_size), dtype=np.float32)
    
    for sentence in corpus:
        for i, center_word in enumerate(sentence):
            # 左侧上下文
            start = max(0, i - window_size)
            for j in range(start, i):
                context_word = sentence[j]
                distance = i - j
                cooccur_matrix[center_word, context_word] += 1.0 / distance
            
            # 右侧上下文
            end = min(len(sentence), i + window_size + 1)
            for j in range(i + 1, end):
                context_word = sentence[j]
                distance = j - i
                cooccur_matrix[center_word, context_word] += 1.0 / distance
    
    # 转换为COO格式用于训练
    return cooccur_matrix.tocoo()
 
 
# 训练函数
def train_glove(model, cooccur_matrix, epochs=50, lr=0.05, x_max=100, alpha=0.75):
    """
    训练GloVe模型
    
    Args:
        model: GloVe模型
        cooccur_matrix: 共现矩阵（scipy.sparse格式）
        epochs: 训练轮数
        lr: 学习率
        x_max: 权重函数参数
        alpha: 权重函数参数
    """
    optimizer = torch.optim.Adagrad(model.parameters(), lr=lr)
    
    # 提取非零元素
    i_indices = torch.LongTensor(cooccur_matrix.row)
    j_indices = torch.LongTensor(cooccur_matrix.col)
    co_occur_values = torch.FloatTensor(cooccur_matrix.data)
    log_co_occur = torch.log(co_occur_values)
    
    # 计算权重
    weights = compute_weight(co_occur_values.numpy(), x_max, alpha)
    weights = torch.FloatTensor(weights)
    
    num_samples = len(i_indices)
    
    print("开始训练GloVe...")
    for epoch in range(epochs):
        # 打乱数据
        perm = torch.randperm(num_samples)
        i_shuffled = i_indices[perm]
        j_shuffled = j_indices[perm]
        log_co_shuffled = log_co_occur[perm]
        weights_shuffled = weights[perm]
        
        total_loss = 0
        batch_size = 4096
        num_batches = (num_samples + batch_size - 1) // batch_size
        
        for batch_idx in range(num_batches):
            start = batch_idx * batch_size
            end = min((batch_idx + 1) * batch_size, num_samples)
            
            i_batch = i_shuffled[start:end]
            j_batch = j_shuffled[start:end]
            log_co_batch = log_co_shuffled[start:end]
            weights_batch = weights_shuffled[start:end]
            
            optimizer.zero_grad()
            loss = model(i_batch, j_batch, log_co_batch, weights_batch)
            loss.backward()
            optimizer.step()
            
            total_loss += loss.item()
        
        avg_loss = total_loss / num_batches
        if (epoch + 1) % 10 == 0:
            print(f"Epoch {epoch + 1}/{epochs}, Loss: {avg_loss:.4f}")
    
    print("训练完成！")
 
 
# 使用示例
if __name__ == "__main__":
    print("GloVe模型实现示例")
    
    # 模拟数据
    vocab_size = 5000
    embedding_dim = 50
    
    # 创建模型
    model = GloVeModel(vocab_size, embedding_dim)
    print(f"模型参数数量: {sum(p.numel() for p in model.parameters())}")

|
import numpy as np
from scipy.stats import spearmanr
 
def evaluate_analogy(word_vectors, vocab, test_cases):
    """
    评估词类比任务
    
    Args:
        word_vectors: 词向量矩阵 (vocab_size, embedding_dim)
        vocab: 词到索引的字典
        test_cases: [(a, b, c, d)] 类比测试用例
    
    Returns:
        准确率
    """
    correct = 0
    total = 0
    
    for a, b, c, d in test_cases:
        if all(w in vocab for w in [a, b, c, d]):
            # 计算 b - a + c
            vec_a = word_vectors[vocab[a]]
            vec_b = word_vectors[vocab[b]]
            vec_c = word_vectors[vocab[c]]
            target_vec = vec_b - vec_a + vec_c
            
            # 计算与所有词的相似度
            similarities = np.dot(word_vectors, target_vec)
            similarities /= (np.linalg.norm(word_vectors, axis=1) * np.linalg.norm(target_vec))
            
            # 排除输入词
            for exclude_word in [a, b, c]:
                similarities[vocab[exclude_word]] = -np.inf
            
            # 找出最相似的词
            predicted_idx = np.argmax(similarities)
            predicted_word = list(vocab.keys())[list(vocab.values()).index(predicted_idx)]
            
            if predicted_word == d:
                correct += 1
            total += 1
    
    return correct / total if total > 0 else 0
 
 
def evaluate_similarity(word_vectors, vocab, test_pairs):
    """
    评估词相似度任务
    
    Args:
        word_vectors: 词向量矩阵
        vocab: 词到索引的字典
        test_pairs: [(word1, word2, human_score)] 测试对
    
    Returns:
        Spearman相关系数
    """
    model_scores = []
    human_scores = []
    
    for word1, word2, human_score in test_pairs:
        if word1 in vocab and word2 in vocab:
            vec1 = word_vectors[vocab[word1]]
            vec2 = word_vectors[vocab[word2]]
            
            # 计算余弦相似度
            cos_sim = np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))
            
            model_scores.append(cos_sim)
            human_scores.append(human_score)
    
    # 计算Spearman相关系数
    correlation, p_value = spearmanr(model_scores, human_scores)
    return correlation

|
class AliasNegativeSampler:
    """使用别名采样的高效负采样器"""
    def __init__(self, word_counts, power=0.75):
        # 你的实现
        pass
    
    def sample(self, k):
        """O(1)复杂度采样k个负样本"""
        pass

|
from sklearn.manifold import TSNE
import matplotlib.pyplot as plt
 
def visualize_embeddings(word_vectors, words, output_file='embeddings.png'):
    """
    可视化词向量
    
    Args:
        word_vectors: 词向量数组 (n, dim)
        words: 词列表
        output_file: 输出图片路径
    """
    # 你的实现
    pass