赞
踩
从零入门NLP竞赛 是 Datawhale 2024 年 AI 夏令营第二期的学习活动(“NLP”方向),基于讯飞开放平台“基于术语词典干预的机器翻译挑战赛”开展的实践学习——适合想 入门并实践 深度学习、解决NLP问题、机器翻译 的学习者参与
本赛题是一个经典的NLP问题,赛事目标为——【通过术语词典构建机器翻译模型】
在特定领域或行业中,由于机器翻译难以保证术语的一致性,导致翻译效果还不够理想。对于术语名词、人名地名等机器翻译不准确的结果,可以通过术语词典进行纠正,避免了混淆或歧义,最大限度提高翻译质量。
自然语言处理(Natural Language Processing,NLP)是语言学与人工智能的分支,试图让计算机能够完成处理语言、理解语言和生成语言等任务。
大致可以将NLP 任务分为四类:
序列标注:比如中文分词,词性标注,命名实体识别,语义角色标注等都可以归入这一类问题。这类任务的共同点是句子中每个单词要求模型根据上下文都要给出一个分类类别;
分类任务:比如我们常见的文本分类,情感计算等都可以归入这一类。这类任务特点是不管文章有多长,总体给出一个分类类别即可;
句子关系判断:比如问答推理,语义改写,自然语言推理等任务都是这个模式,它的特点是给定两个句子,模型判断出两个句子是否具备某种语义关系;
生成式任务:比如机器翻译,文本摘要,写诗造句,看图说话等都属于这一类。它的特点是输入文本内容后,需要自主生成另外一段文字。
训练集(Training Set):
作用:训练集用于训练模型,使模型能够学习输入数据与输出结果之间的映射关系。模型会根据训练集中的样本调整其参数,以最小化预测误差。
目标:让模型在训练数据上尽可能地拟合好,学习到数据的内在规律。
开发集/验证集(Development/Validation Set):
作用:开发集用于在模型训练过程中调整超参数、选择模型架构以及防止过拟合。它作为独立于训练集的数据,用于评估模型在未见过的数据上的表现。
目标:通过在开发集上的性能评估,选择最佳的模型配置,避免模型在训练集上过度拟合,确保模型的泛化能力。
测试集(Test Set):
作用:测试集用于最终评估模型的性能,是在模型训练和调参完全完成后,用来衡量模型实际应用效果的一组数据。它是最接近真实世界数据的评估标准。
目标:提供一个公正、无偏见的性能估计,反映模型在未知数据上的泛化能力。
Datawhale有一个NLP开源项目
对于参赛队伍提交的测试集翻译结果文件,采用自动评价指标 BLEU-4 进行评价,具体工具使用 sacrebleu开源版本。
在机器翻译领域,BLEU(Bilingual Evaluation Understudy)是一种常用的自动评价指标,用于衡量计算机生成的翻译与一组参考译文之间的相似度。这个指标特别关注 n-grams(连续的n个词)的精确匹配,可以被认为是对翻译准确性和流利度的一种统计估计。计算BLEU分数时,首先会统计生成文本中n-grams的频率,然后将这些频率与参考文本中的n-grams进行比较。如果生成的翻译中包含的n-grams与参考译文中出现的相同,则认为是匹配的。最终的BLEU分数是一个介于0到1之间的数值,其中1表示与参考译文完美匹配,而0则表示完全没有匹配。
BLEU-4 特别指的是在计算时考虑四元组(即连续四个词)的匹配情况。
BLEU 评估指标的特点:
优点:计算速度快、计算成本低、容易理解、与具体语言无关、和人类给的评估高度相关。
缺点:不考虑语言表达(语法)上的准确性;测评精度会受常用词的干扰;短译句的测评精度有时会较高;没有考虑同义词或相似表达的情况,可能会导致合理翻译被否定。
除了翻译之外,BLEU评分结合深度学习方法可应用于其他的语言生成问题,例如:语言生成、图片标题生成、文本摘要、语音识别。
- !pip install torchtext
- import torch
- import torch.nn as nn
- import torch.optim as optim
- from torch.utils.data import Dataset, DataLoader
- from torchtext.data.utils import get_tokenizer
- from collections import Counter
- import random
- from torch.utils.data import Subset, DataLoader
- import time
-
- # 定义数据集类
- # 修改TranslationDataset类以处理术语
- class TranslationDataset(Dataset):
- def __init__(self, filename, terminology):
- self.data = []
- with open(filename, 'r', encoding='utf-8') as f:
- for line in f:
- en, zh = line.strip().split('\t')
- self.data.append((en, zh))
-
- self.terminology = terminology
-
- # 创建词汇表,注意这里需要确保术语词典中的词也被包含在词汇表中
- self.en_tokenizer = get_tokenizer('basic_english')
- self.zh_tokenizer = list # 使用字符级分词
-
- en_vocab = Counter(self.terminology.keys()) # 确保术语在词汇表中
- zh_vocab = Counter()
-
- for en, zh in self.data:
- en_vocab.update(self.en_tokenizer(en))
- zh_vocab.update(self.zh_tokenizer(zh))
-
- # 添加术语到词汇表
- self.en_vocab = ['<pad>', '<sos>', '<eos>'] + list(self.terminology.keys()) + [word for word, _ in en_vocab.most_common(10000)]
- self.zh_vocab = ['<pad>', '<sos>', '<eos>'] + [word for word, _ in zh_vocab.most_common(10000)]
-
- self.en_word2idx = {word: idx for idx, word in enumerate(self.en_vocab)}
- self.zh_word2idx = {word: idx for idx, word in enumerate(self.zh_vocab)}
-
-
- def __len__(self):
- return len(self.data)
-
- 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
-
- def collate_fn(batch):
- en_batch, zh_batch = [], []
- for en_item, zh_item in batch:
- en_batch.append(en_item)
- zh_batch.append(zh_item)
-
- # 对英文和中文序列分别进行填充
- en_batch = nn.utils.rnn.pad_sequence(en_batch, padding_value=0, batch_first=True)
- zh_batch = nn.utils.rnn.pad_sequence(zh_batch, padding_value=0, batch_first=True)
-
- return en_batch, zh_batch
-
- class Encoder(nn.Module):
- def __init__(self, input_dim, emb_dim, hid_dim, n_layers, dropout):
- super().__init__()
- self.embedding = nn.Embedding(input_dim, emb_dim)
- self.rnn = nn.GRU(emb_dim, hid_dim, n_layers, dropout=dropout, batch_first=True)
- self.dropout = nn.Dropout(dropout)
-
- def forward(self, src):
- # src shape: [batch_size, src_len]
- embedded = self.dropout(self.embedding(src))
- # embedded shape: [batch_size, src_len, emb_dim]
- outputs, hidden = self.rnn(embedded)
- # outputs shape: [batch_size, src_len, hid_dim]
- # hidden shape: [n_layers, batch_size, hid_dim]
- return outputs, hidden
-
- class Decoder(nn.Module):
- def __init__(self, output_dim, emb_dim, hid_dim, n_layers, dropout):
- super().__init__()
- self.output_dim = output_dim
- self.embedding = nn.Embedding(output_dim, emb_dim)
- self.rnn = nn.GRU(emb_dim, hid_dim, n_layers, dropout=dropout, batch_first=True)
- self.fc_out = nn.Linear(hid_dim, output_dim)
- self.dropout = nn.Dropout(dropout)
-
- def forward(self, input, hidden):
- # input shape: [batch_size, 1]
- # hidden shape: [n_layers, batch_size, hid_dim]
-
- embedded = self.dropout(self.embedding(input))
- # embedded shape: [batch_size, 1, emb_dim]
-
- output, hidden = self.rnn(embedded, hidden)
- # output shape: [batch_size, 1, hid_dim]
- # hidden shape: [n_layers, batch_size, hid_dim]
-
- prediction = self.fc_out(output.squeeze(1))
- # prediction shape: [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 shape: [batch_size, src_len]
- # trg shape: [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)
-
- _, hidden = self.encoder(src)
-
- input = trg[:, 0].unsqueeze(1) # Start token
-
- for t in range(1, trg_len):
- output, hidden = self.decoder(input, hidden)
- outputs[:, t, :] = output
- teacher_force = random.random() < teacher_forcing_ratio
- top1 = output.argmax(1)
- input = trg[:, t].unsqueeze(1) if teacher_force else top1.unsqueeze(1)
-
- return outputs
-
- # 新增术语词典加载部分
- def load_terminology_dictionary(dict_file):
- terminology = {}
- with open(dict_file, 'r', encoding='utf-8') as f:
- for line in f:
- en_term, ch_term = line.strip().split('\t')
- terminology[en_term] = ch_term
- return terminology
-
- def train(model, iterator, optimizer, criterion, clip):
- model.train()
- epoch_loss = 0
- for i, (src, trg) in enumerate(iterator):
- 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()
- torch.nn.utils.clip_grad_norm_(model.parameters(), clip)
- optimizer.step()
- epoch_loss += loss.item()
- return epoch_loss / len(iterator)
-
- # 主函数
- if __name__ == '__main__':
- start_time = time.time() # 开始计时
-
- device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
-
- #terminology = load_terminology_dictionary('../dataset/en-zh.dic')
- terminology = load_terminology_dictionary('../dataset/en-zh.dic')
-
- # 加载数据
- dataset = TranslationDataset('../dataset/train.txt',terminology = terminology)
- # 选择数据集的前N个样本进行训练
- N = 1000 #int(len(dataset) * 1) # 或者你可以设置为数据集大小的一定比例,如 int(len(dataset) * 0.1)
- subset_indices = list(range(N))
- subset_dataset = Subset(dataset, subset_indices)
- train_loader = DataLoader(subset_dataset, batch_size=32, shuffle=True, collate_fn=collate_fn)
-
- # 定义模型参数
- INPUT_DIM = len(dataset.en_vocab)
- OUTPUT_DIM = len(dataset.zh_vocab)
- ENC_EMB_DIM = 256
- DEC_EMB_DIM = 256
- HID_DIM = 512
- N_LAYERS = 2
- ENC_DROPOUT = 0.5
- DEC_DROPOUT = 0.5
-
- # 初始化模型
- enc = Encoder(INPUT_DIM, ENC_EMB_DIM, HID_DIM, N_LAYERS, ENC_DROPOUT)
- dec = Decoder(OUTPUT_DIM, DEC_EMB_DIM, HID_DIM, N_LAYERS, DEC_DROPOUT)
- model = Seq2Seq(enc, dec, device).to(device)
-
- # 定义优化器和损失函数
- optimizer = optim.Adam(model.parameters())
- criterion = nn.CrossEntropyLoss(ignore_index=dataset.zh_word2idx['<pad>'])
-
- # 训练模型
- N_EPOCHS = 10
- CLIP = 1
-
- for epoch in range(N_EPOCHS):
- train_loss = train(model, train_loader, optimizer, criterion, CLIP)
- print(f'Epoch: {epoch+1:02} | Train Loss: {train_loss:.3f}')
-
- # 在训练循环结束后保存模型
- torch.save(model.state_dict(), './translation_model_GRU.pth')
-
- end_time = time.time() # 结束计时
-
- # 计算并打印运行时间
- elapsed_time_minute = (end_time - start_time)/60
- print(f"Total running time: {elapsed_time_minute:.2f} minutes")

- import torch
- from sacrebleu.metrics import BLEU
- from typing import List
-
- # 假设我们已经定义了TranslationDataset, Encoder, Decoder, Seq2Seq类
-
- def load_sentences(file_path: str) -> List[str]:
- with open(file_path, 'r', encoding='utf-8') as f:
- return [line.strip() for line in f]
-
- # 更新translate_sentence函数以考虑术语词典
- def translate_sentence(sentence: str, model: Seq2Seq, dataset: TranslationDataset, terminology, device: torch.device, max_length: int = 50):
- model.eval()
- tokens = dataset.en_tokenizer(sentence)
- tensor = torch.LongTensor([dataset.en_word2idx.get(token, dataset.en_word2idx['<sos>']) for token in tokens]).unsqueeze(0).to(device) # [1, seq_len]
-
- with torch.no_grad():
- _, hidden = model.encoder(tensor)
-
- translated_tokens = []
- input_token = torch.LongTensor([[dataset.zh_word2idx['<sos>']]]).to(device) # [1, 1]
-
- for _ in range(max_length):
- output, hidden = model.decoder(input_token, hidden)
- top_token = output.argmax(1)
- translated_token = dataset.zh_vocab[top_token.item()]
-
- if translated_token == '<eos>':
- break
-
- # 如果翻译的词在术语词典中,则使用术语词典中的词
- if translated_token in terminology.values():
- for en_term, ch_term in terminology.items():
- if translated_token == ch_term:
- translated_token = en_term
- break
-
- translated_tokens.append(translated_token)
- input_token = top_token.unsqueeze(1) # [1, 1]
-
- return ''.join(translated_tokens)
-
- def evaluate_bleu(model: Seq2Seq, dataset: TranslationDataset, src_file: str, ref_file: str, terminology,device: torch.device):
- model.eval()
- src_sentences = load_sentences(src_file)
- ref_sentences = load_sentences(ref_file)
-
- translated_sentences = []
- for src in src_sentences:
- translated = translate_sentence(src, model, dataset, terminology, device)
- translated_sentences.append(translated)
-
- bleu = BLEU()
- score = bleu.corpus_score(translated_sentences, [ref_sentences])
-
- return score
-
- # 主函数
- if __name__ == '__main__':
- device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
-
- # 加载术语词典
- terminology = load_terminology_dictionary('../dataset/en-zh.dic')
-
- # 创建数据集实例时传递术语词典
- dataset = TranslationDataset('../dataset/train.txt', terminology)
-
-
- # 定义模型参数
- INPUT_DIM = len(dataset.en_vocab)
- OUTPUT_DIM = len(dataset.zh_vocab)
- ENC_EMB_DIM = 256
- DEC_EMB_DIM = 256
- HID_DIM = 512
- N_LAYERS = 2
- ENC_DROPOUT = 0.5
- DEC_DROPOUT = 0.5
-
- # 初始化模型
- enc = Encoder(INPUT_DIM, ENC_EMB_DIM, HID_DIM, N_LAYERS, ENC_DROPOUT)
- dec = Decoder(OUTPUT_DIM, DEC_EMB_DIM, HID_DIM, N_LAYERS, DEC_DROPOUT)
- model = Seq2Seq(enc, dec, device).to(device)
-
- # 加载训练好的模型
- model.load_state_dict(torch.load('./translation_model_GRU.pth'))
-
- # 评估BLEU分数
- bleu_score = evaluate_bleu(model, dataset, '../dataset/dev_en.txt', '../dataset/dev_zh.txt', terminology = terminology,device = device)
- print(f'BLEU-4 score: {bleu_score.score:.2f}')

- def inference(model: Seq2Seq, dataset: TranslationDataset, src_file: str, save_dir:str, terminology, device: torch.device):
- model.eval()
- src_sentences = load_sentences(src_file)
-
- translated_sentences = []
- for src in src_sentences:
- translated = translate_sentence(src, model, dataset, terminology, device)
- #print(translated)
- translated_sentences.append(translated)
- #print(translated_sentences)
-
- # 将列表元素连接成一个字符串,每个元素后换行
- text = '\n'.join(translated_sentences)
-
- # 打开一个文件,如果不存在则创建,'w'表示写模式
- with open(save_dir, 'w', encoding='utf-8') as f:
- # 将字符串写入文件
- f.write(text)
-
- #return translated_sentences
-
- # 主函数
- if __name__ == '__main__':
- device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
-
- # 加载术语词典
- terminology = load_terminology_dictionary('../dataset/en-zh.dic')
- # 加载数据集和模型
- dataset = TranslationDataset('../dataset/train.txt',terminology = terminology)
-
- # 定义模型参数
- INPUT_DIM = len(dataset.en_vocab)
- OUTPUT_DIM = len(dataset.zh_vocab)
- ENC_EMB_DIM = 256
- DEC_EMB_DIM = 256
- HID_DIM = 512
- N_LAYERS = 2
- ENC_DROPOUT = 0.5
- DEC_DROPOUT = 0.5
-
- # 初始化模型
- enc = Encoder(INPUT_DIM, ENC_EMB_DIM, HID_DIM, N_LAYERS, ENC_DROPOUT)
- dec = Decoder(OUTPUT_DIM, DEC_EMB_DIM, HID_DIM, N_LAYERS, DEC_DROPOUT)
- model = Seq2Seq(enc, dec, device).to(device)
-
- # 加载训练好的模型
- model.load_state_dict(torch.load('./translation_model_GRU.pth'))
-
- save_dir = '../dataset/submit.txt'
- inference(model, dataset, src_file="../dataset/test_en.txt", save_dir = save_dir, terminology = terminology, device = device)
- print(f"翻译完成!文件已保存到{save_dir}")

- import torch
- import torch.nn as nn
- import torch.optim as optim
- from torch.utils.data import Dataset, DataLoader
- from torchtext.data.utils import get_tokenizer
- from collections import Counter
- import random
- from torch.utils.data import Subset, DataLoader
- import time
- from sacrebleu.metrics import BLEU
- from typing import List
重点使用的库:
torch
:主要用于张量操作和神经网络模型的构建与训练Subset
:用于从数据集中选择子集get_tokenizer
:用于获取文本分词器的函数Counter
:用于统计词频- class TranslationDataset(Dataset):
- def __init__(self, filename, terminology):
- # 初始化数据集,包括加载数据、创建词汇表等
- pass
-
- def __len__(self):
- # 返回数据集大小
- pass
-
- def __getitem__(self, idx):
- # 按索引返回数据对的张量表示
- pass
__len__
方法返回数据集大小,__getitem__
方法按索引返回数据对的张量表示。- def collate_fn(batch):
- # 对批次中的数据对进行填充
- pass
nn.utils.rnn.pad_sequence
函数进行填充操作。- class Encoder(nn.Module):
- def __init__(self, input_dim, emb_dim, hid_dim, n_layers, dropout):
- # 初始化编码器,包括词嵌入、GRU层和Dropout
- pass
-
- def forward(self, src):
- # 编码器的前向传播过程
- pass
nn.Embedding
层将输入序列索引映射为词嵌入。nn.GRU
层进行序列编码,并应用了 Dropout
进行正则化。- class Decoder(nn.Module):
- def __init__(self, output_dim, emb_dim, hid_dim, n_layers, dropout):
- # 初始化解码器,包括词嵌入、GRU层、线性层和Dropout
- pass
-
- def forward(self, input, hidden):
- # 解码器的前向传播过程
- pass
nn.Embedding
层将目标语言序列索引映射为词嵌入。nn.GRU
层进行序列解码,并应用了 Dropout
进行正则化。- class Seq2Seq(nn.Module):
- def __init__(self, encoder, decoder, device):
- # 初始化Seq2Seq模型,包括编码器、解码器和设备
- pass
-
- def forward(self, src, trg, teacher_forcing_ratio=0.5):
- # Seq2Seq模型的前向传播过程
- pass
teacher_forcing_ratio
控制是否使用教师强制(teacher forcing)。- def load_terminology_dictionary(dict_file):
- # 从文件加载术语词典,并返回一个字典
- pass
- def train(model, iterator, optimizer, criterion, clip):
- # 模型训练函数,包括前向传播、损失计算、反向传播和参数更新
- pass
- def translate_sentence(sentence, model, dataset, terminology, device, max_length=50):
- # 使用训练好的模型翻译单个句子
- pass
- def evaluate_bleu(model, dataset, src_file, ref_file, terminology, device):
- # 评估模型在指定数据集上的BLEU分数
- pass
- def inference(model, dataset, src_file, save_dir, terminology, device):
- # 对指定文件中的所有句子进行翻译,并将结果保存到文件
- pass
- if __name__ == '__main__':
- # 主函数,包括模型训练、加载、评估或推理等整体流程
- pass
- # 主函数
- if __name__ == '__main__':
- start_time = time.time() # 开始计时
-
- device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
-
- #terminology = load_terminology_dictionary('../dataset/en-zh.dic')
- terminology = load_terminology_dictionary('../dataset/en-zh.dic')
-
- # 加载数据
- dataset = TranslationDataset('../dataset/train.txt',terminology = terminology)
- # 选择数据集的前N个样本进行训练
- N = 2000 #int(len(dataset) * 1) # 或者你可以设置为数据集大小的一定比例,如 int(len(dataset) * 0.1)
- subset_indices = list(range(N))
- subset_dataset = Subset(dataset, subset_indices)
- train_loader = DataLoader(subset_dataset, batch_size=32, shuffle=True, collate_fn=collate_fn)
-
- # 定义模型参数
- INPUT_DIM = len(dataset.en_vocab)
- OUTPUT_DIM = len(dataset.zh_vocab)
- ENC_EMB_DIM = 256
- DEC_EMB_DIM = 256
- HID_DIM = 512
- N_LAYERS = 2
- ENC_DROPOUT = 0.5
- DEC_DROPOUT = 0.5
-
- # 初始化模型
- enc = Encoder(INPUT_DIM, ENC_EMB_DIM, HID_DIM, N_LAYERS, ENC_DROPOUT)
- dec = Decoder(OUTPUT_DIM, DEC_EMB_DIM, HID_DIM, N_LAYERS, DEC_DROPOUT)
- model = Seq2Seq(enc, dec, device).to(device)
-
- # 定义优化器和损失函数
- optimizer = optim.Adam(model.parameters())
- criterion = nn.CrossEntropyLoss(ignore_index=dataset.zh_word2idx['<pad>'])
-
- # 训练模型
- N_EPOCHS = 50
- CLIP = 1
-
- for epoch in range(N_EPOCHS):
- train_loss = train(model, train_loader, optimizer, criterion, CLIP)
- print(f'Epoch: {epoch+1:02} | Train Loss: {train_loss:.3f}')
-
- # 在训练循环结束后保存模型
- torch.save(model.state_dict(), './translation_model_GRU.pth')
-
- end_time = time.time() # 结束计时
-
- # 计算并打印运行时间
- elapsed_time_minute = (end_time - start_time)/60
- print(f"Total running time: {elapsed_time_minute:.2f} minutes")

可以注意到修改了N和N_EPOCHS两个参数。
两者作用是将数据集中前N个样本抓取训练了N_EPOCHS轮。
hahaha都看到这里了,要是觉得有用的话就辛苦动动小手点个赞吧!
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。