2018年10月13日笔记

tensorflow是谷歌google的深度学习框架，tensor中文叫做张量，flow叫做流。
CNN是convolutional neural network的简称，中文叫做卷积神经网络。
文本分类是NLP（自然语言处理）的经典任务。

0.编程环境

操作系统：Win10
python版本：3.6
集成开发环境：jupyter notebook
tensorflow版本：1.6

1.致谢声明

本文是作者学习《使用卷积神经网络以及循环神经网络进行中文文本分类》的成果，感激前辈；
github链接：https://github.com/gaussic/text-classification-cnn-rnn

2.配置环境

使用卷积神经网络模型要求有较高的机器配置，如果使用CPU版tensorflow会花费大量时间。
读者在有nvidia显卡的情况下，安装GPU版tensorflow会提高计算速度50倍。
安装教程链接：https://blog.csdn.net/qq_36556893/article/details/79433298
如果没有nvidia显卡，但有visa信用卡，请阅读我的另一篇文章《在谷歌云服务器上搭建深度学习平台》，链接：https://www.jianshu.com/p/893d622d1b5a

3.下载并解压数据集

数据集下载链接: https://pan.baidu.com/s/1cfHHr5aLk76AE7qi7L122g 提取码: tvgn
压缩文件夹基于CNN的搜狐新闻文本分类中有8个文件，如下图所示：

image.png

4.完整代码

完整代码已经在数据集文件中给出，即cnn_package.ipynb文件和getTrainTestDataSet.ipynb文件；
getTrainTestDataSet.ipynb文件中代码的作用是将文本文件转换为二进制文件，即4个pickle文件；
从工程开发的角度考虑，本文作者在cnn_package.ipynb文件中封装了一个类TextClassification，对于样本数量在10万左右的分类任务较为适用。
后面章节中将讲解实现细节。

5.获取数据

第1行代码导入库pandas，起别名pd；
第2行代码导入库pickle；
第4-5行代码分别读取训练集文件sohu_train.txt和测试集文件sohu_test.txt。
第6行代码将获取训练集内容列表train_content_list；
第7行代码将获取训练集标签列表train_label_list；
第8行代码将获取测试集内容列表test_content_list；
第9行代码将获取测试集标签列表test_label_list。
第10-17行代码调用pickle库的dump方法将python的对象转换为二进制文件。

import pandas as pd
import pickle

train_df = pd.read_csv('sohu_train.txt', sep='\t', header=None)
test_df = pd.read_csv('sohu_test.txt', sep='\t', header=None)
train_content_list = [k for k in train_df[1]]
train_label_list = [k for k in train_df[0]]
test_content_list = [k for k in test_df[1]]
test_label_list = [k for k in test_df[0]]
with open('train_content_list.pickle', 'wb') as file:
    pickle.dump(train_content_list, file)
with open('train_label_list.pickle', 'wb') as file:
    pickle.dump(train_label_list, file)
with open('test_content_list.pickle', 'wb') as file:
    pickle.dump(test_content_list, file)
with open('test_label_list.pickle', 'wb') as file:
    pickle.dump(test_label_list, file)

6.编写类TextClassification

认为编写类有困难，可以先阅读本文作者的另外一篇文章《基于tensorflow+CNN的新浪新闻文本分类》，链接：https://www.jianshu.com/p/b1000d5345bb，这一篇文章的代码没有进行封装，而且讲解较为详细。
读者阅读下文中的行数时，可以先把代码复制到jupyter notebook的代码块中。
在代码块中按Esc键，进入命令模式，代码块左边的竖线会显示蓝色，如下图所示：

image.png

from sklearn.model_selection import train_test_split
import pickle
from collections import Counter
import tensorflow.contrib.keras as kr
from sklearn.preprocessing import LabelEncoder
import tensorflow as tf
import random
import numpy as np
import pandas as pd
from sklearn.metrics import confusion_matrix
from sklearn.metrics import precision_recall_fscore_support
import warnings
warnings.filterwarnings('ignore')
import time

class TextConfig():
    vocab_size = 5000
    seq_length = 600
    embedding_dim = 64  # 词向量维度
    num_filters = 256  # 卷积核数目
    kernel_size = 5  # 卷积核尺
    hidden_dim = 128  # 全连接层神经元
    dropout_keep_prob = 0.5  # dropout保留比例
    learning_rate = 1e-3  # 学习率
    batch_size = 32  # 每批训练大小
    num_iteration = 5000 #迭代次数
    print_per_batch = num_iteration / 20 #打印间隔

class TextClassification():
    def config(self):
        textConfig = TextConfig()
        self.vocab_size = textConfig.vocab_size
        self.seq_length = textConfig.seq_length
        self.embedding_dim = textConfig.embedding_dim
        self.num_filters = textConfig.num_filters
        self.kernel_size = textConfig.kernel_size
        self.hidden_dim = textConfig.hidden_dim
        self.dropout_keep_prob = textConfig.dropout_keep_prob
        self.learning_rate = textConfig.learning_rate
        self.batch_size = textConfig.batch_size
        self.print_per_batch = textConfig.print_per_batch
        self.num_iteration = textConfig.num_iteration
    
    def __init__(self, *args):
        self.config()
        if len(args) == 2:
            content_list = args[0]
            label_list = args[1]
            train_X, test_X, train_y, test_y = train_test_split(content_list, label_list)
            self.train_content_list = train_X
            self.train_label_list = train_y
            self.test_content_list = test_X
            self.test_label_list = test_y
            self.content_list = self.train_content_list + self.test_content_list
        elif len(args) == 4:
            self.train_content_list = args[0]
            self.train_label_list = args[1]
            self.test_content_list = args[2]
            self.test_label_list = args[3]
            self.content_list = self.train_content_list + self.test_content_list
        else:
            print('false to init TextClassification object')
        self.autoGetNumClasses()
    
    def autoGetNumClasses(self):
        label_list = self.train_label_list + self.test_label_list
        self.num_classes = np.unique(label_list).shape[0]
    
    def getVocabularyList(self, content_list, vocabulary_size):
        allContent_str = ''.join(content_list)
        counter = Counter(allContent_str)
        vocabulary_list = [k[0] for k in counter.most_common(vocabulary_size)]
        return ['PAD'] + vocabulary_list

    def prepareData(self):
        vocabulary_list = self.getVocabularyList(self.content_list, self.vocab_size)
        if len(vocabulary_list) < self.vocab_size:
            self.vocab_size = len(vocabulary_list)
        contentLength_list = [len(k) for k in self.train_content_list]
        if max(contentLength_list) < self.seq_length:
            self.seq_length = max(contentLength_list)
        self.word2id_dict = dict([(b, a) for a, b in enumerate(vocabulary_list)])
        self.labelEncoder = LabelEncoder()
        self.labelEncoder.fit(self.train_label_list)

    def content2idList(self, content):
        return [self.word2id_dict[word] for word in content if word in self.word2id_dict]

    def content2X(self, content_list):
        idlist_list = [self.content2idList(content) for content in content_list]
        X = kr.preprocessing.sequence.pad_sequences(idlist_list, self.seq_length)
        return X

    def label2Y(self, label_list):
        y = self.labelEncoder.transform(label_list)
        Y = kr.utils.to_categorical(y, self.num_classes)
        return Y

    def buildModel(self):
        tf.reset_default_graph()
        self.X_holder = tf.placeholder(tf.int32, [None, self.seq_length])
        self.Y_holder = tf.placeholder(tf.float32, [None, self.num_classes])
        embedding = tf.get_variable('embedding', [self.vocab_size, self.embedding_dim])
        embedding_inputs = tf.nn.embedding_lookup(embedding, self.X_holder)
        conv = tf.layers.conv1d(embedding_inputs, self.num_filters, self.kernel_size)
        max_pooling = tf.reduce_max(conv, reduction_indices=[1])
        full_connect = tf.layers.dense(max_pooling, self.hidden_dim)
        full_connect_dropout = tf.contrib.layers.dropout(full_connect, keep_prob=self.dropout_keep_prob)
        full_connect_activate = tf.nn.relu(full_connect_dropout)
        softmax_before = tf.layers.dense(full_connect_activate, self.num_classes)
        self.predict_Y = tf.nn.softmax(softmax_before)
        cross_entropy = tf.nn.softmax_cross_entropy_with_logits_v2(labels=self.Y_holder, logits=softmax_before)
        self.loss = tf.reduce_mean(cross_entropy)
        optimizer = tf.train.AdamOptimizer(self.learning_rate)
        self.train = optimizer.minimize(self.loss)
        self.predict_y = tf.argmax(self.predict_Y, 1)
        isCorrect = tf.equal(tf.argmax(self.Y_holder, 1), self.predict_y)
        self.accuracy = tf.reduce_mean(tf.cast(isCorrect, tf.float32))

    def trainModel(self):
        self.prepareData()
        self.buildModel()
        init = tf.global_variables_initializer()
        self.session = tf.Session()
        self.session.run(init)
        train_X = self.content2X(self.train_content_list)
        train_Y = self.label2Y(self.train_label_list)
        test_X = self.content2X(self.test_content_list)
        test_Y = self.label2Y(self.test_label_list)
        startTime = time.time()
        for i in range(self.num_iteration):
            selected_index = random.sample(list(range(len(train_Y))), k=self.batch_size)
            batch_X = train_X[selected_index]
            batch_Y = train_Y[selected_index]
            self.session.run(self.train, {self.X_holder: batch_X, self.Y_holder: batch_Y})
            step = i + 1
            if step % self.print_per_batch == 0 or step == 1:
                selected_index = random.sample(list(range(len(test_Y))), k=200)
                batch_X = test_X[selected_index]
                batch_Y = test_Y[selected_index]
                loss_value, accuracy_value = self.session.run([self.loss, self.accuracy],\
                    {self.X_holder: batch_X, self.Y_holder: batch_Y})
                used_time = time.time() - startTime
                print('step:%d loss:%.4f accuracy:%.4f used time:%.2f seconds' %
                      (step, loss_value, accuracy_value, used_time))

    def predict(self, content_list):
        if type(content_list) == str:
            content_list = [content_list]
        batch_X = self.content2X(content_list)
        predict_y = self.session.run(self.predict_y, {self.X_holder:batch_X})
        predict_label_list = self.labelEncoder.inverse_transform(predict_y)
        return predict_label_list

    def predictAll(self):
        predict_label_list = []
        batch_size = 100
        for i in range(0, len(self.test_content_list), batch_size):
            content_list = self.test_content_list[i: i + batch_size]
            predict_label = self.predict(content_list)
            predict_label_list.extend(predict_label)
        return predict_label_list

    def printConfusionMatrix(self):
        predict_label_list = self.predictAll()
        df = pd.DataFrame(confusion_matrix(self.test_label_list, predict_label_list),
                     columns=self.labelEncoder.classes_,
                     index=self.labelEncoder.classes_)
        print('\n Confusion Matrix:')
        print(df)

    def printReportTable(self):
        predict_label_list = self.predictAll()
        reportTable = self.eval_model(self.test_label_list,
                                 predict_label_list,
                                 self.labelEncoder.classes_)
        print('\n Report Table:')
        print(reportTable)
        
    def eval_model(self, y_true, y_pred, labels):
        # 计算每个分类的Precision, Recall, f1, support
        p, r, f1, s = precision_recall_fscore_support(y_true, y_pred)
        # 计算总体的平均Precision, Recall, f1, support
        tot_p = np.average(p, weights=s)
        tot_r = np.average(r, weights=s)
        tot_f1 = np.average(f1, weights=s)
        tot_s = np.sum(s)
        res1 = pd.DataFrame({
            u'Label': labels,
            u'Precision': p,
            u'Recall': r,
            u'F1': f1,
            u'Support': s
        })
        res2 = pd.DataFrame({
            u'Label': ['总体'],
            u'Precision': [tot_p],
            u'Recall': [tot_r],
            u'F1': [tot_f1],
            u'Support': [tot_s]
        })
        res2.index = [999]
        res = pd.concat([res1, res2])
        return res[['Label', 'Precision', 'Recall', 'F1', 'Support']]

7.调用类中方法

第1-10行代码调用pickle库的load方法读取pickle文件中的数据；
第11-14行代码实例化TextClassification对象；
第15行代码调用模型对象的trainModel方法，即做模型训练；
第16行代码调用模型对象的printConfusionMatrix方法，即打印混淆矩阵；
第17行代码调用模型对象的printReportTable方法，即打印报告表；

import pickle

with open('train_content_list.pickle', 'rb') as file:
    train_content_list = pickle.load(file)
with open('train_label_list.pickle', 'rb') as file:
    train_label_list = pickle.load(file)
with open('test_content_list.pickle', 'rb') as file:
    test_content_list = pickle.load(file)
with open('test_label_list.pickle', 'rb') as file:
    test_label_list = pickle.load(file)
model = TextClassification(train_content_list,
                           train_label_list,
                           test_content_list,
                           test_label_list)
model.trainModel()
model.printConfusionMatrix()
model.printReportTable()

上面一段代码的运行结果如下图所示，警告部分不影响程序运行：

image.png

image.png

7.总结

1.本文是作者第7个NLP小型项目，数据共有36000条。
2.分类模型的评估指标F1score为0.95左右，总体来说这个分类模型很优秀，能够投入实际应用。
3.本文进行了类的封装，小型中文文本分类项目经过数据处理得到内容列表content_list和标签列表label_list之后，即可直接使用类做模型训练和预测，并且得到详细的预测结果报告表。