RLlib を使ってナップサック問題を強化学習2

強化学習 Python

局所最適に陥っていたと思われる 前回 に対して、以下の改善案 ※ を思いついたので試してみました。

  • より困難な目標を達成した場合に報酬（価値）へボーナスを加算
 ※ 局所最適から脱して、より良い結果を目指す効果を期待

今回のサンプルコードは http://github.com/fits/try_samples/tree/master/blog/20201019/

サンプル1 改良版（ボーナス加算）

単一操作（品物の 1つを -1 or +1 するか何もしない）を行動とした（前回の）サンプル1 にボーナスを加算する処理を加えてみました。

とりあえず、価値の合計が 375（0-1 ナップサック問題としての最適解）を超えた場合に報酬へ +200 するようにしてみます。

前回から、変数や関数名を一部変更していますが、基本的な処理内容に変更はありません。

また、PPOTrainer では episode_reward_mean / vf_clip_param の値が 200 を超えると警告ログを出すようなので（ppo.py の warn_about_bad_reward_scales）、config で vf_clip_param（デフォルト値は 10）の値を変更するようにしています。

sample1_bonus.ipynb
・・・
def next_state(items, state, action):
    idx = action // 2
    act = action % 2

    if idx < len(items):
        state[idx] += (1 if act == 1 else -1)

    return state

def calc_value(items, state, max_weight, burst_value):
    reward = 0
    weight = 0
    
    for i in range(len(state)):
        reward += items[i][0] * state[i]
        weight += items[i][1] * state[i]
    
    if weight > max_weight or min(state) < 0:
        reward = burst_value
    
    return reward, weight

class Knapsack(gym.Env):
    def __init__(self, config):
        self.items = config["items"]
        self.max_weight = config["max_weight"]
        self.episode_steps = config["episode_steps"]
        self.burst_reward = config["burst_reward"]
        self.bonus_rules = config["bonus_rules"]
        
        n = self.episode_steps
        
        self.action_space = Discrete(len(self.items) * 2 + 1)
        self.observation_space = Box(low = -n, high = n, shape = (len(self.items), ))
        
        self.reset()

    def reset(self):
        self.current_steps = 0
        self.state = [0 for _ in self.items]
        
        return self.state

    def step(self, action):
        self.state = next_state(self.items, self.state, action)
        
        r, _ = calc_value(self.items, self.state, self.max_weight, self.burst_reward)
        reward = r
        
        # 段階的なボーナス加算
        for (v, b) in self.bonus_rules:
            if r > v:
                reward += b
        
        self.current_steps += 1
        done = self.current_steps >= self.episode_steps
        
        return self.state, reward, done, {}

items = [
    [105, 10],
    [74, 7],
    [164, 15],
    [32, 3],
    [235, 22]
]

config = {
    "env": Knapsack, 
    "vf_clip_param": 60,
    "env_config": {
        "items": items, "episode_steps": 10, "max_weight": 35, "burst_reward": -100, 
        # ボーナスの設定
        "bonus_rules": [ (375, 200) ]
    }
}

・・・

trainer = PPOTrainer(config = config)

・・・
# 30回の学習
for _ in range(30):
    r = trainer.train()
    ・・・

・・・

rs = []

# 1000回試行
for _ in range(1000):
    
    s = [0 for _ in range(len(items))]
    r_tmp = config["env_config"]["burst_reward"]

    for _ in range(config["env_config"]["episode_steps"]):
        a = trainer.compute_action(s)
        s = next_state(items, s, a)

        r, w = calc_value(items, s, config["env_config"]["max_weight"], config["env_config"]["burst_reward"])
        
        r_tmp = max(r, r_tmp)

    rs.append(r_tmp)

collections.Counter(rs)

上記の結果（30回の学習後に 1000回試行してそれぞれの最高値をカウント）は以下のようになりました。

結果
Counter({376: 957, 375: 42, 334: 1})

最適解の 376 が出るようになっており、ボーナスの効果はそれなりにありそうです。

ただし、毎回このような結果になるわけではなく、前回と同じように 375（0-1 ナップサック問題としての最適解）止まりとなる場合もあります。

検証

次に、ナップサック問題の内容を変えて検証してみます。

ここでは、「2.5 ナップサック問題 - 数理システム」の例題を題材として、状態の範囲やボーナスの内容を変えると結果にどのような差が生じるのかを確認します。

ナップサック問題の内容

価値 サイズ
120 10
130 12
80 7
100 9
250 21
185 16

最大容量（サイズ） 65 における最適解は以下の通りです。

価値 770 の組み合わせ（最適解）
3, 0, 2, 0, 1, 0
価値 745 の組み合わせ（0-1 ナップサック問題の最適解）
0, 1, 1, 1, 1, 1

1. 単一操作（品物の 1つを -1 or +1 するか何もしない）

行動は前回の サンプル1 と同様の以下とします。

  • 品物のどれか 1つを -1 or +1 するか、何も変更しない

ここでは、以下のような状態範囲とボーナスを試しました。

状態範囲（品物毎の個数の範囲）
状態タイプ 最小値 最大値
a -10 10
b 0 5
c 0 3
ボーナス定義
ボーナス定義タイプ v > 750 v > 760 v > 765
0 0 0 0
1 100 100 100
2 100 200 400

ボーナスは段階的に加算し、ボーナス定義タイプ 2 で価値が仮に 770 だった場合は、700（100 + 200 + 400）を加算する事にします。

また、状態（品物毎の個数）はその範囲を超えないよう最小値もしくは最大値で止まるようにしました。

なお、ここからは Jupyter Notebook ではなく Python スクリプトとして実行します。

test1.py
import sys
import numpy as np

import gym
from gym.spaces import Discrete, Box

import ray
from ray.rllib.agents.ppo import PPOTrainer

import collections

N = int(sys.argv[1])
EPISODE_STEPS = int(sys.argv[2])
STATE_TYPE = sys.argv[3]
BONUS_TYPE = sys.argv[4]

items = [
    [120, 10],
    [130, 12],
    [80, 7],
    [100, 9],
    [250, 21],
    [185, 16]
]

state_types = {
    "a": (-10, 10),
    "b": (0, 5),
    "c": (0, 3)
}

bonus_types = {
    "0": [],
    "1": [(750, 100), (760, 100), (765, 100)],
    "2": [(750, 100), (760, 200), (765, 400)]
}

vf_clip_params = {
    "0": 800,
    "1": 1100,
    "2": 1500
}

def next_state(items, state, action, state_range):
    idx = action // 2
    act = action % 2

    if idx < len(items):
        v = state[idx] + (1 if act == 1 else -1)
        # 状態が範囲内に収まるように調整
        state[idx] = min(state_range[1], max(state_range[0], v))

    return state

def calc_value(items, state, max_weight, burst_value):
    reward = 0
    weight = 0
    
    for i in range(len(state)):
        reward += items[i][0] * state[i]
        weight += items[i][1] * state[i]
    
    if weight > max_weight or min(state) < 0:
        reward = burst_value
    
    return reward, weight

class Knapsack(gym.Env):
    def __init__(self, config):
        self.items = config["items"]
        self.max_weight = config["max_weight"]
        self.episode_steps = config["episode_steps"]
        self.burst_reward = config["burst_reward"]
        self.state_range = config["state_range"]
        self.bonus_rules = config["bonus_rules"]
        
        self.action_space = Discrete(len(self.items) * 2 + 1)
        
        self.observation_space = Box(
            low = self.state_range[0], 
            high = self.state_range[1], 
            shape = (len(self.items), )
        )

        self.reset()

    def reset(self):
        self.current_steps = 0
        self.state = [0 for _ in self.items]
        
        return self.state

    def step(self, action):
        self.state = next_state(self.items, self.state, action, self.state_range)
        
        r, _ = calc_value(self.items, self.state, self.max_weight, self.burst_reward)
        reward = r

        for (v, b) in self.bonus_rules:
            if r > v:
                reward += b
        
        self.current_steps += 1
        done = self.current_steps >= self.episode_steps
        
        return self.state, reward, done, {}

config = {
    "env": Knapsack, 
    "vf_clip_param": vf_clip_params[BONUS_TYPE],
    "env_config": {
        "items": items, "max_weight": 65, "burst_reward": -100, 
        "episode_steps": EPISODE_STEPS, 
        "state_range": state_types[STATE_TYPE], 
        "bonus_rules": bonus_types[BONUS_TYPE]
    }
}

ray.init()

trainer = PPOTrainer(config = config)

for _ in range(N):
    r = trainer.train()
    print(f'iter = {r["training_iteration"]}')

print(f'N = {N}, EPISODE_STEPS = {EPISODE_STEPS}, state_type = {STATE_TYPE}, bonus_type = {BONUS_TYPE}')

rs = []

for _ in range(1000):
    s = [0 for _ in range(len(items))]
    r_tmp = config["env_config"]["burst_reward"]

    for _ in range(config["env_config"]["episode_steps"]):
        a = trainer.compute_action(s)
        s = next_state(items, s, a, config["env_config"]["state_range"])

        r, w = calc_value(
            items, s, 
            config["env_config"]["max_weight"], config["env_config"]["burst_reward"]
        )
        
        r_tmp = max(r, r_tmp)

    rs.append(r_tmp)

print( collections.Counter(rs) )

ray.shutdown()
実行例
> python test1.py 50 10 a 0

学習回数 50、1エピソードのステップ数 10 で学習した後、1000回の試行で最も件数の多かった価値を列挙する処理を 3回実施した結果です。（() 内の数値は 1000回の内にその値が最高値だった件数）

結果（学習回数 = 50、エピソードのステップ数 = 10）
状態タイプ ボーナス定義タイプ 状態の最小値 状態の最大値 765 超過時の総ボーナス 1回目 2回目 3回目
a-0 a 0 -10 10 0 735 (994) 735 (935) 735 (916)
a-1 a 1 -10 10 +300 745 (965) 745 (977) 735 (976)
a-2 a 2 -10 10 +700 735 (945) 735 (971) 770 (1000)
b-0 b 0 0 5 0 750 (931) 750 (829) 750 (1000)
b-1 b 1 0 5 +300 765 (995) 765 (998) 750 (609)
b-2 b 2 0 5 +700 765 (1000) 765 (995) 765 (998)
c-0 c 0 0 3 0 750 (998) 750 (996) 750 (1000)
c-1 c 1 0 3 +300 765 (1000) 750 (993) 765 (1000)
c-2 c 2 0 3 +700 765 (999) 765 (1000) 770 (999)

やはり、ボーナスは有効そうですが、状態タイプ a のように状態の範囲が広く、ボーナスの発生頻度が低くなるようなケースでは有効に働かない可能性も高そうです。

ボーナス定義タイプ 2 で最適解の 770 が出るようになっているものの、頻出するようなものでも無く、たまたま学習が上手くいった場合にのみ発生しているような印象でした。

また、b-1 の 3回目で件数が他と比べて低くなっていますが、こちらは学習が（順調に進まずに）足りていない状態だと考えられます。

2. 一括操作（全品物をそれぞれ -1 or 0 or +1 する）

次に、行動を以下のように変えて同じように検証してみます。

  • 全ての品物を対象にそれぞれを -1 or 0 or +1 する

こちらは、ボーナス加算タイプを 1種類追加しました。

状態範囲（品物毎の個数の範囲）
状態タイプ 最小値 最大値
a -10 10
b 0 5
c 0 3
ボーナス加算
ボーナス加算タイプ v > 750 v > 760 v > 765
0 0 0 0
1 100 100 100
2 100 200 400
3 200 400 800

行動の変更に伴い action_space を Box で定義しています。

test2.py
・・・

state_types = {
    "a": (-10, 10),
    "b": (0, 5),
    "c": (0, 3)
}

bonus_types = {
    "0": [],
    "1": [(750, 100), (760, 100), (765, 100)],
    "2": [(750, 100), (760, 200), (765, 400)],
    "3": [(750, 200), (760, 400), (765, 800)]
}

・・・

def next_state(items, state, action, state_range):
    for i in range(len(action)):
        v = state[i] + round(action[i])
        state[i] = min(state_range[1], max(state_range[0], v))

    return state

・・・

class Knapsack(gym.Env):
    def __init__(self, config):
        self.items = config["items"]
        self.max_weight = config["max_weight"]
        self.episode_steps = config["episode_steps"]
        self.burst_reward = config["burst_reward"]
        self.state_range = config["state_range"]
        self.bonus_rules = config["bonus_rules"]
        # 品物毎の -1 ～ 1
        self.action_space = Box(low = -1, high = 1, shape = (len(self.items), ))
        
        self.observation_space = Box(
            low = self.state_range[0], 
            high = self.state_range[1], 
            shape = (len(self.items), )
        )

        self.reset()

    def reset(self):
        self.current_steps = 0
        self.state = [0 for _ in self.items]
        
        return self.state

    def step(self, action):
        self.state = next_state(self.items, self.state, action, self.state_range)
        
        r, _ = calc_value(self.items, self.state, self.max_weight, self.burst_reward)
        reward = r

        for (v, b) in self.bonus_rules:
            if r > v:
                reward += b
        
        self.current_steps += 1
        done = self.current_steps >= self.episode_steps
        
        return self.state, reward, done, {}

・・・

こちらの方法では、学習回数 50 では明らかに足りなかったので 100 にして実施しました。

結果（学習回数 = 100、エピソードのステップ数 = 10）
状態タイプ ボーナス加算タイプ 状態の最小値 状態の最大値 765 超過時の総ボーナス 1回目 2回目 3回目
a-0 a 0 -10 10 0 735 (477) 735 (531) 735 (714)
a-1 a 1 -10 10 +300 735 (689) 735 (951) 745 (666)
a-2 a 2 -10 10 +700 735 (544) 735 (666) 735 (719)
a-3 a 3 -10 10 +1400 745 (633) 735 (735) 735 (875)
b-0 b 0 0 5 0 735 (364) 760 (716) 740 (590)
b-1 b 1 0 5 +300 735 (935) 760 (988) 655 (685)
b-2 b 2 0 5 +700 760 (1000) 735 (310) 770 (963)
b-3 b 3 0 5 +1400 675 (254) 770 (1000) 770 (909)
c-0 c 0 0 3 0 735 (762) 740 (975) 740 (669)
c-1 c 1 0 3 +300 740 (935) 740 (842) 735 (963)
c-2 c 2 0 3 +700 770 (999) 770 (1000) 715 (508)
c-3 c 3 0 3 +1400 770 (1000) 770 (1000) 770 (1000)

学習の足りていない所が散見されますが（学習も不安定）、特定のタイプで最適解の 770 が割と頻繁に出るようになりました。

ただ、c-3 の場合でも 770 が出やすくなっているものの、確実にそのように学習するわけではありませんでした。

結局のところ、状態・行動・報酬の設計次第という事かもしれません。