0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

深層学習Day3(ラビットチャレンジ)

Posted at

Section1:再帰型ニューラルネットワークの概念

再帰型ニューラルネットワーク(RNN:Recurrent Neural Network)は、時系列や系列データ(文、音声、株価など)を扱うために設計されたニューラルネットワーク
入力の「過去の情報」を記憶しながら次のステップに反映できるニューラルネットワーク
通常のニューラルネットワークとは異なり、時間的なつながり(履歴)を持つデータを処理

データタイプ
テキスト 単語や文(自然言語)
音声 音の波形やスペクトログラム
時系列 株価、センサー信号、気温

通常のニューラルネットワーク(前向き型)は、固定長の入力 → 固定長の出力で履歴を考慮しない。
RNNは次のように「前の隠れ状態(memory)」を時間ステップごとに再利用。
image.png

RNNの課題

勾配消失・勾配爆発問題
長い時系列になると、誤差が前の層まで届かない(勾配が消える)
逆に勾配が発散して数値が崩れることもある

長期依存の学習が苦手
例:10語前の単語の影響を保持するのが難しい

(参考)解決策:改良型RNN

モデル 特徴
LSTM(Long Short-Term Memory) ゲート機構で長期依存の記憶が可能に
GRU(Gated Recurrent Unit) LSTMを軽量化したもの(ゲートは2つ)
双方向RNN 過去と未来両方を考慮できる
Attention / Transformer RNNを使わず、全体に重みをつけて文脈を捉える

実装演習

バイナリ加算をRNNに学習

import numpy as np
from common import functions
import matplotlib.pyplot as plt

# def d_tanh(x):



# データを用意
# 2進数の桁数
binary_dim = 8
# 最大値 + 1
largest_number = pow(2, binary_dim)
# largest_numberまで2進数を用意
binary = np.unpackbits(np.array([range(largest_number)],dtype=np.uint8).T,axis=1)

input_layer_size = 2
hidden_layer_size = 16
output_layer_size = 1

weight_init_std = 1
learning_rate = 0.1

iters_num = 10000
plot_interval = 100

# ウェイト初期化 (バイアスは簡単のため省略)
W_in = weight_init_std * np.random.randn(input_layer_size, hidden_layer_size)
W_out = weight_init_std * np.random.randn(hidden_layer_size, output_layer_size)
W = weight_init_std * np.random.randn(hidden_layer_size, hidden_layer_size)

# Xavier


# He



# 勾配
W_in_grad = np.zeros_like(W_in)
W_out_grad = np.zeros_like(W_out)
W_grad = np.zeros_like(W)

u = np.zeros((hidden_layer_size, binary_dim + 1))
z = np.zeros((hidden_layer_size, binary_dim + 1))
y = np.zeros((output_layer_size, binary_dim))

delta_out = np.zeros((output_layer_size, binary_dim))
delta = np.zeros((hidden_layer_size, binary_dim + 1))

all_losses = []

for i in range(iters_num):
    
    # A, B初期化 (a + b = d)
    a_int = np.random.randint(largest_number/2)
    a_bin = binary[a_int] # binary encoding
    b_int = np.random.randint(largest_number/2)
    b_bin = binary[b_int] # binary encoding
    
    # 正解データ
    d_int = a_int + b_int
    d_bin = binary[d_int]
    
    # 出力バイナリ
    out_bin = np.zeros_like(d_bin)
    
    # 時系列全体の誤差
    all_loss = 0    
    
    # 時系列ループ
    for t in range(binary_dim):
        # 入力値
        X = np.array([a_bin[ - t - 1], b_bin[ - t - 1]]).reshape(1, -1)
        # 時刻tにおける正解データ
        dd = np.array([d_bin[binary_dim - t - 1]])
        
        u[:,t+1] = np.dot(X, W_in) + np.dot(z[:,t].reshape(1, -1), W)
        z[:,t+1] = functions.sigmoid(u[:,t+1])

        y[:,t] = functions.sigmoid(np.dot(z[:,t+1].reshape(1, -1), W_out))


        #誤差
        loss = functions.mean_squared_error(dd, y[:,t])
        
        delta_out[:,t] = functions.d_mean_squared_error(dd, y[:,t]) * functions.d_sigmoid(y[:,t])        
        
        all_loss += loss

        out_bin[binary_dim - t - 1] = np.round(y[:,t])
    
    
    for t in range(binary_dim)[::-1]:
        X = np.array([a_bin[-t-1],b_bin[-t-1]]).reshape(1, -1)        

        delta[:,t] = (np.dot(delta[:,t+1].T, W.T) + np.dot(delta_out[:,t].T, W_out.T)) * functions.d_sigmoid(u[:,t+1])

        # 勾配更新
        W_out_grad += np.dot(z[:,t+1].reshape(-1,1), delta_out[:,t].reshape(-1,1))
        W_grad += np.dot(z[:,t].reshape(-1,1), delta[:,t].reshape(1,-1))
        W_in_grad += np.dot(X.T, delta[:,t].reshape(1,-1))
    
    # 勾配適用
    W_in -= learning_rate * W_in_grad
    W_out -= learning_rate * W_out_grad
    W -= learning_rate * W_grad
    
    W_in_grad *= 0
    W_out_grad *= 0
    W_grad *= 0
    

    if(i % plot_interval == 0):
        all_losses.append(all_loss)        
        print("iters:" + str(i))
        print("Loss:" + str(all_loss))
        print("Pred:" + str(out_bin))
        print("True:" + str(d_bin))
        out_int = 0
        for index,x in enumerate(reversed(out_bin)):
            out_int += x * pow(2, index)
        print(str(a_int) + " + " + str(b_int) + " = " + str(out_int))
        print("------------")

lists = range(0, iters_num, plot_interval)
plt.plot(lists, all_losses, label="loss")
plt.show()

横軸:学習イテレーション、y軸:損失,MSE
初期は高い誤差からスタートし、4000イテレーションあたりから急激に降下し最終的に0で安定した⇒学習はうまく進んでいる
image.png

Section2:LSTM

再帰型ニューラルネットワーク(RNN)の一種で、長期依存関係を学習できるように改良されたモデル

項目 内容
正式名称 Long Short-Term Memory(長短期記憶)
提案者 Hochreiter & Schmidhuber(1997年)
分野 時系列予測・自然言語処理・音声認識など
特徴 長期依存の記憶が可能なRNNモデル

通常のRNNは、誤差逆伝播時に勾配が消失・爆発しやすく、過去の情報(長期依存)を学習しにくいという欠点。
LSTMはこれを「セル状態(記憶)」と「ゲート制御」によって克服。

LSTMのイメージ図

image.png

ゲート 役割
Forget Gate $f_t$ どの情報を忘れるか決める
Input Gate $i_t$ どの情報を追加するか決める
Output Gate $o_t$ どの情報を出力に使うか決める
セル状態 $c_t$ 長期記憶として保持される情報(直列)

数式まとめ
image.png
⊙:要素ごとの積(Hadamard積)

(参考)LSTMの利点・欠点

利点 内容
✅ 長期依存の保持 勾配消失を回避し、遠い過去の情報も記憶できる
✅ 自動で記憶の出し入れを学習 各ゲートが使う/使わないを学習
✅ 自然言語処理に強い 文脈や意味的な関係を保持しやすい
欠点 内容
❌パラメータ数が多い 通常のRNNの約4倍(ゲートごとに重みがある)
❌計算コストが高い 時間・メモリの面でやや重い
❌新モデルに比べ劣る場合も Transformer系の方がスケーラブルで高精度な場合が多い

Section3:GRU

GRU(Gated Recurrent Unit)は、LSTM(Long Short-Term Memory)を簡素化した再帰型ニューラルネットワーク(RNN)の改良モデル

項目 内容
名称 Gated Recurrent Unit(ゲート付き再帰ユニット)
提案年 2014年(Choら)
目的 LSTMと同様に長期記憶を扱うRNNの改良型
特徴 LSTMよりゲートが少なく、計算が軽い

確認テスト(GRUとLSTMの違い)

image.png

標準RNNは 勾配消失問題 により長期依存が苦手
⇒LSTMはその問題を解決したが構造が複雑(3つのゲート+セル状態)
⇒GRUは「LSTMの性能を保ちつつ、構造を簡略化」したモデル

項目 GRU LSTM
ゲート数 2(更新・リセット) 3(入力・出力・忘却)
セル状態 なし(隠れ状態のみ) あり(隠れ状態+セル状態)
パラメータ数 少なめ 多い
計算速度 速い(軽量) やや重い(精度高)
精度 LSTMに近い性能(タスクによる) 長期依存にはより強い傾向

(参考)GRUの内部構造と数式

更新ゲート(Update Gate):どの程度「前の情報」を保持するか
リセットゲート(Reset Gate):どの程度「過去の情報」を忘れるか
image.png
image.png

(参考)GRUの主な用途

分野 用途
自然言語処理 機械翻訳、チャットボット、文生成など
音声認識 音声波形 → テキスト変換
時系列予測 株価、気温、IoTセンサーなど
動作認識 動画中の行動分類など(時系列的動きの処理)

実装演習

import numpy as np
import re
import glob
import collections
import pickle

class Corpus:
    def __init__(self):
        self.unknown_word_symbol = "<???>" # 出現回数の少ない単語は未知語として定義しておく
        self.unknown_word_threshold = 3 # 未知語と定義する単語の出現回数の閾値
        self.corpus_file = "./corpus/**/*.txt"
        self.corpus_encoding = "utf-8"
        self.dictionary_filename = "./data_for_predict/word_dict.dic"
        self.chunk_size = 5
        self.load_dict()
        words = []
        for filename in glob.glob(self.corpus_file, recursive=True):
            with open(filename, "r", encoding=self.corpus_encoding) as f:
                # word breaking
                text = f.read()
                # 全ての文字を小文字に統一し、改行をスペースに変換
                text = text.lower().replace("\n", " ")
                # 特定の文字以外の文字を空文字に置換する
                text = re.sub(r"[^a-z '\-]", "", text)
                # 複数のスペースはスペース一文字に変換
                text = re.sub(r"[ ]+", " ", text)

                # 前処理: '-' で始まる単語は無視する
                words = [ word for word in text.split() if not word.startswith("-")]

        self.data_n = len(words) - self.chunk_size
        self.data = self.seq_to_matrix(words)

    def prepare_data(self):
        """
        訓練データとテストデータを準備する。
        data_n = ( text データの総単語数 ) - chunk_size
        input: (data_n, chunk_size, vocabulary_size)
        output: (data_n, vocabulary_size)
        """

        # 入力と出力の次元テンソルを準備
        all_input = np.zeros([self.chunk_size, self.vocabulary_size, self.data_n])
        all_output = np.zeros([self.vocabulary_size, self.data_n])

        # 準備したテンソルに、コーパスの one-hot 表現(self.data) のデータを埋めていく
        # i 番目から ( i + chunk_size - 1 ) 番目までの単語が1組の入力となる
        # このときの出力は ( i + chunk_size ) 番目の単語
        for i in range(self.data_n):
            all_output[:, i] = self.data[:, i + self.chunk_size] # (i + chunk_size) 番目の単語の one-hot ベクトル
            for j in range(self.chunk_size):
                all_input[j, :, i] = self.data[:, i + self.chunk_size - j - 1]

        # 後に使うデータ形式に合わせるために転置を取る
        all_input = all_input.transpose([2, 0, 1])
        all_output = all_output.transpose()

        # 訓練データ:テストデータを 4 : 1 に分割する
        training_num = ( self.data_n * 4 ) // 5
        return all_input[:training_num], all_output[:training_num], all_input[training_num:], all_output[training_num:]


    def build_dict(self):
        # コーパス全体を見て、単語の出現回数をカウントする
        counter = collections.Counter()
        for filename in glob.glob(self.corpus_file, recursive=True):
            with open(filename, "r", encoding=self.corpus_encoding) as f:

                # word breaking
                text = f.read()
                # 全ての文字を小文字に統一し、改行をスペースに変換
                text = text.lower().replace("\n", " ")
                # 特定の文字以外の文字を空文字に置換する
                text = re.sub(r"[^a-z '\-]", "", text)
                # 複数のスペースはスペース一文字に変換
                text = re.sub(r"[ ]+", " ", text)

                # 前処理: '-' で始まる単語は無視する
                words = [word for word in text.split() if not word.startswith("-")]

                counter.update(words)

        # 出現頻度の低い単語を一つの記号にまとめる
        word_id = 0
        dictionary = {}
        for word, count in counter.items():
            if count <= self.unknown_word_threshold:
                continue

            dictionary[word] = word_id
            word_id += 1
        dictionary[self.unknown_word_symbol] = word_id

        print("総単語数:", len(dictionary))

        # 辞書を pickle を使って保存しておく
        with open(self.dictionary_filename, "wb") as f:
            pickle.dump(dictionary, f)
            print("Dictionary is saved to", self.dictionary_filename)

        self.dictionary = dictionary
        print(self.dictionary)

    def load_dict(self):
        with open(self.dictionary_filename, "rb") as f:
            self.dictionary = pickle.load(f)
            self.vocabulary_size = len(self.dictionary)
            self.input_layer_size = len(self.dictionary)
            self.output_layer_size = len(self.dictionary)
            print("総単語数: ", self.input_layer_size)

    def get_word_id(self, word):
        # print(word)
        # print(self.dictionary)
        # print(self.unknown_word_symbol)
        # print(self.dictionary[self.unknown_word_symbol])
        # print(self.dictionary.get(word, self.dictionary[self.unknown_word_symbol]))
        return self.dictionary.get(word, self.dictionary[self.unknown_word_symbol])

    # 入力された単語を one-hot ベクトルにする
    def to_one_hot(self, word):
        index = self.get_word_id(word)
        data = np.zeros(self.vocabulary_size)
        data[index] = 1
        return data

    def seq_to_matrix(self, seq):
        # print(seq)
        data = np.array([self.to_one_hot(word) for word in seq]) # (data_n, vocabulary_size)
        return data.transpose() # (vocabulary_size, data_n)

def build_dict():
    cp = Corpus()
    cp.build_dict()
import time
import datetime

class Language:
    """
    input layer: self.vocabulary_size
    hidden layer: rnn_size = 30
    output layer: self.vocabulary_size
    """

    def __init__(self):
        self.corpus = Corpus()
        self.dictionary = self.corpus.dictionary
        self.vocabulary_size = len(self.dictionary) # 単語数
        self.input_layer_size = self.vocabulary_size # 入力層の数
        self.hidden_layer_size = 30 # 隠れ層の RNN ユニットの数
        self.output_layer_size = self.vocabulary_size # 出力層の数
        self.batch_size = 128 # バッチサイズ
        self.chunk_size = 5 # 展開するシーケンスの数。c_0, c_1, ..., c_(chunk_size - 1) を入力し、c_(chunk_size) 番目の単語の確率が出力される。
        self.learning_rate = 0.001 # 学習率
        self.epochs = 50 # 学習するエポック数
        self.forget_bias = 1.0 # LSTM における忘却ゲートのバイアス
        self.model_filename = "./data_for_predict/predict_model.ckpt"
        self.unknown_word_symbol = self.corpus.unknown_word_symbol

        # RNN 入力前の Embedding のパラメータ 
        self.hidden_w = tf.Variable(tf.random.truncated_normal([self.input_layer_size, self.hidden_layer_size], stddev=0.01))
        self.hidden_b = tf.Variable(tf.ones([self.hidden_layer_size]))

        # RNN 出力後の 全結合層のパラメータ
        self.output_w = tf.Variable(tf.random.truncated_normal([self.hidden_layer_size, self.output_layer_size], stddev=0.01))
        self.output_b = tf.Variable(tf.ones([self.output_layer_size]))

        # RNN 
        #self.rnn = tf.keras.layers.SimpleRNN(self.hidden_layer_size, activation='tanh', return_sequences=True)
        self.rnn = tf.keras.layers.SimpleRNN(self.hidden_layer_size, activation='tanh')
        # SimpleRNN Layer の weight を 強制的に生成させる 
        self.rnn(np.zeros((self.chunk_size, self.batch_size, self.hidden_layer_size),np.float32)) 

        self.trainable_variables = [self.hidden_w, self.hidden_b, self.output_w, self.output_b, *self.rnn.trainable_variables]

        self.optimizer = None

    def load_weights(self, ckpt_path):
        ckpt = tf.train.load_checkpoint(ckpt_path)

        # checkpoint から明示的に変数名を指定して保存
        self.hidden_w=tf.Variable(ckpt.get_tensor("hidden_w/.ATTRIBUTES/VARIABLE_VALUE"))
        self.hidden_b=tf.Variable(ckpt.get_tensor("hidden_b/.ATTRIBUTES/VARIABLE_VALUE"))
        self.output_w=tf.Variable(ckpt.get_tensor("output_w/.ATTRIBUTES/VARIABLE_VALUE"))
        self.output_b=tf.Variable(ckpt.get_tensor("output_b/.ATTRIBUTES/VARIABLE_VALUE"))
        k1 = tf.Variable(ckpt.get_tensor("rnn_kernel/.ATTRIBUTES/VARIABLE_VALUE"))
        k2 = tf.Variable(ckpt.get_tensor("rnn_reccurent_kernel/.ATTRIBUTES/VARIABLE_VALUE"))
        b  = tf.Variable(ckpt.get_tensor("rnn_bias/.ATTRIBUTES/VARIABLE_VALUE"))
        self.rnn.set_weights([k1,k2,b])
        return
    
    def save_weights(self, model_file):
        ckpt_tf2 = tf.train.Checkpoint(hidden_w=self.hidden_w, hidden_b=self.hidden_b, 
                               output_w=self.output_w, output_b=self.output_b, 
                               rnn_kernel=self.rnn.weights[0], rnn_reccurent_kernel=self.rnn.weights[1], rnn_bias=self.rnn.weights[2])
        save_path = ckpt_tf2.save(model_file)
        print(save_path, "was saved")
        return
        
    @tf.function
    def inference(self, input_data, initial_state):
        """
        :param input_data: (batch_size, chunk_size, vocabulary_size) 次元のテンソル
        :param initial_state: (batch_size, hidden_layer_size) 次元の行列
        :return:
        """
        batch_size, chunk_size, vocab_size = input_data.shape
        
        # 現時点での入力データは (batch_size, chunk_size, input_layer_size) という3次元のテンソル
        # chunkc_size * batch_size 分の単語に対して一気に 演算を行うため tf.transpose, tf.reshape を駆使してサイズ調整する

        # shape 調整
        input_data = tf.transpose(a=input_data, perm=[1, 0, 2]) # 転置。(chunk_size, batch_size, vocabulary_size)
        input_data = tf.reshape(input_data, [-1, self.input_layer_size]) # 変形。(chunk_size * batch_size, input_layer_size)
        # 単語(シンボル)の ベクトル化
        input_data = tf.matmul(input_data, self.hidden_w) + self.hidden_b # 重みWとバイアスBを適用。 (chunk_size * batch_size, hidden_layer_size)
        # shape を 元に戻す
        input_data = tf.reshape(input_data, [chunk_size, batch_size, self.hidden_layer_size]) # 変形。(chunk_size,  batch_size, hidden_layer_size)
        input_data = tf.transpose(a=input_data, perm=[1, 0, 2]) # 転置。(batch_size, chunk_size, hidden_layer_size)
            
        # RNN の演算 予測が行えればよいので 最後の単語のみ得る
        output = self.rnn(input_data, initial_state=initial_state)
        
        # 最後に隠れ層から出力層につながる重みとバイアスを処理する
        # 最終的に softmax 関数で処理し、確率として解釈される。
        # softmax 関数はこの関数の外で定義する。
        output = tf.matmul(output, self.output_w) + self.output_b

        # # print weights
        # print(self.hidden_w[0,0]) 
        # print(self.hidden_b[0]) 
        # print(self.output_w[0,0]) 
        # print(self.output_b[0]) 
        # print(self.rnn.weights[0][0,0]) 
        # print(self.rnn.weights[1][0,0]) 
        # print(self.rnn.weights[2][0]) 
                 
        return output

    def training(self):
        # 今回は最適化手法として Adam を選択する。
        # ここの Adam の部分を変えることで、Adagrad, Adadelta などの他の最適化手法を選択することができる
        optimizer = tf.optimizers.Adam(learning_rate=self.learning_rate)
        return optimizer

    @tf.function
    def loss(self, logits, labels):
        cost = tf.reduce_mean(input_tensor=tf.nn.softmax_cross_entropy_with_logits(logits=logits, labels=tf.stop_gradient(labels)))
        return cost

    @tf.function
    def accuracy(self, prediction, labels):
        correct = tf.equal(tf.argmax(input=prediction, axis=1), tf.argmax(input=labels, axis=1))
        accuracy = tf.reduce_mean(input_tensor=tf.cast(correct, tf.float32))
        return accuracy

    @tf.function
    def train_step(self, inputs, labels, initial_state):
        with tf.GradientTape() as tape:
            prediction = self.inference(inputs, initial_state)
            loss = self.loss(prediction, labels)

        gradients = tape.gradient(loss, self.trainable_variables)
        self.optimizer.apply_gradients(zip(gradients, self.trainable_variables))
        acc  = self.accuracy(prediction, labels)
        return loss, acc

    def train(self, model_file):
        """
        :param save_ckpt: 学習した重み係数を保存する checkpoint の名前
        :return:
        """
        # 訓練・テストデータの用意
        trX, trY, teX, teY = self.corpus.prepare_data()
        training_num = trX.shape[0]

        # ここから実際に学習を走らせる
        # エポックを回す
        log_train_acc = []
        log_train_loss = []
        # log_val_acc = []
        # log_val_loss = [] 
        self.optimizer = self.training()
        for epoch in range(self.epochs):
            step = 0
            epoch_loss = 0
            epoch_acc = 0

            # 訓練データをバッチサイズごとに分けて学習させる (= optimizer を走らせる)
            # エポックごとの損失関数の合計値や(訓練データに対する)精度も計算しておく
            while (step + 1) * self.batch_size < training_num:
                start_idx = step * self.batch_size
                end_idx = (step + 1) * self.batch_size

                batch_xs = tf.Variable(trX[start_idx:end_idx, :, :].astype(np.float32))
                batch_ys = tf.Variable(trY[start_idx:end_idx, :].astype(np.float32))
                initial_state = tf.Variable(np.zeros([self.batch_size, self.hidden_layer_size],dtype=np.float32))
                c, a = self.train_step(batch_xs, batch_ys, initial_state)
                # print("Epoch:", epoch, ", step:", step, "-- loss:", c, " -- accuracy:", a)
                epoch_loss += c
                epoch_acc += a
                step += 1
            # コンソールに損失関数の値や精度を出力しておく
            print("Epoch", epoch, "completed ouf of", self.epochs, "-- loss:", epoch_loss/step, " -- accuracy:",
                    epoch_acc / step)
            log_train_acc.append( (epoch_acc / step).numpy())
            log_train_loss.append((epoch_loss/step ).numpy() )
            
        # 最後にテストデータでの精度を計算して表示する
        inputs = tf.Variable(teX.astype(np.float32))
        initial_state = tf.Variable(np.zeros([teX.shape[0], self.hidden_layer_size],dtype=np.float32))
        labels = tf.Variable(teY.astype(np.float32))
        prediction = self.inference(inputs,initial_state)
        a = self.accuracy(prediction, labels)
        c = self.loss(prediction ,labels)
        # log_val_acc.append( a.numpy() )
        # log_val_loss.append( c.numpy() )

        history = {"train_acc": log_train_acc, "train_loss": log_train_loss
        #, "val_acc":log_val_acc, "val_loss":log_val_loss
        }
        print("Accuracy on test:", a.numpy())
        
        # 学習したモデルも保存しておく
        self.save_weights(model_file)
        return history
    
    def predict(self, seq):
        """
        文章を入力したときに次に来る単語を予測する
        :param seq: 予測したい単語の直前の文字列。chunk_size 以上の単語数が必要。
        :return: 
        """
        @tf.function
        def get_predictions(input_data, initial_state):
            return tf.nn.softmax(self.inference(input_data, initial_state))

        @tf.function
        def get_predicted_labels(predictions):
            return tf.argmax(predictions, axis=1)    

        # ----------- 入力データの作成
        # seq を one-hot 表現に変換する。
        words = [word for word in seq.split() if not word.startswith("-")]
        x = np.zeros([1, self.chunk_size, self.input_layer_size], dtype=np.float32)
        for i in range(self.chunk_size):
            word = seq[len(words) - self.chunk_size + i]
            index = self.dictionary.get(word, self.dictionary[self.unknown_word_symbol])
            x[0][i][index] = 1
        x = tf.Variable(x)
        initial_state = tf.Variable(np.zeros((1,self.hidden_layer_size), dtype=np.float32))
        
        # ----------- ロードしたモデルを使って各単語の出力確率を計算 (tensorflow による計算)
        u = get_predictions(x, initial_state)
         
        # ----------  結果表示
        keys = list(self.dictionary.keys())    

        # 各単語の確率の表示
        display_num = self.vocabulary_size # 10        
        print("各単語の出現確率 (降順)")
        sorted_index = np.argsort(-u[0])
        for i in sorted_index[:display_num]:
            c = self.unknown_word_symbol if i == (self.vocabulary_size - 1) else keys[i]
            print(c, ":", u[0][i].numpy())

        # 最も確率が大きいものを予測結果とする
        v = get_predicted_labels(u)
        print()
        print("Prediction:", seq + " " + ("<???>" if v[0] == (self.vocabulary_size - 1) else keys[v[0]]))


        return
ln = Language()

# 学習済みのパラメータをロード
ln.load_weights("./data_for_predict/predict_model")

# 保存したモデルを使って単語の予測をする
ln.predict("some of them looks like")

実装結果の一例
下記のように、似たような単語は数値として近い値となっていることがわかる
factors : 6.8704006e-20
state : 5.58566e-20
without : 5.092596e-20
us : 5.0687957e-20
examined : 3.9792115e-20
added : 3.4359525e-20
seven : 3.314005e-20
sixth : 2.4965946e-20
clinic : 2.3189118e-20
then : 2.1545402e-20
power : 1.424901e-20
months : 1.08236704e-20
except : 1.9111182e-21
well-known : 1.865203e-21
fair : 6.308525e-22
below : 2.9081745e-23

Section4:双方向RNN(Bidirectional RNN)

双方向RNN(Bidirectional RNN, BiRNN)は、通常のRNNが持つ「過去方向のみの情報伝搬」という制約を克服するためのモデル
→ 過去と未来の両方を見て出力を決定できるため、より文脈を深く理解できる

image.png
1つのRNNセルではなく、正方向と逆方向の2つのRNNを同時に動かす
最終的な出力は、両方向の隠れ状態を連結 or 合成して得る

数式
image.png

(参考)RNNの利点・欠点

メリット 説明
✅ 文脈の理解が深くなる 入力系列の未来と過去の両方を考慮できる
✅ 精度向上 機械翻訳・文分類・音声認識で良い効果
✅ LSTM/GRUとも組み合わせ可能 双方向LSTM / 双方向GRUも非常に一般的
デメリット 説明
❌ 推論で未来情報を使えない 時系列予測など「未来を見れないタスク」には不向き
❌ 計算コスト増加 RNNが2倍になるため、計算・メモリ負荷が増える
❌ リアルタイム処理に不適 全系列を見ないと逆方向が使えないため、遅延が生じる

双方向RNNの用途例

分野 具体例
自然言語処理 文分類、感情分析、品詞タグ付け、機械翻訳
音声処理 音声認識(音の前後を見て文字に変換)
医療 生体信号の解析(ECGなど)
映像処理 動作認識(前後のフレームを参照)

実装演習

演習チャレンジ
image.png
A.4(双方向RNNでは純方向と逆方向に伝播したときの中間層表現を合わせたものが特徴量となるので)

import numpy as np

def bidirectional_rnn_net(xs, W_f, U_f, W_b, U_b, V):
    """
    W_f, U_f : forward RNN weights, shape = (hidden_size, input_size)
    W_b, U_b : backward RNN weights, shape = (hidden_size, input_size)
    V       : output weights, shape = (output_size, 2 * hidden_size)
    """
    xs_f = np.zeros_like(xs)
    xs_b = np.zeros_like(xs)
    
    # 順方向と逆方向の入力系列の準備
    for i, x in enumerate(xs):
        xs_f[i] = x
        xs_b[i] = x[::-1]
    
    # 順方向と逆方向のRNN出力(中間層の系列出力)
    hs_f = _rnn(xs_f, W_f, U_f)
    hs_b = _rnn(xs_b, W_b, U_b)
    
    # 順・逆の中間出力を結合(axis=1 で横方向に連結)
    hs = [np.concatenate([h_f, h_b[::-1]], axis=1) for h_f, h_b in zip(hs_f, hs_b)]
    
    # 出力層へ
    ys = hs.dot(V.T)
    return ys

Section5:Seq2Seq(Sequence to Sequence)

Seq2Seq(Sequence-to-Sequence)モデルは、入力された系列データ(シーケンス)を、別の系列データに変換するためのニューラルネットワークアーキテクチャ

項目 内容
名称 Sequence-to-Sequence(系列から系列へ)
提案 Google(2014年, Sutskeverら)
特徴 入力と出力の長さが異なるシーケンスを扱える
構成 エンコーダ(Encoder)+ デコーダ(Decoder) で構成

[入力系列] → [エンコーダRNN] → [文脈ベクトル] → [デコーダRNN] → [出力系列]のイメージ
image.png

エンコーダ(Encoder)

RNN/LSTM/GRUなどで構成
入力系列を1語ずつ読み取り、最終隠れ状態を「意味の要約(コンテキスト)」とする

デコーダ(Decoder)

エンコーダの最終状態を初期状態として、1語ずつ出力を予測
出力を再帰的に次の入力として使用(Teacher Forcingもあり)

(参考)利点・欠点

項目 内容
✅ 柔軟性 入力と出力の長さが異なるタスク(翻訳・要約など)に対応
✅ RNN拡張と相性が良い LSTM/GRU/BiRNNと組み合わせやすい
❌ 情報圧縮の限界 エンコーダの最終状態(1つ)だけでは長文の情報が不足しがち(これを解決したのが「Attention」)

応用例

分野 タスク 具体例
機械翻訳 英語 → 日本語 "I am hungry" → "私はお腹がすいている"
文要約 長文 → 要約 論文の要約、自動キャプション生成
質問応答 質問 → 回答 "日本の首都は?" → "東京"
チャットボット 発話 → 応答 "こんにちは" → "こんにちは!調子はどう?"

Section6:Word2vec

Word2Vec は、単語をベクトル(数値)で表現する手法であり、単語間の意味的な類似性を計算可能にすることを目的としたニューラルネットワークベースの手法
意味的な関係が空間的に反映される⇒アナロジー(類推)を計算できる

従来の「単語の表現」は以下のようなもの:One-hot ベクトル

単語数分の次元(例:語彙数10,000 → 長さ10,000のベクトル)
各単語が1つの次元で「1」、他は「0」
問題点:意味的な距離が全く反映されない(例:cat と dog のベクトルの距離が rabbit と pen と同じ)

(参考)Word2Vec 2つの学習モデル

1.CBOW(Continuous Bag of Words)
文脈(周囲の単語)から中心の単語を予測
例:[The, cat, ___, on, the] → mat を予測

2.Skip-gram(スキップグラム)
中心の単語から文脈(周囲の単語)を予測
例:中心 = cat → 周囲 = ["The", "sat", "on"]

比較項目 CBOW Skip-gram
入力 周囲の単語 中心の単語
出力 中心の単語 周囲の単語
特徴 学習が速い 精度が高い(特に希少語)

Section7:Attention Mechanism

Attention Mechanism(アテンション機構)とは、入力系列の中で「重要な部分」に重みづけして処理する仕組み

従来の Seq2Seq モデルでは:
エンコーダが入力系列全体を1つのベクトル(文脈ベクトル)に圧縮
長い入力文になると、情報が失われやすく精度が低下

⇒Attentionでは「入力の各部分を動的に参照できるようにする」

Attentionのイメージ
英語: I love you
日本語: 私は あなたを 愛してる
→ 「愛してる」を出力するとき、「love」に一番注目

(参考)利点・欠点

項目 内容
柔軟な参照 長文中の重要な情報をピンポイントで活用
並列計算が可能 特に Self-Attention は並列計算に強い
性能向上 機械翻訳、要約、文分類で大きな改善
問題点 対応策
計算コスト 入力が長いと $O(n^2)$ になる
注意が分散しすぎる Multi-head attentionや正則化で対応

Attentionの応用例

分野 具体例
翻訳 特定の入力単語に注目しながら訳出
要約 文中のキーワードを抽出しやすくなる
質問応答 質問に関係する文章の箇所に注意を向ける
画像キャプション 特定の画像領域にフォーカスして説明文生成

VQ-VAE(Vector Quantized Variational AutoEncoder)

VQ-VAE(Vector Quantized Variational Autoencoder)は、離散的な潜在変数(コードブック)を使ってデータを圧縮・生成するモデル

項目 内容
正式名称 Vector Quantized Variational Autoencoder
提案者 DeepMind(Aaron van den Oordら, 2017)
特徴 潜在空間が「離散的」なオートエンコーダ
主な用途 音声合成、画像生成、テキスト生成(VQ-GAN、VQ-VAE-2など)

image.png
従来の VAE は:
潜在空間が連続値(通常は正規分布)
離散的な情報(単語、記号、カテゴリ)の扱いが難しい

⇒VQ-VAE は 潜在表現を「離散コード」に量子化(Vector Quantization)することで自然言語・音声・画像など離散的構造を持つデータにも強くなった。

image.png

(参考)VQ-VAEの利点・欠点

項目 内容
離散表現 自然言語やカテゴリ情報のようなデータに強い
再構成品質 VAE よりシャープな再構成が可能(ボケが少ない)
拡張性 GANやTransformerと組み合わせやすい(VQ-VAE-2、VQ-GANなど)
課題 内容
量子化の非微分性 最近傍探索は微分不可能 → stop-gradientで近似
コードブックの未使用 学習中に一部のコードが使われなくなる可能性がある
解決策 EMA(Exponential Moving Average)による更新など(VQ-VAE-2で使用)

(参考)VQ-VAEの応用例

分野 用途
音声合成 WaveNet + VQ-VAE:高品質音声生成
画像生成 VQ-VAE-2, VQ-GAN:高解像度画像生成
自然言語処理 離散トークンに分割 → Transformerで処理
ゲーム・強化学習 行動の離散表現として活用

[フレームワーク演習]双方向RNN / 勾配のクリッピング

ハンズオン:spoken_digitデータセットを例にした再帰型ニューラルネットワーク
spoken_digitデータセットは、mnistの音声版です。
データの特徴は、

  • 長さがバラバラの2500個の音声ファイルが入っています。
  • それぞれの音声ファイルは、0から9の数字を読み上げたものです。
  • 5人の話者が、各数字を50回ずつ読み上げています。
  • 各音声ファイルには、対応する数字がラベルとして収録されています。

データセットの読込は、tensorflow_datasetsを使って行います。tensorflow_datasetsは予めある程度の前処理を行ったデータを手軽にダウンロード・準備・使用することができるライブラリです。

ライブラリの機能で、ダウンロードしたデータセットを、シャッフルすることや、分割することができます。

ここでは、データセットの全体の、

  • 70%をトレーニング用
  • 15%を検証用
  • 15%をテスト用

として分割します。

1件だけデータを読み取ります。

データには、Pythonのdict形式で、

  • audio: 数千単位時間(サンプル時間;タイムステップ 非常に短い時間の音声)の音声データ
  • audio/filname: 音声ファイルのファイル名
  • label: 対応する数字のラベル

の情報が含まれています。

今回使用するのは、audiolabelです。

ここで、audioについて詳しくデータを見ます。audioはモデルの入力データになります。

matplotlibを使うことで、音声データの波形を可視化することができます。横方向(x軸方向)に時間、縦方向(y軸方向)に強弱が示されます。

数千単位時間分でサンプリングされた音声データであることが確認できます。

image.png

データをモデルに入力できるようにするため、データに前処理を行います。

今回は、音声データをすべて1000単位時間にそろえます。音声データの時間をそろえるには、長すぎるデータは途中で切り落とし、短すぎるデータには、後ろに0を付け足すことで長さを揃えます。

また、ミニバッチによる学習を行うため、8個でミニバッチを構築します。

tf.data.Datasetには、データセットの内容を編集する機能があります。データセットの中に含まれる1つ1つのデータに対して処理を行うには.map()関数を使います。

ここでは、すべての音声データの長さを揃え、入力データと教師ラベルのペアにします。
image.png

実装演習

双方向LSTMを用いたモデルを紹介します。Kerasで双方向に接続されたネットワークを定義するためには、系列データに対するレイヤーをBidirectionalレイヤーで準備することで、定義することができます。

import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras import layers



tf.keras.backend.clear_session()

model_4 = tf.keras.models.Sequential()

model_4.add(layers.Input((NUM_DATA_POINTS, 1)))
model_4.add(layers.Bidirectional(layers.LSTM(64)))

model_4.add(layers.Dense(10, activation='softmax'))

model_4.summary()


model_4.predict(sample[0]).shape

image.png

学習方法は、.compile()を行い、.fit()に学習データを渡し実行します。

model_4.compile(
    loss=tf.keras.losses.SparseCategoricalCrossentropy(),
    optimizer=tf.keras.optimizers.Adam(),
    metrics=['accuracy']
)

model_4.fit(
    dataset_prep_train,
    validation_data=dataset_prep_valid,
)

結果:
219/219 ━━━━━━━━━━━━━━━━━━━━ 15s 55ms/step - accuracy: 0.1282 - loss: 2.3533 - val_accuracy: 0.1867 - val_loss: 2.1216

次に、勾配クリッピングを行う

import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras import layers



tf.keras.backend.clear_session()

model_5 = tf.keras.models.Sequential()

model_5.add(layers.Input((NUM_DATA_POINTS, 1)))
model_5.add(layers.LSTM(64))

model_5.add(layers.Dense(10, activation='softmax'))

model_5.summary()


model_5.predict(sample[0]).shape

image.png

勾配クリッピングを行うには、Optimizerの引数に、クリッピングの指定を行う

model_5.compile(
    loss=tf.keras.losses.SparseCategoricalCrossentropy(),
    optimizer=tf.keras.optimizers.Adam(clipvalue=0.5),
    metrics=['accuracy']
)

model_5.fit(
    dataset_prep_train,
    validation_data=dataset_prep_valid,
)

219/219 ━━━━━━━━━━━━━━━━━━━━ 9s 32ms/step - accuracy: 0.1009 - loss: 2.3958 - val_accuracy: 0.1280 - val_loss: 2.2941

[フレームワーク演習]Seq2Seq

Seq2Seq(Encoder-Decoder)モデルを用いたsin-cosの変換
sin関数の値からcos関数の値をSeq2Seqモデルに予測させることを試みる。
入力データとして、seq_inを、seq_outを出力データ(教師データ)として準備する

x = np.linspace(-3 * np.pi, 3 * np.pi, 100)
seq_in = np.sin(x)
seq_out = np.cos(x)

plt.plot(x, seq_in, label='$y=\sin x$')
plt.plot(x, seq_out, label='$y=\cos x$')
plt.legend()
plt.grid()
plt.show()

sin関数の値が入力値、cos関数の値が推論したい値となる
image.png

実装演習

モデルを定義するパラメータを準備する。
NUM_ENC_TOKENS: 入力データの次元数
NUM_DEC_TOKENS: 出力データの次元数
NUM_HIDDEN_PARAMS: 単純RNN層の出力次元数(コンテキストの次元数にもなる)
NUM_STEPS: モデルへ入力するデータの時間的なステップ数。

学習を行うためのモデルを定義する。
エンコーダーの出力として、ステート(コンテキスト)のみをデコーダー側へ渡している点に注目。
エンコーダーとデコーダーはコンテキスト以外に繋がりのない分離したモデル構造となっている

tf.keras.backend.clear_session()

e_input = tf.keras.layers.Input(shape=(NUM_STEPS, NUM_ENC_TOKENS), name='e_input')
_, e_state = tf.keras.layers.SimpleRNN(NUM_HIDDEN_PARAMS, return_state=True, name='e_rnn')(e_input)

d_input = tf.keras.layers.Input(shape=(NUM_STEPS, NUM_DEC_TOKENS), name='d_input')
d_rnn = tf.keras.layers.SimpleRNN(NUM_HIDDEN_PARAMS, return_sequences=True, return_state=True, name='d_rnn')
d_rnn_out, _ = d_rnn(d_input, initial_state=[e_state])

d_dense = tf.keras.layers.Dense(NUM_DEC_TOKENS, activation='linear', name='d_output')
d_output = d_dense(d_rnn_out)

model_train = tf.keras.models.Model(inputs=[e_input, d_input], outputs=d_output)
model_train.compile(optimizer='adam', loss='mean_squared_error')

model_train.summary()

モデル定義
image.png

モデルの定義に合わせて学習用データを準備する。

ex: エンコーダーの入力として使用する値。
dx: デコーダーの入力として渡す値。最終的に出力したい値の1つ前のステップの値。
dy: 最終的に推論したい値。dxと比べて時間的に1ステップ先の値となっている

n = len(x) - NUM_STEPS
ex = np.zeros((n, NUM_STEPS))
dx = np.zeros((n, NUM_STEPS))
dy = np.zeros((n, NUM_STEPS))

for i in range(0, n):
  ex[i] = seq_in[i:i + NUM_STEPS]
  dx[i, 1:] = seq_out[i:i + NUM_STEPS - 1]
  dy[i] = seq_out[i: i + NUM_STEPS]

ex = ex.reshape(n, NUM_STEPS, 1)
dx = dx.reshape(n, NUM_STEPS, 1)
dy = dy.reshape(n, NUM_STEPS, 1)

学習を行う ミニバッチのサイズ: 16 エポック数: 80回

BATCH_SIZE = 16
EPOCHS = 80

history = model_train.fit([ex, dx], dy, batch_size=BATCH_SIZE, epochs=EPOCHS, validation_split=0.2, verbose=False)

モデルの推論を行う関数を準備する。
関数内では、
エンコーダーで初期のコンテキストを取得する。
デコーダーで初期のコンテキストと1文字目を元に推論を開始する。
デコーダーから最終出力とコンテキストが出力され、次のステップでのデコーダーのコンテキスト・文字

def predict(input_data):
  state_value = model_pred_e.predict(input_data)
  _dy = np.zeros((1, 1, 1))
  
  output_data = []
  for i in range(0, NUM_STEPS):
    y_output, state_value = pred_d_model.predict([_dy, state_value])
    
    output_data.append(y_output[0, 0, 0])
    _dy[0, 0, 0] = y_output

  return output_data

init_points = [0, 24, 49, 74]

for i in init_points:
  _x = ex[i : i + 1]
  _y = predict(_x)
    
  if i == 0:
    plt.plot(x[i : i + NUM_STEPS], _y, color="red", label='output')
  else:
    plt.plot(x[i : i + NUM_STEPS], _y, color="red")

plt.plot(x, seq_out, color = 'blue', linestyle = "dashed", label = 'correct')
plt.grid()
plt.legend()
plt.show()  

実装結果が以下のようになった。
若干の誤差があるものの、コサインカーブを描く出力が得られた
image.png

[フレームワーク演習]data-augmentation

画像認識モデルの訓練には多様なデータが必要だが、実際には十分な量の画像を集めるのが難しいため、データ拡張が用いられる。これは画像に反転や回転などの処理を加えて新たなデータを生成する手法。たとえば、水平反転により左右の視点を変えることができ、学習効果が高まる。ただし、「6」と「9」のように反転によって意味が変わる例もあるため、課題に応じた拡張方法の選択が重要。

実装演習

データの水増し(データ拡張)に関して、具体的に画像を見ながら確認
まず、元画像を確認

def load_and_preprocess_image(path: str) -> tf.Tensor:
    """
    指定されたファイルパスから画像を読み込み、前処理を施します。

    Args:
        path (str): 読み込む画像のファイルパス。

    Returns:
        tf.Tensor: 前処理された画像。float32型のテンソルで、値は[0, 1]の範囲に正規化されています。
    """
    # 画像ファイルを読み込む
    image = tf.io.read_file(path)

    # JPEG画像を3つのカラーチャンネルでデコードする
    image = tf.image.decode_jpeg(image, channels=3)

    # 画像をfloat32型に変換し、[0, 1]の範囲に正規化する
    image = tf.image.convert_image_dtype(image, tf.float32)

    return image



# データセットを作成
dataset_original = tf.data.Dataset.from_tensor_slices([IMAGE_PATH])

# データセットに関数をマップ
dataset = dataset_original.map(load_and_preprocess_image)

for image in dataset.take(1):
    plt.imshow(image)
    plt.axis('off')
    plt.show()

元画像
image.png

"Horizontal Flip"(水平反転)は、画像データ拡張の一つで、画像を左右に反転させる処理を指します。この処理により、オブジェクトや特徴が画像内で左右に対称的である場合、新しいデータを生成できます。

# 画像をランダムに左右反転する関数
def random_flip_left_right(image):
  # 結果を固定したい場合には、seed=数字でseed値を固定してください
  return tf.image.random_flip_left_right(image)

# データセットに画像処理関数を適用
dataset = dataset_original.map(load_and_preprocess_image)
dataset = dataset.map(random_flip_left_right)

for image in dataset.take(1):
    plt.imshow(image)
    plt.axis('off')
    plt.show()

image.png

"Crop"(切り抜き)は、画像データ拡張の一つで、画像内の一部分を切り抜く処理を指します。この処理により、元の画像から異なる視点やスケールでの画像データを生成できます。
画像を切り抜く際、切り抜かれた領域は元の画像の一部であり、その一部分の詳細や特徴に焦点を当てることができます。これは、物体検出やセグメンテーションなどのタスクで特に有用

# 画像をランダムに左右反転する関数を作成します。
def random_flip_and_crop_image(image: tf.Tensor, target_height: int, target_width: int, seed: int = None) -> tf.Tensor:
    """
    画像にランダムなフリップとクロップ操作を適用します。

    Args:
        image (tf.Tensor): 入力画像のテンソル。
        target_height (int): クロップ後の目標の高さ。
        target_width (int): クロップ後の目標の幅。
        seed (int, optional): シード値。ランダム性を制御します。デフォルトはNone。

    Returns:
        tf.Tensor: ランダムなフリップとクロップが適用された画像のテンソル。
    """
    # 元画像のサイズを取得
    original_size = tf.shape(image)[:2]

    target_size = tf.convert_to_tensor([target_height, target_width], dtype=tf.int32)
    size_condition = tf.logical_and(
        tf.greater_equal(original_size[0], target_size[0]),
        tf.greater_equal(original_size[1], target_size[1])
    )

    # 入力された切り抜きサイズが、元画像よりも大きい場合、元画像のサイズに調整
    crop_size = tf.where(size_condition, target_size, original_size)

    # チャネル数を追加(カラー画像の場合は3チャネル)
    crop_size = tf.concat([crop_size, [3]], axis=0)

    # ランダムクロップを適用
    # シード値を指定してランダム性を制御
    cropped_image = tf.image.random_crop(image, size=crop_size, seed=seed)

    return cropped_image

# 切り出すサイズを設定します。
target_height = 100
target_width = 100

# データセットに画像処理関数を適用
dataset = dataset_original.map(load_and_preprocess_image)
dataset = dataset.map(lambda x: random_flip_and_crop_image(x, target_height, target_width))

for image in dataset.take(1):
    plt.imshow(image)
    plt.axis('off')
    plt.show()

image.png

ガウシアンフィルタ(Gaussian Filter)は、画像処理に広く使用されるフィルタで、画像のノイズを減少させたり、画像を滑らかにしたり(ぼかしたり)するために使われます。このフィルタは、ガウス分布から名前を取っており、ガウシアンカーネルを用いて画像に畳み込み操作を適用します。

$G(x,y) = \frac{1}{2\pi\sigma^2} e^{-\frac{x^2 + y^2}{2\sigma^2}}$

ここで、

  • $ G(x,y) $ は2次元のガウス関数
  • $ x $ と $ y $ は座標
  • $ \sigma $ は標準偏差

ガウシアンフィルタには、画像処理におけるいくつかの重要な応用があります:

  • ノイズの低減: ガウシアンフィルタは、画像内のノイズを減少させるために一般的に使用されます。画像を平滑化することで、高周波ノイズが抑えられ、よりクリーンで視覚的に魅力的な画像が得られます。

  • 画像の強調: 一部の場合、ガウシアンぼかしを画像に適用して、細かい不完全さや詳細を軽減し、より滑らかな外観にすることがあります。

  • スケールスペース表現: ガウスぼかしは、コンピュータビジョンおよび物体検出で画像のスケールスペース表現を作成するために使用され、さまざまなスケールで物体を識別するのに役立ちます。

ガウシアンフィルタの適用量は、カーネルサイズと標準偏差に依存します。適切なカーネルサイズとシグマ値の選択は、特定の画像処理タスクの要件に応じて行われます。シグマ値を大きくし、カーネルを小さくすると、より軽微な平滑化効果が得られます。逆に、シグマ値を小さくし、カーネルを大きくすると、より強力な平滑化が行われます。

def gaussian_filter(channels: int, kernel_size: int, sigma: float) -> tf.Tensor:
    """
    ガウシアンフィルタを生成します。

    Args:
        channels (int): チャネル数。
        kernel_size (int): カーネルのサイズ(奇数の値)。
        sigma (float): ガウシアン分布の標準偏差。

    Returns:
        tf.Tensor: 生成されたガウシアンフィルタのカーネル。
    """
    ax = tf.range(-kernel_size // 2 + 1.0, kernel_size // 2 + 1.0)
    xx, yy = tf.meshgrid(ax, ax)
    kernel = tf.exp(-(xx**2 + yy**2) / (2.0 * sigma**2))
    kernel = kernel / tf.reduce_sum(kernel)
    kernel = kernel[..., tf.newaxis, tf.newaxis]
    kernel = tf.tile(kernel, [1, 1, channels, 1])
    return kernel

def apply_gaussian_blur(image: tf.Tensor, channels: int, kernel_size: int, sigma: float) -> tf.Tensor:
    """
    ガウシアンブラーを適用して画像をぼかします。

    Args:
        image (tf.Tensor): 入力画像のテンソル。
        channels (int): チャネル数。
        kernel_size (int): カーネルのサイズ(奇数の値)。
        sigma (float): ガウシアン分布の標準偏差。

    Returns:
        tf.Tensor: ガウシアンブラーが適用された画像のテンソル。
    """
    kernel = gaussian_filter(channels, kernel_size, sigma)

    image = tf.expand_dims(image, 0)
    blurred_image = tf.nn.depthwise_conv2d(image, kernel, strides=[1, 1, 1, 1], padding='SAME')
    blurred_image = tf.squeeze(blurred_image, 0)

    return blurred_image

channels = 3
kernel_size = 11
sigma = 100

dataset = dataset_original.map(load_and_preprocess_image)
dataset = dataset.map(lambda x: apply_gaussian_blur(x, channels, kernel_size, sigma))

for image in dataset.take(1):
    plt.imshow(image)
    plt.axis('off')
    plt.show()

image.png

[フレームワーク演習]activate_functions

ニューラルネットワークの順伝播(forward)では、線形変換で得た値に対して、非線形な変換を行う。非線形な変換を行う際に用いられる関数を、活性化関数という。
入力データ(説明変数)を$x$、総入力データを$u$、中間層出力データを$z$、出力データ(予測データ)を$y$、重みを$W$、バイアスを$b$とする、入力層サイズ$3$、中間層サイズ$2 \times 1$、出力層サイズ$2$のニューラルネットワークは、
image.png
のように図示することができる。

図の中の数式
image.png
の、$f$を「中間層に用いる活性化関数」、$g$を「出力層に用いる活性化関数」という。

実装演習

中間層に用いる活性化関数
・ステップ関数
入力値が閾値以上のときは 1 (活性化)、それ以外のときは 0 (不活性化)
閾値は基本的には 0
image.png

def step_function(x):
    """forward
       
       step
       ステップ関数
       (閾値0)
    """
    return np.where(x >= 0.0, 1.0, 0.0)

def d_step_function(x):
    """backward
       
       derivative of step
       ステップ関数の導関数
       (閾値0)
    """
    dx = np.where(x == 0.0, np.nan, 0.0)
    return dx

# def d_step_function(x):
#     """derivative of step
#        ステップ関数の導関数
#        (閾値0, xが0.0のときの微分値を定義した劣微分)
#     """
#     dx = np.zeros_like(x)
#     return dx

x = np.arange(-600, 601, 1) * 0.01
f, d, = step_function, d_step_function
y1, y2 = f(x), d(x)

_, ax = plt.subplots()
ax.plot(x, y1, label=f.__doc__.split("\n")[0].strip())
ax.plot(x, y2, label=d.__doc__.split("\n")[0].strip(), linewidth=1.0)
ax.set_xlabel("$x$")
ax.set_ylabel("$y_{1}=f(x), y_{2}=f^{\prime}(x)$")
ax.set_title(f.__doc__.split("\n")[2].strip())
ax.legend()

plt.show()

image.png

・シグモイド関数
image.png

def sigmoid(x):
    """forward
       
       sigmoid
       シグモイド関数
    """
    return 1.0 / (1.0 + np.exp(-x))

def d_sigmoid(x):
    """backward
       
       derivative of sigmoid
       シグモイド関数の導関数
    """
    dx = sigmoid(x) * (1.0 - sigmoid(x))
    return dx

x = np.arange(-600, 601, 1) * 0.01
f, d = sigmoid, d_sigmoid
y1, y2 = f(x), d(x)

_, ax = plt.subplots()
ax.plot(x, y1, label=f.__doc__.split("\n")[0].strip())
ax.plot(x, y2, label=d.__doc__.split("\n")[0].strip(), linewidth=1.0)
ax.set_xlabel("$x$")
ax.set_ylabel("$y_{1}=f(x), y_{2}=f^{\prime}(x)$")
ax.set_title(f.__doc__.split("\n")[2].strip())
ax.legend()

plt.show()

image.png

・ReLU
別名: 正規化線形関数、ランプ関数
入力値が 0 以上のときは入力値と同じ値、それ以外のときは 0
image.png

def relu(x):
    """forward
       
       ReLU
       正規化線形関数
    """
    return np.maximum(0, x)

def d_relu(x):
    """backward
       
       derivative of ReLU
       正規化線形関数の導関数
    """
    dx = np.where(x > 0.0, 1.0, np.where(x < 0.0, 0.0, np.nan))
    return dx

# def d_relu(x):
#     """backward
#        
#        derivative of ReLU
#        正規化線形関数の導関数
#        (xが0.0のときの微分値を定義した劣微分)
#     """
#     dx = step_function(x)
#     return dx

x = np.arange(-600, 601, 1) * 0.01
f, d = relu, d_relu
y1, y2 = f(x), d(x)

_, ax = plt.subplots()
ax.plot(x, y1, label=f.__doc__.split("\n")[0].strip())
ax.plot(x, y2, label=d.__doc__.split("\n")[0].strip(), linewidth=1.0)
ax.set_xlabel("$x$")
ax.set_ylabel("$y_{1}=f(x), y_{2}=f^{\prime}(x)$")
ax.set_title(f.__doc__.split("\n")[2].strip())
ax.legend()

plt.show()

image.png

出力層に用いる活性化関数
・2値分類:シグモイド関数
・多値分類:ソフトマックス関数
image.png

def softmax(x):
    """forward
       
       softmax
       ソフトマックス関数
    """
    if x.ndim == 2:
        x = x.T
        x = x - np.max(x, axis=0)
        y = np.exp(x) / np.sum(np.exp(x), axis=0)
        return y.T

    x = x - np.max(x)  # オーバーフロー対策
    return np.exp(x) / np.sum(np.exp(x))

def d_softmax(x):
    """backward
       
       derivative of softmax
       ソフトマックス関数の導関数
    """
    y = softmax(x)
    dx = -y[:,:,None] * y[:,None,:]       # ヤコビ行列を計算 (i≠jの場合)
    iy, ix = np.diag_indices_from(dx[0])  # 対角要素の添字を取得
    dx[:,iy,ix] = y * (1.0 - y)           # 対角要素値を修正 (i=jの場合)
    return dx

x = np.pad(np.arange(-600, 601, 1).reshape((-1, 1)) * 0.01, ((0, 0), (0, 1)), 'constant')
g = softmax
y = g(x)

_, ax = plt.subplots()
for j in range(x.shape[1]):
    ax.plot(x[:,j], y[:,j], label=r" $j={}$".format(j))
ax.set_xlabel("$x_{i0}$")
ax.set_ylabel("$y_{i}=g(x)_{i}$")
ax.set_title(g.__doc__.split("\n")[2].strip())
ax.legend()

plt.show()

image.png

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?