2024Datawhale AI夏令营---基于术语词典干预的机器翻译挑战赛--学习笔记

#Datawhale #NLP

1.背景介绍：

机器翻译（Machine Translation，简称MT）是自然语言处理领域的一个重要分支，其目标是将一种语言的文本自动转换为另一种语言的文本。机器翻译的发展可以追溯到20世纪50年代，经历了从基于规则的方法、统计方法到深度学习方法的演变过程。

当前，机器翻译正朝着更加智能化和个性化方向发展。一方面，结合上下文理解、情感分析等技术，提高翻译的准确性和自然度；另一方面，通过用户反馈和个性化学习，提供更加符合用户需求的翻译服务。同时，跨语言信息检索、多模态翻译等新兴领域也正在成为研究热点。

总的来说，机器翻译的发展历程是从规则驱动到数据驱动，再到智能驱动的过程，反映了自然语言处理技术的进步和应用需求的变化。

2.数据集简介：

基于术语词典干预的机器翻译挑战赛选择以英文为源语言，中文为目标语言的机器翻译。本次大赛除英文到中文的双语数据，还提供英中对照的术语词典。参赛队伍需要基于提供的训练数据样本从多语言机器翻译模型的构建与训练，并基于测试集以及术语词典，提供最终的翻译结果

赛题数据划分：

训练集：双语数据 - 中英14万余双语句对
开发集：英中1000双语句对
测试集：英中1000双语句对
术语词典：英中2226条
训练集（training set）用于运行你的学习算法。

开发集（development set）用于调整参数，选择特征，以及对学习算法作出其它决定。有时也称为留出交叉验证集（hold-out cross validation set）。
测试集（test set）用于评估算法的性能，但不会据此改变学习算法或参数。

下面是每日任务打卡以及学习笔记：

task1:了解机器翻译 & 理解赛题

跑通baseline并上传数据：

baseline主要分为三个板块：模型训练，模型推理（验证集），模型推理（测试集）

模型训练：

值得注意的是，这里其实用的jupternotebook写的代码，就是一块一块的，一块定义一个函数并且写上notebook笔记注释，在最后一块写上if name主函数调用运行即可，运行的时候切记要将所有的代码块从头到尾一个个运行一遍，才能运行主函数，否则前面的代码编译器未编译，主函数里面调用的函数识别不出来。

这里我是图方便，全部合在一起了，到时候直接运行就好，和jupyter那个要从头到尾跑一遍代码块一样的道理，这就是python的特质，没有什么引入接口函数，必须要从头到尾来一遍才能读进去。（maybe）

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('E:/2024年项目资料/2024Datawhale AI夏令营---基于术语词典干预的机器翻译挑战赛/dataset/en-zh.dic')

    # 加载数据
    dataset = TranslationDataset('E:/2024年项目资料/2024Datawhale AI夏令营---基于术语词典干预的机器翻译挑战赛/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(), 'E:/2024年项目资料/2024Datawhale AI夏令营---基于术语词典干预的机器翻译挑战赛/model/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")

直接运行主函数：

结果如下：这里我是按默认的参数只跑了10个epoch,训练了1000对中英翻译对，效果其实是会很差的。

模型推理（验证集）：

这里是用验证集先对模型进行一次评测，大家都懂验证集是用来干嘛的，就是拿来给模型调参的，这里只是一个演示，跑通了后面会发挥其作用用于调参。

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('E:/2024年项目资料/2024Datawhale AI夏令营---基于术语词典干预的机器翻译挑战赛/dataset/en-zh.dic')
    
    # 创建数据集实例时传递术语词典
    dataset = TranslationDataset('E:/2024年项目资料/2024Datawhale AI夏令营---基于术语词典干预的机器翻译挑战赛/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('E:/2024年项目资料/2024Datawhale AI夏令营---基于术语词典干预的机器翻译挑战赛/model/translation_model_GRU.pth'))

    # 评估BLEU分数
    bleu_score = evaluate_bleu(model, dataset, 'E:/2024年项目资料/2024Datawhale AI夏令营---基于术语词典干预的机器翻译挑战赛/dataset/dev_en.txt', 'E:/2024年项目资料/2024Datawhale AI夏令营---基于术语词典干预的机器翻译挑战赛/dataset/dev_zh.txt', terminology = terminology,device = device)
    print(f'BLEU-4 score: {bleu_score.score:.2f}')

这里验证集用的评分指标是自动评价指标 BLEU-4 ，具体工具使用 sacrebleu开源版本。

什么是 BLEU-4 ？下面简单的做一个介绍

BLEU，全称为Bilingual Evaluation Understudy（双语评估替换），是一种对生成语句进行评估的指标。BLEU 评分是由Kishore Papineni等人2002年的论文《BLEU: a Method for Automatic Evaluation of Machine Translation》中提出的。

在机器翻译领域，BLEU（Bilingual Evaluation Understudy）是一种常用的自动评价指标，用于衡量计算机生成的翻译与一组参考译文之间的相似度。这个指标特别关注 n-grams（连续的n个词）的精确匹配，可以被认为是对翻译准确性和流利度的一种统计估计。计算BLEU分数时，首先会统计生成文本中n-grams的频率，然后将这些频率与参考文本中的n-grams进行比较。如果生成的翻译中包含的n-grams与参考译文中出现的相同，则认为是匹配的。最终的BLEU分数是一个介于0到1之间的数值，其中1表示与参考译文完美匹配，而0则表示完全没有匹配。

BLEU-4 特别指的是在计算时考虑四元组（即连续四个词）的匹配情况。

BLEU 评估指标的特点：

优点：计算速度快、计算成本低、容易理解、与具体语言无关、和人类给的评估高度相关。
缺点：不考虑语言表达（语法）上的准确性；测评精度会受常用词的干扰；短译句的测评精度有时会较高；没有考虑同义词或相似表达的情况，可能会导致合理翻译被否定。

除翻译之外，BLEU评分结合深度学习方法可应用于其他的语言生成问题，例如：语言生成、图片标题生成、文本摘要、语音识别。

运行主函数，结果如下：

嘿嘿，直接是0，哈哈哈，效果确实很差

模型推理（测试集）：

下面就是正式的将我们的模型训练后，再经过验证集的调参之后，用于正式的测试啦

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('E:/2024年项目资料/2024Datawhale AI夏令营---基于术语词典干预的机器翻译挑战赛/dataset/en-zh.dic')
    # 加载数据集和模型
    dataset = TranslationDataset('E:/2024年项目资料/2024Datawhale AI夏令营---基于术语词典干预的机器翻译挑战赛/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('E:/2024年项目资料/2024Datawhale AI夏令营---基于术语词典干预的机器翻译挑战赛/model/translation_model_GRU.pth'))
    
    save_dir = 'E:/2024年项目资料/2024Datawhale AI夏令营---基于术语词典干预的机器翻译挑战赛/dataset/submit.txt'
    inference(model, dataset, src_file="E:/2024年项目资料/2024Datawhale AI夏令营---基于术语词典干预的机器翻译挑战赛/dataset/test_en.txt", save_dir = save_dir, terminology = terminology, device = device)
    print(f"翻译完成！文件已保存到{save_dir}")

test-data已经准备好了，这里就是将模型用于测试集进行推理吧，不能叫评测，结果如下：

额这结果确实是意料之中的差啊哈哈哈哈，没事没事，这相当于让你看10遍一个1000对中英翻译，就能学会英语翻译另外1000个句子一样哈哈哈，impossible