深層強化学習の実装例のご紹介

env = gym.make("FrozenLake-v0")
n_obs = env.observation_space.n
n_act = env.action_space.n

q = np.zeros([n_obs, n_act])

epoch_cnt = 20000
max_steps = 200
epsilon = 0.001
gamma = 0.9
alpha = 0.9

rewards = np.zeros(epoch_cnt)

for epoch in tqdm(range(epoch_cnt)):
    pobs = env.reset()
    done = False
    for step in range(max_steps):
        pact = np.argmax(q[pobs, :])
        pact = np.random.choice(np.where(q[pobs, :] == q[pobs, pact])[0])
        if np.random.rand() <= epsilon:
            pact = env.action_space.sample()
        obs, reward, done, _ = env.step(pact)
        if not done:
            q[pobs, pact] += alpha * (reward - q[pobs, pact] + gamma * np.max(q[obs, :]))
        else:
            q[pobs, pact] += alpha * (reward - q[pobs, pact])
        pobs = obs
        rewards[epoch] = reward
        if done:
            break

rates = np.average(rewards.reshape([epoch_cnt//1000, 1000]), axis = 1)
plt.plot(rates)
plt.show()

import copy
import time
import numpy as np
import matplotlib
import matplotlib.pyplot as plt
import chainer
import chainer.links as L
import chainer.functions as F
import gym

# NNクラス定義
class NN(chainer.Chain):
    def __init__(self):
        super(NN, self).__init__(
            xc = L.Linear(16, 100),
            ch = L.Linear(100, 100),
            hy = L.Linear(100, 4)
        )
    def __call__(self, x):
        x = chainer.Variable(x)
        h = F.relu(self.xc(x))
        h = F.relu(self.ch(h))
        y = F.relu(self.hy(h))
        return y
    def reset(self):
        self.zerograds()

# 状態変換関数定義
def convert(obs):
    tmp = np.zeros(16)
    tmp[obs] = 1
    obs = np.array(tmp, dtype="float32")
    return obs

# 環境
env = gym.make("FrozenLake-v0")

# モデル
Q = NN()  # 近似Q関数
Q_ast = copy.deepcopy(Q)
optimizer = chainer.optimizers.Adam()
optimizer.setup(Q)

# 定数
EPOCH_NUM = 1000
MEMORY_SIZE = 1000
BATCH_SIZE = 100
EPSILON = 1
EPSILON_DECREASE = 0.005
EPSILON_MIN = 0.01
START_REDUCE_EPSILON = 1000
TRAIN_FREQ = 10
UPDATE_TARGET_Q_FREQ = 20
GAMMA = 0.99

total_step = 0
memory = []
total_rewards = []
total_losses = []

# 学習開始
print("\t".join(["epoch", "EPSILON", "reward", "total_step", "loss", "elapsed_time"]))
start = time.time()

for epoch in range(EPOCH_NUM):
    pobs = env.reset()
    pobs = convert(pobs)
    done = False
    total_reward = 0
    total_loss = 0

    while not done:
        # 行動選択
        pact = env.action_space.sample()
        if np.random.rand() > EPSILON:
            pact = Q(pobs.reshape((1, 16)))
            pact = np.argmax(pact.data)

        # 行動
        obs, reward, done, _ = env.step(pact)
        obs = convert(obs)

        # メモリに蓄積
        memory.append((pobs, pact, reward, obs, done))
        if len(memory) > MEMORY_SIZE:
            memory.pop(0)

        # 学習
        if len(memory) == MEMORY_SIZE:
            if total_step % TRAIN_FREQ == 0:
                np.random.shuffle(memory)
                memory_idx = range(len(memory))
                for i in memory_idx[::BATCH_SIZE]:
                    batch = np.array(memory[i:i+BATCH_SIZE])
                    pobss = np.array(batch[:,0].tolist(), dtype="float32").reshape((BATCH_SIZE, 16))
                    pacts = np.array(batch[:,1].tolist(), dtype="int32")
                    rewards = np.array(batch[:,2].tolist(), dtype="int32")
                    obss = np.array(batch[:,3].tolist(), dtype="float32").reshape((BATCH_SIZE, 16))
                    dones = np.array(batch[:,4].tolist(), dtype="bool")
                    # set y
                    q = Q(pobss)
                    maxq = list(map(np.max, Q_ast(obss).data))
                    target = copy.deepcopy(q.data)
                    for j in range(BATCH_SIZE):
                        target[j, pacts[j]] = rewards[j]+GAMMA*maxq[j]*(not dones[j])
                    # Perform a gradient descent step
                    Q.reset()
                    loss = F.mean_squared_error(q, chainer.Variable(target))
                    total_loss += loss.data
                    loss.backward()
                    optimizer.update()

        # Q関数の更新
        if total_step % UPDATE_TARGET_Q_FREQ == 0:
            Q_ast = copy.deepcopy(Q)

        # εの減少
        if EPSILON > EPSILON_MIN:
            if total_step > START_REDUCE_EPSILON:
                EPSILON -= EPSILON_DECREASE

        # 次の行動へ
        total_reward += reward
        total_step += 1
        pobs = obs

    total_rewards.append(total_reward)
    total_losses.append(total_loss)

    if (epoch+1) % 100 == 0:
        elapsed_time = time.time()-start
        r = sum(total_rewards[((epoch+1)-10):(epoch+1)])
        l = sum(total_losses[((epoch+1)-10):(epoch+1)])
        print("\t".join(map(str,[epoch+1, EPSILON, r, total_step, l, str(elapsed_time)+"[sec]"])))
        start = time.time()

plt.figure(figsize=(10,5))
resize = (len(total_rewards)//10, 10)
tmp = np.array(total_rewards, dtype="float32").reshape(resize)
tmp = np.average(tmp, axis=1)
plt.plot(tmp)
plt.show()

env = gym.make("CartPole-v0")
print("observation space num: ", env.observation_space.shape[0])
print("action space num: ", env.action_space.n)
print("-"*50)
pobs = env.reset()
done = False
while not done:
    act = env.action_space.sample()
    obs, reward, done, _ = env.step(act)
    print(pobs, act, reward, obs, done)
    pobs = obs

import copy
import time
import numpy as np
import matplotlib
import matplotlib.pyplot as plt
import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
from torch.autograd import Variable
import gym
from gym import wrappers

# 環境
MONITOR = False
env = gym.make("CartPole-v0")
if MONITOR:
    env = wrappers.Monitor(env, "./result", force=True)

obs_num = env.observation_space.shape[0]
acts_num = env.action_space.n
HIDDEN_SIZE = 100

class NN(nn.Module):
    def __init__(self):
        super(NN, self).__init__()
        self.fc1 = nn.Linear(obs_num, HIDDEN_SIZE)
        self.fc2 = nn.Linear(HIDDEN_SIZE, HIDDEN_SIZE)
        self.fc3 = nn.Linear(HIDDEN_SIZE, HIDDEN_SIZE)
        self.fc4 = nn.Linear(HIDDEN_SIZE, acts_num)
    def __call__(self, x):
        h = F.relu(self.fc1(x))
        h = F.relu(self.fc2(h))
        h = F.relu(self.fc3(h))
        y = F.relu(self.fc4(h))
        return y

# 定数
EPOCH_NUM = 3000
STEP_MAX = 200
MEMORY_SIZE = 200
BATCH_SIZE = 50
EPSILON = 1.0
EPSILON_DECREASE = 0.001
EPSILON_MIN = 0.1
START_REDUCE_EPSILON = 200
TRAIN_FREQ = 10
UPDATE_TARGET_Q_FREQ = 20
GAMMA = 0.97
LOG_FREQ = 1000

# モデル
Q = NN()
Q_ast = copy.deepcopy(Q)
optimizer = optim.RMSprop(Q.parameters(), lr=0.00015, alpha=0.95, eps=0.01)

total_step = 0
memory = []
total_rewards = []

# 学習開始
print("\t".join(["epoch", "epsilon", "reward", "total_step", "elapsed_time"]))
start = time.time()
for epoch in range(EPOCH_NUM):
    pobs = env.reset()
    step = 0
    done = False
    total_reward = 0

    while not done and step < STEP_MAX:
        if MONITOR:
            env.render()
        # 行動選択
        pact = env.action_space.sample()
        # ε-greedy法
        if np.random.rand() > EPSILON:
            # 最適な行動を予測
            pobs_ = np.array(pobs, dtype="float32").reshape((1, obs_num))
            pobs_ = Variable(torch.from_numpy(pobs_))
            pact = Q(pobs_)
            maxs, indices = torch.max(pact.data, 1)
            pact = indices.numpy()[0]

        # 行動
        obs, reward, done, _ = env.step(pact)
        if done:
            reward = -1

        # メモリに蓄積
        memory.append((pobs, pact, reward, obs, done))
        if len(memory) > MEMORY_SIZE:
            memory.pop(0)

        # 学習
        if len(memory) == MEMORY_SIZE:
            # 経験リプレイ
            if total_step % TRAIN_FREQ == 0:
                memory_ = np.random.permutation(memory)
                memory_idx = range(len(memory_))
                for i in memory_idx[::BATCH_SIZE]:
                    batch = np.array(memory_[i:i+BATCH_SIZE])
                    pobss = np.array(batch[:,0].tolist(), dtype="float32").reshape((BATCH_SIZE, obs_num))
                    pacts = np.array(batch[:,1].tolist(), dtype="int32")
                    rewards = np.array(batch[:,2].tolist(), dtype="int32")
                    obss = np.array(batch[:,3].tolist(), dtype="float32").reshape((BATCH_SIZE, obs_num))
                    dones = np.array(batch[:,4].tolist(), dtype="bool")
                    # set y
                    pobss_ = Variable(torch.from_numpy(pobss))
                    q = Q(pobss_)
                    obss_ = Variable(torch.from_numpy(obss))
                    maxs, indices = torch.max(Q_ast(obss_).data, 1)
                    maxq = maxs.numpy()
                    target = copy.deepcopy(q.data.numpy())
                    for j in range(BATCH_SIZE):
                        target[j, pacts[j]] = rewards[j]+GAMMA*maxq[j]*(not dones[j])
                    # Perform a gradient descent step
                    optimizer.zero_grad()
                    loss = nn.MSELoss()(q, Variable(torch.from_numpy(target)))
                    loss.backward()
                    optimizer.step()

            # Q関数の更新
            if total_step % UPDATE_TARGET_Q_FREQ == 0:
                Q_ast = copy.deepcopy(Q)

        # εの減少
        if EPSILON > EPSILON_MIN and total_step > START_REDUCE_EPSILON:
            EPSILON -= EPSILON_DECREASE

        # 次の行動へ
        total_reward += reward
        step += 1
        total_step += 1
        pobs = obs

    total_rewards.append(total_reward)

    if (epoch+1) % LOG_FREQ == 0:
        r = sum(total_rewards[((epoch+1)-LOG_FREQ):(epoch+1)])/LOG_FREQ
        elapsed_time = time.time()-start
        print("\t".join(map(str,[epoch+1, EPSILON, r, total_step, str(elapsed_time)+"[sec]"])))
        start = time.time()

if MONITOR:
    env.render(close=True)