プログラミング素人のはてなブログ

プログラミングも電気回路も専門外の技術屋の末端が勉強したことや作品をアウトプットするブログ。コードに間違いなど見つけられたら、気軽にコメントください。 C#、python3、ラズパイ始めました。

ニューラルネットワークに平均を理解させる

ちょっと大きめにでたタイトルではあるかもしれませんが。

課題定義

平均値とは電卓でもExcelでもPythonでも…、簡単に計算できるものではありますが、ニューラルネットワークの判断根拠というのは、こういったものとはちょっと違うと思います。

要するに、ニューラルネットワークは数字が与えらて「平均を求めろ!」と指示されたれたからといって、かならずしも足し算や割り算をして平均を求めるというわけではないのだと理解しています。

そこで、ここでは10個の数字の組み合わせを入力として与え、その平均を”答え”として与えることで、ニューラルネットワークが平均値を”理解”できるか試してみました。

そうは言っても、分類問題でしかないのでできて当たり前ではあります。

設計方針

以前作ったMnistのプログラムを改良します。
s51517765.hatenadiary.jp

入力
0~99までの整数のなかからランダムに10個

出力
0~99までの整数をOne hot encoding

ただし、ここでは100分割にしましたが実際の正解は少数があるので、正解ラベルはintにCastし±0.5未満であればOKとみなします。

プログラムの作成

Trainデータは3~max_numまでの乱数を10個入力します。
乱数の最大値を99に固定してしまうと、”答え”が偏ってしまうのではないかと思いこのようにしてみました。
numpyのListをx_train = np.array([])のように初期化し、np.appendで追加していきます。y_train = np.append(y_train, np.average(x_tmp))で平均が算出できます。y_train = y_train.astype(np.int32) で整数にします。

    x_train = np.array([])
    y_train = np.array([])

    # Trainデータ作成
    for j in range(train_size):
        x_tmp = np.zeros((list_size))
        max = random.randint(3, max_num - 1)
        for i in range(list_size):
            x_tmp[i] = random.randint(0, max - 1)
        y_train = np.append(y_train, np.average(x_tmp))
        x_train = np.append(x_train, x_tmp)
        y_train = y_train.astype(np.int32)  # 正解ラベルを整数に
    x_train = x_train.reshape(train_size, list_size)
    print(x_train)
    print(y_train)

モデルと学習結果を読み込む関数を作ってモデルビルドを行うことで、判定の関数からも共通して読み込むことができます。
また、学習結果を読み込むようにすることで、一旦学習を終了しても、追加学習をさせることができます。

def model_read():
    # モデルを読み込む
    model = Sequential()
    model.add(Dense(32, activation='relu', input_dim=list_size))
    model.add(Dense(32, activation='relu', input_dim=32))
    model.add(Dense(max_num, activation='softmax')) #softmax
    model.compile(optimizer='rmsprop', loss='categorical_crossentropy', metrics=['accuracy'])
    # 学習済みモデルの重みのデータを読み込み
    if os.path.exists(model_weight):
        model.load_weights(model_weight)  # 最初は重みデータファイルがないとErrorになる
    return model

学習結果を評価する関数ですが、x_test = np.append(x_test, x_tmp/max_num) 入力のところで正規化することに気づかず、学習結果は90%以上行っているのに、この評価関数では全く!?という状態になって悩みました。

同じような失敗をしている例がありまして、気づくことができました。
teratail.com

def check():
    model = model_read()
    x_test = np.array([])
    max = random.randint(3, max_num - 1)
    for i in range(list_size):
        x_tmp= random.randint(0, max - 1)
        x_test = np.append(x_test, x_tmp/max_num) #正規化して評価
    print(x_test)
    x_test = x_test.reshape(1, list_size)  # 1次元化
    # 判定
    res = model.predict([x_test])
    print("res = ", res)
    y = res.argmax()  # 値の中で最も値が大きいものが答え
    print("Prediction is ", y)
    ave = np.average(x_test)*100 #正規化された計算結果をもとの表示にもどす
    print("Anser is  ", int(ave))
    if abs(ave -  y) < 0.5:
        print("〇")
        return 1#正解数のカウント
    else:
        print("×")


コード全体

from keras.utils.np_utils import to_categorical
from keras.models import Sequential
from keras.layers import Dense
import random
import numpy as np
import os

train_size = 20000
test_size = 100
list_size = 10
max_num = 100  # 0-99

model_weight = 'cnn_ave.hdf5' #モデルを保存する名前

x_tmp = np.zeros((list_size))
# x_train=np.zeros((test_size,list_size)) #行, 列
np.set_printoptions(precision=1, suppress=True)  # 指数表示禁止、少数表示
np.set_printoptions(threshold=300000)  # 要素の省略禁止

def model_build():
    x_train = np.array([])
    x_test = np.array([])
    y_train = np.array([])
    y_test = np.array([])

    # Trainデータ作成
    for j in range(train_size):
        x_tmp = np.zeros((list_size))
        max = random.randint(3, max_num - 1)
        for i in range(list_size):
            x_tmp[i] = random.randint(0, max - 1)
        y_train = np.append(y_train, np.average(x_tmp))
        x_train = np.append(x_train, x_tmp)
        y_train = y_train.astype(np.int32)  # 正解ラベルを整数に
    x_train = x_train.reshape(train_size, list_size)

    # Testデータ作成
    for j in range(test_size):
        max = random.randint(2, max_num - 1)
        max = 30
        for i in range(list_size):
            x_tmp[i] = random.randint(0, max - 1)
        y_test = np.append(y_test, np.average(x_tmp))
        x_test = np.append(x_test, x_tmp)
        y_test = y_test.astype(np.int32)  # 正解ラベルを整数に
    x_test = x_test.reshape(test_size, list_size)

    # 0~1の範囲に変換(正規化)
    x_train = x_train.astype('float32') / max_num
    x_test = x_test.astype('float32') / max_num

    # 正解ラベルをone-hot-encoding
    y_train = to_categorical(y_train, max_num)
    y_test = to_categorical(y_test, max_num)

    model = model_read()
    model.compile(optimizer='rmsprop', loss='categorical_crossentropy', metrics=['accuracy'])
    model.fit(x_train, y_train, batch_size=32, epochs=300, verbose=1)  # データを学習
    model.save_weights(model_weight)  # 学習結果を保存
    score = model.evaluate(x_test, y_test)

def model_read():
    # モデルを読み込む
    model = Sequential()
    model.add(Dense(32, activation='relu', input_dim=list_size))
    model.add(Dense(32, activation='relu', input_dim=32))
    model.add(Dense(max_num, activation='softmax')) #softmax
    model.compile(optimizer='rmsprop', loss='categorical_crossentropy', metrics=['accuracy'])
    # 学習済みモデルの重みのデータを読み込み
    if os.path.exists(model_weight):
        model.load_weights(model_weight)  # 最初は重みデータファイルがないとErrorになる
    return model

def check():
    model = model_read()
    x_test = np.array([])
    max = random.randint(3, max_num - 1)
    for i in range(list_size):
        x_tmp= random.randint(0, max - 1)
        x_test = np.append(x_test, x_tmp/max_num) #正規化して評価
    print(x_test)
    x_test = x_test.reshape(1, list_size)  # 1次元化
    # 判定
    res = model.predict([x_test])
    print("res = ", res)
    y = res.argmax()  # 値の中で最も値が大きいものが答え
    print("Prediction is ", y)
    ave = np.average(x_test)*100 #正規化された計算結果をもとの表示にもどす
    print("Anser is  ", int(ave))
    if abs(ave -  y) < 0.5:
        print("〇")
        return 1
    else:
        print("×")

if __name__ == "__main__":
    model_build()
    correct_answer_rate = 0
    for i in range(30):
        if check() == 1:  # テスト実行
            correct_answer_rate += 1
    print("correct_answer_rate= ", int(correct_answer_rate/30*100),"%")