目录1.序列分类问题2.序列标注问题当输入数据是时间序列时 例如文本中的词序列股票价格的时间序列语音信号序列等序列中元素出现的顺序会影响对序列的理解RNN是一种序列模型它能够捕捉序列中的时间依赖性。如果序列中元素的顺序发生变化模型输出的结果是不同的。捕捉上下文信息。序列中每个元素计算出的特征不仅与该元素有关还与序列中的其他元素有关。来回顾一下学过的模型结构全连接层把序列数据 (形状为(T,d),T为序列长度d为序列中每个元素的向量长度) 展平为T x d的特征向量再输入到全连接层。这种做法的确使得模型能够一次性“看见”全部的序列信息但存在明显问题全连接层假设每个输入节点代表一个特征维度但这在很多序列中例如文本是不成立的。参数量大如果模型可以处理的最大序列长度为max_length序列中每个元素特征向量的长度为d则输入节点数量为max_length x d。(若max_length512, d300则输入节点数为153600)模型效率低下没有通过共享参数提取局部特征也无法应对局部特征的平移。这一点可以参考CNN和全连接层的对比。CNN序列数据可以通过1D CNN进行建模。CNN提取的是局部特征无法“看到”远处的信息。如果想要扩大CNN的感受野必须要堆叠多个卷积层和池化层。但随着CNN感受野的扩大时间分辨率也随之下降无法完成序列标注任务。CNN可以完成序列分类任务。RNN是一种新的网络结构它按时间步接受序列中的每个元素通过隐藏层来逐步汇总到目前时间步 (1 ~ t) 的序列信息并基于隐藏层计算输出。RNN的网络结构如下RNN在每个时间步的计算公式为tanh 相比 sigmoid梯度消失会稍微好一点但原始 RNN 依然扛不住很长的序列后面才出来 LSTM/GRU 解决这个问题。RNN网络的计算顺序为:(1) 基于计算(2) 基于计算......(4) 基于计算思考如果我们将输入序列的顺序打乱那么的计算结果会发生变化吗——会1.序列分类问题如果是序列分类问题则将通过softmax预测文本属于各个类别的概率即其中损失函数为交叉熵若的标注为y则该样本的损失函数为2.序列标注问题如果是序列标注问题则将每一步输出的通过softmax预测每个元素属于各个类别的概率此时每个时间步t都会基于计算输出其中如果设每个时间步的标注分别为那么时间步t的损失函数为该样本的损失函数为思考损失函数的自变量是什么在损失函数中是变量吗现在我们利用nn.Linear来搭建一个RNN网络演示一下如何利用RNN来完成文本分类任务和序列标注任务。PyTorch RNN相关的类RNNCell — PyTorch 2.14 documentationRNN — PyTorch 2.14 documentationRNNCellRNN思考RNN的参数量和计算复杂度分别是多少?计算量乘法(d_out * d_in d_out * d_out) * T参数量d_out * d_in d_out * d_out 2 * d_out只能串行计算不能并行计算。思考RNN可以堆叠多层吗可以也可CNN RNN 混搭。import torch.nn as nn import torch from torch.utils.data import Dataset, DataLoader, random_split import os from tokenizer import MyTokenizer from typing import Union import fasttext import sys import torch.nn as nn import torch class MyRNNCell(torch.nn.Module): def __init__(self, input_dim, output_dim): super().__init__() self.input_dim input_dim self.output_dim output_dim self.linear_x nn.Linear(input_dim, output_dim) self.linear_h nn.Linear(output_dim, output_dim) def forward(self, x, h): return nn.functional.tanh(self.linear_x(x) self.linear_h(h)) class MyRNN(nn.Module): def __init__(self, input_dim, output_dim): super().__init__() self.input_dim input_dim self.output_dim output_dim self.rnncell MyRNNCell(input_dim, output_dim) def forward(self, inputs): # inputs: (batch_size, seq_len, input_dim) # outpus: (batch_size, seq_len, output_dim) length inputs.shape[1] outputs torch.zeros(inputs.shape[0], length, self.output_dim) with torch.no_grad(): for i in range(length): if i 0: outputs[:, i, :] self.rnncell(inputs[:, i, :], torch.zeros(inputs.shape[0], self.output_dim)) else: outputs[:, i, :] self.rnncell(inputs[:, i, :], outputs[:, i-1, :]) return outputs class MySequenceClassificationModel(nn.Module): def __init__(self, vocab_size, embedding_dim, hidden_dim, output_dim, pad_id0, embedding_matrixNone): super().__init__() if embedding_matrix is None: self.embedding nn.Embedding(vocab_size, embedding_dim, padding_idxpad_id) else: self.embedding nn.Embedding.from_pretrained(embeddingsembedding_matrix, freezeFalse, padding_idxpad_id) self.rnn MyRNN(embedding_dim, hidden_dim) self.linear nn.Linear(hidden_dim, output_dim) def forward(self, inputs): # inputs现在是一个dict # inputs[data]: 词 id tensor, (batch_size, max_length) # inputs[last_token_pos]: 最后一个token的位置 (batch_size) embedding_outputs self.embedding(inputs[data]) # (batch_size, max_length, embedding_dim) rnn_outputs self.rnn(embedding_outputs) # batch_size, seq_len, hidden_dim # pool at inputs[last_token_pos] pooling_outputs rnn_outputs[torch.arange(rnn_outputs.shape[0]), inputs[last_token_pos], :] return self.linear(pooling_outputs) class MySequenceClassificationModel2(nn.Module): def __init__(self, vocab_size, embedding_dim, hidden_dim, output_dim, pad_id0, embedding_matrixNone): super().__init__() if embedding_matrix is None: self.embedding nn.Embedding(vocab_size, embedding_dim, padding_idxpad_id) else: self.embedding nn.Embedding.from_pretrained(embeddingsembedding_matrix, freezeFalse, padding_idxpad_id) self.rnn nn.LSTM(input_sizeembedding_dim, hidden_sizehidden_dim, batch_firstTrue) self.linear nn.Linear(hidden_dim, output_dim) def forward(self, inputs): # inputs现在是一个dict # inputs[data]: 词 id tensor, (batch_size, max_length) # inputs[last_token_pos]: 最后一个token的位置 (batch_size) embedding_outputs self.embedding(inputs[data]) # (batch_size, max_length, embedding_dim) rnn_outputs, _ self.rnn(embedding_outputs) # batch_size, seq_len, hidden_dim # pool at inputs[last_token_pos] pooling_outputs rnn_outputs[torch.arange(rnn_outputs.shape[0]), inputs[last_token_pos], :] return self.linear(pooling_outputs) class MyTextDataset(Dataset): def __init__(self, file_path, labels None, max_length32, pad_id0): super().__init__() data [] label_set set() with open(file_path, r) as f: for line in f.readlines(): try: label, text line.strip().split(\t) ids [int(x) for x in text.split()] length len(ids) if len(ids) max_length: ids ids[:max_length] length max_length else: ids ids [pad_id] * (max_length - len(ids)) data.append((label, ids, length-1)) label_set.add(label) except: continue self.labels None if labels is not None: self.labels labels else : self.labels list(label_set) self.labels.sort() self.data [] for item in data: item2 dict() item2[label] self.labels.index(item[0]) item2[data] torch.LongTensor(item[1]) item2[last_token_pos] item[2] self.data.append(item2) def __len__(self): return len(self.data) def __getitem__(self, index): return self.data[index] def same_seed(seed): torch.manual_seed(seed) torch.cuda.manual_seed(seed) torch.cuda.manual_seed_all(seed) torch.backends.cudnn.deterministic True torch.backends.cudnn.benchmark False def train(model : nn.Module, train_dataset : Dataset, valid_dataset : Dataset, loss_fn : nn.Module, optimizer : torch.optim.Optimizer, config : dict) - None: # data loader train_dataloader DataLoader(train_dataset, batch_sizeconfig[train_batch_size], shuffleTrue, drop_lastTrue) eval_dataloader DataLoader(valid_dataset, batch_sizeconfig[eval_batch_size], shuffleFalse, drop_lastFalse) # move model to device model model.to(config[device]) # 训练的主流程 best_valid_loss 1e10 no_improve_epochs 0 step 0 for epoch in range(config[epochs]): model.train() train_loss [] for data in train_dataloader: # 把数据导入到设备中 for key in data: data[key] data[key].to(config[device]) # forward preds model(data) loss loss_fn(preds, data[label]) train_loss.append(loss) # tensorboard可视化 #writer.add_scalar(minibatch loss, loss, step) # backward loss.backward() # update optimizer.step() step 1 if step % 500 0: print(fepoch: {epoch}, step: {step}, training loss: {loss.item()}) sys.stdout.flush() # 梯度清零 optimizer.zero_grad() train_mean_loss sum(train_loss) / len(train_loss) # 在验证集上测试 valid_mean_loss, valid_mean_accuracy test(model, eval_dataloader, loss_fn, config) print(fepoch: {epoch}, train_loss: {train_mean_loss:.5f}, valid_loss: {valid_mean_loss:.5f}, valid_accuracy: {valid_mean_accuracy:.5f}) #writer.add_scalars(epoch loss, {train: train_mean_loss, valid: valid_mean_loss}, epoch) if valid_mean_loss best_valid_loss: best_valid_loss valid_mean_loss no_improve_epochs 0 torch.save(model.state_dict(), config[save_path]) else: no_improve_epochs 1 if no_improve_epochs config[max_no_improve_epochs]: print(fearly stop at epoch: {epoch}) break def test(model : nn.Module, data : Union[DataLoader, Dataset], loss_fn : nn.Module, config : dict) - list : if next(iter(model.parameters())).device.type ! config[device]: model model.to(config[device]) model.eval() if isinstance(data, DataLoader): dataloader data else: dataloader DataLoader(data, batch_sizeconfig[eval_batch_size], shuffleFalse, drop_lastFalse) loss [] accuracy [] batch_num [] with torch.no_grad(): for data in dataloader: for key in data: data[key] data[key].to(config[device]) preds model(data) loss.append(loss_fn(preds, data[label])) accuracy.append((torch.argmax(preds, dim-1).reshape(-1) data[label]).float().mean()) batch_num.append(len(data)) mean_loss sum([loss[i] * batch_num[i] for i in range(len(loss))]) / sum(batch_num) mean_accuracy sum([accuracy[i] * batch_num[i] for i in range(len(accuracy))]) / sum(batch_num) return mean_loss, mean_accuracy if __name__ __main__: config { seed: 0, lr: 0.001, weight_decay: 0.001, train_batch_size: 32, eval_batch_size: 32, epochs: 10, max_no_improve_epochs: 10, valid_ratio: 0.1, save_path: ./model/textrnn.ckpt, device: cuda if torch.cuda.is_available() else cpu, word2vec: cc.zh.300.bin, vocab: vocab.txt, train_file: data/toutiao_cat_data_train_1_tokenized.txt, test_file: data/toutiao_cat_data_test_tokenized.txt, max_length: 32 } print(-----------------config-----------------) print(config) # set seed same_seed(config[seed]) # 检查存放模型的目录是否存在如果不存在则创建 model_dir os.path.split(config[save_path])[0] if not os.path.isdir(model_dir): os.makedirs(model_dir) # load vocab tokenizer MyTokenizer(config[vocab], padpad) # dataset train_dataset MyTextDataset(config[train_file], None, config[max_length], tokenizer.pad_id) test_dataset MyTextDataset(config[test_file], train_dataset.labels, config[max_length], tokenizer.pad_id) labels train_dataset.labels print(\n.join(labels)) train_size int(len(train_dataset) * (1 -config[valid_ratio])) valid_size len(train_dataset) - train_size train_dataset, valid_dataset random_split(train_dataset, [train_size, valid_size]) # load word2vec embedding_matrix None if config[word2vec] ! : model fasttext.load_model(config[word2vec]) embedding_matrix [model.get_word_vector(tokenizer.index2word[i]) for i in range(tokenizer.vocab_size)] embedding_matrix torch.tensor(embedding_matrix) embedding_matrix[tokenizer.pad_id, :] torch.zeros(300) # create model model MySequenceClassificationModel2(tokenizer.vocab_size, 300, 300, len(labels), tokenizer.pad_id, embedding_matrix) # loss function loss_fn nn.CrossEntropyLoss(reductionmean) # optimizer optimizer torch.optim.AdamW(model.parameters(), lrconfig[lr], weight_decayconfig[weight_decay]) # train train(model, train_dataset, valid_dataset, loss_fn, optimizer, config) # test model.load_state_dict(torch.load(config[save_path], weights_onlyTrue)) test_loss, test_accuracy test(model, test_dataset, loss_fn, config) print(ftest_loss: {test_loss:.5f}, test_accuracy: {test_accuracy:.5f})