Tensorflow文字分類練習

初學tensorflow，借鑑了很多別人的經驗，參考部落格對評論分類(感謝博主的一系列好文)，本人也嘗試著實現了對文字資料的分類。

1、資料

這裡借用此部落格提供的負類資料和正類資料對程式進行驗證(再次感謝此博主)。這些資料的每一個樣本是對電影的正面或負面的評價。

2、nltk包的安裝和使用

對文字資料進行處理，需要藉助自然語言處理包NLTK (Natural Language Toolkit) 對每一個樣本進行預處理。

(1) 安裝 nltk

nltk的安裝可採用如下程式碼：

# pip install nltk 

(2) 下載 nltk data， 這是必要的nltk資料包，實現分詞、詞性標註、命名實體識別等功能都要用到這個資料包

$ python  # 進入python
>>> import nltk  # 匯入nltk
>>> nltk.download()  # 下載 nltk data

注意 nltk.download() 會彈出 NLTK Downloader 下載介面，此過程持續時間較長，請耐心等待。

下載過程中可能會出現某些 package 下載失敗的情況，此時可點選 All Packages 標籤，進而雙擊下載失敗的 package 可單獨下載，一般情況下都能下載成功。如果依然有問題，可移步至 nltk data 進行手動下載。

(3) 測試 nltk 是否安裝成功

$ python
from nltk.book import *
* **
text1
…
…
…
text9

若出現以上結果，則恭喜你， nltk 安裝成功。

3. 分類程式碼實現

# 用 one-hot vector 表示每個評論樣本
import numpy as np
from nltk.tokenize import word_tokenize
from nltk.stem import WordNetLemmatizer
from collections import Counter
import tensorflow as tf
import random

# 從檔案中獲得所包含的所有單詞，以及每句話所包含的單詞列表
 

def _get_words(fname):
    with open(fname, 'rb') as f:
        lines = f.readlines()

    lex = []  # 檔案中包含的所有單詞
    wwords = [] # 每句話包含的單詞列表
    for line in lines:
        try:
            # 將每個句子拆成一個個單詞
            words = word_tokenize(line.lower()) 
            # 將每個詞還原成原始詞性
            lemmatizer = WordNetLemmatizer()
            words = [lemmatizer.lemmatize(word) for word in words]
            lex += words
            wwords.append(words)
        except:
            pass

    return lex, wwords

# 彙集所有檔案所包含的詞，生成詞典
def get_dict(fnames):
    lexs = []
    wwords = []
    for fname in fnames:
        tmp_lex, tmp_words = _get_words(fname)
        lexs += tmp_lex
        wwords.append(tmp_words)

    word_count = Counter(lexs)
    dicts = []
    # 去掉常見詞和生僻詞，也可用比例進行限制
    for word in word_count:
        if word_count[word] >= 6000 or word_count[word] <= 200:
            continue

        dicts.append(word)

    return dicts, wwords

# 將每個樣本用向量表示，向量長度=詞典長度，樣本中出現的詞在詞典的對應位置用1表示，否則用0表示   
def vector_file(dicts, wwords, tags):
    datas =[]
    for words, tag in zip(wwords, tags):    
        codes = np.zeros((len(words), len(dicts)))
        for line_id, line_words in enumerate(words):
            for ix, word in enumerate(line_words):
                if word in dicts:
                    codes[line_id, ix] = 1

            datas.append([codes[line_id], tag])

    datas = np.array(datas)
    return datas

# 將資料集拆成訓練集和測試集
def get_train_test(fnames, tags, ratio=0.1):
    dicts, wwords = get_dict(fnames)
    data = vector_file(dicts, wwords, tags)

    te_size = np.int(ratio*len(data))

    tr_data = data[:-te_size]
    te_data = data[-te_size:] 

    return tr_data, te_data
#==========================================
# 前饋神經網路，hidden_layers 中每個元素代表對應隱藏層包含的神經元數；input_layer 為輸入向量的維數，這裡等於詞典的長度；output_layer 為樣本的類別數
def nn(data, input_layer, hidden_layers, output_layer):
    # ly_wbs 存放每層網路的引數
    ly_wbs = []
    ly_wb = {'w': tf.Variable(tf.random_normal([input_layer, hidden_layers[0]])), 
             'b': tf.Variable(tf.random_normal([hidden_layers[0]]))}
    ly_wbs.append(ly_wb)
    for ix in xrange(len(hidden_layers)):
        if ix < len(hidden_layers)-1:
            ly_wb = {'w': tf.Variable(tf.random_normal([hidden_layers[ix], hidden_layers[ix+1]])), 
                     'b': tf.Variable(tf.random_normal([hidden_layers[ix+1]]))}
            ly_wbs.append(ly_wb)

        else:
            ly_wb = {'w': tf.Variable(tf.random_normal([hidden_layers[-1], output_layer])), 
                     'b': tf.Variable(tf.random_normal([output_layer]))}
            ly_wbs.append(ly_wb)

    # 通過前饋網路，由輸入 data 得到神經網路的預測輸出  
    out = data        
    for ly_wb in ly_wbs:
        out = tf.add(tf.matmul(out, ly_wb['w']), ly_wb['b'])

    return out

# 訓練神經網路           
def train(X, Y, 
          tr_data, te_data, 
          input_layer, hidden_layers, output_layer, 
          batch_size=50, epoch=10):
    # 構建 tensor graph
    # 預測 op
    preY = nn(X, input_layer, hidden_layers, output_layer)
    # cost op，採用 softmax cross entropy
    cost_func = tf.reduce_mean(tf.nn.softmax_cross_entropy_with_logits(labels=Y, logits=preY))
    # optim op，可選不同的優化方法
    optim = tf.train.AdamOptimizer(learning_rate=0.001).minimize(cost_func)

    # 執行 ops
    with tf.Session() as sess:
        sess.run(tf.global_variables_initializer())

        i = 0
        for ix in xrange(epoch):
            epoch_loss = 0
            random.shuffle(tr_data)
            tr_x = tr_data[:, 0]
            tr_y = tr_data[:, 1]
            while i < len(tr_data):
                start = i
                end = i + batch_size
                if end > len(tr_x):
                    end = len(tr_x)

                batch_x = tr_x[start:end]
                batch_y = tr_y[start:end]

                _, c = sess.run([optim, cost_func], feed_dict={X:list(batch_x), Y:list(batch_y)})
                epoch_loss += c
                i += batch_size

            print 'epoch %d, loss %f' % (ix+1, c)

        te_x = te_data[:, 0]
        te_y = te_data[:, 1]

        # 計算分類準確率
        correct = tf.equal(tf.argmax(preY, 1), tf.argmax(Y, 1))
        accuracy = tf.reduce_mean(tf.cast(correct, 'float'))
        print 'test accuracy:', accuracy.eval({X:list(te_x), Y:list(te_y)})
#==========================================
# 主程式，資料預處理與分類
def main():
    pos_file = 'pos.txt'
    neg_file = 'neg.txt'

    fnames, tags = [pos_file, neg_file], [[0, 1], [1, 0]]
    tr_data, te_data = get_train_test(fnames, tags, ratio=0.1)

    input_layer, hidden_layers, output_layer = len(tr_data[0][0]), (1000, 800), len(tr_data[0][1])
    X = tf.placeholder('float', [None, input_layer])
    Y = tf.placeholder('float')
    train(X, Y, 
          tr_data, te_data, 
          input_layer, hidden_layers, output_layer, 
          batch_size=50, epoch=10) 

#=========================================================
# test codes
#=========================================================
if __name__ == '__main__':
    main()

以上程式可以完成文字分類任務，但準確率很低。究其原因，可能有三個：1）樣本量小；2）文字的預處理僅使用了簡單的 one-hot vector，處理過於粗糙；3）分類器使用了基本的三層前饋神經網路，引數也沒有調到最優。所以只能說提升空間還很大，仍需繼續努力…

不過當練手還是有所得的，姑且記下。