Chainerで文章分類をしてみた。

環境

chainer1.5付近
mac or ubuntu

Chainerの主なクラス

・Variable
変数の値の変化を記録する。
・Function
ネットワークの管理を行う。
・Link
ネットワークのパラメータを持つ。
・Chain
Linkをまとめたもの。
・Optimizer
ネットワークを操作する。

初心者なりにやってみた

ユニット

ユニット１個で書いてみました。
図を書きたかったけど、非常に大変で挫折。

１つのユニットで順伝播

# -*- coding: utf-8 -*-
from chainer import optimizers, cuda, serializers
import chainer.functions as F
import chainer.links as L
from chainer import FunctionSet, Variable
import numpy as np
##ユニット一つのモデルを作成してみた##
model = FunctionSet(l1 = L.Linear(4, 1)) # # １つのユニット。４つの入力と１つの出力。
x_data = np.random.rand(1, 4) * 100 # 4つのランダムな配列を作成
x_data = x_data.astype(np.float32) # 変換をする必要があった
x = Variable(x_data) # Variableはキャスト

print(float(model.l1(x).data))

順伝播

隠れ層１層のネットワークです。
784の入力で隠れ層が１０００で出力が１０。
mnistのチュートリアルのところから抜き取ったコードのぽいです。
softmax_cross_entropyの活性化関数で順伝播のみしてます。
逆伝播は何回か回したいので後で書いてみます。

順伝播

from chainer import FunctionSet, Variable
import chainer.functions as F
import numpy as np

# 多層パーセプトロンの定義
model = FunctionSet(l1=F.Linear( 784, 1000),
                    l2=F.Linear(1000, 1000),
                    l3=F.Linear(1000, 10))

def forward(x_data, y_data):
    x, t = Variable(x_data), Variable(y_data) # Variableはキャスト

    h1 = F.relu(model.l1(x))
    h2 = F.relu(model.l2(h1))
    y  = model.l3(h2)
    return F.softmax_cross_entropy(y, t)

x_data = np.random.rand(1, 784) * 100
y_data = np.array([0])            # ミニバッチを初期化
x_data = x_data.astype(np.float32) # 変換をする必要があった
y_data = y_data.astype(np.int32) # 変換をする必要があった
loss = forward(x_data, y_data)  # 順伝播
# loss.backward()                 # 逆伝播
print(float((loss.data)))

活性化関数

ニューロンを発火させて遊んでみました。
何回も叩くと発火したり発火しなかったり関数の違いがあっておもしろいです。

活性化関数

# -*- coding: utf-8 -*-
from chainer import optimizers, cuda, serializers
import chainer.functions as F
import chainer.links as L
from chainer import FunctionSet, Variable
import numpy as np
import matplotlib.pyplot as plt

def test_plot(func, max):
     xd = Variable(np.array([range(-max,max)], dtype=np.float32).T)
     yd = func(xd)
     plt.figure(figsize=(6,6))
     plt.plot(xd.data,yd.data)
     plt.show()

##ユニット一つのモデルを作成してみた##
model = FunctionSet(l1 = L.Linear(4, 1)) # # １つのユニット。４つの入力と１つの出力。
x_data = np.random.rand(1, 4) * 100 # 4つのランダムな配列を作成
x_data = x_data.astype(np.float32) # 変換をする必要があった
x = Variable(x_data) # Variableはキャスト

# 何度も実行した場合に活性化関数の動きを観察する。
# 数字が大きくなると発火を意味する。
print(float(model.l1(x).data))
u = model.l1(x)
z = F.sigmoid(u) # ロジスティックシグモイド関数
print(z.data)
z = F.tanh(u) # 双曲線正接関数
print(z.data)
z = F.relu(u) # 正規化線形関数
print(z.data)

# 活性化関数のグラフを表示してみる
# 困った時にお馴染みのランプ関数のサンプル
test_plot(F.relu, 10)
# シグモイド
test_plot(F.sigmoid, 10)
# タンジェント
test_plot(F.tanh, 10)

型変換

正解ラベル(本書だと望ましい出力d)の型を整える必要があります。と書いてありました。
実際に変換をしないでエラーになることが何度もあったのでメモしておきます。
上の方で書いた[１つのユニットで順伝播]でもキャストしないとエラーになっていました。

型変換

d = d.astype(numpy.float32) # 回帰
d = d.astype(numpy.int32) # 分類

誤差関数

yをネットワークの出力、dを望ましい出力。
学習データと正解のラベリングをペアで渡す。

誤差関数

loss = F.mean_squared_error(y, d) 
loss = F.softmax_cross_entropy(y, d)

フィードフォワードでテキスト分類

・パラメータ調整の履歴
隠れ層ユニット数調節
活性化関数の変更
学習回数の変更

テキストフィルタリング

# coding: utf-8
import numpy as np
from sklearn.cross_validation import train_test_split
from collections import defaultdict
import six
import sys
import chainer
import chainer.links as L
from chainer import optimizers, cuda, serializers
import chainer.functions as F
import argparse
from gensim import corpora, matutils
import MeCab

"""
単純なNNでテキスト分類 (posi-nega)
 - 隠れ層は2つ
 - 文書ベクトルにはBoWモデルを使用
 @author ichiroex
"""
# coding: utf-8
import numpy as np
from numpy import hstack, vsplit, hsplit, reshape
from sklearn.cross_validation import train_test_split
from collections import defaultdict
import six
import sys
import chainer
import chainer.links as L
from chainer import optimizers, cuda, serializers
import chainer.functions as F
import argparse
from gensim import corpora, matutils
import MeCab
from matplotlib import pyplot as plt


"""
単純なNNでテキスト分類 (posi-nega)
 - 隠れ層は2つ
 - 文書ベクトルにはBoWモデルを使用
 @author ichiroex
"""

class InitData():

    def __init__(self):
        source = []
        target = []
        comment = []
        word = []

    def to_words(self, sentence):
        """
        入力: 'すべて自分のほうへ'
        出力: tuple(['すべて', '自分', 'の', 'ほう', 'へ'])
        """
        tagger = MeCab.Tagger('mecabrc')  # 別のTaggerを使ってもいい
        mecab_result = tagger.parse(sentence)
        info_of_words = mecab_result.split('\n')
        words = []
        for info in info_of_words:
            # macabで分けると、文の最後に’’が、その手前に'EOS'が来る
            if info == 'EOS' or info == '':
                break
                # info => 'な\t助詞,終助詞,*,*,*,*,な,ナ,ナ'
            info_elems = info.split(',')
            # 6番目に、無活用系の単語が入る。もし6番目が'*'だったら0番目を入れる
            if info_elems[6] == '*':
                # info_elems[0] => 'ヴァンロッサム\t名詞'
                words.append(info_elems[0][:-3])
                continue
            words.append(info_elems[6])
        return tuple(words)

    def load_data(self, fname):

        source = []
        target = []
        comment = []
        word = []
        f = open(fname, "r")

        document_list = [] #各行に一文書. 文書内の要素は単語
        for l in f.readlines():
            sample = l.strip().split(" ", 1)        #ラベルと単語列を分ける
            # print(sample)
            label = int(sample[0])                  #ラベル
            target.append(label)
            # document_list.append(sample[1].split()) #単語分割して文書リストに追加
            document_list.append(self.to_words(sample[1])) #単語分割して文書リストに追加
            comment.append(sample[1])
            word.append(self.to_words(sample[1]))
            # print(sample[1])


        print(document_list)

        #単語辞書を作成
        dictionary = {}   
        dictionary = corpora.Dictionary(document_list)
        dictionary.filter_extremes(no_below=0, no_above=100) 
        # no_below: 使われている文書がno_below個以下の単語を無視
        # no_above: 使われてる文章の割合がno_above以上の場合無視

        #文書のベクトル化
        for document in document_list:
            tmp = dictionary.doc2bow(document) #文書をBoW表現
            vec = list(matutils.corpus2dense([tmp], num_terms=len(dictionary)).T[0]) 
            source.append(vec)


        dataset = {}
        dataset['source'] = np.array(source)
        dataset['target'] = np.array(target)
        dataset['comment'] = comment;
        dataset['word'] = word;
        print("------------")
        # print(dataset['source'])
        # print(dataset['target'])
        print(dataset['comment'])
        print("全データ数:", len(dataset['source'])) # データの数
        print ("辞書に登録された単語数:", len(dictionary.items())) # 辞書に登録された単語の数

        return dataset, dictionary





if __name__ == '__main__':

    #学習グラフ用
    losses =[]
    accuracies =[]

    initdata = InitData()

    #引数の設定
    parser = argparse.ArgumentParser()
    parser.add_argument('--gpu  '    , dest='gpu'        , type=int, default=0,            help='1: use gpu, 0: use cpu')
    parser.add_argument('--data '    , dest='data'       , type=str, default='input.dat',  help='an input data file')
    parser.add_argument('--epoch'    , dest='epoch'      , type=int, default=100,          help='number of epochs to learn')
    parser.add_argument('--batchsize', dest='batchsize'  , type=int, default=1,           help='learning minibatch size')
    parser.add_argument('--units'    , dest='units'      , type=int, default=100,           help='number of hidden unit')

    args = parser.parse_args()

    batchsize   = args.batchsize    # minibatch size
    n_epoch     = args.epoch        # エポック数(パラメータ更新回数)

    # Prepare dataset
    dataset, dictionary = initdata.load_data(args.data)

    dataset['source'] = dataset['source'].astype(np.float32) #文書ベクトル
    dataset['target'] = dataset['target'].astype(np.int32)   #ラベル

    x_train, x_test, y_train, y_test, c_train, c_test, word_train, word_test = train_test_split(dataset['source'], dataset['target'], dataset['comment'], dataset['word'], test_size=0.15)

    print("------------")
    print("x_train", x_train)
    print("x_test", x_test)
    # print(y_train)
    # print(y_test)

    N_test = y_test.size         # test data size
    N = len(x_train)             # train data size
    in_units = x_train.shape[1]  # 入力層のユニット数 (語彙数)
    print("学習データ数:", N)

    n_units = args.units # 隠れ層のユニット数
    n_label = 2          # 出力層のユニット数

    #モデルの定義
    model = chainer.Chain(l1=L.Linear(in_units, n_units),
                          l2=L.Linear(n_units, n_units),
                           l3=L.Linear(n_units,  n_label))

    # #GPUを使うかどうか
    # if args.gpu > 0:
        # cuda.check_cuda_available()
        # cuda.get_device(args.gpu).use()
        # model.to_gpu()
        # xp = np if args.gpu <= 0 else cuda.cupy #args.gpu <= 0: use cpu, otherwise: use gpu

    xp = np

    batchsize = args.batchsize
    n_epoch   = args.epoch

    def forward(x, t, train=True):
        h1 = F.relu(model.l1(x))
        h2 = F.relu(model.l2(h1))
        y = model.l3(h2)
        # h1 = F.dropout(F.relu(model.l1(x)), train=True)
        # h2 = F.dropout(F.relu(model.l2(h1)), train=True)
        # y = model.l3(h2)
        # h1 = F.dropout(F.relu(model.l1(x)))
        # h2 = F.dropout(F.relu(model.l2(h1)))
        # y = model.l3(h2)


        # print("l1:",h1.data.size)
        # print("l2:",h2.data.size)
        # print("l3:",y.data.size)
        # print(y.data)
        return F.softmax_cross_entropy(y, t), F.accuracy(y, t), y.data

    # Setup optimizer
    optimizer = optimizers.Adam()
    optimizer.setup(model)

    # Learning loop---------------------------------------------
    for epoch in six.moves.range(1, n_epoch + 1):

        print ('epoch', epoch)

        # training---------------------------------------------
        perm = np.random.permutation(N) #ランダムな整数列リストを取得
        sum_train_loss     = 0.0
        sum_train_accuracy = 0.0
        for i in six.moves.range(0, N, batchsize):

            #perm を使い x_train, y_trainからデータセットを選択 (毎回対象となるデータは異なる)
            x = chainer.Variable(xp.asarray(x_train[perm[i:i + batchsize]])) #source
            t = chainer.Variable(xp.asarray(y_train[perm[i:i + batchsize]])) #target
            c = chainer.Variable(xp.asarray(c_train[perm[i:i + batchsize]]))
            w = chainer.Variable(xp.asarray(word_train[perm[i:i + batchsize]]))

            # print("x_train ",x_train)
            # print("x_train ",np.vsplit(x_train, 2))

            model.zerograds()            # 勾配をゼロ初期化
            loss, acc, y = forward(x, t)    # 順伝搬
            sum_train_loss      += float(cuda.to_cpu(loss.data)) * len(t)   # 平均誤差計算用
            sum_train_accuracy  += float(cuda.to_cpu(acc.data )) * len(t)   # 平均正解率計算用
            loss.backward()              # 誤差逆伝播
            optimizer.update()           # 最適化 
            losses.append(loss.data) #誤差関数グラフ用


        print('train mean loss={}, accuracy={}'.format(
            sum_train_loss / N, sum_train_accuracy / N)) #平均誤差
        print(np.argmax(y), y, c.data, w.data) #出力結果一覧


        # evaluation---------------------------------------------
        sum_test_loss     = 0.0
        sum_test_accuracy = 0.0
        for i in six.moves.range(0, N_test, batchsize):

            # all test data
            x = chainer.Variable(xp.asarray(x_test[i:i + batchsize]))
            t = chainer.Variable(xp.asarray(y_test[i:i + batchsize]))
            c = chainer.Variable(xp.asarray(c_train[i:i + batchsize]))
            w = chainer.Variable(xp.asarray(word_train[i:i + batchsize]))

            loss, acc, y = forward(x, t, train=False)

            sum_test_loss     += float(cuda.to_cpu(loss.data)) * len(t)
            sum_test_accuracy += float(cuda.to_cpu(acc.data))  * len(t)
            accuracies.append(acc.data) #汎化性能グラフ用

        print(' test mean loss={}, accuracy={}'.format(
            sum_test_loss / N_test, sum_test_accuracy / N_test)) #平均誤差
        print(np.argmax(y), y, c.data, w.data) #出力結果一覧
        print("-----------------------------------")


    #modelとoptimizerを保存---------------------------------------------
    print ('save the model')
    serializers.save_npz('pn_classifier_ffnn.model', model)
    print ('save the optimizer')
    serializers.save_npz('pn_classifier_ffnn.state', optimizer)

    plt.plot(losses, label = "sum_train_loss")
    plt.plot(accuracies, label = "sum_train_accuracy")
    plt.yscale('log')
    plt.legend()
    plt.grid(True)
    plt.title("loss")
    plt.xlabel("epoch")
    plt.ylabel("loss")
    plt.show()

    # print ('load the model and optimizer')
    # serializers.load_npz('pn_classifier_ffnn.model', model)
    # serializers.load_npz('pn_classifier_ffnn.state', optimizer)

まだ問題として汎化能力測定が残っています。

corpus2denseを使っていますが、chainerのembeddingを使用してもいいかと思ってます。
embedding関数は次元の圧縮された単語情報を返します。
http://qiita.com/odashi_t/items/a1be7c4964fbea6a116e

ーー用意したデーターー
疑問文を1 
普通の文を0 
【input.dat】

0 明日天気良ければいいねー
0 天気のいい日はバーベキューでしょ。
0 梅雨の時期には雨が降らないと、水不足になる。
0 温暖化の影響で日本は暑くなっている。
0 天気がいい日はハイキングでしょ。
0 天気がいい日はウォーキングよね。

よくあるエラー

Expect: in_types[0].ndim >= 2
Actual: 0 < 2

入力x,dの次元数があっていない。
Actual: 0(実際の入力次元数) < 2(期待する次元数)

（対応例）
X = chainer.Variable(np.array(x[0], dtype=np.float32))
 ↓
X = chainer.Variable(np.atleast_2d(np.array(x[0], dtype=np.float32)))

データの読み込みを早くする

データをpklで保存する

data_train = [[1,1,1,2,2],[1,1,1,2,2]]
test_train = [2,1]
target_train = [[1,1,1,2,2],[1,1,1,2,2]]
target_test = [2,1]
data_sample = {'data': np.append(data_train, data_test, axis=0),
             'target': np.append(target_train, target_test, axis=0)}
with open('data_sample.pkl', 'wb') as output:
        six.moves.cPickle.dump(data_sample, output, -1)

データを読み込む

if not os.path.exists('data_sample.pkl'):
        download_data()
    with open('data_sample.pkl', 'rb') as data_sample_pickle:
        data_sample = six.moves.cPickle.load(data_dample_pickle)

書き方

Optimizerの使い方には二通り。

直接書く方法

x = np.random.uniform(-1, 1, (2, 4)).astype('f')
model.cleargrads()
# compute gradient here...
loss = F.sum(model(chainer.Variable(x)))
loss.backward()
optimizer.update()

関数をupdateに渡す

def lossfun(arg1, arg2):
    calculate loss
    loss = F.sum(model(arg1 - arg2))
    return loss
arg1 = np.random.uniform(-1, 1, (2, 4)).astype('f')
arg2 = np.random.uniform(-1, 1, (2, 4)).astype('f')
optimizer.update(lossfun, chainer.Variable(arg1), chainer.Variable(arg2))

mnistサンプル（chainer1.7）

mnistデータ28*28の画像で784次元。
60000枚の学習データ。

N = 60000
x_train, x_test = np.split(mnist['data'],   [N])
y_train, y_test = np.split(mnist['target'], [N])

log

load MNIST dataset
x_train (60000, 784)
y_train (60000,)

バッチサイズが100でスライス。

perm = np.random.permutation(N)
for i in six.moves.range(0, N, batchsize):
        x = chainer.Variable(xp.asarray(x_train[perm[i:i + batchsize]]))
        t = chainer.Variable(xp.asarray(y_train[perm[i:i + batchsize]]))

log

x shape: (100, 784)
t shape: (100,)

784次元を入力として更新

model = L.Classifier(net.MnistMLP(784, n_units, 10))
optimizer.update(model, x, t)

log

model内の次元
('x:', (100, 784))
('h1:', (100, 1000))
('h2:', (100, 1000))
('l3:', (100, 10))

オートエンコーダを簡単に書くライブラリ

ソースコード

参考ページ

お勉強をさせていただきました。