tensorflow学习教程之文本分类详析

# -*- coding: utf-8 -*-

import pandas as pd
import numpy as np
import tensorflow as tf
from collections import Counter
from sklearn.datasets import fetch_20newsgroups

def get_word_2_index(vocab):
 word2index = {}
 for i,word in enumerate(vocab):
 word2index[word] = i
 return word2index


def get_batch(df,i,batch_size):
 batches = []
 results = []
 texts = df.data[i*batch_size : i*batch_size+batch_size]
 categories = df.target[i*batch_size : i*batch_size+batch_size]
 for text in texts:
 layer = np.zeros(total_words,dtype=float)
 for word in text.split(' '):
  layer[word2index[word.lower()]] += 1
 batches.append(layer)
 
 for category in categories:
 y = np.zeros((3),dtype=float)
 if category == 0:
  y[0] = 1.
 elif category == 1:
  y[1] = 1.
 else:
  y[2] = 1.
 results.append(y)
 return np.array(batches),np.array(results)

def multilayer_perceptron(input_tensor, weights, biases):
 #hidden层RELU函数激励
 layer_1_multiplication = tf.matmul(input_tensor, weights['h1'])
 layer_1_addition = tf.add(layer_1_multiplication, biases['b1'])
 layer_1 = tf.nn.relu(layer_1_addition)
 
 layer_2_multiplication = tf.matmul(layer_1, weights['h2'])
 layer_2_addition = tf.add(layer_2_multiplication, biases['b2'])
 layer_2 = tf.nn.relu(layer_2_addition)
 
 # Output layer 
 out_layer_multiplication = tf.matmul(layer_2, weights['out'])
 out_layer_addition = out_layer_multiplication + biases['out']
 return out_layer_addition

#main
#从sklearn.datas获取数据
cate = ["comp.graphics","sci.space","rec.sport.baseball"]
newsgroups_train = fetch_20newsgroups(subset='train', categories=cate)
newsgroups_test = fetch_20newsgroups(subset='test', categories=cate)

# 计算训练和测试数据总数
vocab = Counter()
for text in newsgroups_train.data:
 for word in text.split(' '):
 vocab[word.lower()]+=1
 
for text in newsgroups_test.data:
 for word in text.split(' '):
 vocab[word.lower()]+=1

total_words = len(vocab)
word2index = get_word_2_index(vocab)

n_hidden_1 = 100 # 一层hidden层神经元个数
n_hidden_2 = 100 # 二层hidden层神经元个数
n_input = total_words 
n_classes = 3  # graphics, sci.space and baseball 3层输出层即将文本分为三类
#占位
input_tensor = tf.placeholder(tf.float32,[None, n_input],name="input")
output_tensor = tf.placeholder(tf.float32,[None, n_classes],name="output") 
#正态分布存储权值和偏差值
weights = {
 'h1': tf.Variable(tf.random_normal([n_input, n_hidden_1])),
 'h2': tf.Variable(tf.random_normal([n_hidden_1, n_hidden_2])),
 'out': tf.Variable(tf.random_normal([n_hidden_2, n_classes]))
}
biases = {
 'b1': tf.Variable(tf.random_normal([n_hidden_1])),
 'b2': tf.Variable(tf.random_normal([n_hidden_2])),
 'out': tf.Variable(tf.random_normal([n_classes]))
}

#初始化
prediction = multilayer_perceptron(input_tensor, weights, biases)

# 定义 loss and optimizer 采用softmax函数
# reduce_mean计算平均误差
loss = tf.reduce_mean(tf.nn.softmax_cross_entropy_with_logits(logits=prediction, labels=output_tensor))
optimizer = tf.train.AdamOptimizer(learning_rate=learning_rate).minimize(loss)

#初始化所有变量
init = tf.global_variables_initializer()

#部署 graph
with tf.Session() as sess:
 sess.run(init)
 training_epochs = 100
 display_step = 5
 batch_size = 1000
 # Training
 for epoch in range(training_epochs):
 avg_cost = 0.
 total_batch = int(len(newsgroups_train.data) / batch_size)
 for i in range(total_batch):
  batch_x,batch_y = get_batch(newsgroups_train,i,batch_size)
  c,_ = sess.run([loss,optimizer], feed_dict={input_tensor: batch_x,output_tensor:batch_y})
  # 计算平均损失
  avg_cost += c / total_batch
 # 每5次epoch展示一次loss
 if epoch % display_step == 0:
  print("Epoch:", '%d' % (epoch+1), "loss=", "{:.6f}".format(avg_cost))
 print("Finished!")

 # Test model
 correct_prediction = tf.equal(tf.argmax(prediction, 1), tf.argmax(output_tensor, 1))
 # 计算准确率
 accuracy = tf.reduce_mean(tf.cast(correct_prediction, "float"))
 total_test_data = len(newsgroups_test.target)
 batch_x_test,batch_y_test = get_batch(newsgroups_test,0,total_test_data)
 print("Accuracy:", accuracy.eval({input_tensor: batch_x_test, output_tensor: batch_y_test}))

# 计算训练和测试数据总数
vocab = Counter()
for text in newsgroups_train.data:
 for word in text.split(' '):
 vocab[word.lower()]+=1
 
for text in newsgroups_test.data:
 for word in text.split(' '):
 vocab[word.lower()]+=1

total_words = len(vocab)
word2index = get_word_2_index(vocab)

def get_batch(df,i,batch_size):
 batches = []
 results = []
 texts = df.data[i*batch_size : i*batch_size+batch_size]
 categories = df.target[i*batch_size : i*batch_size+batch_size]
 for text in texts:
 layer = np.zeros(total_words,dtype=float)
 for word in text.split(' '):
  layer[word2index[word.lower()]] += 1
 batches.append(layer)
 
 for category in categories:
 y = np.zeros((3),dtype=float)
 if category == 0:
  y[0] = 1.
 elif category == 1:
  y[1] = 1.
 else:
  y[2] = 1.
 results.append(y)
 return np.array(batches),np.array(results)

weights = {
 'h1': tf.Variable(tf.random_normal([n_input, n_hidden_1])),
 'h2': tf.Variable(tf.random_normal([n_hidden_1, n_hidden_2])),
 'out': tf.Variable(tf.random_normal([n_hidden_2, n_classes]))
}
biases = {
 'b1': tf.Variable(tf.random_normal([n_hidden_1])),
 'b2': tf.Variable(tf.random_normal([n_hidden_2])),
 'out': tf.Variable(tf.random_normal([n_classes]))
}

#初始化所有变量
init = tf.global_variables_initializer()

#部署 graph
with tf.Session() as sess:
 sess.run(init)
 training_epochs = 100
 display_step = 5
 batch_size = 1000
 # Training
 for epoch in range(training_epochs):
 avg_cost = 0.
 total_batch = int(len(newsgroups_train.data) / batch_size)
 for i in range(total_batch):
  batch_x,batch_y = get_batch(newsgroups_train,i,batch_size)
  c,_ = sess.run([loss,optimizer], feed_dict={input_tensor: batch_x,output_tensor:batch_y})
  # 计算平均损失
  avg_cost += c / total_batch
 # 每5次epoch展示一次loss
 if epoch % display_step == 0:
  print("Epoch:", '%d' % (epoch+1), "loss=", "{:.6f}".format(avg_cost))
 print("Finished!")

loss = tf.reduce_mean(tf.nn.softmax_cross_entropy_with_logits(logits=prediction, labels=output_tensor))
optimizer = tf.train.AdamOptimizer(learning_rate=learning_rate).minimize(loss)