基于tensorflow+CNN的垃圾邮件文本分类
2018年10月12日笔记
tensorflow是谷歌google的深度学习框架，tensor中文叫做张量，flow叫做流。
CNN是convolutional neural network的简称，中文叫做卷积神经网络。
文本分类是NLP（自然语言处理）的经典任务。
0.编程环境
操作系统：Win10
tensorflow版本：1.6
tensorboard版本：1.6
python版本：3.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/10QtokJ8_tkK6I3GifalxWg 提取码: uytb
压缩文件CNN垃圾邮件分类中有3个文件，如下图所示：
image.png
1.cnn_package.ipynb是本文的代码文件，读者可以直接运行
2.mailContent_list.pickle文件是本文作者处理好的邮件内容文件，可以用pickle.load方法加载；
3.mailLabel_list.pickle文件是本文作者处理好的邮件标签文件，可以用pickle.load方法加载。
数据集中共有2种分类：垃圾邮件用spam表示，正常邮件用ham表示。
垃圾邮件样本40000多条，正常邮件样本20000多条，样本总共60000多条。
4.完整代码
完整代码已经在数据集文件中给出，即cnn_package.ipynb文件；
从工程开发的角度考虑，本文作者封装了一个类TextClassification，对于样本数量在10万左右的分类任务较为适用。
后面章节中将讲解实现细节。
5.编写类TextClassification
认为编写类有困难，可以先阅读本文作者的另外一篇文章《基于tensorflow+CNN的新浪新闻文本分类》，链接：https://cloud.tencent.com/developer/article/1351828，这一篇文章的代码没有进行封装，而且讲解较为详细。
读者阅读下文中的行数时，可以先把代码复制到jupyter notebook的代码块中。
在代码块中按Esc键，进入命令模式，代码块左边的竖线会显示蓝色，如下图所示：
image.png
在命令模式下，点击L键，会显示代码行数。
推荐博客《Text-CNN 文本分类》从模型原理上辅助理解，链接：https://blog.csdn.net/chuchus/article/details/77847476
本文作者解释每行代码含义如下：
第1-15行代码导入程序运行必需的库；
第16-27行代码定义文本设置类TextConfig，经过本文作者实践，这是1种较为合适的编码方式；
第30-42行代码定义类中方法config，实例化TextConfig对象后，将其中的参数赋值给模型对象；
第44-62行代码定义对象实例化方法，如果参数数量是2个，其中第1个参数是内容列表content_list，第2个参数是标签列表label_list，则调用sklearn.preprocessing库的train_test_split方法划分训练集、测试集；
如果参数数量是4个，其中第1个参数是训练集内容列表train_content_list，第2个参数是训练集标签列表train_label_list，第3个参数是测试集内容列表test_content_list，第4个参数是测试集标签列表test_label_list，则都将它们保存为对象的属性；
第63行表示调用第65-67行代码的类中方法autoGetNumClasses，即自动获取类别的数量，并赋值给对象的属性num_classes；
第69-73行代码定义类中方法getVocabularyList，将列表中的字符串合并为1个字符串allContent_str，调用collections库的Counter方法，把allContent_str作为参数，即对字符串中的字符做统计计数，最后返回出现次数排名前vocabulary_size的数，即前5000的数；
第75-84行代码定义类中方法prepareData，即准备数据，根据数据的实际情况调整模型属性：词汇表大小vocab_size、序列长度seq_length、字转id字典word2id_dict、标签编码对象labelEncoder；
第86-87行代码定义类中方法content2idList，即文本内容转换为id列表；
第89-92行代码定义类中方法content2X，将文本内容列表content_list转换为特征矩阵X；
第94-97行代码定义类中方法label2Y，将文本标签列表label_list转换为预测目标值Y，具体方法是先调用LabelEncoder对象的transform方法做标签编码，然后调用kr.utils.to_categorical方法做Ont-Hot编码；
第99-118行代码定义类中方法buildModel，即搭建卷积神经网络模型，再次提醒，理解此部分代码有困难，可以先理解本文作者没有封装的代码，链接：https://cloud.tencent.com/developer/article/1351828；
第120-145行代码定义类中方法trainModel，即训练模型；
模型总共迭代训练num_iteration次，即5000次，每隔250次打印步数step、模型在测试集的损失值loss_value和准确率accuracy_value
第147-153行代码定义类中方法predict，传入参数是文本内容content_list，方法返回结果是标签列表label_list；
第155-162行代码定义类中方法predictAll，此方法作用是避免预测的样本过多导致显存不足，所以分批预测，每批100个样本，使用列表的extend方法将每批的预测结果合并；
第164-170行代码定义类中方法printConfusionMatrix，即打印模型在测试集的预测混淆矩阵；
第172-204行代码定义类中方法printReportTable，即打印模型在测试集的预测报告表。
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']]6.调用类中方法
第1-6行代码调用pickle库的load方法读取pickle文件中的数据；
第7行代码实例化TextClassification对象；
第8行代码调用模型对象的trainModel方法，即做模型训练；
第9行代码调用模型对象的printConfusionMatrix方法，即打印混淆矩阵；
第10行代码调用模型对象的printReportTable方法，即打印报告表；
import pickle
with open('mailContent_list.pickle', 'rb') as file:
content_list = pickle.load(file)
with open('mailLabel_list.pickle', 'rb') as file:
label_list = pickle.load(file)
model = TextClassification(content_list, label_list)
model.trainModel()
model.printConfusionMatrix()
model.printReportTable()上面一段代码的运行结果如下图所示，警告部分不影响程序运行：
image.png
7.总结
1.本文是作者第6个NLP小型项目，数据共有60000多条。
2.分类模型的评估指标F1score为0.994左右，总体来说这个分类模型很优秀，能够投入实际应用。
3.本文进行了类的封装，小型中文文本分类项目经过数据处理得到内容列表content_list和标签列表label_list之后，即可直接使用类做模型训练和预测，并且得到详细的预测结果报告表。