19_Word2Vec详解：训练你的词嵌入

vec("国王") - vec("男人") + vec("女人") ≈ vec("女王")

上下文词 → 中心词
[word(t-2), word(t-1), word(t+1), word(t+2)] → word(t)

中心词 → 上下文词
word(t) → [word(t-2), word(t-1), word(t+1), word(t+2)]

输入(中心词one-hot向量) → 投影层(查找词向量) → 输出层(预测上下文词概率)

L = \sum_{t=1}^{T} \sum_{-C ≤ j ≤ C, j≠0} log P(w_{t+j} | w_t)

P(w_O | w_I) = exp(v'_{w_O} · v_{w_I}) / \sum_{w=1}^{V} exp(v'_w · v_{w_I})

log σ(v'_c · v_w) + \sum_{i=1}^{k} log σ(-v'_{c_i} · v_w)

P(w) = f(w)^0.75 / \sum_{u∈V} f(u)^0.75

import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
import collections
import random
from collections import defaultdict
import matplotlib.pyplot as plt
from sklearn.manifold import TSNE

# 设置随机种子以确保结果可复现
def set_seed(seed=42):
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    if torch.cuda.is_available():
        torch.cuda.manual_seed_all(seed)

def preprocess_text(text, min_count=5):
    """预处理文本，构建词汇表并将文本转换为索引序列"""
    # 分词并转为小写
    words = text.lower().split()
    
    # 统计词频
    word_counts = collections.Counter(words)
    
    # 过滤低频词
    valid_words = [word for word in words if word_counts[word] >= min_count]
    
    # 构建词汇表
    word_to_idx = {word: i for i, word in enumerate(set(valid_words))}
    idx_to_word = {i: word for word, i in word_to_idx.items()}
    
    # 将文本转换为索引序列
    indexed_text = [word_to_idx[word] for word in valid_words]
    
    # 计算词频，用于负采样
    word_freq = defaultdict(int)
    for word in valid_words:
        word_freq[word_to_idx[word]] += 1
    
    return indexed_text, word_to_idx, idx_to_word, word_freq

# 示例文本（实际应用中会使用更大的语料库）
sample_text = """
Word2Vec is a technique for natural language processing. 
Word2Vec was developed by Google researchers. 
It is used to produce word embeddings. 
Word embeddings are vector representations of words. 
These vectors capture semantic relationships between words. 
Skip-Gram and CBOW are two models in Word2Vec. 
Skip-Gram predicts context words given a center word. 
CBOW predicts a center word given context words. 
Negative sampling is an optimization technique for Word2Vec. 
It helps in training the model efficiently with large vocabularies.
"""

set_seed()
indexed_text, word_to_idx, idx_to_word, word_freq = preprocess_text(sample_text)
vocab_size = len(word_to_idx)
print(f"词汇表大小: {vocab_size}")
print(f"处理后的文本长度: {len(indexed_text)}")
print(f"前10个单词索引: {indexed_text[:10]}")

def generate_training_data(indexed_text, word_freq, window_size=2, negative_samples=5):
    """生成训练数据，包括正样本和负样本"""
    training_data = []
    
    # 准备负采样的概率分布
    total_count = sum(word_freq.values())
    word_prob = {word: (freq / total_count) ** 0.75 for word, freq in word_freq.items()}
    word_prob_sum = sum(word_prob.values())
    word_prob = {word: prob / word_prob_sum for word, prob in word_prob.items()}
    
    # 构建用于负采样的词汇表和概率列表
    vocab_indices = list(word_prob.keys())
    vocab_probs = list(word_prob.values())
    
    # 生成训练样本
    for i, center_word in enumerate(indexed_text):
        # 确定上下文窗口的边界
        start = max(0, i - window_size)
        end = min(len(indexed_text), i + window_size + 1)
        
        # 生成正样本（中心词-上下文词对）
        for j in range(start, end):
            if j != i:
                context_word = indexed_text[j]
                training_data.append((center_word, context_word, 1))  # 1表示正样本
                
                # 为每个正样本生成负样本
                for _ in range(negative_samples):
                    # 随机选择负样本，确保不是中心词或上下文词
                    while True:
                        neg_word = np.random.choice(vocab_indices, p=vocab_probs)
                        if neg_word != center_word and neg_word != context_word:
                            break
                    training_data.append((center_word, neg_word, 0))  # 0表示负样本
    
    return training_data

window_size = 2
negative_samples = 5
training_data = generate_training_data(indexed_text, word_freq, window_size, negative_samples)
print(f"生成的训练样本数量: {len(training_data)}")
print(f"前5个训练样本: {training_data[:5]}")

class SkipGramModel(nn.Module):
    def __init__(self, vocab_size, embedding_dim):
        super(SkipGramModel, self).__init__()
        # 输入词嵌入（中心词）
        self.in_embed = nn.Embedding(vocab_size, embedding_dim)
        # 输出词嵌入（上下文词）
        self.out_embed = nn.Embedding(vocab_size, embedding_dim)
        
        # 初始化权重
        nn.init.xavier_uniform_(self.in_embed.weight)
        nn.init.xavier_uniform_(self.out_embed.weight)
    
    def forward(self, center_words, target_words, labels):
        # 获取中心词的嵌入
        center_embeds = self.in_embed(center_words)  # [batch_size, embedding_dim]
        # 获取目标词的嵌入
        target_embeds = self.out_embed(target_words)  # [batch_size, embedding_dim]
        
        # 计算点积
        scores = torch.sum(center_embeds * target_embeds, dim=1)  # [batch_size]
        # 使用sigmoid函数计算概率
        # 对于正样本，我们希望scores接近1；对于负样本，我们希望scores接近0
        loss = nn.functional.binary_cross_entropy_with_logits(
            scores, labels.float(), reduction='mean')
        
        return loss
    
    def get_word_embedding(self, word_idx):
        """获取指定词的嵌入向量"""
        return self.in_embed.weight[word_idx].detach().numpy()

# 模型参数
embedding_dim = 100
epochs = 100
batch_size = 32
learning_rate = 0.01

# 创建模型、损失函数和优化器
model = SkipGramModel(vocab_size, embedding_dim)
criterion = nn.BCEWithLogitsLoss()
optimizer = optim.Adam(model.parameters(), lr=learning_rate)

def train_model(model, training_data, epochs, batch_size, optimizer):
    """训练Skip-Gram模型"""
    model.train()
    
    # 将训练数据转换为张量
    center_words = torch.tensor([pair[0] for pair in training_data], dtype=torch.long)
    target_words = torch.tensor([pair[1] for pair in training_data], dtype=torch.long)
    labels = torch.tensor([pair[2] for pair in training_data], dtype=torch.long)
    
    # 创建数据加载器
    dataset = torch.utils.data.TensorDataset(center_words, target_words, labels)
    data_loader = torch.utils.data.DataLoader(dataset, batch_size=batch_size, shuffle=True)
    
    # 记录损失
    losses = []
    
    for epoch in range(epochs):
        total_loss = 0
        
        for batch in data_loader:
            center_batch, target_batch, label_batch = batch
            
            # 清零梯度
            optimizer.zero_grad()
            
            # 前向传播
            loss = model(center_batch, target_batch, label_batch)
            
            # 反向传播
            loss.backward()
            
            # 更新权重
            optimizer.step()
            
            total_loss += loss.item() * len(batch)
        
        # 计算平均损失
        avg_loss = total_loss / len(training_data)
        losses.append(avg_loss)
        
        if (epoch + 1) % 10 == 0:
            print(f"Epoch [{epoch+1}/{epochs}], Loss: {avg_loss:.4f}")
    
    return losses

# 训练模型
losses = train_model(model, training_data, epochs, batch_size, optimizer)

# 绘制损失曲线
plt.figure(figsize=(10, 6))
plt.plot(losses)
plt.title('Training Loss')
plt.xlabel('Epoch')
plt.ylabel('Loss')
plt.grid(True)
plt.savefig('skipgram_loss.png')
plt.show()

def find_similar_words(model, word_to_idx, idx_to_word, word, top_k=5):
    """查找与给定词最相似的词"""
    if word not in word_to_idx:
        print(f"词 '{word}' 不在词汇表中")
        return []
    
    word_idx = word_to_idx[word]
    word_embed = model.get_word_embedding(word_idx)
    
    # 计算与所有词的相似度
    similarities = []
    for i in range(len(idx_to_word)):
        if i != word_idx:
            sim = np.dot(word_embed, model.get_word_embedding(i)) / (
                np.linalg.norm(word_embed) * np.linalg.norm(model.get_word_embedding(i))
            )
            similarities.append((idx_to_word[i], sim))
    
    # 按相似度排序
    similarities.sort(key=lambda x: x[1], reverse=True)
    
    return similarities[:top_k]

def visualize_word_embeddings(model, idx_to_word, n_words=30):
    """可视化词向量"""
    # 获取词嵌入矩阵
    embeddings = np.zeros((min(n_words, len(idx_to_word)), embedding_dim))
    words = []
    
    for i in range(min(n_words, len(idx_to_word))):
        embeddings[i] = model.get_word_embedding(i)
        words.append(idx_to_word[i])
    
    # 使用t-SNE降维
    tsne = TSNE(n_components=2, random_state=42)
    embeddings_tsne = tsne.fit_transform(embeddings)
    
    # 绘制散点图
    plt.figure(figsize=(12, 10))
    for i, word in enumerate(words):
        plt.scatter(embeddings_tsne[i, 0], embeddings_tsne[i, 1])
        plt.annotate(word, xy=(embeddings_tsne[i, 0], embeddings_tsne[i, 1]), 
                     xytext=(5, 2), textcoords='offset points')
    
    plt.title('Word Embeddings Visualization')
    plt.grid(True)
    plt.savefig('word_embeddings.png')
    plt.show()

# 查找相似词
similar_words = find_similar_words(model, word_to_idx, idx_to_word, 'word2vec')
print(f"与'word2vec'相似的词:")
for word, similarity in similar_words:
    print(f"{word}: {similarity:.4f}")

# 可视化词嵌入
visualize_word_embeddings(model, idx_to_word)

import tensorflow as tf
from tensorflow.keras.layers import Embedding, Dense, Input
from tensorflow.keras.models import Model
from tensorflow.keras.optimizers import Adam
import numpy as np
import collections
import random
from collections import defaultdict
import matplotlib.pyplot as plt
from sklearn.manifold import TSNE

# 设置随机种子
def set_seed(seed=42):
    random.seed(seed)
    np.random.seed(seed)
    tf.random.set_seed(seed)

# 这里使用与PyTorch实现相同的预处理函数和数据生成函数
# 我们复用之前的preprocess_text和generate_training_data函数

# 实现TensorFlow版本的Skip-Gram模型
def create_skipgram_model(vocab_size, embedding_dim):
    # 输入层
    center_word_input = Input(shape=(1,), name='center_word')
    target_word_input = Input(shape=(1,), name='target_word')
    
    # 嵌入层
    embedding_layer = Embedding(
        input_dim=vocab_size,
        output_dim=embedding_dim,
        embeddings_initializer='glorot_uniform',
        name='word_embedding'
    )
    
    # 中心词嵌入
    center_embedding = embedding_layer(center_word_input)
    # 目标词嵌入
    target_embedding = embedding_layer(target_word_input)
    
    # 点积层
    dot_product = tf.keras.layers.Dot(axes=2, name='dot_product')([center_embedding, target_embedding])
    # 扁平化
    dot_product_flat = tf.keras.layers.Flatten()(dot_product)
    # 输出层（sigmoid激活）
    output = Dense(1, activation='sigmoid', name='output')(dot_product_flat)
    
    # 创建模型
    model = Model(
        inputs=[center_word_input, target_word_input],
        outputs=output
    )
    
    # 编译模型
    model.compile(
        optimizer=Adam(learning_rate=0.01),
        loss='binary_crossentropy',
        metrics=['accuracy']
    )
    
    return model

# 提取词嵌入的模型
def create_embedding_extractor(model):
    return Model(
        inputs=model.get_layer('word_embedding').input,
        outputs=model.get_layer('word_embedding').output
    )

# 模型参数
embedding_dim = 100
epochs = 100
batch_size = 32

# 创建模型
skipgram_model = create_skipgram_model(vocab_size, embedding_dim)
skipgram_model.summary()

# 准备训练数据
center_words = np.array([pair[0] for pair in training_data], dtype=np.int32)
target_words = np.array([pair[1] for pair in training_data], dtype=np.int32)
labels = np.array([pair[2] for pair in training_data], dtype=np.float32)

# 训练模型
history = skipgram_model.fit(
    [center_words, target_words],
    labels,
    batch_size=batch_size,
    epochs=epochs,
    validation_split=0.1,
    verbose=2
)

# 绘制训练历史
plt.figure(figsize=(12, 5))
plt.subplot(1, 2, 1)
plt.plot(history.history['loss'], label='Training Loss')
plt.plot(history.history['val_loss'], label='Validation Loss')
plt.title('Model Loss')
plt.xlabel('Epoch')
plt.ylabel('Loss')
plt.legend()
plt.grid(True)

plt.subplot(1, 2, 2)
plt.plot(history.history['accuracy'], label='Training Accuracy')
plt.plot(history.history['val_accuracy'], label='Validation Accuracy')
plt.title('Model Accuracy')
plt.xlabel('Epoch')
plt.ylabel('Accuracy')
plt.legend()
plt.grid(True)

plt.tight_layout()
plt.savefig('tensorflow_skipgram_training.png')
plt.show()

# 提取词嵌入
embedding_extractor = create_embedding_extractor(skipgram_model)

# 查找相似词
def find_similar_words_tf(embedding_extractor, word_to_idx, idx_to_word, word, top_k=5):
    if word not in word_to_idx:
        print(f"词 '{word}' 不在词汇表中")
        return []
    
    word_idx = word_to_idx[word]
    word_embedding = embedding_extractor.predict([[word_idx]])[0][0]
    
    similarities = []
    for i in range(len(idx_to_word)):
        if i != word_idx:
            other_embedding = embedding_extractor.predict([[i]])[0][0]
            sim = np.dot(word_embedding, other_embedding) / (
                np.linalg.norm(word_embedding) * np.linalg.norm(other_embedding)
            )
            similarities.append((idx_to_word[i], sim))
    
    similarities.sort(key=lambda x: x[1], reverse=True)
    return similarities[:top_k]

# 可视化词嵌入
def visualize_word_embeddings_tf(embedding_extractor, idx_to_word, n_words=30):
    embeddings = np.zeros((min(n_words, len(idx_to_word)), embedding_dim))
    words = []
    
    for i in range(min(n_words, len(idx_to_word))):
        embeddings[i] = embedding_extractor.predict([[i]])[0][0]
        words.append(idx_to_word[i])
    
    tsne = TSNE(n_components=2, random_state=42)
    embeddings_tsne = tsne.fit_transform(embeddings)
    
    plt.figure(figsize=(12, 10))
    for i, word in enumerate(words):
        plt.scatter(embeddings_tsne[i, 0], embeddings_tsne[i, 1])
        plt.annotate(word, xy=(embeddings_tsne[i, 0], embeddings_tsne[i, 1]), 
                     xytext=(5, 2), textcoords='offset points')
    
    plt.title('TensorFlow Word Embeddings Visualization')
    plt.grid(True)
    plt.savefig('tensorflow_word_embeddings.png')
    plt.show()

# 查找相似词
similar_words_tf = find_similar_words_tf(embedding_extractor, word_to_idx, idx_to_word, 'word2vec')
print(f"\nTensorFlow: 与'word2vec'相似的词:")
for word, similarity in similar_words_tf:
    print(f"{word}: {similarity:.4f}")

# 可视化词嵌入
visualize_word_embeddings_tf(embedding_extractor, idx_to_word)

P(c|w) = \prod_{n \in path(c)} \sigma(±v'_n · v_w)