Python使用循環(huán)神經(jīng)網(wǎng)絡(luò)解決文本分類問題的方法詳解

# 實現(xiàn)LSTM網(wǎng)絡(luò)
  # 生成Cell網(wǎng)格所需參數(shù)
  def _generate_paramas(x_size, h_size, b_size):
    x_w = tf.get_variable('x_weight', x_size)
    h_w = tf.get_variable('h_weight', h_size)
    bias = tf.get_variable('bias', b_size, initializer=tf.constant_initializer(0.0))
    return x_w, h_w, bias
 
  scale = 1.0 / math.sqrt(embedding_size + lstm_nodes[-1]) / 3.0
  lstm_init = tf.random_uniform_initializer(-scale, scale)
  with tf.variable_scope('lstm_nn', initializer=lstm_init):
    # 輸入門參數(shù)
    with tf.variable_scope('input'):
      x_in, h_in, b_in = _generate_paramas(
        x_size=[embedding_size, lstm_nodes[0]],
        h_size=[lstm_nodes[0], lstm_nodes[0]],
        b_size=[1, lstm_nodes[0]]
      )
    # 輸出門參數(shù)
    with tf.variable_scope('output'):
      x_out, h_out, b_out = _generate_paramas(
        x_size=[embedding_size, lstm_nodes[0]],
        h_size=[lstm_nodes[0], lstm_nodes[0]],
        b_size=[1, lstm_nodes[0]]
      )
    # 遺忘門參數(shù)
    with tf.variable_scope('forget'):
      x_f, h_f, b_f = _generate_paramas(
        x_size=[embedding_size, lstm_nodes[0]],
        h_size=[lstm_nodes[0], lstm_nodes[0]],
        b_size=[1, lstm_nodes[0]]
      )
    # 中間狀態(tài)參數(shù)
    with tf.variable_scope('mid_state'):
      x_m, h_m, b_m = _generate_paramas(
        x_size=[embedding_size, lstm_nodes[0]],
        h_size=[lstm_nodes[0], lstm_nodes[0]],
        b_size=[1, lstm_nodes[0]]
      )
 
    # 兩個初始化狀態(tài)，隱含狀態(tài)state和初始輸入h
    state = tf.Variable(tf.zeros([batch_size, lstm_nodes[0]]), trainable=False)
    h = tf.Variable(tf.zeros([batch_size, lstm_nodes[0]]), trainable=False)
    # 遍歷LSTM每輪循環(huán)，即每個詞的輸入過程
    for i in range(max_words):
      # 取出每輪輸入，三維數(shù)組embedd_inputs的第二維代表訓練的輪數(shù)
      embedded_input = embedded_inputs[:, i, :]
      # 將取出的結(jié)果reshape為二維
      embedded_input = tf.reshape(embedded_input, [batch_size, embedding_size])
      # 遺忘門計算
      forget_gate = tf.sigmoid(tf.matmul(embedded_input, x_f) + tf.matmul(h, h_f) + b_f)
      # 輸入門計算
      input_gate = tf.sigmoid(tf.matmul(embedded_input, x_in) + tf.matmul(h, h_in) + b_in)
      # 輸出門
      output_gate = tf.sigmoid(tf.matmul(embedded_input, x_out) + tf.matmul(h, h_out) + b_out)
      # 中間狀態(tài)
      mid_state = tf.tanh(tf.matmul(embedded_input, x_m) + tf.matmul(h, h_m) + b_m)
      # 計算隱含狀態(tài)state和輸入h
      state = state * forget_gate + input_gate * mid_state
      h = output_gate + tf.tanh(state)
    # 最后遍歷的結(jié)果就是LSTM的輸出
    last_output = h

# 將文件中的句子通過jieba庫拆分為單個詞
def segment_word(input_file, output_file):
  # 循環(huán)遍歷訓練數(shù)據(jù)集的每一個文件
  for root, folders, files in os.walk(input_file):
    print('root:', root)
    for folder in folders:
      print('dir:', folder)
    for file in files:
      file_dir = os.path.join(root, file)
      with open(file_dir, 'rb') as in_file:
        # 讀取文件中的文本
        sentence = in_file.read()
        # 通過jieba函數(shù)庫將句子拆分為單個詞組
        words = jieba.cut(sentence)
        # 文件夾路徑最后兩個字即為分類名
        content = root[-2:] + '\t'
        # 去除詞組中的空格，排除為空的詞組
        for word in words:
          word = word.strip(' ')
          if word != '':
            content += word + ' '
      # 換行并將文本寫入輸出文件
      content += '\n'
      with open(output_file, 'a') as outfile:
        outfile.write(content.strip(' '))

# 統(tǒng)計每個詞出現(xiàn)的頻率
def get_list(segment_file, out_file):
  # 通過詞典保存每個詞組出現(xiàn)的頻率
  word_dict = {}
  with open(segment_file, 'r') as seg_file:
    lines = seg_file.readlines()
    # 遍歷文件的每一行
    for line in lines:
      line = line.strip('\r\n')
      # 將一行按空格拆分為每個詞，統(tǒng)計詞典
      for word in line.split(' '):
        # 如果這個詞組沒有在word_dict詞典中出現(xiàn)過，則新建詞典項并設(shè)為0
        word_dict.setdefault(word, 0)
        # 將詞典word_dict中詞組word對應(yīng)的項計數(shù)加一
        word_dict[word] += 1
    # 將詞典中的列表排序，關(guān)鍵字為列表下標為1的項，且逆序
    sorted_list = sorted(word_dict.items(), key=lambda d: d[1], reverse=True)
    with open(out_file, 'w') as outfile:
      # 將排序后的每條詞典項寫入文件
      for item in sorted_list:
        outfile.write('%s\t%d\n' % (item[0], item[1]))

# 定義超參數(shù)
embedding_size = 32 # 每個詞組向量的長度
max_words = 10 # 一個句子最大詞組長度
lstm_layers = 2 # lstm網(wǎng)絡(luò)層數(shù)
lstm_nodes = [64, 64] # lstm每層結(jié)點數(shù)
fc_nodes = 64 # 全連接層結(jié)點數(shù)
batch_size = 100 # 每個批次樣本數(shù)據(jù)
lstm_grads = 1.0 # lstm網(wǎng)絡(luò)梯度
learning_rate = 0.001 # 學習率
word_threshold = 10 # 詞表頻率門限，低于該值的詞語不統(tǒng)計
num_classes = 4 # 最后的分類結(jié)果有4類

class Word_list:
  def __init__(self, filename):
    # 用詞典類型來保存需要統(tǒng)計的詞組及其頻率
    self._word_dic = {}
    with open(filename, 'r',encoding='GB2312',errors='ignore') as f:
      lines = f.readlines()
    for line in lines:
      word, freq = line.strip('\r\n').split('\t')
      freq = int(freq)
      # 如果詞組的頻率小于閾值，跳過不統(tǒng)計
      if freq < word_threshold:
        continue
      # 詞組列表中每個詞組都是不重復(fù)的，按序添加到word_dic中即可，下一個詞組id就是當前word_dic的長度
      word_id = len(self._word_dic)
      self._word_dic[word] = word_id
 
  def sentence2id(self, sentence):
    # 將以空格分割的句子返回word_dic中對應(yīng)詞組的id，若不存在返回-1
    sentence_id = [self._word_dic.get(word, -1)
            for word in sentence.split()]
    return sentence_id
 
 
train_list = Word_list(train_list_dir)

class TextData:
  def __init__(self, segment_file, word_list):
    self.inputs = []
    self.labels = []
    # 通過詞典管理文本類別
    self.label_dic = {'體育': 0, '校園': 1, '女性': 2, '出版': 3}
    self.index = 0
 
    with open(segment_file, 'r') as f:
      lines = f.readlines()
      for line in lines:
        # 文本按制表符分割，前面為類別，后面為句子
        label, content = line.strip('\r\n').split('\t')[0:2]
        self.content_size = len(content)
        # 將類別轉(zhuǎn)換為數(shù)字id
        label_id = self.label_dic.get(label)
        # 將句子轉(zhuǎn)化為embedding數(shù)組
        content_id = word_list.sentence2id(content)
        # 如果句子的詞組長超過最大值，截取max_words長度以內(nèi)的id值
        content_id = content_id[0:max_words]
        # 如果不夠則填充-1，直到max_words長度
        padding_num = max_words - len(content_id)
        content_id = content_id + [-1 for i in range(padding_num)]
        self.inputs.append(content_id)
        self.labels.append(label_id)
    self.inputs = np.asarray(self.inputs, dtype=np.int32)
    self.labels = np.asarray(self.labels, dtype=np.int32)
    self._shuffle_data()
 
  # 對數(shù)據(jù)按照(input,label)對來打亂順序
  def _shuffle_data(self):
    r_index = np.random.permutation(len(self.inputs))
    self.inputs = self.inputs[r_index]
    self.labels = self.labels[r_index]
 
  # 返回一個批次的數(shù)據(jù)
  def next_batch(self, batch_size):
    # 當前索引+批次大小得到批次的結(jié)尾索引
    end_index = self.index + batch_size
    # 如果結(jié)尾索引大于樣本總數(shù)，則打亂所有樣本從頭開始
    if end_index > len(self.inputs):
      self._shuffle_data()
      self.index = 0
      end_index = batch_size
    # 按索引返回一個批次的數(shù)據(jù)
    batch_inputs = self.inputs[self.index:end_index]
    batch_labels = self.labels[self.index:end_index]
    self.index = end_index
    return batch_inputs, batch_labels
 
  # 獲取詞表數(shù)目
  def get_size(self):
    return self.content_size
 
# 訓練數(shù)據(jù)集對象
train_set = TextData(train_segment_dir, train_list)
# print(data_set.next_batch(10))
# 訓練數(shù)據(jù)集詞組條數(shù)
train_list_size = train_set.get_size()

# 創(chuàng)建計算圖模型
def create_model(list_size, num_classes):
  # 定義輸入輸出占位符
  inputs = tf.placeholder(tf.int32, (batch_size, max_words))
  outputs = tf.placeholder(tf.int32, (batch_size,))
  # 定義是否dropout的比率
  keep_prob = tf.placeholder(tf.float32, name='keep_rate')
  # 記錄訓練的總次數(shù)
  global_steps = tf.Variable(tf.zeros([], tf.float32), name='global_steps', trainable=False)
 
  # 將輸入轉(zhuǎn)化為embedding編碼
  with tf.variable_scope('embedding',
              initializer=tf.random_normal_initializer(-1.0, 1.0)):
    embeddings = tf.get_variable('embedding', [list_size, embedding_size], tf.float32)
    # 將指定行的embedding數(shù)值抽取出來
    embedded_inputs = tf.nn.embedding_lookup(embeddings, inputs)
 
  # 實現(xiàn)LSTM網(wǎng)絡(luò)
  scale = 1.0 / math.sqrt(embedding_size + lstm_nodes[-1]) / 3.0
  lstm_init = tf.random_uniform_initializer(-scale, scale)
  with tf.variable_scope('lstm_nn', initializer=lstm_init):
    # 構(gòu)建兩層的lstm，每層結(jié)點數(shù)為lstm_nodes[i]
    cells = []
    for i in range(lstm_layers):
      cell = tf.contrib.rnn.BasicLSTMCell(lstm_nodes[i], state_is_tuple=True)
      # 實現(xiàn)Dropout操作
      cell = tf.contrib.rnn.DropoutWrapper(cell, output_keep_prob=keep_prob)
      cells.append(cell)
    # 合并兩個lstm的cell
    cell = tf.contrib.rnn.MultiRNNCell(cells)
    # 將embedded_inputs輸入到RNN中進行訓練
    initial_state = cell.zero_state(batch_size, tf.float32)
    # runn_output:[batch_size,num_timestep,lstm_outputs[-1]
    rnn_output, _ = tf.nn.dynamic_rnn(cell, embedded_inputs, initial_state=initial_state)
    last_output = rnn_output[:, -1, :]
 
  # 構(gòu)建全連接層
  fc_init = tf.uniform_unit_scaling_initializer(factor=1.0)
  with tf.variable_scope('fc', initializer=fc_init):
    fc1 = tf.layers.dense(last_output, fc_nodes, activation=tf.nn.relu, name='fc1')
    fc1_drop = tf.contrib.layers.dropout(fc1, keep_prob)
    logits = tf.layers.dense(fc1_drop, num_classes, name='fc2')
 
  # 定義評估指標
  with tf.variable_scope('matrics'):
    # 計算損失值
    softmax_loss = tf.nn.sparse_softmax_cross_entropy_with_logits(logits=logits, labels=outputs)
    loss = tf.reduce_mean(softmax_loss)
    # 計算預(yù)測值，求第1維中最大值的下標，例如[1,1,5,3,2] argmax=> 2
    y_pred = tf.argmax(tf.nn.softmax(logits), 1, output_type=tf.int32)
    # 求準確率
    correct_prediction = tf.equal(outputs, y_pred)
    accuracy = tf.reduce_mean(tf.cast(correct_prediction, tf.float32))
 
  # 定義訓練方法
  with tf.variable_scope('train_op'):
    train_var = tf.trainable_variables()
    # for var in train_var:
    #   print(var)
    # 對梯度進行裁剪防止梯度消失或者梯度爆炸
    grads, _ = tf.clip_by_global_norm(tf.gradients(loss, train_var), clip_norm=lstm_grads)
    # 將梯度應(yīng)用到變量上去
    optimizer = tf.train.AdamOptimizer(learning_rate=learning_rate)
    train_op = optimizer.apply_gradients(zip(grads, train_var), global_steps)
 
  # 以元組的方式將結(jié)果返回
  return ((inputs, outputs, keep_prob),
      (loss, accuracy),
      (train_op, global_steps))
 
# 調(diào)用構(gòu)建函數(shù)，接收解析返回的參數(shù)
placeholders, matrics, others = create_model(train_list_size, num_classes)
inputs, outputs, keep_prob = placeholders
loss, accuracy = matrics
train_op, global_steps = others

# 進行訓練
init_op = tf.global_variables_initializer()
train_keep_prob = 0.8    # 訓練集的dropout比率
train_steps = 10000
 
with tf.Session() as sess:
  sess.run(init_op)
 
  for i in range(train_steps):
    # 按批次獲取訓練集數(shù)據(jù)
    batch_inputs, batch_labels = train_set.next_batch(batch_size)
    # 運行計算圖
    res = sess.run([loss, accuracy, train_op, global_steps],
            feed_dict={inputs: batch_inputs, outputs: batch_labels,
                 keep_prob: train_keep_prob})
    loss_val, acc_val, _, g_step_val = res
    if g_step_val % 20 == 0:
      print('第%d輪訓練，損失：%3.3f,準確率：%3.5f' % (g_step_val, loss_val, acc_val))