序列To序列（Seq2Seq）

2022-10-03 本文已影响0人小黄不头秃

(一)序列To序列 / Seq2Seq

这是什么意思呢？也就是说输入是一串序列，输出也是一串序列。就像是生物学中的DNA转录成RNA序列一样。其实他最早被用于机器翻译，给定一个源语言的句子，自动翻译成目标语言。两个序列可以有不同的长度。

序列To序列 / Seq2Seq，使用的也是编码器，解码器架构。编码器是一个RNN，读取输入句子，解码器使用另一个RNN来输出。并且双向循环神经网络可以用于编码器。

（1）编码器解码器细节

上一篇介绍了什么是编码器-解码器架构，这里我们仔细观察一下两者之间的连接。

编码器是没有输出的RNN，编码器的最后一层会将隐状态输送到解码器的RNN作为输入，并且解码器有自己的输入
编码器和解码器的RNN是两个不同的网络
编码器最后的时间步隐状态用作解码器的初始隐状态

（2）训练和推理（预测）

训练时解码器使用目标句子作为输入。

推理的时候，它是由上一个时刻的输出作为下一个时刻的输入，不断迭代下去。
具体实现见代码。

（3）判断预测句子的好坏

衡量序列的好坏——BLEU。
$p_n$ 是预测中所有n-gram的精度，n-gram就是n元的意思，就是在序列中连续的子序列的个数。就是说不论顺序，预测值的n-gram的组合在真实n-gram中的比例。

例如：
标签序列：A,B,C,D,E,F
预测序列：A,B,B,C,D,E
则： $p_1 = \frac{4}{5}$ 、 $p_2 = \frac{3}{4}$ 、 $p_3 = \frac{1}{3}$ 、 $p_4 = 0$

BLEU的定义：（越大越好）

（二）代码实现

import collections 
import math 
import torch 
from torch import nn 
from d2l import torch as d2l

注意这里的输入的形状是(batch_size,num_steps)，每个num_step是用下标表示的一个词。
然后embedding的参数列表（词的总个数，编码维度）用来实现词与词向量的映射。

例如：nn.Embedding(28, 2)，一共有28个词，每个词用长度为2的向量表示。
给一个（10，10）的输入，批量为10，时间步设为10的输入，输出就会是（10，10，2）他将里面的所有词都转变为二维向量。

embedding = nn.Embedding(10, 2)  # 10个词，每个词用2维词向量表示
input = torch.arange(0, 6).view(3, 2).long()  # 3个句子，每句子有2个词
    # input = t.autograd.Variable(input)
output = embedding(input)
print(output.size())
print(embedding.weight.size())

# 编码器
#@save
class Seq2SeqEncoder(d2l.Encoder):
    """用于序列到序列学习的循环神经网络编码器"""
    def __init__(self, vocab_size, embed_size, num_hiddens, num_layers,dropout=0, **kwargs):
        super(Seq2SeqEncoder, self).__init__(**kwargs)
        # 嵌入层
        self.embedding = nn.Embedding(vocab_size, embed_size)
        self.rnn = nn.GRU(embed_size, num_hiddens, num_layers,
                          dropout=dropout)

    def forward(self, X, *args):
        # 输入'X'的形状：(batch_size,num_steps)
        # 输出'X'的形状：(batch_size,num_steps,embed_size)
        X = self.embedding(X)
        # 在循环神经网络模型中，第一个轴对应于时间步
        X = X.permute(1, 0, 2)
        # 如果未提及状态，则默认为0
        output, state = self.rnn(X)
        # output的形状:(num_steps,batch_size,num_hiddens)
        # state的形状:(num_layers,batch_size,num_hiddens)
        return output, state

encoder = Seq2SeqEncoder(vocab_size=10, embed_size=8, num_hiddens=16,
                         num_layers=2)
encoder.eval()
X = torch.zeros((4, 7), dtype=torch.long)
output, state = encoder(X)
output.shape, state.shape

class Seq2SeqDecoder(d2l.Decoder):
    """用于序列到序列学习的循环神经网络解码器"""
    def __init__(self, vocab_size, embed_size, num_hiddens, num_layers,
                 dropout=0, **kwargs):
        super(Seq2SeqDecoder, self).__init__(**kwargs)
        self.embedding = nn.Embedding(vocab_size, embed_size)
        self.rnn = nn.GRU(embed_size + num_hiddens, num_hiddens, num_layers,dropout=dropout)
        self.dense = nn.Linear(num_hiddens, vocab_size)

    def init_state(self, enc_outputs, *args):
        # outputs是指（output, state）
        return enc_outputs[1]

    def forward(self, X, state):
        # 输入'X'的形状：(batch_size,num_steps)
        # 输出'X'的形状：(batch_size,num_steps,embed_size)
        X = self.embedding(X).permute(1, 0, 2) # (num_steps,batch_size,embed_size)
        # 广播context，使其具有与X相同的num_steps (num_steps,batch_size,num_hiddens)
        context = state[-1].repeat(X.shape[0], 1, 1) # （通道数的重复倍数，列的重复倍数，行的重复倍数）
        # context的形状(num_steps,batch_size,num_hiddens)
        X_and_context = torch.cat((X, context), 2) # 这就是为什么GRU的输入长度是embed_size + num_hiddens
        output, state = self.rnn(X_and_context, state)
        output = self.dense(output).permute(1, 0, 2)
        # output的形状:(batch_size,num_steps,vocab_size)
        # state的形状:(num_layers,batch_size,num_hiddens)
        return output, state

decoder = Seq2SeqDecoder(vocab_size=10, embed_size=8, num_hiddens=16,
                         num_layers=2)
decoder.eval()
state = decoder.init_state(encoder(X))
# print(state.shape) # torch.Size([2, 4, 16])
output, state = decoder(X, state)
output.shape, state.shape, X.shape

# 通过零值化，屏蔽不相关项。
def sequence_mask(X, valid_len, value=0):
    """在序列中屏蔽不相关的项"""
    maxlen = X.size(1)
    # print(maxlen) # 3
    # [None,:]是按行增加一个维度，[:, None]按列增加一个维度
    mask = torch.arange((maxlen), dtype=torch.float32,device=X.device)[None, :] < valid_len[:, None]
    # print(torch.arange((maxlen), dtype=torch.float32,device=X.device)[None, :]) # tensor([[0., 1., 2.]])
    # print(valid_len[:, None]) 
        # tensor([[1],
        #     [2]])
    X[~mask] = value
    return X

X = torch.tensor([[1, 2, 3], [4, 5, 6]])
sequence_mask(X, torch.tensor([1, 2]))

class MaskedSoftmaxCELoss(nn.CrossEntropyLoss):
    """带遮蔽的softmax交叉熵损失函数"""
    # pred的形状：(batch_size,num_steps,vocab_size)
    # label的形状：(batch_size,num_steps)
    # valid_len的形状：(batch_size,)
    def forward(self, pred, label, valid_len):
        weights = torch.ones_like(label)
        weights = sequence_mask(weights, valid_len)# 只留下有效长度内权重，其他的置为0
        self.reduction='none'
        unweighted_loss = super(MaskedSoftmaxCELoss, self).forward(pred.permute(0, 2, 1), label) # permute后的形状(batch_size,vocab_size,num_steps)
        weighted_loss = (unweighted_loss * weights).mean(dim=1)
        return weighted_loss

loss = MaskedSoftmaxCELoss()
print(loss(torch.ones((3, 4, 10), dtype=torch.float32), torch.ones((3, 4), dtype=torch.long), torch.tensor([4, 2, 0])))

l = nn.CrossEntropyLoss(reduction="none")
print(l(torch.ones((3, 10, 4), dtype=torch.float32), torch.ones((3, 4), dtype=torch.long)))

def train_seq2seq(net, data_iter, lr, num_epochs, tgt_vocab, device):
    """训练序列到序列模型"""
    def xavier_init_weights(m):
        if type(m) == nn.Linear:
            nn.init.xavier_uniform_(m.weight)
        if type(m) == nn.GRU:
            for param in m._flat_weights_names:
                if "weight" in param:
                    nn.init.xavier_uniform_(m._parameters[param])

    net.apply(xavier_init_weights)
    net.to(device)
    optimizer = torch.optim.Adam(net.parameters(), lr=lr)
    loss = MaskedSoftmaxCELoss()
    net.train()
    animator = d2l.Animator(xlabel='epoch', ylabel='loss',
                     xlim=[10, num_epochs])
    for epoch in range(num_epochs):
        timer = d2l.Timer()
        metric = d2l.Accumulator(2)  # 训练损失总和，词元数量
        for batch in data_iter:
            optimizer.zero_grad()
            X, X_valid_len, Y, Y_valid_len = [x.to(device) for x in batch]
            bos = torch.tensor([tgt_vocab['<bos>']] * Y.shape[0],
                          device=device).reshape(-1, 1)# (batch_size, 1)
            dec_input = torch.cat([bos, Y[:, :-1]], 1)  # 强制教学
            # print(bos.shape,dec_input.shape) # torch.Size([64, 1]) torch.Size([64, 10])
            Y_hat, _ = net(X, dec_input, X_valid_len)
            l = loss(Y_hat, Y, Y_valid_len)
            l.sum().backward()  # 损失函数的标量进行“反向传播”
            d2l.grad_clipping(net, 1)
            num_tokens = Y_valid_len.sum()
            optimizer.step()
            with torch.no_grad():
                metric.add(l.sum(), num_tokens)
        if (epoch + 1) % 10 == 0:
            animator.add(epoch + 1, (metric[0] / metric[1],))
    print(f'loss {metric[0] / metric[1]:.3f}, {metric[1] / timer.stop():.1f} '
        f'tokens/sec on {str(device)}')

class EncoderDecoder(nn.Module):
    """The base class for the encoder-decoder architecture.

    Defined in :numref:`sec_encoder-decoder`"""
    def __init__(self, encoder, decoder, **kwargs):
        super(EncoderDecoder, self).__init__(**kwargs)
        self.encoder = encoder
        self.decoder = decoder

    def forward(self, enc_X, dec_X, *args):
        enc_outputs = self.encoder(enc_X, *args)
        dec_state = self.decoder.init_state(enc_outputs, *args)
        return self.decoder(dec_X, dec_state)

embed_size, num_hiddens, num_layers, dropout = 32, 32, 2, 0.1
batch_size, num_steps = 64, 10
lr, num_epochs, device = 0.005, 300, d2l.try_gpu()

train_iter, src_vocab, tgt_vocab = d2l.load_data_nmt(batch_size, num_steps)
encoder = Seq2SeqEncoder(len(src_vocab), embed_size, num_hiddens, num_layers,
                        dropout)
decoder = Seq2SeqDecoder(len(tgt_vocab), embed_size, num_hiddens, num_layers,
                        dropout)
net = d2l.EncoderDecoder(encoder, decoder)
train_seq2seq(net, train_iter, lr, num_epochs, tgt_vocab, device)

#@save
def predict_seq2seq(net, src_sentence, src_vocab, tgt_vocab, num_steps,
                    device, save_attention_weights=False):
    """序列到序列模型的预测"""
    # 在预测时将net设置为评估模式
    net.eval()
    src_tokens = src_vocab[src_sentence.lower().split(' ')] + [src_vocab['<eos>']]# 在末尾加上结束标记
    enc_valid_len = torch.tensor([len(src_tokens)], device=device)
    src_tokens = d2l.truncate_pad(src_tokens, num_steps, src_vocab['<pad>'])# 填充字符
    # 添加批量轴
    enc_X = torch.unsqueeze(torch.tensor(src_tokens, dtype=torch.long, device=device), dim=0) # 增加一个维度
    enc_outputs = net.encoder(enc_X, enc_valid_len)
    dec_state = net.decoder.init_state(enc_outputs, enc_valid_len) # 初始化解码器
    # 添加批量轴
    dec_X = torch.unsqueeze(torch.tensor([tgt_vocab['<bos>']], dtype=torch.long, device=device), dim=0) # 解码器的输入我们放入一个开始标记符
    output_seq, attention_weight_seq = [], []
    for _ in range(num_steps):
        Y, dec_state = net.decoder(dec_X, dec_state)
        # 我们使用具有预测最高可能性的词元，作为解码器在下一时间步的输入
        # Y的形状：(batch_size,num_steps,vocab_size), batch_size = 1
        dec_X = Y.argmax(dim=2) # 每个批量，获取每个num_step最大下标，并作为下一个预测值的开始（batch_size,num_step）,下标和tgt_vocab一致
        # print(Y.shape) # torch.Size([1, 1, 201])
        # print(dec_X.shape) # torch.Size([1, 1])
        pred = dec_X.squeeze(dim=0).type(torch.int32).item()# 去掉一个维度
        # 保存注意力权重（稍后讨论）
        if save_attention_weights:
            attention_weight_seq.append(net.decoder.attention_weights)
        # 一旦序列结束词元被预测，输出序列的生成就完成了
        if pred == tgt_vocab['<eos>']:
            break
        output_seq.append(pred)
    return ' '.join(tgt_vocab.to_tokens(output_seq)), attention_weight_seq

def bleu(pred_seq, label_seq, k):  #@save
    """计算BLEU"""
    pred_tokens, label_tokens = pred_seq.split(' '), label_seq.split(' ')
    len_pred, len_label = len(pred_tokens), len(label_tokens)
    score = math.exp(min(0, 1 - len_label / len_pred))
    for n in range(1, k + 1):
        num_matches, label_subs = 0, collections.defaultdict(int)
        for i in range(len_label - n + 1):
            label_subs[' '.join(label_tokens[i: i + n])] += 1
        for i in range(len_pred - n + 1):
            if label_subs[' '.join(pred_tokens[i: i + n])] > 0:
                num_matches += 1
                label_subs[' '.join(pred_tokens[i: i + n])] -= 1
        score *= math.pow(num_matches / (len_pred - n + 1), math.pow(0.5, n))
    return score

engs = ['go .', "i lost .", 'he\'s calm .', 'i\'m home .']
fras = ['va !', 'j\'ai perdu .', 'il est calme .', 'je suis chez moi .']
for eng, fra in zip(engs, fras):
    translation, attention_weight_seq = predict_seq2seq(
        net, eng, src_vocab, tgt_vocab, num_steps, device)
    print(f'{eng} => {translation}, bleu {bleu(translation, fra, k=2):.3f}')