化学系エンジニアがAIを学ぶ

PyTorchでディープラーニング、強化学習を学び、主に化学工学の問題に取り組みます

メモ: Pillowを用いた深層学習用の画像データ数値化

画像データの深層学習を行うためのPillowを用いた画像データ数値化のメモ。 次の適当な画像データを、深層学習に使えるように数値化する。

f:id:schemer1341:20191027221206p:plain:w300 (testfig.png)
from PIL import Image
import numpy as np

im = Image.open("./testfig.png") # 画像ファイル読み込み
im = im.convert("L") # 画像のモード変更 "L"は8bitグレイスケール
im = im.resize((256, 256)) # 画像サイズ変更
data = np.array(im) # 画像を配列として数値データ化
im.show() # 処理後の画像表示

カラー、透過度等も含めたその他の画像モードは下記リンク参考

Concepts — Pillow (PIL Fork) 6.2.1 documentation

その他参考

https://mmm-ssss.com/2018/11/12/deeplearning_3/#toc

PIL/Pillow チートシート - Qiita

ディープラーニングによる化学プラントの異常検知の簡単な例題

はじめに

例えば連続で運転し化学品を製造するプラントにおいて、何らかの原因で反応温度や圧力が異常に上昇したり、製品中の不純物が増加するなどして運転が継続できず、プラント停止を余儀なくされることがある。この場合、プラント設備の復旧や製品在庫管理・販売計画見直しに多大な人手・費用が必要となる。

異常の兆候に早い段階で気付くことができれば、対策を講じることで必要な人手・費用を抑えることができる。このためAI(ディープラーニング)による化学プラントの異常検知が注目されており、例えば下記の三井化学の例がある。ここでは簡単な例題でディープラーニングによる異常検知をやってみる。

人工知能(AI)を用いて、化学プラントの製造過程で製品の品質予測に成功|2016|ニュースリリース|三井化学株式会社

例題

下記の原料Aから生成物Bを製造する反応器を考える。①原料流量、②原料温度、③反応圧力、④熱媒体温度を制御して、⑤出口温度をある一定の温度範囲内に収めるように運転している(範囲から外れると製品品質悪化)。

f:id:schemer1341:20190915012130p:plain:w500

以下のグラフはある1000日間の運転データである。900日を過ぎたあたりで出口温度がこのプラントの管理上限値である320℃を超え異常が認識された。出口温度の上昇は反応副生物の分解によるもので、急激に反応が起こるのでなく、ゆっくりと時間をかけて分解量が増加していくことがわかっているものとする。

f:id:schemer1341:20190916230731p:plain

反応量が少ないうちに出口温度の僅かな上昇で異常を検知できることが好ましいが、制御値①②③④の振れが出口温度を振れさせてしまい、副生物分解による僅かな温度上昇を見分けることは難しい状況である。

ディープラーニングを用いて早い段階で異常を検知することができるか?

※このデータは下記コードから人工的に生成した。気になる人は参考に。制御値①②③④はランダムで振れを発生させ、出口温度は①②③④を変数とした所定の関数から算出している。さらに700日目から日数に応じて増加する異常値を出口温度に加えている。ここで使用したデータそのものもリンク先に置いてある。

chemical-engineer-learns-ai/20190921_anomaly_detection at master · nakamura-13/chemical-engineer-learns-ai · GitHub

異常検知の手順

次の手順でディープラーニングを用いて異常を検知する。

  1. 正常な期間のデータをディープラーニングで学習し、制御値①②③④と出口温度を相関するモデルを作る
  2. 上記1.のモデルを使って異常な期間の出口温度の予測値を得る
  3. 異常な期間の実際の出口温度とディープラーニングで予測した出口温度の差を異常度とする

正常期間データを学習したモデルから得た異常期間の予測値は副生物分解による温度上昇を含まない値となる。したがって、この値と異常期間の実際の値の差を取ることで異常さの度合いが見えるはずである。

正常期間データの学習

700日目までは正常だったと考えて、その期間のデータを再現するようネットワークを学習させた。

f:id:schemer1341:20190918004339p:plain:w500

なお用いたネットワークは下記の通りで、ノード数20の中間層を2つ入れた。

## ニューラルネットワークモデル設定
class Model(nn.Module):
    def __init__(self):
        super(Model, self).__init__()
        self.l1 = nn.Linear(4, 20) # 入力4つ (原料流量、原料温度、反応圧力、熱媒体温度)
        self.l2 = nn.Linear(20, 20)
        self.l3 = nn.Linear(20, 1) # 出力1つ (出口温度)

    def forward(self, x):
        x = F.relu(self.l1(x))
        x = F.relu(self.l2(x))
        return self.l3(x)

学習させた結果、次のグラフとなった。色々と試行錯誤したものの初期の乖離は解消されなかった。しかしながら、それ以外の期間についてはほぼデータを再現できている。

f:id:schemer1341:20190918010556p:plain:w500

学習したモデルによる出口温度予測および異常度の算出

異常と考えられる期間も含めたデータとモデル予測値は次のグラフとなる。異常を認識したタイミングよりも早い時点でデータとモデル予測値に差が生じていることが見て取れる。

f:id:schemer1341:20190919000843p:plain:w500

データとモデル予測値の差の絶対値を「異常度」と定義し、それをグラフにプロットすると次の図となる。異常度がいくつになったら異常と判断するかがポイントとなってくるが、例えば初期を除き10℃以上を異常と考えるなら、100日程度早めに異常兆候を知ることができる。100日あれば事前に十分な対策を講じることができるだろう。

f:id:schemer1341:20190921175844p:plain:w500

おわりに

実際のプラントでは様々なノイズが制御値や計測値に影響するため、こう簡単にはいかないかもしれないが、実務でも適用をトライしたいと思う。

使用したコード

chemical-engineer-learns-ai/20190921_anomaly_detection at master · nakamura-13/chemical-engineer-learns-ai · GitHub

深層強化学習の簡単な例 〜Double DQN適用〜

はじめに

以前の記事で簡単な問題を対象にDQN、Experience Replayを適用してみたが、今回はさらにDouble DQNを追加してみる。なお下記サイトを参考にした。

Let’s make a DQN: Double Learning and Prioritized Experience Replay – ヤロミル

https://www.renom.jp/ja/notebooks/product/renom_rl/ddqn/notebook.html

Double DQN

通常のDQNでは次の形でニューラルネットワークを更新している。

f:id:schemer1341:20190911214824p:plain:w500

ポイントは \gamma {\rm max}Q(s_{t+1}, a)で、次の状態の最大のQ値を求めるのに自分自身のネットワークを使用している点である。もし仮にQ値の初期値が変に偏っていて、ある行動のQ値が他に比べ極端に高い場合、同じネットワークで更新を繰り返すと、その行動の価値が過大に評価されてしまうことになる。

これを緩和するために提案された方法がDouble DQNである。この方法では、メインのネットワークとは別のネットワーク(Target network)を作り、ネットワーク更新時に次の状態における行動はメインネットワークで決めるが、その行動の価値評価は別のネットワークで行う。なお別のネットワークには過去のメインネットワークを用いる。この方法を用いる場合のネットワーク更新式は次の形となる。

f:id:schemer1341:20190911011958p:plain:w250 f:id:schemer1341:20190911012155p:plain:w570

対象とする問題

前回記事と同一の報酬払出装置を考える。前回のプログラムを修正してDouble DQNを適用する。

深層強化学習(Deep Q Network, DQN)の簡単な例 〜Experience Replay追加〜 - 化学系エンジニアがAIを学ぶ

Double DQN の適用

次の通り、メインネットワークに加え、メインとは別のネットワーク(Target network)を準備する。

import torch
import torch.nn as nn
import torch.nn.functional as F

class DQN(nn.Module):
    def __init__(self):
        super(DQN, self).__init__()
        self.l1 = nn.Linear(1, 3)
        self.l2 = nn.Linear(3, 3)
        self.l3 = nn.Linear(3, 2)

    def forward(self, x):
        x = F.relu(self.l1(x))
        x = F.relu(self.l2(x))
        x = self.l3(x)
        return x

dqn = DQN()   # メインネットワーク
dqn_t = DQN() # もう1つのネットワーク(Target network)
optimizer = torch.optim.SGD(dqn.parameters(), lr=0.01)
criterion = nn.MSELoss()

Double DQNを適用する場合のネットワーク更新は下記の形となる。通常のDQNと異なるのは下記「Q値の算出」にあるa_m、max_q_nextを求める部分だけである。(なおここではExperience Replayを利用している。)

BATCH_SIZE = 50

def update_dqn(replay_memory):
    ## メモリがバッチサイズより小さいときは何もしない
    if len(replay_memory) < BATCH_SIZE:
        return

    ## ミニバッチ取得
    transitions = replay_memory.sample(BATCH_SIZE)
    ## (状態、行動、次の状態、報酬)✕バッチサイズ を (状態xバッチサイズ、行動✕バッチサイズ、、、)に変える
    batch = Transition(*zip(*transitions))
    
    ## 各値をtensorに変換
    state = torch.FloatTensor(batch.state).unsqueeze(1)
    action = torch.LongTensor(batch.action).unsqueeze(1)
    next_state = torch.FloatTensor(batch.next_state).unsqueeze(1)
    reward = torch.FloatTensor(batch.reward).unsqueeze(1)

    ## Q値の算出
    q_now = dqn(state).gather(-1, action) # 今の状態のQ値
    a_m = dqn(next_state).max(-1)[1].unsqueeze(1) # メインネットワークで決めた次のaction,[1]でindexを得る
    max_q_next = dqn_t(next_state).gather(-1, a_m) # Target networkから上で決めたactionのQ値を得る
    gamma = 0.9
    q_target = reward + gamma * max_q_next # 目標のQ値

    # DQNパラメータ更新
    optimizer.zero_grad()
    loss = criterion(q_now, q_target) # 今のQ値と目標のQ値で誤差を取る
    loss.backward()
    optimizer.step()
    
    return loss.item() # lossの確認のために返す

学習時に所定の計算ステップごとにTarget networkにメインネットワークをコピーして更新される形とする(下記コードの一番下)。更新されるまでは学習時に過去のメインネットワークを参照する形となる。

NUM_EPISODES = 1200
NUM_STEPS = 5

log = [] # 結果のプロット用

for episode in range(NUM_EPISODES):
    env = Dispenser(0)
    total_reward = 0 # 1エピソードでの報酬の合計を保持する

    for s in range(NUM_STEPS):
        ## 現在の状態を確認
        state = env.state
        ## 行動を決める
        action = decide_action(state, episode)
        ## 決めた行動に従いステップを進める。次の状態、報酬を得る
        next_state, reward = env.step(action)
        ## 結果をReplayMemoryに記録
        memory.push(state, action, next_state, reward)
        ## DQNを更新
        loss = update_dqn(memory)
        total_reward += reward

    log.append([total_reward, loss])
    if (episode+1)%100==0: print(episode+1, loss)
    ## 2エピソード = 計算10ステップ 毎の頻度でTarget networkにメインネットワークをコピー
    if episode%2==0:
        dqn_t.load_state_dict(dqn.state_dict())

学習結果

多少計算が安定している感じがするが、このくらいの問題ではあまり差はないのかもしれない。 f:id:schemer1341:20190911221649p:plain

コード

chemical-engineer-learns-ai/dispenser_Double-DQN.py at master · nakamura-13/chemical-engineer-learns-ai · GitHub

メモ: PyTorch 重み、バイアスの初期化について

はじめに

PyTorchのニューラルネットワークの重み・バイアスの初期化についてのメモを記す。

重み

重みの内容は次のようにして確認できる。

>>> import torch.nn as nn
>>> l = nn.Linear(1, 3)
>>> l.weight
Parameter containing:
tensor([[ 0.6204],
        [-0.5651],
        [-0.6809]], requires_grad=True)

重みの初期化は次のようにnn.initモジュールの関数を使用することで実施できる。

>>> nn.init.xavier_uniform_(l.weight) # Xavierの初期値
Parameter containing:
tensor([[ 0.5318],
        [-0.8438],
        [-1.1093]], requires_grad=True)

>>> nn.init.kaiming_uniform_(l.weight) # Heの初期値
Parameter containing:
tensor([[ 2.3349],
        [-0.9218],
        [-1.3357]], requires_grad=True)

単純な正規分布・一様分布や定数を設定できる関数もnn.init内に準備されている。

詳しくはこちらを参照: https://pytorch.org/docs/master/nn.init.html

バイアス

バイアスの内容は重みと同様に確認できる。

>>> l.bias
Parameter containing:
tensor([-0.9920,  0.6406, -0.6558], requires_grad=True)

バイアスの初期化も重みと同様のやり方。

>>> nn.init.uniform_(l.bias, 0, 1) # 一様分布
Parameter containing:
tensor([0.8032, 0.7810, 0.1619], requires_grad=True)

>>> nn.init.constant_(l.bias, 0) # 定数
Parameter containing:
tensor([0., 0., 0.], requires_grad=True)

デフォルトの初期値

nn.Linearのソースを見ると重みについてはHeの初期値が入っている模様。

pytorch/linear.py at master · pytorch/pytorch · GitHub

def reset_parameters(self):
    init.kaiming_uniform_(self.weight, a=math.sqrt(5))
    if self.bias is not None:
        fan_in, _ = init._calculate_fan_in_and_fan_out(self.weight)
        bound = 1 / math.sqrt(fan_in)
        init.uniform_(self.bias, -bound, bound)

参考

まるまるにっき | pytorchの重みの初期化の方法

python - How to initialize weights in PyTorch? - Stack Overflow

What is the default initialization of a conv2d layer and linear layer? - PyTorch Forums

ニュース: Preferred Networks、JXTGエネルギーと石油精製プラントの最適化・自動化に関する共同研究を開始

Preferred NetworksがJXTGエネルギーと石油精製プラントの最適化・自動化に関する共同研究を開始したそうです。10億円もの資金が調達されるそうで、一気に技術開発が進むことが期待されます。近い業界に関わる者としては、得られた知見などを可能な範囲でオープンにしてもらえたらと思いますが、難しいでしょうね。。。

it.impressbm.co.jp

iotnews.jp

深層強化学習(Deep Q Network, DQN)の簡単な例 〜Experience Replay追加〜

はじめに

前回の記事でOpenAI Gymを使わず非常に簡単な問題を対象にDQNを適用してみたが、"Experience Replay"を入れていなかった。今回は前回の問題にExperience Replayを追加してみる。なおこれを実施するにあたり、下記サイトを参考にした。

第15回 CartPole課題で深層強化学習DQNを実装|Tech Book Zone Manatee

Reinforcement Learning (DQN) Tutorial — PyTorch Tutorials 1.2.0 documentation

Experience Replay

行動とその結果を経験として記録しておき、その過去の経験をサンプリングして学習データとする方法。通常の逐次で学習する方法に比べ、時間的に離れたデータを使用することができ、偏りを抑えた安定した学習ができるとのこと。複数データをまとめてニューラルネットワークをミニバッチ学習させることもできる。

具体的には「現在の状態」、「行動」、「行動後の状態」、「報酬」のデータの組をある一定の数だけメモリに残しておいて、その中から学習に使うデータをいくつかランダムにサンプリングし、これらを1つのバッチとしてニューラルネットワークに学習させる。

対象とする問題

前回記事と同一の報酬払出装置を考える。プログラムは前回のものにいくつか追加・修正してExperience Replayを適用する。

深層強化学習(Deep Q Network, DQN)の簡単な例 - 化学系エンジニアがAIを学ぶ

Experience Replayの実装

下記のpytorch.orgのチュートリアルにあるように、namedtupleを用いて経験を記録するクラスを追加する。

Reinforcement Learning (DQN) Tutorial — PyTorch Tutorials 1.2.0 documentation

ここでは500回分の行動の結果を記録することにしている(数は適当)。500を超えると古い方から順番に上書きされる形となっている。サンプリング時は指定したバッチサイズ数だけのデータが取り出される。

from collections import namedtuple

# 「現在の状態」、「行動」、「行動後の状態」、「報酬」を記録するnamedtuple
Transition = namedtuple('Transition', ('state', 'action', 'next_state', 'reward'))

import random

class ReplayMemory(object):
    def __init__(self, capacity):
        self.capacity = capacity # メモリの許容サイズ
        self.memory = []         # 経験を保存するリスト
        self.index = 0           # 保存した経験リストのindex

    def push(self, state, action, next_state, reward): # 経験をリストに追加(push)する
        if len(self.memory) < self.capacity: # 許容サイズ以下なら追加
            self.memory.append(None)
        self.memory[self.index] = Transition(state, action, next_state, reward)
        self.index = (self.index + 1) % self.capacity # indexを1つ進める。許容サイズ超えたら古い順から上書き

    def sample(self, batch_size): # バッチサイズ分の経験のランダムサンプリング
        return random.sample(self.memory, batch_size) 

    def __len__(self):
        return len(self.memory)

CAPACITY = 500

memory = ReplayMemory(CAPACITY)

DQNの学習

上記の行動を記録したクラスを利用できる形に、DQNの学習部分を修正する。

BATCH_SIZE = 50 # サンプリングするデータ数

def update_dqn(replay_memory): # 行動を記録したクラスを引数とする
    ## メモリがバッチサイズより小さいときはまだ学習しない
    if len(replay_memory) < BATCH_SIZE:
        return

    ## バッチ取得
    transitions = replay_memory.sample(BATCH_SIZE)
    ## (状態、行動、次の状態、報酬)✕バッチサイズ を (状態xバッチサイズ、行動✕バッチサイズ、、、)に変える
    batch = Transition(*zip(*transitions))
    
    ## 各値をtensorに変換
    state = torch.FloatTensor(batch.state).unsqueeze(1)
    action = torch.LongTensor(batch.action).unsqueeze(1)
    next_state = torch.FloatTensor(batch.next_state).unsqueeze(1)
    reward = torch.FloatTensor(batch.reward).unsqueeze(1)

    ## Q値の算出
    q_now = dqn(state).gather(-1, action) # 今の状態のQ値
    max_q_next = dqn(next_state).max(-1)[0].unsqueeze(1).detach() # 状態移行後の最大のQ値
    gamma = 0.9
    q_target = reward + gamma * max_q_next # 目標のQ値

    # DQNパラメータ更新
    optimizer.zero_grad()
    loss = criterion(q_now, q_target) # 今のQ値と目標のQ値で誤差を取る
    loss.backward()
    optimizer.step()
    
    return loss.item() # lossの確認のために返す

学習結果

他のコードは前回記事と同じである。下図に学習の推移を示す。左は誤差の推移、右は1エピソードにおけるトータル報酬の推移である。実際に計算してみるとわかるが逐次で学習させるより、Experience Replayを適用するほうが学習が安定しやすい。

f:id:schemer1341:20190609202605p:plain

コード

chemical-engineer-learns-ai/dispenser_experience_replay.py at master · nakamura-13/chemical-engineer-learns-ai · GitHub

参考

第15回 CartPole課題で深層強化学習DQNを実装|Tech Book Zone Manatee

Reinforcement Learning (DQN) Tutorial — PyTorch Tutorials 1.2.0 documentation