赞
踩
提示:文章写完后,目录可以自动生成,如何生成可参考右边的帮助文档
大家好,今天要来简单分析下task2的相关概念以及代码,并且记录下自己的学习笔记。task2官方链接
首先,task2仍然使用Seq2Seq模型,但是在数据处理上使用了Spacy对英文分词以及 jieba 对中文分词,同时还使用了Spacy中的预训练语言模型en_core_web_trf。
Spacy是一个NLP领域的文本预处理Python库,包括分词(Tokenization)、词性标注(Part-of-speech Tagging, POS Tagging)、依存分析(Dependency Parsing)、词形还原(Lemmatization)、句子边界检测(Sentence Boundary Detection,SBD)、命名实体识别(Named Entity Recognition, NER)等功能。
Spacy的安装比较方便,只需要执行以下代码即可
pip install spacy -i https://pypi.tuna.tsinghua.edu.cn/simple
然后执行命令下载en_core_Web_trf模型
python -m spacy download en_core_web_trf
不过个人建议直接去github手动下载包
官方链接
随便哪个都行,然后执行
pip install en_core_web_trf-3.7.3-py3-none-any.whl 或者en_core_web_trf-3.7.3.tar.gz
然后我们来尝试加载下模型
import spacy # 导包
# 加载模型
nlp = spacy.load("en_core_web_sm")
# 使用模型,传入句子即可
doc = nlp("Apple is looking at buying U.K. startup for $1 billion")
# 获取分词结果
print([token.text for token in doc])
输出为
['Apple', 'is', 'looking', 'at', 'buying', 'U.K.', 'startup', 'for', '$', '1', 'billion']
如果运行后有像我这样的错误
TypeError: ForwardRef._evaluate() missing 1 required keyword-only argument: 'recursive_guard'
那么可以先卸载,然后直接在pycharm的设置中找到python解释器手动安装,然后下载模型即可。(注意版本需要对应)
另外Spacy也有中文的预训练模型,开头是zh,大家可以自行探索。
所谓数据预处理就对数据集进行清理、标准化和转换,以达到更好的训练效果。这里只列出一些常用的处理方式,具体的需要根据实际来设计。
def __getitem__(self, idx):
en, zh = self.data[idx]
en_tensor = torch.tensor([self.en_word2idx.get(word, self.en_word2idx['<sos>']) for word in self.en_tokenizer(en)] + [self.en_word2idx['<eos>']])
zh_tensor = torch.tensor([self.zh_word2idx.get(word, self.zh_word2idx['<sos>']) for word in self.zh_tokenizer(zh)] + [self.zh_word2idx['<eos>']])
return en_tensor, zh_tensor
GRU编码器和解码器在上一篇中我有简单介绍,这次对RNN模型进行一些补充。
编码器的内容和上一篇中的Seq2Seq模型一样,对于解码器增加两个公式
z
t
z_{t}
zt是解码器的输入,
s
t
s_{t}
st是隐藏状态,公式一表示t时刻的隐藏状态
s
t
s_{t}
st是由解码器通过调用上一时刻t-1的输入
z
t
−
1
z_{t-1}
zt−1和隐藏状态
s
t
−
1
s_{t-1}
st−1进行线性变换通过tanh激活函数生成。也就是说
s
t
s_{t}
st与上一时刻的输入
z
t
−
1
z_{t-1}
zt−1、隐藏状态
s
t
−
1
s_{t-1}
st−1有关。
公式二是在输出层使用Softmax函数对数据进行归一化处理将数据限制在[0,1]之间。
将一个词转换成向量的操作,每个词在词向量表中都有对应的向量表示,我们需要做的就是在表中找到它,然后替换。当然每个词向量的维数都是固定的,一般都是300维,参考自己的词表。这个操作也被叫做Embedding lookup(词嵌入查找)。
当一个句子被输入到计算机时,程序会将每个词视为一个token t ,每个token都有一个词嵌入 A 。但是这些词嵌入没有上下文。注意力机制的思想是应用某种权重或相似性,让初始 词嵌入 A 获得更多上下文信息,从而获得最终带上下文的词嵌入 Y 。
我们用这一思想来找到权重向量 W ,通过将词嵌入相乘(点积)以获得更多上下文。
接下来我们来看看代码,我这里用的是第二个版本的task2代码,第一个版本输出是有问题的。
先导包
import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
from torch.nn.utils import clip_grad_norm_
from torch.utils.data import Dataset, DataLoader
from torchtext.data.utils import get_tokenizer
from torchtext.vocab import build_vocab_from_iterator
from typing import List, Tuple
import jieba
import random
import sacrebleu
import time
import math
进行数据预处理
# 定义tokenizer en_tokenizer = get_tokenizer('spacy', language='en_core_web_trf') zh_tokenizer = lambda x: list(jieba.cut(x)) # 使用jieba分词 # 读取数据函数 def read_data(file_path: str) -> List[str]: with open(file_path, 'r', encoding='utf-8') as f: return [line.strip() for line in f] # 数据预处理函数 def preprocess_data(en_data: List[str], zh_data: List[str]) -> List[Tuple[List[str], List[str]]]: processed_data = [] for en, zh in zip(en_data, zh_data): en_tokens = en_tokenizer(en.lower())[:MAX_LENGTH] zh_tokens = zh_tokenizer(zh)[:MAX_LENGTH] if en_tokens and zh_tokens: # 确保两个序列都不为空 processed_data.append((en_tokens, zh_tokens)) return processed_data # 构建词汇表 def build_vocab(data: List[Tuple[List[str], List[str]]]): en_vocab = build_vocab_from_iterator( (en for en, _ in data), specials=['<unk>', '<pad>', '<bos>', '<eos>'] ) zh_vocab = build_vocab_from_iterator( (zh for _, zh in data), specials=['<unk>', '<pad>', '<bos>', '<eos>'] ) en_vocab.set_default_index(en_vocab['<unk>']) zh_vocab.set_default_index(zh_vocab['<unk>']) return en_vocab, zh_vocab # 处理数据集 class TranslationDataset(Dataset): def __init__(self, data: List[Tuple[List[str], List[str]]], en_vocab, zh_vocab): self.data = data self.en_vocab = en_vocab self.zh_vocab = zh_vocab def __len__(self): return len(self.data) def __getitem__(self, idx): en, zh = self.data[idx] en_indices = [self.en_vocab['<bos>']] + [self.en_vocab[token] for token in en] + [self.en_vocab['<eos>']] zh_indices = [self.zh_vocab['<bos>']] + [self.zh_vocab[token] for token in zh] + [self.zh_vocab['<eos>']] return en_indices, zh_indices
对数据进行填充
def collate_fn(batch): en_batch, zh_batch = [], [] for en_item, zh_item in batch: if en_item and zh_item: # 确保两个序列都不为空 # print("都不为空") en_batch.append(torch.tensor(en_item)) zh_batch.append(torch.tensor(zh_item)) else: print("存在为空") if not en_batch or not zh_batch: # 如果整个批次为空,返回空张量 return torch.tensor([]), torch.tensor([]) # src_sequences = [item[0] for item in batch] # trg_sequences = [item[1] for item in batch] en_batch = nn.utils.rnn.pad_sequence(en_batch, batch_first=True, padding_value=en_vocab['<pad>']) zh_batch = nn.utils.rnn.pad_sequence(zh_batch, batch_first=True, padding_value=zh_vocab['<pad>']) # en_batch = pad_sequence(en_batch, batch_first=True, padding_value=en_vocab['<pad>']) # zh_batch = pad_sequence(zh_batch, batch_first=True, padding_value=zh_vocab['<pad>']) return en_batch, zh_batch
# 数据加载函数 def load_data(train_path: str, dev_en_path: str, dev_zh_path: str, test_en_path: str): # 读取训练数据 train_data = read_data(train_path) train_en, train_zh = zip(*(line.split('\t') for line in train_data)) # 读取开发集和测试集 dev_en = read_data(dev_en_path) dev_zh = read_data(dev_zh_path) test_en = read_data(test_en_path) # 预处理数据 train_processed = preprocess_data(train_en, train_zh) dev_processed = preprocess_data(dev_en, dev_zh) test_processed = [(en_tokenizer(en.lower())[:MAX_LENGTH], []) for en in test_en if en.strip()] # 构建词汇表 global en_vocab, zh_vocab en_vocab, zh_vocab = build_vocab(train_processed) # 创建数据集 train_dataset = TranslationDataset(train_processed, en_vocab, zh_vocab) dev_dataset = TranslationDataset(dev_processed, en_vocab, zh_vocab) test_dataset = TranslationDataset(test_processed, en_vocab, zh_vocab) from torch.utils.data import Subset # 假设你有10000个样本,你只想用前1000个样本进行测试 indices = list(range(N)) train_dataset = Subset(train_dataset, indices) # 创建数据加载器 train_loader = DataLoader(train_dataset, batch_size=BATCH_SIZE, shuffle=True, collate_fn=collate_fn, drop_last=True) dev_loader = DataLoader(dev_dataset, batch_size=BATCH_SIZE, collate_fn=collate_fn, drop_last=True) test_loader = DataLoader(test_dataset, batch_size=BATCH_SIZE, collate_fn=collate_fn, drop_last=True) return train_loader, dev_loader, test_loader, en_vocab, zh_vocab # 数据填充方法 def pad_sequence(batch,batch_first,padding_value): # 对源语言和目标语言序列分别进行填充 src_sequences = [item[0] for item in batch] trg_sequences = [item[1] for item in batch] src_padded = nn.utils.rnn.pad_sequence(src_sequences, batch_first=True, padding_value=en_vocab['<pad>']) trg_padded = nn.utils.rnn.pad_sequence(trg_sequences, batch_first=True, padding_value=zh_vocab['<pad>']) return src_padded, trg_padded
构建模型
class Encoder(nn.Module): def __init__(self, input_dim, emb_dim, hid_dim, n_layers, dropout): super().__init__() self.hid_dim = hid_dim self.n_layers = n_layers self.embedding = nn.Embedding(input_dim, emb_dim) self.gru = nn.GRU(emb_dim, hid_dim, n_layers, dropout=dropout, batch_first=True) self.dropout = nn.Dropout(dropout) def forward(self, src): # src = [batch size, src len] embedded = self.dropout(self.embedding(src)) # embedded = [batch size, src len, emb dim] outputs, hidden = self.gru(embedded) # outputs = [batch size, src len, hid dim * n directions] # hidden = [n layers * n directions, batch size, hid dim] return outputs, hidden # 注意力机制实现 class Attention(nn.Module): def __init__(self, hid_dim): super().__init__() self.attn = nn.Linear(hid_dim * 2, hid_dim) self.v = nn.Linear(hid_dim, 1, bias=False) def forward(self, hidden, encoder_outputs): # hidden = [1, batch size, hid dim] # encoder_outputs = [batch size, src len, hid dim] batch_size = encoder_outputs.shape[0] src_len = encoder_outputs.shape[1] hidden = hidden.repeat(src_len, 1, 1).transpose(0, 1) # hidden = [batch size, src len, hid dim] energy = torch.tanh(self.attn(torch.cat((hidden, encoder_outputs), dim=2))) # energy = [batch size, src len, hid dim] attention = self.v(energy).squeeze(2) # attention = [batch size, src len] return F.softmax(attention, dim=1) class Decoder(nn.Module): def __init__(self, output_dim, emb_dim, hid_dim, n_layers, dropout, attention): super().__init__() self.output_dim = output_dim self.hid_dim = hid_dim self.n_layers = n_layers self.attention = attention self.embedding = nn.Embedding(output_dim, emb_dim) self.gru = nn.GRU(hid_dim + emb_dim, hid_dim, n_layers, dropout=dropout, batch_first=True) self.fc_out = nn.Linear(hid_dim * 2 + emb_dim, output_dim) self.dropout = nn.Dropout(dropout) def forward(self, input, hidden, encoder_outputs): # input = [batch size, 1] # hidden = [n layers, batch size, hid dim] # encoder_outputs = [batch size, src len, hid dim] input = input.unsqueeze(1) embedded = self.dropout(self.embedding(input)) # embedded = [batch size, 1, emb dim] a = self.attention(hidden[-1:], encoder_outputs) # a = [batch size, src len] a = a.unsqueeze(1) # a = [batch size, 1, src len] weighted = torch.bmm(a, encoder_outputs) # weighted = [batch size, 1, hid dim] rnn_input = torch.cat((embedded, weighted), dim=2) # rnn_input = [batch size, 1, emb dim + hid dim] output, hidden = self.gru(rnn_input, hidden) # output = [batch size, 1, hid dim] # hidden = [n layers, batch size, hid dim] embedded = embedded.squeeze(1) output = output.squeeze(1) weighted = weighted.squeeze(1) prediction = self.fc_out(torch.cat((output, weighted, embedded), dim=1)) # prediction = [batch size, output dim] return prediction, hidden class Seq2Seq(nn.Module): def __init__(self, encoder, decoder, device): super().__init__() self.encoder = encoder self.decoder = decoder self.device = device def forward(self, src, trg, teacher_forcing_ratio=0.5): # src = [batch size, src len] # trg = [batch size, trg len] batch_size = src.shape[0] trg_len = trg.shape[1] trg_vocab_size = self.decoder.output_dim outputs = torch.zeros(batch_size, trg_len, trg_vocab_size).to(self.device) encoder_outputs, hidden = self.encoder(src) input = trg[:, 0] for t in range(1, trg_len): output, hidden = self.decoder(input, hidden, encoder_outputs) outputs[:, t] = output teacher_force = random.random() < teacher_forcing_ratio top1 = output.argmax(1) input = trg[:, t] if teacher_force else top1 return outputs
# 初始化模型
def initialize_model(input_dim, output_dim, emb_dim, hid_dim, n_layers, dropout, device):
attn = Attention(hid_dim)
enc = Encoder(input_dim, emb_dim, hid_dim, n_layers, dropout)
dec = Decoder(output_dim, emb_dim, hid_dim, n_layers, dropout, attn)
model = Seq2Seq(enc, dec, device).to(device)
return model
定义优化器,这里使用的是固定学习率,也可以使用动态学习率来得到更好的效果
# 定义优化器 def initialize_optimizer(model, learning_rate=0.001): return optim.Adam(model.parameters(), lr=learning_rate) # 运行时间 def epoch_time(start_time, end_time): elapsed_time = end_time - start_time elapsed_mins = int(elapsed_time / 60) elapsed_secs = int(elapsed_time - (elapsed_mins * 60)) return elapsed_mins, elapsed_secs def train(model, iterator, optimizer, criterion, clip): model.train() epoch_loss = 0 for i, batch in enumerate(iterator): #print(f"Training batch {i}") src, trg = batch #print(f"Source shape before: {src.shape}, Target shape before: {trg.shape}") if src.numel() == 0 or trg.numel() == 0: #print("Empty batch detected, skipping...") continue # 跳过空的批次 src, trg = src.to(DEVICE), trg.to(DEVICE) optimizer.zero_grad() output = model(src, trg) output_dim = output.shape[-1] output = output[:, 1:].contiguous().view(-1, output_dim) trg = trg[:, 1:].contiguous().view(-1) loss = criterion(output, trg) loss.backward() clip_grad_norm_(model.parameters(), clip) optimizer.step() epoch_loss += loss.item() print(f"Average loss for this epoch: {epoch_loss / len(iterator)}") return epoch_loss / len(iterator) def evaluate(model, iterator, criterion): model.eval() epoch_loss = 0 with torch.no_grad(): for i, batch in enumerate(iterator): #print(f"Evaluating batch {i}") src, trg = batch if src.numel() == 0 or trg.numel() == 0: continue # 跳过空批次 src, trg = src.to(DEVICE), trg.to(DEVICE) output = model(src, trg, 0) # 关闭 teacher forcing output_dim = output.shape[-1] output = output[:, 1:].contiguous().view(-1, output_dim) trg = trg[:, 1:].contiguous().view(-1) loss = criterion(output, trg) epoch_loss += loss.item() return epoch_loss / len(iterator) # 翻译函数 def translate_sentence(src_indexes, src_vocab, tgt_vocab, model, device, max_length=50): model.eval() src_tensor = src_indexes.unsqueeze(0).to(device) # 添加批次维度 # with torch.no_grad(): # encoder_outputs = model.encoder(model.positional_encoding(model.src_embedding(src_tensor) * math.sqrt(model.d_model))) trg_indexes = [tgt_vocab['<bos>']] for i in range(max_length): trg_tensor = torch.LongTensor(trg_indexes).unsqueeze(0).to(device) # print("src_tensor:",src_tensor) # print("trg_tensor:",trg_tensor) with torch.no_grad(): output = model(src_tensor, trg_tensor) pred_token = output.argmax(2)[:, -1].item() trg_indexes.append(pred_token) if pred_token == tgt_vocab['<eos>']: break trg_tokens = [tgt_vocab.get_itos()[i] for i in trg_indexes] return trg_tokens[1:-1] # 移除<bos>和<eos>标记 # 主训练循环 def train_model(model, train_iterator, valid_iterator, optimizer, criterion, N_EPOCHS = 10, CLIP = 1, save_path = '../model/best-model.pt'): best_valid_loss = float('inf') for epoch in range(N_EPOCHS): start_time = time.time() #print(f"Starting Epoch {epoch + 1}") train_loss = train(model, train_iterator, optimizer, criterion, CLIP) valid_loss = evaluate(model, valid_iterator, criterion) end_time = time.time() epoch_mins, epoch_secs = epoch_time(start_time, end_time) if valid_loss < best_valid_loss: best_valid_loss = valid_loss torch.save(model.state_dict(), save_path) print(f'Epoch: {epoch+1:02} | Time: {epoch_mins}m {epoch_secs}s') print(f'\tTrain Loss: {train_loss:.3f} | Train PPL: {math.exp(train_loss):7.3f}') print(f'\t Val. Loss: {valid_loss:.3f} | Val. PPL: {math.exp(valid_loss):7.3f}')
开始训练,注意这里DEVICE中的cuda要设置为0,还有BATCH_SIZE要修改成1,这里设置最大句子长度为100是因为数据集中的句子统计结果有90%长度都小于100,当然如果想覆盖所有句子,可以对长度不够的句子做填充,来保证数据长度统一。
另外,本次训练保存的是最优模型,防止中间训练时有不好的循环出现影响最后训练效果。
# 定义常量 MAX_LENGTH = 100 # 最大句子长度 BATCH_SIZE = 1 DEVICE = torch.device('cuda:0' if torch.cuda.is_available() else 'cpu') N = 2000 # 采样训练集的数量 train_path = '../dataset/train.txt' dev_en_path = '../dataset/dev_en.txt' dev_zh_path = '../dataset/dev_zh.txt' test_en_path = '../dataset/test_en.txt' train_loader, dev_loader, test_loader, en_vocab, zh_vocab = load_data( train_path, dev_en_path, dev_zh_path, test_en_path ) print(f"英语词汇表大小: {len(en_vocab)}") print(f"中文词汇表大小: {len(zh_vocab)}") print(f"训练集大小: {len(train_loader.dataset)}") print(f"开发集大小: {len(dev_loader.dataset)}") print(f"测试集大小: {len(test_loader.dataset)}") # 主函数 if __name__ == '__main__': N_EPOCHS = 30 CLIP=1 # 模型参数 INPUT_DIM = len(en_vocab) OUTPUT_DIM = len(zh_vocab) EMB_DIM = 128 HID_DIM = 256 N_LAYERS = 2 DROPOUT = 0.5 # 初始化模型 model = initialize_model(INPUT_DIM, OUTPUT_DIM, EMB_DIM, HID_DIM, N_LAYERS, DROPOUT, DEVICE) print(f'The model has {sum(p.numel() for p in model.parameters() if p.requires_grad):,} trainable parameters') # 定义损失函数 criterion = nn.CrossEntropyLoss(ignore_index=zh_vocab['<pad>']) # 初始化优化器 optimizer = initialize_optimizer(model) # 训练模型 save_path = '../model/best-model.pt' train_model(model, train_loader, dev_loader, optimizer, criterion, N_EPOCHS, CLIP, save_path = save_path) print(f"训练完成!模型已保存到:{save_path}")
在测试集上进行翻译试试
save_dir = '../results/submit_task2.txt'
with open(save_dir, 'w') as f:
translated_sentences = []
for batch in test_loader: # 遍历所有数据
src, _ = batch
src = src.to(DEVICE)
translated = translate_sentence(src[0], en_vocab, zh_vocab, model, DEVICE, max_length=50) # 翻译结果,max_length生成翻译的最大长度
#print(translated)
results = "".join(translated)
f.write(results + '\n') # 将结果写入文件
print(f"翻译完成,结果已保存到{save_dir}")
我们看下结果
可以看到在接近20轮的时候损失已经基本稳定在3.3左右了。
而BLEU-4的分数比task1高了不少
翻译的结果还是一个人工智障((lll¬ω¬))。。
让我们来看看讯飞的评分
甚至比task1还要低了/(ㄒoㄒ)/~~
本文就先到这里,具体是什么原因导致的评分很低我还没研究出来,等研究出来了再做更新。
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。