自然语言处理之机器翻译：Sequence-to-Sequence(Seq2Seq)：编码器-解码器架构

    import numpy as np
    from keras.models import Model
    from keras.layers import Input, LSTM, Dense
    
    # 定义输入和输出的维度
    num_encoder_tokens = 10000
    num_decoder_tokens = 8000
    max_encoder_seq_length = 100
    max_decoder_seq_length = 100
    
    # 构建编码器
    encoder_inputs = Input(shape=(None, num_encoder_tokens))
    encoder = LSTM(256, return_state=True)
    encoder_outputs, state_h, state_c = encoder(encoder_inputs)
    encoder_states = [state_h, state_c]
    
    # 构建解码器
    decoder_inputs = Input(shape=(None, num_decoder_tokens))
    decoder_lstm = LSTM(256, return_sequences=True, return_state=True)
    decoder_outputs, _, _ = decoder_lstm(decoder_inputs, initial_state=encoder_states)
    decoder_dense = Dense(num_decoder_tokens, activation='softmax')
    decoder_outputs = decoder_dense(decoder_outputs)
    
    # 定义和编译模型
    model = Model([encoder_inputs, decoder_inputs], decoder_outputs)
    model.compile(optimizer='rmsprop', loss='categorical_crossentropy')
    
    # 假设我们有预处理后的训练数据
    encoder_input_data = np.zeros((num_samples, max_encoder_seq_length, num_encoder_tokens), dtype='float32')
    decoder_input_data = np.zeros((num_samples, max_decoder_seq_length, num_decoder_tokens), dtype='float32')
    decoder_target_data = np.zeros((num_samples, max_decoder_seq_length, num_decoder_tokens), dtype='float32')
    
    # 训练模型
    model.fit([encoder_input_data, decoder_input_data], decoder_target_data,
          batch_size=batch_size,
          epochs=epochs,
          validation_split=0.2)

    import torch
    import torch.nn as nn
    import torch.optim as optim
    
    # 定义编码器
    class EncoderRNN(nn.Module):
    def __init__(self, input_size, hidden_size):
        super(EncoderRNN, self).__init__()
        self.hidden_size = hidden_size
        self.embedding = nn.Embedding(input_size, hidden_size)
        self.gru = nn.GRU(hidden_size, hidden_size)
    
    def forward(self, input, hidden):
        embedded = self.embedding(input).view(1, 1, -1)
        output = embedded
        output, hidden = self.gru(output, hidden)
        return output, hidden
    
    def initHidden(self):
        return torch.zeros(1, 1, self.hidden_size, device=device)
    
    # 定义解码器
    class DecoderRNN(nn.Module):
    def __init__(self, hidden_size, output_size):
        super(DecoderRNN, self).__init__()
        self.hidden_size = hidden_size
        self.embedding = nn.Embedding(output_size, hidden_size)
        self.gru = nn.GRU(hidden_size, hidden_size)
        self.out = nn.Linear(hidden_size, output_size)
        self.softmax = nn.LogSoftmax(dim=1)
    
    def forward(self, input, hidden):
        output = self.embedding(input).view(1, 1, -1)
        output = torch.relu(output)
        output, hidden = self.gru(output, hidden)
        output = self.softmax(self.out(output[0]))
        return output, hidden
    
    def initHidden(self):
        return torch.zeros(1, 1, self.hidden_size, device=device)
    
    # 定义训练函数
    def train(input_tensor, target_tensor, encoder, decoder, encoder_optimizer, decoder_optimizer, criterion, max_length=MAX_LENGTH):
    encoder_hidden = encoder.initHidden()
    encoder_optimizer.zero_grad()
    decoder_optimizer.zero_grad()
    input_length = input_tensor.size(0)
    target_length = target_tensor.size(0)
    encoder_outputs = torch.zeros(max_length, encoder.hidden_size, device=device)
    loss = 0
    for ei in range(input_length):
        encoder_output, encoder_hidden = encoder(input_tensor[ei], encoder_hidden)
        encoder_outputs[ei] = encoder_output[0, 0]
    decoder_input = torch.tensor([[SOS_token]], device=device)
    decoder_hidden = encoder_hidden
    for di in range(target_length):
        decoder_output, decoder_hidden = decoder(decoder_input, decoder_hidden)
        loss += criterion(decoder_output, target_tensor[di])
        decoder_input = target_tensor[di]  # 教师强制
    loss.backward()
    encoder_optimizer.step()
    decoder_optimizer.step()
    return loss.item() / target_length
    
    # 数据预处理和训练循环的代码省略...

    import torch
    import torch.nn as nn
    
    class Encoder(nn.Module):
    def __init__(self, input_dim, emb_dim, enc_hid_dim, dec_hid_dim, dropout):
        super().__init__()
        
        self.embedding = nn.Embedding(input_dim, emb_dim)
        
        self.rnn = nn.GRU(emb_dim, enc_hid_dim, bidirectional = True)
        
        self.fc = nn.Linear(enc_hid_dim * 2, dec_hid_dim)
        
        self.dropout = nn.Dropout(dropout)
        
    def forward(self, src):
        # src = [src sent len, batch size]
        
        embedded = self.dropout(self.embedding(src))
        
        # embedded = [src sent len, batch size, emb dim]
        
        outputs, hidden = self.rnn(embedded)
        
        # outputs = [src sent len, batch size, hid dim * num directions]
        # hidden = [n layers * num directions, batch size, hid dim]
        
        # hidden is stacked [forward_1, backward_1, forward_2, backward_2, ...]
        # outputs are always from the last layer
        
        # hidden [-2, :, : ] is the last of the forwards RNN 
        # hidden [-1, :, : ] is the last of the backwards RNN
        
        # initial decoder hidden is final hidden state of the forwards and backwards 
        #  encoder RNNs fed through a linear layer
        hidden = torch.tanh(self.fc(torch.cat((hidden[-2,:,:], hidden[-1,:,:]), dim = 1)))
        
        # outputs = [src sent len, batch size, enc hid dim * 2]
        # hidden = [batch size, dec hid dim]
        
        return outputs, hidden

    class Decoder(nn.Module):
    def __init__(self, output_dim, emb_dim, dec_hid_dim, enc_hid_dim, dropout):
        super().__init__()
        
        self.output_dim = output_dim
        self.embedding = nn.Embedding(output_dim, emb_dim)
        
        self.rnn = nn.GRU(emb_dim + enc_hid_dim * 2, dec_hid_dim)
        
        self.fc_out = nn.Linear(emb_dim + dec_hid_dim + enc_hid_dim * 2, output_dim)
        
        self.dropout = nn.Dropout(dropout)
        
    def forward(self, input, hidden, context):
        # input = [batch size]
        # hidden = [batch size, dec hid dim]
        # context = [src sent len, batch size, enc hid dim * 2]
        
        input = input.unsqueeze(0)
        
        # input = [1, batch size]
        
        embedded = self.dropout(self.embedding(input))
        
        # embedded = [1, batch size, emb dim]
        
        emb_con = torch.cat((embedded, context), dim = 2)
        
        # emb_con = [1, batch size, emb dim + enc hid dim * 2]
        
        output, hidden = self.rnn(emb_con, hidden.unsqueeze(0))
        
        # output = [seq len, batch size, dec hid dim * n directions]
        # hidden = [n layers * n directions, batch size, dec hid dim]
        
        # seq len, n layers and n directions will always be 1 in this decoder, therefore:
        # output = [1, batch size, dec hid dim]
        # hidden = [1, batch size, dec hid dim]
        
        output = torch.cat((embedded.squeeze(0), hidden.squeeze(0), context.squeeze(0)), dim = 1)
        
        # output = [batch size, emb dim + dec hid dim + enc hid dim * 2]
        
        prediction = self.fc_out(output)
        
        # prediction = [batch size, output dim]
        
        return prediction, hidden.squeeze(0)

    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 = [src sent len, batch size]
        # trg = [trg sent len, batch size]
        # teacher_forcing_ratio is probability to use teacher forcing
        # e.g. if teacher_forcing_ratio is 0.75 we use teacher forcing 75% of the time
        
        batch_size = trg.shape[1]
        trg_len = trg.shape[0]
        trg_vocab_size = self.decoder.output_dim
        
        # tensor to store decoder outputs
        outputs = torch.zeros(trg_len, batch_size, trg_vocab_size).to(self.device)
        
        # last hidden state of the encoder is used as the initial hidden state of the decoder
        enc_outputs, hidden = self.encoder(src)
        
        # first input to the decoder is the <sos> tokens
        input = trg[0,:]
        
        for t in range(1, trg_len):
            
            # insert input token embedding, previous hidden state and the encoder outputs
            # receive output tensor (predictions) and new hidden state
            output, hidden = self.decoder(input, hidden, enc_outputs)
            
            # place predictions in a tensor holding predictions for each token
            outputs[t] = output
            
            # decide if we are going to use teacher forcing or not
            teacher_force = random.random() < teacher_forcing_ratio
            
            # get the highest predicted token from our predictions
            top1 = output.argmax(1) 
            
            # if teacher forcing, use actual next token as next input
            # if not, use predicted token
            input = trg[t] if teacher_force else top1
        
        return outputs

    import torch
    import torch.nn as nn
    
    # 定义损失函数
    criterion = nn.CrossEntropyLoss()
    
    # 假设模型的输出是一个序列，每个时间步的输出是一个词汇表大小的向量
    # output: [batch_size, sequence_length, vocab_size]
    # target: [batch_size, sequence_length]
    output = torch.randn(32, 10, 5000, requires_grad=True)
    target = torch.randint(5000, (32, 10))
    
    # 计算损失
    loss = criterion(output.view(-1, 5000), target.view(-1))
    print(f"Loss: {loss.item()}")

    import torch
    import torch.nn as nn
    
    class Seq2Seq(nn.Module):
    def __init__(self, encoder, decoder, device):
        super(Seq2Seq, self).__init__()
        self.encoder = encoder
        self.decoder = decoder
        self.device = device
    
    def forward(self, src, trg, teacher_forcing_ratio=0.5):
        batch_size = src.shape[1]
        trg_len = trg.shape[0]
        trg_vocab_size = self.decoder.output_dim
    
        # 初始化输出序列
        outputs = torch.zeros(trg_len, batch_size, trg_vocab_size).to(self.device)
    
        # 编码器的输出
        enc_output, hidden = self.encoder(src)
    
        # 解码器的输入
        input = trg[0,:]
    
        for t in range(1, trg_len):
            output, hidden = self.decoder(input, hidden, enc_output)
            outputs[t] = output
            # 决定是否使用教师强制
            teacher_force = random.random() < teacher_forcing_ratio
            top1 = output.argmax(1)
            input = trg[t] if teacher_force else top1
    
        return outputs

    import torch
    import torch.nn as nn
    
    class Attention(nn.Module):
    def __init__(self, hidden_size):
        super(Attention, self).__init__()
        self.hidden_size = hidden_size
        self.attn = nn.Linear(hidden_size * 2, hidden_size)
        self.v = nn.Parameter(torch.rand(hidden_size))
    
    def forward(self, decoder_hidden, encoder_outputs):
        # encoder_outputs: [seq_len, batch_size, hidden_size]
        # decoder_hidden: [1, batch_size, hidden_size]
        
        seq_len = encoder_outputs.size(0)
        batch_size = encoder_outputs.size(1)
        
        # 重复解码器隐藏状态，使其与编码器输出的序列长度相同
        decoder_hidden = decoder_hidden.repeat(seq_len, 1, 1)
        
        # 计算注意力能量
        energy = torch.tanh(self.attn(torch.cat((decoder_hidden, encoder_outputs), 2)))
        
        # 计算注意力权重
        attn_energies = torch.bmm(energy, self.v.unsqueeze(2)).squeeze(2)
        attn_weights = nn.functional.softmax(attn_energies, dim=0)
        
        return attn_weights

    def compute_context_vector(attn_weights, encoder_outputs):
    # attn_weights: [seq_len, batch_size]
    # encoder_outputs: [seq_len, batch_size, hidden_size]
    
    # 将注意力权重与编码器输出相乘，得到加权的编码器输出
    weighted_outputs = attn_weights.unsqueeze(2) * encoder_outputs
    
    # 对加权的编码器输出进行求和，得到加权上下文向量
    context_vector = weighted_outputs.sum(dim=0)
    
    return context_vector

    class AttnDecoder(nn.Module):
    def __init__(self, hidden_size, output_size, dropout_p=0.1, max_length=100):
        super(AttnDecoder, self).__init__()
        self.hidden_size = hidden_size
        self.output_size = output_size
        self.dropout_p = dropout_p
        self.max_length = max_length
        
        self.embedding = nn.Embedding(output_size, hidden_size)
        self.attn = Attention(hidden_size)
        self.gru = nn.GRU(hidden_size * 2, hidden_size)
        self.out = nn.Linear(hidden_size * 2, output_size)
        self.dropout = nn.Dropout(dropout_p)
    
    def forward(self, input, hidden, encoder_outputs):
        # input: [1, batch_size]
        # hidden: [1, batch_size, hidden_size]
        # encoder_outputs: [seq_len, batch_size, hidden_size]
        
        embedded = self.embedding(input).view(1, input.size(0), -1)
        embedded = self.dropout(embedded)
        
        attn_weights = self.attn(hidden, encoder_outputs)
        context = compute_context_vector(attn_weights, encoder_outputs)
        
        # 将嵌入词和加权上下文向量拼接
        rnn_input = torch.cat((embedded, context.unsqueeze(0)), 2)
        
        # 更新解码器状态
        output, hidden = self.gru(rnn_input, hidden)
        
        # 计算输出词的概率
        output = output.squeeze(0)
        output = torch.cat((output, context), 1)
        output = self.out(output)
        output = nn.functional.log_softmax(output, dim=1)
        
        return output, hidden, attn_weights

    import tensorflow as tf
    
    # 创建一个简单的编码器
    class Encoder(tf.keras.Model):
    def __init__(self, vocab_size, embedding_dim, enc_units, batch_sz):
        super(Encoder, self).__init__()
        self.batch_sz = batch_sz
        self.enc_units = enc_units
        self.embedding = tf.keras.layers.Embedding(vocab_size, embedding_dim)
        self.gru = tf.keras.layers.GRU(self.enc_units,
                                       return_sequences=True,
                                       return_state=True,
                                       recurrent_initializer='glorot_uniform')
        self.dropout = tf.keras.layers.Dropout(0.2)
    
    def call(self, x, hidden):
        x = self.embedding(x)
        x = self.dropout(x)
        output, state = self.gru(x, initial_state=hidden)
        return output, state
    
    # 初始化参数
    vocab_size = 10000
    embedding_dim = 256
    enc_units = 1024
    batch_sz = 64
    
    # 创建编码器实例
    encoder = Encoder(vocab_size, embedding_dim, enc_units, batch_sz)
    
    # 假设输入数据
    input_data = tf.random.uniform((batch_sz, 10), maxval=vocab_size, dtype=tf.int32)
    hidden = tf.zeros((batch_sz, enc_units))
    
    # 调用编码器
    output, state = encoder(input_data, hidden)

    import tensorflow as tf
    
    # 创建一个简单的解码器
    class Decoder(tf.keras.Model):
    def __init__(self, vocab_size, embedding_dim, dec_units, batch_sz):
        super(Decoder, self).__init__()
        self.batch_sz = batch_sz
        self.dec_units = dec_units
        self.embedding = tf.keras.layers.Embedding(vocab_size, embedding_dim)
        self.gru = tf.keras.layers.GRU(self.dec_units,
                                       return_sequences=True,
                                       return_state=True,
                                       recurrent_initializer='glorot_uniform')
        self.fc = tf.keras.layers.Dense(vocab_size, kernel_regularizer=tf.keras.regularizers.l2(0.01))
    
    def call(self, x, hidden):
        x = self.embedding(x)
        output, state = self.gru(x, initial_state=hidden)
        output = tf.reshape(output, (-1, output.shape[2]))
        x = self.fc(output)
        return x, state
    
    # 初始化参数
    vocab_size = 10000
    embedding_dim = 256
    dec_units = 1024
    batch_sz = 64
    
    # 创建解码器实例
    decoder = Decoder(vocab_size, embedding_dim, dec_units, batch_sz)
    
    # 假设输入数据
    input_data = tf.random.uniform((batch_sz, 10), maxval=vocab_size, dtype=tf.int32)
    hidden = tf.zeros((batch_sz, dec_units))
    
    # 调用解码器
    output, state = decoder(input_data, hidden)

    import tensorflow as tf
    
    # 定义学习率衰减函数
    def lr_decay(epoch):
    if epoch < 10:
        return 0.001
    else:
        return 0.001 * tf.math.exp(0.1 * (10 - epoch))
    
    # 创建学习率衰减回调
    lr_scheduler = tf.keras.callbacks.LearningRateScheduler(lr_decay)
    
    # 创建模型并编译
    model = tf.keras.Sequential([
    tf.keras.layers.Dense(64, activation='relu', input_shape=(10,)),
    tf.keras.layers.Dense(10, activation='softmax')
    ])
    model.compile(optimizer=tf.keras.optimizers.Adam(0.001),
              loss='sparse_categorical_crossentropy',
              metrics=['accuracy'])
    
    # 假设训练数据
    x_train = tf.random.normal((1000, 10))
    y_train = tf.random.uniform((1000,), maxval=10, dtype=tf.int32)
    
    # 训练模型，使用学习率衰减回调
    model.fit(x_train, y_train, epochs=20, callbacks=[lr_scheduler])

    import tensorflow as tf
    
    # 创建动态学习率调整回调
    reduce_lr = tf.keras.callbacks.ReduceLROnPlateau(monitor='val_loss', factor=0.2,
                                                 patience=5, min_lr=0.001)
    
    # 创建模型并编译
    model = tf.keras.Sequential([
    tf.keras.layers.Dense(64, activation='relu', input_shape=(10,)),
    tf.keras.layers.Dense(10, activation='softmax')
    ])
    model.compile(optimizer=tf.keras.optimizers.Adam(0.01),
              loss='sparse_categorical_crossentropy',
              metrics=['accuracy'])
    
    # 假设训练和验证数据
    x_train = tf.random.normal((1000, 10))
    y_train = tf.random.uniform((1000,), maxval=10, dtype=tf.int32)
    x_val = tf.random.normal((200, 10))
    y_val = tf.random.uniform((200,), maxval=10, dtype=tf.int32)
    
    # 训练模型，使用动态学习率调整回调
    model.fit(x_train, y_train, epochs=20, validation_data=(x_val, y_val), callbacks=[reduce_lr])

    import torch
    import torch.nn as nn
    from torch.autograd import Variable
    
    # 定义编码器
    class EncoderRNN(nn.Module):
    def __init__(self, input_size, hidden_size):
        super(EncoderRNN, self).__init__()
        self.hidden_size = hidden_size
        self.embedding = nn.Embedding(input_size, hidden_size)
        self.gru = nn.GRU(hidden_size, hidden_size)
    
    def forward(self, input, hidden):
        embedded = self.embedding(input).view(1, 1, -1)
        output = embedded
        output, hidden = self.gru(output, hidden)
        return output, hidden
    
    def initHidden(self):
        return Variable(torch.zeros(1, 1, self.hidden_size))
    
    # 定义解码器
    class DecoderRNN(nn.Module):
    def __init__(self, hidden_size, output_size):
        super(DecoderRNN, self).__init__()
        self.hidden_size = hidden_size
        self.embedding = nn.Embedding(output_size, hidden_size)
        self.gru = nn.GRU(hidden_size, hidden_size)
        self.out = nn.Linear(hidden_size, output_size)
        self.softmax = nn.LogSoftmax(dim=1)
    
    def forward(self, input, hidden):
        output = self.embedding(input).view(1, 1, -1)
        output = F.relu(output)
        output, hidden = self.gru(output, hidden)
        output = self.softmax(self.out(output[0]))
        return output, hidden
    
    def initHidden(self):
        return Variable(torch.zeros(1, 1, self.hidden_size))

    Q: 你好吗？
    A: 我很好，谢谢！
    
    Q: 你叫什么名字？
    A: 我是你的聊天机器人。
    
    Q: 你会说英语吗？
    A: 是的，我会说英语。

    import torch
    import torch.nn as nn
    
    # 定义带有注意力机制的解码器
    class AttnDecoderRNN(nn.Module):
    def __init__(self, hidden_size, output_size, dropout_p=0.1, max_length=50):
        super(AttnDecoderRNN, self).__init__()
        self.hidden_size = hidden_size
        self.output_size = output_size
        self.dropout_p = dropout_p
        self.max_length = max_length
    
        self.embedding = nn.Embedding(self.output_size, self.hidden_size)
        self.attn = nn.Linear(self.hidden_size * 2, self.max_length)
        self.attn_combine = nn.Linear(self.hidden_size * 2, self.hidden_size)
        self.dropout = nn.Dropout(self.dropout_p)
        self.gru = nn.GRU(self.hidden_size, self.hidden_size)
        self.out = nn.Linear(self.hidden_size, self.output_size)
    
    def forward(self, input, hidden, encoder_outputs):
        embedded = self.embedding(input).view(1, 1, -1)
        embedded = self.dropout(embedded)
    
        attn_weights = F.softmax(
            self.attn(torch.cat((embedded[0], hidden[0]), 1)), dim=1)
        attn_applied = torch.bmm(attn_weights.unsqueeze(0),
                                 encoder_outputs.unsqueeze(0))
    
        output = torch.cat((embedded[0], attn_applied[0]), 1)
        output = self.attn_combine(output).unsqueeze(0)
    
        output = F.relu(output)
        output, hidden = self.gru(output, hidden)
    
        output = F.log_softmax(self.out(output[0]), dim=1)
        return output, hidden, attn_weights

    EN: Hello, how are you?
    FR: Bonjour, comment vas-tu?
    
    EN: I am learning to code.
    FR: Je suis en train d'apprendre à coder.

    # 导入必要的库
    import tensorflow as tf
    from tensorflow.keras.layers import Input, LSTM, Dense
    from tensorflow.keras.models import Model
    
    # 定义编码器
    encoder_inputs = Input(shape=(None, num_encoder_tokens))
    encoder = LSTM(latent_dim, return_state=True)
    encoder_outputs, state_h, state_c = encoder(encoder_inputs)
    encoder_states = [state_h, state_c]
    
    # 定义解码器
    decoder_inputs = Input(shape=(None, num_decoder_tokens))
    decoder_lstm = LSTM(latent_dim, return_sequences=True, return_state=True)
    decoder_outputs, _, _ = decoder_lstm(decoder_inputs, initial_state=encoder_states)
    decoder_dense = Dense(num_decoder_tokens, activation='softmax')
    decoder_outputs = decoder_dense(decoder_outputs)
    
    # 构建模型
    model = Model([encoder_inputs, decoder_inputs], decoder_outputs)
    
    # 编译模型
    model.compile(optimizer='rmsprop', loss='categorical_crossentropy')
    
    # 训练模型
    model.fit([encoder_input_data, decoder_input_data], decoder_target_data,
          batch_size=batch_size,
          epochs=epochs,
          validation_split=0.2)