2026/4/18 11:03:19
网站建设
项目流程
青海网站建设公司多少钱,太原网页设计师,wordpress开,深圳公司网站建设服务为先超越Word2Vec#xff1a;基于对比学习的现代词嵌入实现与实践
引言#xff1a;词嵌入的演进与新挑战
词嵌入作为自然语言处理的基石#xff0c;自Word2Vec、GloVe等经典方法问世以来#xff0c;已经历了显著的发展。然而#xff0c;传统方法存在固有的局限性#xff1a;它…超越Word2Vec基于对比学习的现代词嵌入实现与实践引言词嵌入的演进与新挑战词嵌入作为自然语言处理的基石自Word2Vec、GloVe等经典方法问世以来已经历了显著的发展。然而传统方法存在固有的局限性它们通常基于简单的共现统计或浅层神经网络难以捕捉复杂的上下文语义特别是在处理一词多义、稀有词和领域特定术语时表现欠佳。近年来对比学习Contrastive Learning为词嵌入领域带来了革命性的新思路。与预测下一个词的传统方法不同对比学习通过拉近语义相似的样本、推远不相似的样本来学习表示。本文将深入探讨基于对比学习的词嵌入实现提供一个完整的、生产级别的Python实现并分析其在现代NLP应用中的优势。对比学习词嵌入的核心思想传统方法的局限传统的词嵌入方法如Word2Vec主要基于分布假说“出现在相似上下文中的词具有相似含义”但这种方法存在以下问题静态表示每个词只有一个固定的向量表示上下文忽略无法处理一词多义现象负采样局限基于频率的负采样可能无法提供有意义的对比信号对比学习的优势对比学习通过构建正负样本对直接优化表示空间的结构正样本对(anchor, positive) ← 语义相似的词 负样本对(anchor, negative) ← 语义不相似的词 目标使正样本在表示空间中更接近负样本更远离基于对比学习的词嵌入实现系统架构设计我们设计一个双塔结构的对比学习模型包含以下核心组件编码器网络将词序列转换为固定维度的向量投影头将编码器输出映射到对比学习空间损失函数使用InfoNCENormalized Temperature-scaled Cross Entropy损失数据增强策略创建语义不变的数据变换环境配置与依赖# requirements.txt torch2.0.1 transformers4.30.0 numpy1.24.3 pandas2.0.2 scikit-learn1.3.0 tqdm4.65.0 datasets2.13.1 # 新增用于高级文本处理和数据增强 textaugment2.0.0 sentence-transformers2.2.2数据准备与增强策略import torch from torch.utils.data import Dataset, DataLoader from typing import List, Tuple, Dict import random from textaugment import EDA, Word2vec import numpy as np class ContrastiveWordDataset(Dataset): 对比学习专用的词级别数据集 支持多种数据增强策略生成正样本 def __init__(self, texts: List[str], vocab: Dict[str, int], window_size: int 3, num_negatives: int 10, augmentation_strategy: str mixed): Args: texts: 原始文本列表 vocab: 词汇表字典 {word: index} window_size: 上下文窗口大小 num_negatives: 每个正样本对应的负样本数 augmentation_strategy: 数据增强策略 self.texts texts self.vocab vocab self.window_size window_size self.num_negatives num_negatives self.augmentation_strategy augmentation_strategy # 构建词索引 self.word_to_indices self._build_word_indices() # 初始化数据增强器 self.augmenters self._init_augmenters() # 预计算词频率用于负采样 self.word_freq self._compute_word_frequencies() def _build_word_indices(self) - Dict[str, List[Tuple[int, int]]]: 记录每个词在文本中出现的位置 word_indices {} for text_idx, text in enumerate(self.texts): words text.split() for word_pos, word in enumerate(words): if word in self.vocab: if word not in word_indices: word_indices[word] [] word_indices[word].append((text_idx, word_pos)) return word_indices def _init_augmenters(self): 初始化不同的数据增强器 augmenters {} if self.augmentation_strategy in [synonym, mixed]: # 使用预训练词向量进行同义词替换 augmenters[synonym] Word2vec() if self.augmentation_strategy in [eda, mixed]: # EDA: 同义词替换、随机插入、随机交换、随机删除 augmenters[eda] EDA() return augmenters def _augment_word(self, word: str, context: List[str]) - str: 对目标词进行数据增强 strategies [] if synonym in self.augmenters and random.random() 0.5: try: augmented self.augmenters[synonym].augment(word) if augmented ! word: strategies.append(augmented) except: pass if eda in self.augmenters and random.random() 0.5: # 使用EDA策略生成变体 try: # 创建小文本进行增强 temp_text .join(context [word]) augmented self.augmenters[eda].random_swap(temp_text) augmented_words augmented.split() if augmented_words and augmented_words[-1] ! word: strategies.append(augmented_words[-1]) except: pass return random.choice(strategies) if strategies else word def _get_context_words(self, text_idx: int, word_pos: int) - List[str]: 获取目标词的上下文词 words self.texts[text_idx].split() start max(0, word_pos - self.window_size) end min(len(words), word_pos self.window_size 1) # 排除目标词本身 context words[start:word_pos] words[word_pos1:end] return context def _sample_negative_words(self, anchor_word: str, n: int 10) - List[str]: 基于频率和语义的负采样 高频词和语义不相关的词更容易被采样为负样本 all_words list(self.vocab.keys()) # 计算采样权重频率越高权重越大 weights [self.word_freq.get(w, 1e-6) for w in all_words] # 排除正样本词及其常见同义词 excluded_words {anchor_word} # 这里可以添加同义词排除逻辑 valid_words [w for w in all_words if w not in excluded_words] valid_weights [weights[list(self.vocab.keys()).index(w)] for w in valid_words] # 归一化权重 valid_weights np.array(valid_weights) valid_weights valid_weights / valid_weights.sum() # 采样负样本 negatives np.random.choice( valid_words, sizemin(n, len(valid_words)), pvalid_weights, replaceFalse ) return list(negatives) def __getitem__(self, idx): # 随机选择一个词作为锚点 all_words list(self.word_to_indices.keys()) anchor_word random.choice(all_words) # 随机选择锚点词的一个出现位置 text_idx, word_pos random.choice(self.word_to_indices[anchor_word]) # 获取上下文 context self._get_context_words(text_idx, word_pos) # 生成正样本通过数据增强 positive_word self._augment_word(anchor_word, context) # 采样负样本 negative_words self._sample_negative_words( anchor_word, self.num_negatives ) # 转换为索引 anchor_idx self.vocab.get(anchor_word, 0) positive_idx self.vocab.get(positive_word, 0) negative_indices [self.vocab.get(w, 0) for w in negative_words] # 获取上下文词的索引 context_indices [self.vocab.get(w, 0) for w in context if w in self.vocab] return { anchor: anchor_idx, positive: positive_idx, negatives: negative_indices, context: context_indices, anchor_word: anchor_word, positive_word: positive_word } def __len__(self): return len(self.word_to_indices) * 10 # 每个词采样10次模型架构实现import torch import torch.nn as nn import torch.nn.functional as F from transformers import AutoModel, AutoConfig class ContrastiveWordEmbeddingModel(nn.Module): 基于对比学习的词嵌入模型 使用Transformer编码器和投影头 def __init__(self, vocab_size: int, embedding_dim: int 256, hidden_dim: int 512, num_layers: int 3, num_heads: int 8, dropout: float 0.1, temperature: float 0.07, use_context: bool True): super().__init__() self.vocab_size vocab_size self.embedding_dim embedding_dim self.temperature temperature self.use_context use_context # 词嵌入层 self.word_embedding nn.Embedding(vocab_size, embedding_dim) # 上下文编码器小型Transformer encoder_config AutoConfig.from_pretrained(bert-base-uncased) encoder_config.hidden_size hidden_dim encoder_config.num_hidden_layers num_layers encoder_config.num_attention_heads num_heads encoder_config.intermediate_size hidden_dim * 4 self.context_encoder AutoModel.from_config(encoder_config) # 上下文投影层 self.context_projection nn.Sequential( nn.Linear(hidden_dim, embedding_dim), nn.LayerNorm(embedding_dim), nn.GELU(), nn.Dropout(dropout) ) # 词表示投影头用于对比学习 self.word_projection nn.Sequential( nn.Linear(embedding_dim, embedding_dim * 2), nn.LayerNorm(embedding_dim * 2), nn.GELU(), nn.Dropout(dropout), nn.Linear(embedding_dim * 2, embedding_dim) ) # 上下文感知的词表示融合 if use_context: self.fusion_layer nn.Sequential( nn.Linear(embedding_dim * 2, embedding_dim), nn.LayerNorm(embedding_dim), nn.GELU() ) # 初始化权重 self._init_weights() def _init_weights(self): 初始化模型权重 # 词嵌入层初始化 nn.init.normal_(self.word_embedding.weight, mean0.0, std0.02) # 线性层初始化 for module in [self.context_projection, self.word_projection]: for name, param in module.named_parameters(): if weight in name: nn.init.xavier_uniform_(param) elif bias in name: nn.init.zeros_(param) def encode_context(self, context_ids: torch.Tensor, context_mask: torch.Tensor None) - torch.Tensor: 编码上下文序列 Args: context_ids: [batch_size, seq_len] context_mask: [batch_size, seq_len] Returns: context_embeddings: [batch_size, embedding_dim] if context_mask is None: context_mask (context_ids ! 0).long() # 获取上下文编码 encoder_outputs self.context_encoder( input_idscontext_ids, attention_maskcontext_mask ) # 使用[CLS] token或平均池化 if hasattr(self.context_encoder.config, pooler_output): context_repr encoder_outputs.pooler_output else: # 平均池化 sequence_output encoder_outputs.last_hidden_state attention_mask_expanded context_mask.unsqueeze(-1).float() sum_embeddings torch.sum(sequence_output * attention_mask_expanded, dim1) sum_mask torch.clamp(attention_mask_expanded.sum(dim1), min1e-9) context_repr sum_embeddings / sum_mask # 投影到词嵌入空间 context_embeddings self.context_projection(context_repr) return context_embeddings def forward(self, anchor_ids: torch.Tensor, positive_ids: torch.Tensor, negative_ids: torch.Tensor, context_ids: torch.Tensor None) - Dict[str, torch.Tensor]: 前向传播 Args: anchor_ids: [batch_size] positive_ids: [batch_size] negative_ids: [batch_size, num_negatives] context_ids: [batch_size, context_len] Returns: 包含损失和各项指标的字典 batch_size anchor_ids.size(0) num_negatives negative_ids.size(1) # 获取基础词嵌入 anchor_emb self.word_embedding(anchor_ids) # [batch_size, emb_dim] positive_emb self.word_embedding(positive_ids) negative_emb self.word_embedding(negative_ids.view(-1)).view( batch_size, num_negatives, -1) # [batch_size, num_neg, emb_dim] # 如果有上下文编码上下文并融合 if self.use_context and context_ids is not None: context_emb self.encode_context(context_ids) # 融合词嵌入和上下文信息 anchor_with_context torch.cat([anchor_emb, context_emb], dim-1) positive_with_context torch.cat([positive_emb, context_emb], dim-1) anchor_emb self.fusion_layer(anchor_with_context) positive_emb self.fusion_layer(positive_with_context) # 为负样本重复上下文 context_emb_expanded context_emb.unsqueeze(1).repeat(1, num_negatives, 1) negative_with_context torch.cat( [negative_emb, context_emb_expanded], dim-1) negative_emb self.fusion_layer( negative_with_context.view(-1, self.embedding_dim * 2) ).view(batch_size, num_negatives, -1) # 投影到对比学习空间 anchor_proj self.word_projection(anchor_emb) positive_proj self.word_projection(positive_emb) negative_proj self.word_projection( negative_emb.view(-1, self.embedding_dim) ).view(batch_size, num_negatives, -1) # 计算相似度 # 正样本相似度 pos_similarity F.cosine_similarity( anchor_proj, positive_proj, dim-1 ) / self.temperature # [batch_size] # 负样本相似度 anchor_expanded anchor_proj.unsqueeze(1) # [batch_size, 1, emb_dim] neg_similarity F.cosine_similarity( anchor_expanded, negative_proj, dim-1 ) / self.temperature # [batch_size, num_negatives]