Pythonで強化学習する

2017-10-23 (月) 10:52:07 (26d) | Topic path: Top / 強化学習 / Pythonで強化学習する

はじめに

ここでは、Pythonで強化学習を行います。

例題として、Sutton & Barto の『エージェント・アプローチ人工知能』で用いられている [math]4 \times 3[/math] の迷路を使います。

maze.png

Sのマスが初期状態、+1 と -1 のマスが終端状態です。

行動は東西南北の4種類です。

外側と塗られているマスには壁があり、その方向には進めません。 0.8の確率で選んだ方向に進むことができ、0.1の確率で選んだ方向に対して90度左のマス、0.1の確率でその反対側のマスに進みます。

数値が書かれているマスに到達するとその数値が報酬として得られます。 その他のマスに到達すると報酬 -0.02 が得られます。

強化学習アルゴリズムはQ学習、行動選択は[math]\epsilon[/math]-グリーディーです。

次の環境で確認しました。

  • macOS Sierra 10.12.5
  • Python 3.5.1

準備

NumPyを使いますので、インストールします。

まず、pip3をインストールし、最新版にします。

$ sudo easy_install pip
$ sudo pip3 install --upgrade pip

次に、pip3でNumPyをインストールします。

$ sudo pip3 install numpy

クラス構造

  • 環境を表す Environment
    • 迷路を表す Maze
  • 強化学習エージェントを表す ReinformentLearningAgent
    • Q学習エージェントを表す QLearningAgent

環境 environment.py

まず初めに、環境を表すクラスがどのような変数とメソッドを持っているかを決めておきます。

変数としては、次のものを持つことにします。

  • 状態数 states
  • 行動数 actions

メソッドしては、次のものを持つことにします。

  • 状態を初期化して返す initState
  • 終端状態かどうかを調べる isTerminal
  • 報酬を返す getReward
  • 行動を実行して次の状態と報酬を返す takeAction
  • 状態を文字列で返す strState
  • 行動を文字列で返す strAction
  • ログを出力する printLog

printLog 以外の実際の実装は、子クラスで行います。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# environment.py
#
# Copyright 2017 Tohgoroh Matsui All Rights Reserved.
#
from abc import ABCMeta, abstractmethod


class Environment(metaclass=ABCMeta):
    """強化学習の環境。"""

    states = None   # 状態数
    actions = None  # 行動数

    @classmethod
    @abstractmethod
    def initState(self):
        """初期状態を返す。"""
        pass

    @abstractmethod
    def isTerminal(self):
        """終端状態ならTrue、そうでないならFalseを返す。"""
        pass

    @classmethod
    @abstractmethod
    def getReward(self, state):
        """報酬を返す。"""
        pass

    @classmethod
    @abstractmethod
    def takeAction(self, state, action):
        """行動を実行して状態を更新し、次の状態と報酬を返す。"""
        pass

    @classmethod
    @abstractmethod
    def strState(self, state):
        """状態を表す文字列を返す。"""
        pass

    @classmethod
    @abstractmethod
    def strAction(self, action):
        """行動を表す文字列を返す。"""
        pass

    @classmethod
    def printLog(self, steps, state, action, reward, state_):
        """ログを出力する。"""
        print('%d: %s, %s, %f, %s' % (steps, self.strState(state), self.strAction(action), reward, self.strState(state_)))

強化学習エージェント rlagent.py

強化学習エージェントを表す ReinforcementLearningAgent クラスには、行動選択と評価を実装します。

学習するためのメソッドは抽象クラスとし、子クラスである QLearningAgent に実装します。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# rlagent.py
#
# Copyright 2017 Tohgoroh Matsui All Rights Reserved.
#
from abc import ABCMeta, abstractmethod
import numpy as np


class ReinforcementLearningAgent(metaclass=ABCMeta):
    """強化学習エージェント。"""

    @abstractmethod
    def learn(self):
        """強化学習を用いて学習する。"""
        pass

    def evaluate(self):
        """学習した行動価値が最も高い行動を選択したときの平均ステップ数を返す。"""
        episodes = 0    # エピソード数
        totalSteps = 0  # ステップ数
        while episodes < 100:   # 100エピソード繰り返す:
            steps = 0   # このエピソードにおけるステップ数
            state = self.env.initState() # 状態を初期化する            state = self.getState(rawState) # 状態を状態番号に
            while not self.env.isTerminal(state) and steps < 100:   # 状態が終端状態になるか、または、100ステップになるまで繰り返す:
                action = self.greedy(state)  # グリーディーに行動を選択する
                state_, reward = self.env.takeAction(state, action) # 選択した行動を実行する
                #self.env.printLog(steps, state, action, reward, state_)  # ログを出力する
                state = state_
                steps += 1
                totalSteps += 1
            episodes += 1
        return totalSteps / 100

    def uniformly(self):
        """一様ランダムに行動を選択して返す。"""
        action = np.random.randint(self.env.actions)    # 0からaction-1までの整数をランダムに生成する
        return action

    def greedy(self, state):
        """グリーディーに行動を選択して返す。"""
        s = self.env.getS(state)    # 状態番号
        maxQ = self.aQ[s].max() # 最大のQ値
        aIds = np.where(self.aQ[s] == maxQ)[0]  # Q値がmaxQである要素のインデックスの配列
        r = np.random.randint(len(aIds))    # 0から要素数-1までの整数をランダムに生成する
        action = aIds[r]    # Q値がmaxQである要素のインデックスから行動を選択する
        return action

    def epsilonGreedy(self, state):
        """ε-グリーディー選択を用いて行動を選択して返す。"""
        r = np.random.rand()    # 乱数を生成する
        action = self.uniformly() if  r < self.epsilon else self.greedy(state)  # 確率εで一様ランダムに、確率1-εでグリーディーに行動を選択する
        return action

    def chooseAction(self, state):
        """行動を選択して返す。"""
        action = self.epsilonGreedy(state)  # ε-グリーディー選択で行動を選択する
        return action

    def setEnvironment(self, env):
        """環境をセットする。"""
        self.env = env

    def __init__(self, seed=0, discountRate=0.9, stepSize = 0.1, epsilon=0.1, maxSteps=1000000):
        np.random.seed(seed)
        self.discountRate = discountRate    # 割引率 γ
        self.stepSize = stepSize    # ステップ・サイズ α
        self.epsilon = epsilon  # ε-グリーディー選択のε
        self.maxSteps = maxSteps  # 最大ステップ数
        self.env = None # 環境
        self.aQ = None  # Q値

Q学習エージェント qlagent.py

Q学習エージェントを表す QLearningAgent クラスには、Q学習のアルゴリズムだけを実装します。

Q学習のアルゴリズムは、次のようなものです。

  1. 行動価値 [math]Q(s, a)[/math] を初期化
  2. エピソードごとに繰り返し:
    1. 状態 [math]s[/math] を初期化
    2. [math]s[/math] が終端状態になるまで繰り返し:
      1. [math]Q[/math]から導かれる行動選択確率に基づいて [math]s[/math] から行動 [math]a[/math] を選択
      2. [math]a[/math] を実行し、次の状態 [math]s'[/math] と報酬 [math]r[/math] を観測する
      3. [math]Q(s, a) \leftarrow Q(s, a) + \alpha \left[ r + \gamma * \max_{a'}Q(s', a') - Q(s, a)\right][/math]
      4. [math]s \leftarrow s'[/math]

これを、Pythonで実装すると、次のようになります。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# qlagent.py
#
# Copyright 2017 Tohgoroh Matsui All Rights Reserved.
#
import rlagent
import numpy as np


class QLearningAgent(rlagent.ReinforcementLearningAgent):
    """Q学習エージェント。"""

    def learn(self):
        """Q学習を用いて学習する。"""
        gamma = self.discountRate   # 割引率
        alpha = self.stepSize   # ステップ・サイズ
        episodes = 0    # エピソード数
        totalSteps = 0  # ステップ数
        self.aQ = np.zeros([self.env.states, self.env.actions]) # Q値をゼロに初期化する
        while totalSteps < self.maxSteps: # ステップ数が最大ステップ数になるまで繰り返し:
            steps = 0   # このエピソードにおけるステップ数
            state = self.env.initState() # 状態を初期化する
            while not self.env.isTerminal(state):   # 終端状態になるまで繰り返す:
                action = self.epsilonGreedy(state)  # 行動を選択する
                state_, reward = self.env.takeAction(state, action)  # 行動を実行し、次の状態と報酬を観測する
                #self.env.printLog(steps, state, action, reward, state_)   # ログを出力する
                action_ = self.greedy(state_)   # 次の状態でQ値が最大の行動を調べる
                s, a, = self.env.getS(state), self.env.getA(action) # 状態番号と行動番号
                s_, a_ = self.env.getS(state_), self.env.getA(action_)  # 状態番号を行動番号
                self.aQ[s, a] += alpha * (reward + gamma * self.aQ[s_, a_] - self.aQ[s, a])  # Q値を更新する
                state = state_  # 状態を更新する
                steps += 1
                totalSteps += 1
                if totalSteps % np.power(10, np.floor(np.log10(totalSteps))) == 0:
                    eval = self.evaluate()  # 評価する
                    print('%d, %f' % (totalSteps, eval))
            episodes += 1
        print(self.aQ)

    def __init__(self, seed=0, discountRate=0.9, stepSize=0.1, epsilon=0.1, maxSteps=1000000):
        super().__init__(seed, discountRate, stepSize, epsilon, maxSteps)

迷路 maze.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# maze.py
#
# Copyright 2017 Tohgoroh Matsui All Rights Reserved.
#
import environment as env
import numpy as np


class Maze(env.Environment):
    """Sutton & Barto 4x3迷路問題。状態は (x座標, y座標) で表される。"""

    walls = np.array([[0, 0, 0, 0],
                      [0, 1, 0, 0],
                      [0, 0, 0, 0]]).T  # 壁
    terminals = np.array([[ 0, 0, 0, 1],
                          [ 0, 0, 0, 1],
                          [ 0, 0, 0, 0]]).T # 終端状態
    rewards = np.array([[-0.02, -0.02, -0.02,  1.],
                        [-0.02,  0.,   -0.02, -1.],
                        [-0.02, -0.02, -0.02, -0.02]]).T    # 報酬
    width = len(walls)  # 迷路の幅
    height = len(walls[0])   # 迷路の高さ
    states = width * height   # 状態数
    actions = 4   # 行動数
    deterministic = False # 状態遷移が決定的

    @classmethod
    def initState(cls):
        """初期状態を返す。"""
        return (0, 2)

    @classmethod
    def isTerminal(cls, state):
        """渡された状態が終端状態ならTrue, そうでないならFalseを返す。"""
        x, y = state[0], state[1]
        return True if cls.terminals[x, y] == 1 else False

    @classmethod
    def isWall(cls, state):
        """渡された状態が壁ならTrue, そうでないならFalseを返す。"""
        x, y = state[0], state[1]
        return True if x < 0 or x >= cls.width or y < 0 or y >= cls.height or cls.walls[x, y] == 1 else False

    @classmethod
    def getS(cls, state):
        """状態を状態番号に変換して返す。"""
        x, y = state[0], state[1]
        s = y * cls.width + x
        return s

    @classmethod
    def getA(cls, action):
        """行動を行動番号に変換して返す。"""
        return action

    @classmethod
    def getReward(cls, state):
        """報酬を返す。"""
        x, y = state[0], state[1]
        reward = cls.rewards[x, y]
        return reward

    @classmethod
    def takeAction(cls, state, action):
        """行動を実行して状態を更新し、次の状態と報酬を返す。"""
        if action == 0:
            forward = cls.north(state)
            left = cls.east(state)
            right = cls.west(state)
        elif action == 1:
            forward = cls.east(state)
            left = cls.south(state)
            right = cls.west(state)
        elif action == 2:
            forward = cls.west(state)
            left = cls.north(state)
            right = cls.south(state)
        else:
            forward = cls.south(state)
            left = cls.west(state)
            right = cls.east(state)
        if cls.deterministic:
            state_ = forward
        else:
            r = np.random.random()
            if r < 0.8:
                state_ = forward
            elif r < 0.9:
                state_ = left
            else:
                state_ = right
        reward = cls.getReward(state_)  # 報酬
        return state_, reward

    @classmethod
    def strState(cls, state):
        """状態を表す文字列を返す。"""
        x, y = state[0], state[1]
        return '(%d, %d)' % (x, y)

    @classmethod
    def strAction(cls, action):
        """行動を表す文字列を返す。"""
        if action == 0:
            str = 'north'
        elif action == 1:
            str = 'east'
        elif action == 2:
            str = 'west'
        else:
            str = 'south'
        return str

    @classmethod
    def north(cls, state):
        """北に壁がない場合は北隣の状態を、壁がある場合は現在の状態を返す。"""
        x, y = state[0], state[1]
        y_ = y - 1
        return (x, y_) if not cls.isWall((x, y_)) else state

    @classmethod
    def east(cls, state):
        """東に壁がない場合は東隣の状態を、壁がある場合は現在の状態を返す。"""
        x, y = state[0], state[1]
        x_ = x + 1
        return (x_, y) if not cls.isWall((x_, y)) else state

    @classmethod
    def west(cls, state):
        """西に壁がない場合は西隣の状態を、壁がある場合は現在の状態を返す。"""
        x, y = state[0], state[1]
        x_ = x - 1
        return (x_, y) if not cls.isWall((x_, y)) else state

    @classmethod
    def south(cls, state):
        """南に壁がない場合は南隣の状態を、壁がある場合は現在の状態を返す。"""
        x, y = state[0], state[1]
        y_ = y + 1
        return (x, y_) if not cls.isWall((x, y_)) else state

    def __init__(self, deterministic=False):
        Maze.deterministic = deterministic
        super().__init__()

実行用ファイル mazeql.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# mazeql.py
#
# Copyright 2017 Tohgoroh Matsui All Rights Reserved.
#
import maze
import qlagent
import sys

if __name__ == '__main__':
    argvs = sys.argv    # コマンドライン引数
    seed = 0 if len(argvs) == 1 else int(argvs[1])  # 乱数のシード
    maze = maze.Maze()  # 環境
    agent = qlagent.QLearningAgent(seed=seed)   # Q学習エージェント
    agent.setEnvironment(maze)  # 環境をエージェントにセットする
    agent.learn()   # 学習する

実行結果

$ python3 mazeql.py
1, 49.290000
2, 42.750000
3, 59.680000
4, 64.650000
5, 67.800000
6, 37.910000
7, 46.420000
8, 55.940000
9, 51.880000
10, 44.650000
20, 66.240000
30, 70.940000
40, 71.500000
50, 97.150000
60, 62.780000
70, 100.000000
80, 100.000000
90, 100.000000
100, 14.470000
200, 50.660000
300, 46.580000
400, 9.960000
500, 9.270000
600, 8.780000
700, 9.610000
800, 7.630000
900, 9.740000
1000, 7.850000
2000, 6.960000
3000, 7.010000
4000, 7.170000
5000, 6.510000
6000, 6.600000
7000, 6.610000
8000, 7.310000
9000, 6.580000
10000, 6.680000
20000, 6.700000
30000, 7.700000
40000, 6.920000
50000, 7.190000
60000, 6.970000
70000, 6.790000
80000, 7.080000
90000, 7.070000
100000, 7.400000
200000, 7.460000
300000, 6.770000
400000, 7.730000
500000, 6.970000
600000, 6.790000
700000, 7.220000
800000, 7.590000
900000, 7.270000
1000000, 7.030000
[[ 0.57936398  0.66081859  0.54196507  0.47743206]
 [ 0.68507578  0.80528612  0.59248293  0.6700417 ]
 [ 0.83194323  0.94140941  0.67752173  0.59096585]
 [ 0.          0.          0.          0.        ]
 [ 0.56468961  0.44486706  0.47738216  0.39147569]
 [ 0.          0.          0.          0.        ]
 [ 0.40230731 -0.64613184  0.33453209  0.17906799]
 [ 0.          0.          0.          0.        ]
 [ 0.45612173  0.31248324  0.37599547  0.35867265]
 [ 0.30082596  0.30399564  0.3301133   0.3123197 ]
 [ 0.31227229  0.13243078  0.31015554  0.28215629]
 [-0.6082414  -0.0130165   0.18116616 -0.01257615]]

乱数を使っているので、乱数のシードを変えて何回か実行し、平均を取らないといけません。

添付ファイル: filemazeql.py 5件 [詳細] filemaze.py 5件 [詳細] fileqlagent.py 4件 [詳細] filerlagent.py 4件 [詳細] fileenvironment.py 4件 [詳細] filemaze.png 2件 [詳細]
トップ   編集 凍結 差分 バックアップ 添付 複製 名前変更 リロード   新規 一覧 単語検索 最終更新   ヘルプ   最終更新のRSS