NLP进阶，使用TextRNN和TextRNN_ATT实现文本分类

举报
AI浩 发表于 2021/12/23 00:43:35 2021/12/23
【摘要】 TextRNN TextRNN仅仅是将Word Embedding后，输入到双向LSTM中，然后对最后一位的输出输入到全连接层中，在对其进行softmax分类即可，模型如下图： 代码： clas...

TextRNN

TextRNN仅仅是将Word Embedding后，输入到双向LSTM中，然后对最后一位的输出输入到全连接层中，在对其进行softmax分类即可，模型如下图：

img

代码：

class RNN(nn.Module):
    def __init__(self, vocab_size, embedding_dim, hidden_dim, output_dim,
                 n_layers=2, bidirectional=True, dropout=0.2, pad_idx=0):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embedding_dim, padding_idx=pad_idx)
        self.rnn = nn.LSTM(embedding_dim, hidden_dim, num_layers=n_layers,batch_first=True,
                           bidirectional=bidirectional)
        self.fc = nn.Linear(hidden_dim * 2, output_dim)
        # 这里hidden_dim乘以2是因为是双向，需要拼接两个方向，跟n_layers的层数无关。
        self.dropout = nn.Dropout(dropout)
    def forward(self, text):
        # text.shape=[seq_len, batch_size]
        embedded = self.dropout(self.embedding(text))
        # output: [batch,seq,2*hidden if bidirection else hidden]
        # hidden/cell: [bidirec * n_layers, batch, hidden]
        output, (hidden, cell) = self.rnn(embedded)
        # concat the final forward (hidden[-2,:,:]) and backward (hidden[-1,:,:]) hidden layers
        hidden = self.dropout(torch.cat((hidden[-2, :, :], hidden[-1, :, :]), dim=1))
        # hidden = [batch size, hid dim * num directions]，
        return self.fc(hidden.squeeze(0))  # 在接一个全连接层，最终输出[batch size, output_dim]

  
 
    
  
  • 1
    • 
  
  • 2
    • 
  
  • 3
    • 
  
  • 4
    • 
  
  • 5
    • 
  
  • 6
    • 
  
  • 7
    • 
  
  • 8
    • 
  
  • 9
    • 
  
  • 10
    • 
  
  • 11
    • 
  
  • 12
    • 
  
  • 13
    • 
  
  • 14
    • 
  
  • 15
    • 
  
  • 16
    • 
  
  • 17
    • 
  
  • 18
    • 
  
  • 19
    • 
  
  • 20
    •

TextRNN_ATT

在TextRNN的基础上加入注意力机制，代码：

class RNN_ATTs(nn.Module):
    def __init__(self, vocab_size, embedding_dim, hidden_dim, output_dim,
                 n_layers=2, bidirectional=True, dropout=0.2, pad_idx=0, hidden_size2=64):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embedding_dim, padding_idx=pad_idx)
        self.lstm = nn.LSTM(embedding_dim, hidden_dim, n_layers,
                            bidirectional=bidirectional, batch_first=True, dropout=dropout)
        self.tanh1 = nn.Tanh()
        # self.u = nn.Parameter(torch.Tensor(config.hidden_size * 2, config.hidden_size * 2))
        self.w = nn.Parameter(torch.zeros(hidden_dim * 2))
        self.tanh2 = nn.Tanh()
        self.fc1 = nn.Linear(hidden_dim * 2, hidden_size2)
        self.fc = nn.Linear(hidden_size2, output_dim)

    def forward(self, x):
        emb = self.embedding(x)  # [batch_size, seq_len, embeding]=[128, 32, 300]
        H, _ = self.lstm(emb)  # [batch_size, seq_len, hidden_size * num_direction]=[128, 32, 256]

        M = self.tanh1(H)  # [128, 32, 256]
        # M = torch.tanh(torch.matmul(H, self.u))
        alpha = F.softmax(torch.matmul(M, self.w), dim=1).unsqueeze(-1)  # [128, 32, 1]
        out = H * alpha  # [128, 32, 256]
        out = torch.sum(out, 1)  # [128, 256]
        out = F.relu(out)
        out = self.fc1(out)
        out = self.fc(out)  # [128, 64]
        return out


  
 
    
  
  • 1
    • 
  
  • 2
    • 
  
  • 3
    • 
  
  • 4
    • 
  
  • 5
    • 
  
  • 6
    • 
  
  • 7
    • 
  
  • 8
    • 
  
  • 9
    • 
  
  • 10
    • 
  
  • 11
    • 
  
  • 12
    • 
  
  • 13
    • 
  
  • 14
    • 
  
  • 15
    • 
  
  • 16
    • 
  
  • 17
    • 
  
  • 18
    • 
  
  • 19
    • 
  
  • 20
    • 
  
  • 21
    • 
  
  • 22
    • 
  
  • 23
    • 
  
  • 24
    • 
  
  • 25
    • 
  
  • 26
    • 
  
  • 27
    • 
  
  • 28
    •

数据集

数据集采用cnews数据集，包含三个文件，分别是cnews.train.txt,cnews.val.txt,cnews,test.txt。类别：体育, 娱乐, 家居, 房产, 教育, 时尚, 时政, 游戏, 科技, 财经，共10个类别。网盘地址：

链接：https://pan.baidu.com/s/1awlBYclO_mxntEgL_tUF0g
提取码：rtnv

构建词向量

第一步，读取预料，做分词。

思路：

1、创建默认方式的分词对象seg。

2、打开文件，按照行读取文章。

3、去掉收尾的空格，将label和文章分割开。

4、将分词后的文章放到src_data，label放入labels里。

5、返回结果。

我对代码做了注解，如下：

def read_corpus(file_path):
    """读取语料
    :param file_path:
    :param type:
    :return:
    """
    src_data = []
    labels = []
    seg = pkuseg.pkuseg() #使用默认分词方式。
    with codecs.open(file_path,'r',encoding='utf-8') as fout:
        for line in tqdm(fout.readlines(),desc='reading corpus'):
            if line is not None:
                # line.strip()的意思是去掉每句话句首句尾的空格
                # .split(‘\t’)的意思是根据'\t'把label和文章内容分开，label和内容是通过‘\t’隔开的。
                # \t表示空四个字符，也称缩进，相当于按一下Tab键
                pair = line.strip().split('\t')
                if len(pair) != 2:
                    print(pair)
                    continue
                src_data.append(seg.cut(pair[1]))# 对文章内容分词。
                labels.append(pair[0])
    return (src_data, labels) #返回文章内容的分词结果和labels


  
 
    
  
  • 1
    • 
  
  • 2
    • 
  
  • 3
    • 
  
  • 4
    • 
  
  • 5
    • 
  
  • 6
    • 
  
  • 7
    • 
  
  • 8
    • 
  
  • 9
    • 
  
  • 10
    • 
  
  • 11
    • 
  
  • 12
    • 
  
  • 13
    • 
  
  • 14
    • 
  
  • 15
    • 
  
  • 16
    • 
  
  • 17
    • 
  
  • 18
    • 
  
  • 19
    • 
  
  • 20
    • 
  
  • 21
    • 
  
  • 22
    • 
  
  • 23
    •

经过这个步骤得到了labels和分词后的文章。如下代码：

src_sents, labels = read_corpus('cnews/cnews.train.txt')

  
 
    
  
  • 1
    •

对labels做映射：

    labels = {label: idx for idx, label in enumerate(labels)}

  
 
    
  
  • 1
    •

得到labels对应的idx的字典，idx的值是最后一次插入label的值。

第二步 构建词向量

这一步主要用到vocab.py的from_corpus方法

思路：

1、创建vocab_entry对象。

2、对分词后的文章统计词频，生成一个词和词频构成的字典。

3、从字典中取出Top size - 2个元素。

4、获取元素的词。

5、执行add方法将词放入vocab_entry，生成词和id，id就是词对应的向量值。

代码如下：

    @staticmethod
    def from_corpus(corpus, size, min_feq=3):
        """从给定语料中创建VocabEntry"""
        vocab_entry = VocabEntry()
        # chain函数来自于itertools库，itertools库提供了非常有用的基于迭代对象的函数，而chain函数则是可以串联多个迭代对象来形成一个更大的迭代对象
        # *的作用：返回单个迭代器。
        # word_freq是个字典，key=词，value=词频
        word_freq = Counter(chain(*corpus))  # Counter 是实现的 dict 的一个子类，可以用来方便地计数,统计词频

        valid_words = word_freq.most_common(size - 2)  # most_common()函数用来实现Top n 功能，在这里选出Top size-2个词
        valid_words = [word for word, value in valid_words if value >= min_feq]  # 把符合要求的词找出来放到list里面。
        print('number of word types: {}, number of word types w/ frequency >= {}: {}'
              .format(len(word_freq), min_feq, len(valid_words)))
        for word in valid_words:  # 将词放进VocabEntry里面。
            vocab_entry.add(word)
        return vocab_entry

  
 
    
  
  • 1
    • 
  
  • 2
    • 
  
  • 3
    • 
  
  • 4
    • 
  
  • 5
    • 
  
  • 6
    • 
  
  • 7
    • 
  
  • 8
    • 
  
  • 9
    • 
  
  • 10
    • 
  
  • 11
    • 
  
  • 12
    • 
  
  • 13
    • 
  
  • 14
    • 
  
  • 15
    • 
  
  • 16
    •

创建完成后将词向量保存到json文件中

 vocab = Vocab.build(src_sents, labels, 50000, 3)
    print('generated vocabulary, source %d words' % (len(vocab.vocab)))
    vocab.save('./vocab.json')


  
 
    
  
  • 1
    • 
  
  • 2
    • 
  
  • 3
    • 
  
  • 4
    •

训练

训练使用Train_RNN.py,先看分析main方法的参数。

参数

    parse = argparse.ArgumentParser()

    parse.add_argument("--train_data_dir", default='./cnews/cnews.train.txt', type=str, required=False)
    parse.add_argument("--dev_data_dir", default='./cnews/cnews.val.txt', type=str, required=False)
    parse.add_argument("--test_data_dir", default='./cnews/cnews.test.txt', type=str, required=False)
    parse.add_argument("--output_file", default='deep_model.log', type=str, required=False)
    parse.add_argument("--batch_size", default=4, type=int)
    parse.add_argument("--do_train", default=True, action="store_true", help="Whether to run training.")
    parse.add_argument("--do_test", default=True, action="store_true", help="Whether to run training.")
    parse.add_argument("--learnning_rate", default=5e-4, type=float)
    parse.add_argument("--num_epoch", default=50, type=int)
    parse.add_argument("--max_vocab_size", default=50000, type=int)
    parse.add_argument("--min_freq", default=2, type=int)
    parse.add_argument("--hidden_size", default=256, type=int)
    parse.add_argument("--embed_size", default=300, type=int)
    parse.add_argument("--dropout_rate", default=0.2, type=float)
    parse.add_argument("--warmup_steps", default=0, type=int, help="Linear warmup over warmup_steps.")
    parse.add_argument("--GRAD_CLIP", default=1, type=float)
    parse.add_argument("--vocab_path", default='vocab.json', type=str)

  
 
    
  
  • 1
    • 
  
  • 2
    • 
  
  • 3
    • 
  
  • 4
    • 
  
  • 5
    • 
  
  • 6
    • 
  
  • 7
    • 
  
  • 8
    • 
  
  • 9
    • 
  
  • 10
    • 
  
  • 11
    • 
  
  • 12
    • 
  
  • 13
    • 
  
  • 14
    • 
  
  • 15
    • 
  
  • 16
    • 
  
  • 17
    • 
  
  • 18
    • 
  
  • 19
    •

参数说明：

train_data_dir：训练集路径。

dev_data_dir：验证集路径

test_data_dir：测试集路径

output_file：输出的log路径

batch_size：batchsize的大小。

do_train：是否训练，默认True、

do_test：是否测试，默认True

learnning_rate：学习率

num_epoch：epoch的数量

max_vocab_size：词向量的个数

min_freq：词频，过滤低于这个数值的词

hidden_size：隐藏层的个数

embed_size：Embedding的长度。

dropout_rate：dropout的值。

warmup_steps：设置预热的值。

vocab_path：词向量保存的路径

构建词向量

    vocab = build_vocab(args)
    label_map = vocab.labels
    print(label_map)

  
 
    
  
  • 1
    • 
  
  • 2
    • 
  
  • 3
    •

build_vocab的方法：

def build_vocab(args):
    if not os.path.exists(args.vocab_path):
        src_sents, labels = read_corpus(args.train_data_dir)
        labels = {label: idx for idx, label in enumerate(labels)}
        vocab = Vocab.build(src_sents, labels, args.max_vocab_size, args.min_freq)
        vocab.save(args.vocab_path)
    else:
        vocab = Vocab.load(args.vocab_path)
    return vocab

  
 
    
  
  • 1
    • 
  
  • 2
    • 
  
  • 3
    • 
  
  • 4
    • 
  
  • 5
    • 
  
  • 6
    • 
  
  • 7
    • 
  
  • 8
    • 
  
  • 9
    •

创建模型

创建CNN模型，将模型放到GPU上，调用train方法，训练。

  rnn_model = RNN_ATTs(len(vocab.vocab), args.embed_size, args.hidden_size,
                        len(label_map), n_layers=1, bidirectional=True, dropout=args.dropout_rate)
  rnn_model.to(device)
  train(args, rnn_model, train_data, dev_data, vocab, dtype='RNN')

  
 
    
  
  • 1
    • 
  
  • 2
    • 
  
  • 3
    • 
  
  • 4
    •

对train方法做了一些注解，如下：

def train(args, model, train_data, dev_data, vocab, dtype='CNN'):
    LOG_FILE = args.output_file
    #记录训练log
    with open(LOG_FILE, "a") as fout:
        fout.write('\n')
        fout.write('==========' * 6)
        fout.write('start trainning: {}'.format(dtype))
        fout.write('\n')

    time_start = time.time()
    if not os.path.exists(os.path.join('./runs', dtype)):
        os.makedirs(os.path.join('./runs', dtype))
    tb_writer = SummaryWriter(os.path.join('./runs', dtype))
    # 计算总的迭代次数
    t_total = args.num_epoch * (math.ceil(len(train_data) / args.batch_size))
    #optimizer = bnb.optim.Adam8bit(model.parameters(), lr=0.001, betas=(0.9, 0.995))  # add bnb optimizer
    optimizer = AdamW(model.parameters(), lr=args.learnning_rate, eps=1e-8)#设置优化器
    scheduler = get_linear_schedule_with_warmup(optimizer=optimizer, num_warmup_steps=args.warmup_steps,
                                                num_training_steps=t_total) #设置预热。
    criterion = nn.CrossEntropyLoss()# 设置loss为交叉熵
    global_step = 0
    total_loss = 0.
    logg_loss = 0.
    val_acces = []
    train_epoch = trange(args.num_epoch, desc='train_epoch')
    for epoch in train_epoch:#训练epoch
        model.train()
        for src_sents, labels in batch_iter(train_data, args.batch_size, shuffle=True):
            src_sents = vocab.vocab.to_input_tensor(src_sents, args.device)
            global_step += 1
            optimizer.zero_grad()
            logits = model(src_sents)
            y_labels = torch.tensor(labels, device=args.device)
            example_losses = criterion(logits, y_labels)
            example_losses.backward()
            torch.nn.utils.clip_grad_norm_(model.parameters(), args.GRAD_CLIP)
            optimizer.step()
            scheduler.step()

            total_loss += example_losses.item()
            if global_step % 100 == 0:
                loss_scalar = (total_loss - logg_loss) / 100
                logg_loss = total_loss

                with open(LOG_FILE, "a") as fout:
                    fout.write("epoch: {}, iter: {}, loss: {},learn_rate: {}\n".format(epoch, global_step, loss_scalar,
                                                                                       scheduler.get_lr()[0]))
                print("epoch: {}, iter: {}, loss: {}, learning_rate: {}".format(epoch, global_step, loss_scalar,
                                                                                scheduler.get_lr()[0]))
                tb_writer.add_scalar("lr", scheduler.get_lr()[0], global_step)
                tb_writer.add_scalar("loss", loss_scalar, global_step)

        print("Epoch", epoch, "Training loss", total_loss / global_step)
        eval_loss, eval_result = evaluate(args, criterion, model, dev_data, vocab)  # 评估模型
        with open(LOG_FILE, "a") as fout:
            fout.write("EVALUATE: epoch: {}, loss: {},eval_result: {}\n".format(epoch, eval_loss, eval_result))
        eval_acc = eval_result['acc']
        if len(val_acces) == 0 or eval_acc > max(val_acces):
            # 如果比之前的acc要da，就保存模型
            print("best model on epoch: {}, eval_acc: {}".format(epoch, eval_acc))
            torch.save(model.state_dict(), "classifa-best-{}.th".format(dtype))
            val_acces.append(eval_acc)

    time_end = time.time()
    print("run model of {},taking total {} m".format(dtype, (time_end - time_start) / 60))
    with open(LOG_FILE, "a") as fout:
        fout.write("run model of {},taking total {} m\n".format(dtype, (time_end - time_start) / 60))


  
 
    
  
  • 1
    • 
  
  • 2
    • 
  
  • 3
    • 
  
  • 4
    • 
  
  • 5
    • 
  
  • 6
    • 
  
  • 7
    • 
  
  • 8
    • 
  
  • 9
    • 
  
  • 10
    • 
  
  • 11
    • 
  
  • 12
    • 
  
  • 13
    • 
  
  • 14
    • 
  
  • 15
    • 
  
  • 16
    • 
  
  • 17
    • 
  
  • 18
    • 
  
  • 19
    • 
  
  • 20
    • 
  
  • 21
    • 
  
  • 22
    • 
  
  • 23
    • 
  
  • 24
    • 
  
  • 25
    • 
  
  • 26
    • 
  
  • 27
    • 
  
  • 28
    • 
  
  • 29
    • 
  
  • 30
    • 
  
  • 31
    • 
  
  • 32
    • 
  
  • 33
    • 
  
  • 34
    • 
  
  • 35
    • 
  
  • 36
    • 
  
  • 37
    • 
  
  • 38
    • 
  
  • 39
    • 
  
  • 40
    • 
  
  • 41
    • 
  
  • 42
    • 
  
  • 43
    • 
  
  • 44
    • 
  
  • 45
    • 
  
  • 46
    • 
  
  • 47
    • 
  
  • 48
    • 
  
  • 49
    • 
  
  • 50
    • 
  
  • 51
    • 
  
  • 52
    • 
  
  • 53
    • 
  
  • 54
    • 
  
  • 55
    • 
  
  • 56
    • 
  
  • 57
    • 
  
  • 58
    • 
  
  • 59
    • 
  
  • 60
    • 
  
  • 61
    • 
  
  • 62
    • 
  
  • 63
    • 
  
  • 64
    • 
  
  • 65
    • 
  
  • 66
    • 
  
  • 67
    • 
  
  • 68
    •

重点注释了一下batch_iter方法，如下：

def batch_iter(data, batch_size, shuffle=False):
    """
        batch数据
    :param data: list of tuple
    :param batch_size:
    :param shuffle:
    :return:
    """
    batch_num = math.ceil(len(data) / batch_size)# 计算迭代的次数
    index_array = list(range(len(data))) #按照data的长度，映射list
    if shuffle:#是否打乱顺序
        random.shuffle(index_array)

    for i in range(batch_num):
        indices = index_array[i*batch_size:(i+1)*batch_size]# 选出batchsize个index
        examples = [data[idx] for idx in indices]# 通过index找到对应的data
        examples = sorted(examples,key=lambda x: len(x[1]),reverse=True)#按照label排序
        src_sents = [e[0] for e in examples] #把data中的文章放到src_sents
        labels = [label_map[e[1]] for e in examples] #将标题映射label_map对应的value
        yield src_sents, labels


  
 
    
  
  • 1
    • 
  
  • 2
    • 
  
  • 3
    • 
  
  • 4
    • 
  
  • 5
    • 
  
  • 6
    • 
  
  • 7
    • 
  
  • 8
    • 
  
  • 9
    • 
  
  • 10
    • 
  
  • 11
    • 
  
  • 12
    • 
  
  • 13
    • 
  
  • 14
    • 
  
  • 15
    • 
  
  • 16
    • 
  
  • 17
    • 
  
  • 18
    • 
  
  • 19
    • 
  
  • 20
    • 
  
  • 21
    •

下面一个重要的方法是vocab.vocab.to_input_tensor，核心思路：

1、将数据通过 self.words2indices方法转为词对应的数值。

2、找出一个batch中最长的数据，剩下的数据后面补0，形成统一的长度。

3、将第二步得到的结果放入torch.tensor

代码如下：

 def to_input_tensor(self, sents: List[List[str]], device: torch.device):
        """
        将原始句子list转为tensor,同时将句子PAD成max_len
        :param sents: list of list<str>
        :param device:
        :return:
        """
        sents = self.words2indices(sents)
        sents = pad_sents(sents, self.word2id['<PAD>'])
        sents_var = torch.tensor(sents, device=device)
        return sents_var


  
 
    
  
  • 1
    • 
  
  • 2
    • 
  
  • 3
    • 
  
  • 4
    • 
  
  • 5
    • 
  
  • 6
    • 
  
  • 7
    • 
  
  • 8
    • 
  
  • 9
    • 
  
  • 10
    • 
  
  • 11
    • 
  
  • 12
    •

开始训练：

image-20211112210114989

验证

将do_train改为False，do_test改为True就可以开启验证模型，TextRNN能达到0.96的成绩。

parse.add_argument("--do_train", default=False, action="store_true", help="Whether to run training.")

  
 
    
  
  • 1
    •

image-20211112200928120
完整代码链接：
https://download.csdn.net/download/hhhhhhhhhhwwwwwwwwww/40816205

文章来源: wanghao.blog.csdn.net，作者：AI浩，版权归原作者所有，如需转载，请联系作者。

原文链接：wanghao.blog.csdn.net/article/details/121295961

【版权声明】本文为华为云社区用户转载文章，如果您发现本社区中有涉嫌抄袭的内容，欢迎发送邮件进行举报，并提供相关证据，一经查实，本社区将立刻删除涉嫌侵权内容，举报邮箱： cloudbbs@huaweicloud.com
机器学习 自然语言处理基础
  • 点赞
  • 收藏
  • 关注作者

评论（0）

0/1000
抱歉，系统识别当前为高风险访问，暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称，即可参与社区互动！

*长度不超过10个汉字或20个英文字符，设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符，设置后3个月内不可修改。

确认 取消

加入云驻计划，成为创作者

  • 华为云周边好礼
  • 免费体验产品
  • 特殊身份标识
  • 线下官方门票
  • 内部专家零距离
  • 与10000+优质创作者共同成长
立即加入