1 大綱概述
文本分類這個系列將會有十篇左右，包括基於word2vec預訓練的文本分類，與及基於最新的預訓練模型（ELMo，BERT等）的文本分類。總共有以下系列：
jupyter notebook代碼均在textClassifier倉庫中，python代碼在NLP-Project中的text_classfier中。
2 數據集
數據集為IMDB 電影影評，總共有三個數據文件，在/data/rawData目錄下，包括unlabeledTrainData.tsv，labeledTrainData.tsv，testData.tsv。在進行文本分類時需要有標簽的數據（labeledTrainData），數據預處理如文本分類實戰（一）—— word2vec預訓練詞向量中一樣，預處理后的文件為/data/preprocess/labeledTrain.csv。
3 textCNN 模型結構
textCNN 可以看作是n-grams的表現形式，textCNN介紹可以看這篇，論文Convolutional Neural Networks for Sentence Classification中提出的三種feature size的卷積核可以認為是對應了3-gram，4-gram和5-gram。整體模型結構如下，先用不同尺寸（3， 4， 5）的卷積核去提取特征，在進行最大池化，最后將不同尺寸的卷積核提取的特征拼接在一起作為輸入到softmax中的特征向量。
4 配置訓練參數
我們將模型參數，訓練參數等都配置在Config類中，方便之后調參。
import os import csv import time import datetime import random import json from collections import Counter from math import sqrt import gensim import pandas as pd import numpy as np import tensorflow as tf from sklearn.metrics import roc_auc_score, accuracy_score, precision_score, recall_score
# 配置參數 class TrainingConfig(object): epoches = 5 evaluateEvery = 100 checkpointEvery = 100 learningRate = 0.001 class ModelConfig(object): embeddingSize = 200 numFilters = 128 filterSizes = [2, 3, 4, 5] dropoutKeepProb = 0.5 l2RegLambda = 0.0 class Config(object): sequenceLength = 200 # 取了所有序列長度的均值 batchSize = 128 dataSource = "../data/preProcess/labeledTrain.csv" stopWordSource = "../data/english" numClasses = 1 # 二分類設置為1，多分類設置為類別的數目 rate = 0.8 # 訓練集的比例 training = TrainingConfig() model = ModelConfig() # 實例化配置參數對象 config = Config()
5 生成訓練數據
1）將數據加載進來，將句子分割成詞表示，並去除低頻詞和停用詞。
2）將詞映射成索引表示，構建詞匯-索引映射表，並保存成json的數據格式，之后做inference時可以用到。（注意，有的詞可能不在word2vec的預訓練詞向量中，這種詞直接用UNK表示）
3）從預訓練的詞向量模型中讀取出詞向量，作為初始化值輸入到模型中。
4）將數據集分割成訓練集和測試集
# 數據預處理的類，生成訓練集和測試集 class Dataset(object): def __init__(self, config): self.config = config self._dataSource = config.dataSource self._stopWordSource = config.stopWordSource self._sequenceLength = config.sequenceLength # 每條輸入的序列處理為定長 self._embeddingSize = config.model.embeddingSize self._batchSize = config.batchSize self._rate = config.rate self._stopWordDict = {} self.trainReviews = [] self.trainLabels = [] self.evalReviews = [] self.evalLabels = [] self.wordEmbedding =None self.labelList = [] def _readData(self, filePath): """ 從csv文件中讀取數據集 """ df = pd.read_csv(filePath) if self.config.numClasses == 1: labels = df["sentiment"].tolist() elif self.config.numClasses > 1: labels = df["rate"].tolist() review = df["review"].tolist() reviews = [line.strip().split() for line in review] return reviews, labels def _labelToIndex(self, labels, label2idx): """ 將標簽轉換成索引表示 """ labelIds = [label2idx[label] for label in labels] return labelIds def _wordToIndex(self, reviews, word2idx): """ 將詞轉換成索引 """ reviewIds = [[word2idx.get(item, word2idx["UNK"]) for item in review] for review in reviews] return reviewIds def _genTrainEvalData(self, x, y, word2idx, rate): """ 生成訓練集和驗證集 """ reviews = [] for review in x: if len(review) >= self._sequenceLength: reviews.append(review[:self._sequenceLength]) else: reviews.append(review + [word2idx["PAD"]] * (self._sequenceLength - len(review))) trainIndex = int(len(x) * rate) trainReviews = np.asarray(reviews[:trainIndex], dtype="int64") trainLabels = np.array(y[:trainIndex], dtype="float32") evalReviews = np.asarray(reviews[trainIndex:], dtype="int64") evalLabels = np.array(y[trainIndex:], dtype="float32") return trainReviews, trainLabels, evalReviews, evalLabels def _genVocabulary(self, reviews, labels): """ 生成詞向量和詞匯-索引映射字典，可以用全數據集 """ allWords = [word for review in reviews for word in review] # 去掉停用詞 subWords = [word for word in allWords if word not in self.stopWordDict] wordCount = Counter(subWords) # 統計詞頻 sortWordCount = sorted(wordCount.items(), key=lambda x: x[1], reverse=True) # 去除低頻詞 words = [item[0] for item in sortWordCount if item[1] >= 5] vocab, wordEmbedding = self._getWordEmbedding(words) self.wordEmbedding = wordEmbedding word2idx = dict(zip(vocab, list(range(len(vocab))))) uniqueLabel = list(set(labels)) label2idx = dict(zip(uniqueLabel, list(range(len(uniqueLabel))))) self.labelList = list(range(len(uniqueLabel))) # 將詞匯-索引映射表保存為json數據，之后做inference時直接加載來處理數據 with open("../data/wordJson/word2idx.json", "w", encoding="utf-8") as f: json.dump(word2idx, f) with open("../data/wordJson/label2idx.json", "w", encoding="utf-8") as f: json.dump(label2idx, f) return word2idx, label2idx def _getWordEmbedding(self, words): """ 按照我們的數據集中的單詞取出預訓練好的word2vec中的詞向量 """ wordVec = gensim.models.KeyedVectors.load_word2vec_format("../word2vec/word2Vec.bin", binary=True) vocab = [] wordEmbedding = [] # 添加 "pad" 和 "UNK", vocab.append("PAD") vocab.append("UNK") wordEmbedding.append(np.zeros(self._embeddingSize)) wordEmbedding.append(np.random.randn(self._embeddingSize)) for word in words: try: vector = wordVec.wv[word] vocab.append(word) wordEmbedding.append(vector) except: print(word + "不存在於詞向量中") return vocab, np.array(wordEmbedding) def _readStopWord(self, stopWordPath): """ 讀取停用詞 """ with open(stopWordPath, "r") as f: stopWords = f.read() stopWordList = stopWords.splitlines() # 將停用詞用列表的形式生成，之后查找停用詞時會比較快 self.stopWordDict = dict(zip(stopWordList, list(range(len(stopWordList))))) def dataGen(self): """ 初始化訓練集和驗證集 """ # 初始化停用詞 self._readStopWord(self._stopWordSource) # 初始化數據集 reviews, labels = self._readData(self._dataSource) # 初始化詞匯-索引映射表和詞向量矩陣 word2idx, label2idx = self._genVocabulary(reviews, labels) # 將標簽和句子數值化 labelIds = self._labelToIndex(labels, label2idx) reviewIds = self._wordToIndex(reviews, word2idx) # 初始化訓練集和測試集 trainReviews, trainLabels, evalReviews, evalLabels = self._genTrainEvalData(reviewIds, labelIds, word2idx, self._rate) self.trainReviews = trainReviews self.trainLabels = trainLabels self.evalReviews = evalReviews self.evalLabels = evalLabels data = Dataset(config) data.dataGen()
6 生成batch數據集
采用生成器的形式向模型輸入batch數據集，（生成器可以避免將所有的數據加入到內存中）
# 輸出batch數據集 def nextBatch(x, y, batchSize): """ 生成batch數據集，用生成器的方式輸出 """ perm = np.arange(len(x)) np.random.shuffle(perm) x = x[perm] y = y[perm] numBatches = len(x) // batchSize for i in range(numBatches): start = i * batchSize end = start + batchSize batchX = np.array(x[start: end], dtype="int64") batchY = np.array(y[start: end], dtype="float32") yield batchX, batchY
7 textCNN 模型
# 構建模型 class TextCNN(object): """ Text CNN 用於文本分類 """ def __init__(self, config, wordEmbedding): # 定義模型的輸入 self.inputX = tf.placeholder(tf.int32, [None, config.sequenceLength], name="inputX") self.inputY = tf.placeholder(tf.int32, [None], name="inputY") self.dropoutKeepProb = tf.placeholder(tf.float32, name="dropoutKeepProb") # 定義l2損失 l2Loss = tf.constant(0.0) # 詞嵌入層 with tf.name_scope("embedding"): # 利用預訓練的詞向量初始化詞嵌入矩陣 self.W = tf.Variable(tf.cast(wordEmbedding, dtype=tf.float32, name="word2vec") ,name="W") # 利用詞嵌入矩陣將輸入的數據中的詞轉換成詞向量，維度[batch_size, sequence_length, embedding_size] self.embeddedWords = tf.nn.embedding_lookup(self.W, self.inputX) # 卷積的輸入是思維[batch_size, width, height, channel]，因此需要增加維度，用tf.expand_dims來增大維度 self.embeddedWordsExpanded = tf.expand_dims(self.embeddedWords, -1) # 創建卷積和池化層 pooledOutputs = [] # 有三種size的filter，3， 4， 5，textCNN是個多通道單層卷積的模型，可以看作三個單層的卷積模型的融合 for i, filterSize in enumerate(config.model.filterSizes): with tf.name_scope("conv-maxpool-%s" % filterSize): # 卷積層，卷積核尺寸為filterSize * embeddingSize，卷積核的個數為numFilters # 初始化權重矩陣和偏置 filterShape = [filterSize, config.model.embeddingSize, 1, config.model.numFilters] W = tf.Variable(tf.truncated_normal(filterShape, stddev=0.1), name="W") b = tf.Variable(tf.constant(0.1, shape=[config.model.numFilters]), name="b") conv = tf.nn.conv2d( self.embeddedWordsExpanded, W, strides=[1, 1, 1, 1], padding="VALID", name="conv") # relu函數的非線性映射 h = tf.nn.relu(tf.nn.bias_add(conv, b), name="relu") # 池化層，最大池化，池化是對卷積后的序列取一個最大值 pooled = tf.nn.max_pool( h, ksize=[1, config.sequenceLength - filterSize + 1, 1, 1], # ksize shape: [batch, height, width, channels] strides=[1, 1, 1, 1], padding='VALID', name="pool") pooledOutputs.append(pooled) # 將三種size的filter的輸出一起加入到列表中 # 得到CNN網絡的輸出長度 numFiltersTotal = config.model.numFilters * len(config.model.filterSizes) # 池化后的維度不變，按照最后的維度channel來concat self.hPool = tf.concat(pooledOutputs, 3) # 攤平成二維的數據輸入到全連接層 self.hPoolFlat = tf.reshape(self.hPool, [-1, numFiltersTotal]) # dropout with tf.name_scope("dropout"): self.hDrop = tf.nn.dropout(self.hPoolFlat, self.dropoutKeepProb) # 全連接層的輸出 with tf.name_scope("output"): outputW = tf.get_variable( "outputW", shape=[numFiltersTotal, config.numClasses], initializer=tf.contrib.layers.xavier_initializer()) outputB= tf.Variable(tf.constant(0.1, shape=[config.numClasses]), name="outputB") l2Loss += tf.nn.l2_loss(outputW) l2Loss += tf.nn.l2_loss(outputB) self.logits = tf.nn.xw_plus_b(self.hDrop, outputW, outputB, name="logits") if config.numClasses == 1: self.predictions = tf.cast(tf.greater_equal(self.logits, 0.0), tf.int32, name="predictions") elif config.numClasses > 1: self.predictions = tf.argmax(self.logits, axis=-1, name="predictions") print(self.predictions) # 計算二元交叉熵損失 with tf.name_scope("loss"): if config.numClasses == 1: losses = tf.nn.sigmoid_cross_entropy_with_logits(logits=self.logits, labels=tf.cast(tf.reshape(self.inputY, [-1, 1]), dtype=tf.float32)) elif config.numClasses > 1: losses = tf.nn.sparse_softmax_cross_entropy_with_logits(logits=self.logits, labels=self.inputY) self.loss = tf.reduce_mean(losses) + config.model.l2RegLambda * l2Loss
8 定義計算metrics的函數
""" 定義各類性能指標 """ def mean(item: list) -> float: """ 計算列表中元素的平均值 :param item: 列表對象 :return: """ res = sum(item) / len(item) if len(item) > 0 else 0 return res def accuracy(pred_y, true_y): """ 計算二類和多類的准確率 :param pred_y: 預測結果 :param true_y: 真實結果 :return: """ if isinstance(pred_y[0], list): pred_y = [item[0] for item in pred_y] corr = 0 for i in range(len(pred_y)): if pred_y[i] == true_y[i]: corr += 1 acc = corr / len(pred_y) if len(pred_y) > 0 else 0 return acc def binary_precision(pred_y, true_y, positive=1): """ 二類的精確率計算 :param pred_y: 預測結果 :param true_y: 真實結果 :param positive: 正例的索引表示 :return: """ corr = 0 pred_corr = 0 for i in range(len(pred_y)): if pred_y[i] == positive: pred_corr += 1 if pred_y[i] == true_y[i]: corr += 1 prec = corr / pred_corr if pred_corr > 0 else 0 return prec def binary_recall(pred_y, true_y, positive=1): """ 二類的召回率 :param pred_y: 預測結果 :param true_y: 真實結果 :param positive: 正例的索引表示 :return: """ corr = 0 true_corr = 0 for i in range(len(pred_y)): if true_y[i] == positive: true_corr += 1 if pred_y[i] == true_y[i]: corr += 1 rec = corr / true_corr if true_corr > 0 else 0 return rec def binary_f_beta(pred_y, true_y, beta=1.0, positive=1): """ 二類的f beta值 :param pred_y: 預測結果 :param true_y: 真實結果 :param beta: beta值 :param positive: 正例的索引表示 :return: """ precision = binary_precision(pred_y, true_y, positive) recall = binary_recall(pred_y, true_y, positive) try: f_b = (1 + beta * beta) * precision * recall / (beta * beta * precision + recall) except: f_b = 0 return f_b def multi_precision(pred_y, true_y, labels): """ 多類的精確率 :param pred_y: 預測結果 :param true_y: 真實結果 :param labels: 標簽列表 :return: """ if isinstance(pred_y[0], list): pred_y = [item[0] for item in pred_y] precisions = [binary_precision(pred_y, true_y, label) for label in labels] prec = mean(precisions) return prec def multi_recall(pred_y, true_y, labels): """ 多類的召回率 :param pred_y: 預測結果 :param true_y: 真實結果 :param labels: 標簽列表 :return: """ if isinstance(pred_y[0], list): pred_y = [item[0] for item in pred_y] recalls = [binary_recall(pred_y, true_y, label) for label in labels] rec = mean(recalls) return rec def multi_f_beta(pred_y, true_y, labels, beta=1.0): """ 多類的f beta值 :param pred_y: 預測結果 :param true_y: 真實結果 :param labels: 標簽列表 :param beta: beta值 :return: """ if isinstance(pred_y[0], list): pred_y = [item[0] for item in pred_y] f_betas = [binary_f_beta(pred_y, true_y, beta, label) for label in labels] f_beta = mean(f_betas) return f_beta def get_binary_metrics(pred_y, true_y, f_beta=1.0): """ 得到二分類的性能指標 :param pred_y: :param true_y: :param f_beta: :return: """ acc = accuracy(pred_y, true_y) recall = binary_recall(pred_y, true_y) precision = binary_precision(pred_y, true_y) f_beta = binary_f_beta(pred_y, true_y, f_beta) return acc, recall, precision, f_beta def get_multi_metrics(pred_y, true_y, labels, f_beta=1.0): """ 得到多分類的性能指標 :param pred_y: :param true_y: :param labels: :param f_beta: :return: """ acc = accuracy(pred_y, true_y) recall = multi_recall(pred_y, true_y, labels) precision = multi_precision(pred_y, true_y, labels) f_beta = multi_f_beta(pred_y, true_y, labels, f_beta) return acc, recall, precision, f_beta
9 訓練模型
在訓練時，我們定義了tensorBoard的輸出，並定義了兩種模型保存的方法。
# 訓練模型 # 生成訓練集和驗證集 trainReviews = data.trainReviews trainLabels = data.trainLabels evalReviews = data.evalReviews evalLabels = data.evalLabels wordEmbedding = data.wordEmbedding labelList = data.labelList # 定義計算圖 with tf.Graph().as_default(): session_conf = tf.ConfigProto(allow_soft_placement=True, log_device_placement=False) session_conf.gpu_options.allow_growth=True session_conf.gpu_options.per_process_gpu_memory_fraction = 0.9 # 配置gpu占用率 sess = tf.Session(config=session_conf) # 定義會話 with sess.as_default(): cnn = TextCNN(config, wordEmbedding) globalStep = tf.Variable(0, name="globalStep", trainable=False) # 定義優化函數，傳入學習速率參數 optimizer = tf.train.AdamOptimizer(config.training.learningRate) # 計算梯度,得到梯度和變量 gradsAndVars = optimizer.compute_gradients(cnn.loss) # 將梯度應用到變量下，生成訓練器 trainOp = optimizer.apply_gradients(gradsAndVars, global_step=globalStep) # 用summary繪制tensorBoard gradSummaries = [] for g, v in gradsAndVars: if g is not None: tf.summary.histogram("{}/grad/hist".format(v.name), g) tf.summary.scalar("{}/grad/sparsity".format(v.name), tf.nn.zero_fraction(g)) outDir = os.path.abspath(os.path.join(os.path.curdir, "summarys")) print("Writing to {}\n".format(outDir)) lossSummary = tf.summary.scalar("loss", cnn.loss) summaryOp = tf.summary.merge_all() trainSummaryDir = os.path.join(outDir, "train") trainSummaryWriter = tf.summary.FileWriter(trainSummaryDir, sess.graph) evalSummaryDir = os.path.join(outDir, "eval") evalSummaryWriter = tf.summary.FileWriter(evalSummaryDir, sess.graph) # 初始化所有變量 saver = tf.train.Saver(tf.global_variables(), max_to_keep=5) # 保存模型的一種方式，保存為pb文件 savedModelPath = "../model/textCNN/savedModel" if os.path.exists(savedModelPath): os.rmdir(savedModelPath) builder = tf.saved_model.builder.SavedModelBuilder(savedModelPath) sess.run(tf.global_variables_initializer()) def trainStep(batchX, batchY): """ 訓練函數 """ feed_dict = { cnn.inputX: batchX, cnn.inputY: batchY, cnn.dropoutKeepProb: config.model.dropoutKeepProb } _, summary, step, loss, predictions = sess.run( [trainOp, summaryOp, globalStep, cnn.loss, cnn.predictions], feed_dict) timeStr = datetime.datetime.now().isoformat() if config.numClasses == 1: acc, recall, prec, f_beta = get_binary_metrics(pred_y=predictions, true_y=batchY) elif config.numClasses > 1: acc, recall, prec, f_beta = get_multi_metrics(pred_y=predictions, true_y=batchY, labels=labelList) trainSummaryWriter.add_summary(summary, step) return loss, acc, prec, recall, f_beta def devStep(batchX, batchY): """ 驗證函數 """ feed_dict = { cnn.inputX: batchX, cnn.inputY: batchY, cnn.dropoutKeepProb: 1.0 } summary, step, loss, predictions = sess.run( [summaryOp, globalStep, cnn.loss, cnn.predictions], feed_dict) if config.numClasses == 1: acc, precision, recall, f_beta = get_binary_metrics(pred_y=predictions, true_y=batchY) elif config.numClasses > 1: acc, precision, recall, f_beta = get_multi_metrics(pred_y=predictions, true_y=batchY, labels=labelList) evalSummaryWriter.add_summary(summary, step) return loss, acc, precision, recall, f_beta for i in range(config.training.epoches): # 訓練模型 print("start training model") for batchTrain in nextBatch(trainReviews, trainLabels, config.batchSize): loss, acc, prec, recall, f_beta = trainStep(batchTrain[0], batchTrain[1]) currentStep = tf.train.global_step(sess, globalStep) print("train: step: {}, loss: {}, acc: {}, recall: {}, precision: {}, f_beta: {}".format( currentStep, loss, acc, recall, prec, f_beta)) if currentStep % config.training.evaluateEvery == 0: print("\nEvaluation:") losses = [] accs = [] f_betas = [] precisions = [] recalls = [] for batchEval in nextBatch(evalReviews, evalLabels, config.batchSize): loss, acc, precision, recall, f_beta = devStep(batchEval[0], batchEval[1]) losses.append(loss) accs.append(acc) f_betas.append(f_beta) precisions.append(precision) recalls.append(recall) time_str = datetime.datetime.now().isoformat() print("{}, step: {}, loss: {}, acc: {},precision: {}, recall: {}, f_beta: {}".format(time_str, currentStep, mean(losses), mean(accs), mean(precisions), mean(recalls), mean(f_betas))) if currentStep % config.training.checkpointEvery == 0: # 保存模型的另一種方法，保存checkpoint文件 path = saver.save(sess, "../model/textCNN/model/my-model", global_step=currentStep) print("Saved model checkpoint to {}\n".format(path)) inputs = {"inputX": tf.saved_model.utils.build_tensor_info(cnn.inputX), "keepProb": tf.saved_model.utils.build_tensor_info(cnn.dropoutKeepProb)} outputs = {"predictions": tf.saved_model.utils.build_tensor_info(cnn.predictions)} prediction_signature = tf.saved_model.signature_def_utils.build_signature_def(inputs=inputs, outputs=outputs, method_name=tf.saved_model.signature_constants.PREDICT_METHOD_NAME) legacy_init_op = tf.group(tf.tables_initializer(), name="legacy_init_op") builder.add_meta_graph_and_variables(sess, [tf.saved_model.tag_constants.SERVING], signature_def_map={"predict": prediction_signature}, legacy_init_op=legacy_init_op) builder.save()
10 預測代碼
x = "this movie is full of references like mad max ii the wild one and many others the ladybug´s face it´s a clear reference or tribute to peter lorre this movie is a masterpiece we´ll talk much more about in the future" # 注：下面兩個詞典要保證和當前加載的模型對應的詞典是一致的 with open("../data/wordJson/word2idx.json", "r", encoding="utf-8") as f: word2idx = json.load(f) with open("../data/wordJson/label2idx.json", "r", encoding="utf-8") as f: label2idx = json.load(f) idx2label = {value: key for key, value in label2idx.items()} xIds = [word2idx.get(item, word2idx["UNK"]) for item in x.split(" ")] if len(xIds) >= config.sequenceLength: xIds = xIds[:config.sequenceLength] else: xIds = xIds + [word2idx["PAD"]] * (config.sequenceLength - len(xIds)) graph = tf.Graph() with graph.as_default(): gpu_options = tf.GPUOptions(per_process_gpu_memory_fraction=0.333) session_conf = tf.ConfigProto(allow_soft_placement=True, log_device_placement=False, gpu_options=gpu_options) sess = tf.Session(config=session_conf) with sess.as_default(): checkpoint_file = tf.train.latest_checkpoint("../model/textCNN/model/") saver = tf.train.import_meta_graph("{}.meta".format(checkpoint_file)) saver.restore(sess, checkpoint_file) # 獲得需要喂給模型的參數，輸出的結果依賴的輸入值 inputX = graph.get_operation_by_name("inputX").outputs[0] dropoutKeepProb = graph.get_operation_by_name("dropoutKeepProb").outputs[0] # 獲得輸出的結果 predictions = graph.get_tensor_by_name("output/predictions:0") pred = sess.run(predictions, feed_dict={inputX: [xIds], dropoutKeepProb: 1.0})[0] pred = [idx2label[item] for item in pred] print(pred)
11 總結
構建tensorflow深度學習模型時，最好能有一套自己的代碼框架，上述代碼就是我比較喜歡的框架，總共有四大類：參數配置，訓練數據生成，模型結構，訓練模型。在之后構建其他的模型時，依照這種結構可以很快地實現。建議大家尋找最適合自己的代碼結構，很多時候都可以實現代碼的復用。