CNTK與深度強化學習筆記之二： Cart Pole遊戲示例

如需轉載，請指明出處。

前言

前面一篇文章，CNTK與深度強化學習筆記之一： 環境搭建和基本概念，非常概要的介紹了CNTK，深度強化學習和DQN的一些基本概念。這些概念希望後面還有文章繼續展開深入:)，但是隻看理論不寫代碼，很容易讓人迷惑。學習應該是一個理論和實踐反複的過程。上一章的公式太多，這一章沒有公式，隻有代碼。建議大家這兩章來回看，把理論和代碼對應起來。我們先來一個簡單的例子看一下。這個例子來自CNTK的官方文檔：CNTK 203: Reinforcement Learning Basics，做了一些修改。

上一篇文章之後，有幾個問題可能是比較讓人困惑的，先列舉在這裡，然後我們通過示例看看是如何解決的：

• 一開始沒有任何的訓練資料和标記，深度神經網絡是如何被訓練的呢？是不是能像上文提到的，從一堆垃圾資料裡面，學到有意義的東西？
• 經曆重放技術确實有效嗎？
• ε-greedy exploration算法如何實作，确實有效嗎？

gym的Cart Pole環境

Cart Pole在OpenAI的gym模拟器裡面，是相對比較簡單的一個遊戲。遊戲裡面有一個小車，上有豎着一根杆子。小車需要左右移動來保持杆子豎直。如果杆子傾斜的角度大于15°，那麼遊戲結束。小車也不能移動出一個範圍（中間到兩邊各2.4個機關長度）。如下圖所示：

在gym的Cart Pole環境（env）裡面，左移或者右移小車的action之後，env都會傳回一個+1的reward。到達200個reward之後，遊戲也會結束。

該環境的較長的描述在這裡。在這個連結裡面大家可以看到别人的模型和玩的成績。另外每個state和action值的含義也在這裡：CartPole-v0 wiki。

下面幾個詞後面的代碼會用到（通過變量名展現）：

• observation: 代表了對環境的觀察，即環境的State
• Spaces: 包括action space，表示有哪些action，和observation space，表示有哪些state。

CNTK的DQN模型實作

針對這個遊戲和DQN，我們來看看如何實作模型。下面分段講解代碼。

準備工作

import numpy as np
import math
import os
import random

import gym

import cntk as C

env = gym.make('CartPole-v0')
n_state = env.observation_space.shape[]
n_action = env.action_space.n

print('CartPole-v0 environment: %d states, %d actions' % (n_state, n_action))
           

這段代碼建立了CartPole-v0的環境。n_state儲存了observation數組的大小，即環境用多大的數組來表示狀态。n_action儲存了系統中action的數目。對于Cart Pole來說，這兩個值分别是4和2。

reward_target = 
epoch_baseline = 
reward_discount = 

hidden_size =  # hidden layer size
batch_size = 
learning_rate = 

max_epsilon = 
min_epsilon = 
epsilon_decay = 
           

這段代碼設定了一些參數。reward_target是我們想要拿到的一局遊戲中reward的總數，超過了這個總數程式就可以退出了。reward_discount是折扣率。這個遊戲是一個action，馬上會有一個reward，是以設定成0.99應該是比較合适的。max_epsilon，min_epsilon和epsilon_decay是為了實作上一節提到的ε-greedy exploration算法。reward_target設定為195和epoch_baseline設定為100的原因是，根據官方的定義，如果連續100個epoch的平均reward超過195，那麼CartPole-v0這個問題就被認為是解決了。這樣大家的成績就有了一個衡量标準：使用的epoch越少越好。

Brain

class Brain:
    def __init__(self):
        self.params = {}
        self.model, self.trainer, self.loss = self._create()

    def _create(self):
        observation = C.sequence.input_variable(n_state, np.float32, name='s')
        q_target = C.sequence.input_variable(n_action, np.float32, name='q')

        l1 = C.layers.Dense(hidden_size, activation=C.relu)
        l2 = C.layers.Dense(n_action)
        unbound_model = C.layers.Sequential([l1, l2])
        self.model = unbound_model(observation)

        self.params = dict(W1=l1.W, b1=l1.b, W2=l2.W, b2=l2.b)

        self.loss = C.reduce_mean(C.square(self.model-q_target), axis=)
        meas = C.reduce_mean(C.square(self.model-q_target), axis=)

        lr_schedule = C.learning_rate_schedule(learning_rate, C.UnitType.minibatch)
        learner = C.sgd(self.model.parameters,
                        lr_schedule,
                        gradient_clipping_threshold_per_sample=)

        progress_printer = C.logging.ProgressPrinter()
        self.trainer = C.Trainer(self.model, (self.loss, meas), learner, progress_printer)

        return self.model, self.trainer, self.loss

    def train(self, x, y):
        arguments = dict(zip(self.loss.arguments, [x,y]))
        updated, results = self.trainer.train_minibatch(arguments, outputs=[self.loss.output])

    def predict(self, s):
        return self.model.eval([s])
           

這個類叫做Brain，就是對應的深度神經網絡模型。模型的資料輸入是observation，即目前的環境State。模型有兩層，第一層是Dense層，即一個全連接配接層，有64個節點，使用RELU作為激活函數。第二個Dense層，有n_action個節點，作為模型的輸出，即每個Action對應的Reward。注意這裡的Action，實際上是Action的index。另外還有一個輸入q_target是目标值，即最大的未來獎勵。我們還定義了loss function和measure function，都是模型輸出和q_target的方差。

Memory

class Memory: # stored as ( s, a, r, s_ )
    samples = []

    def __init(self):
        pass

    def add(self, sample):
        self.samples.append(sample)

    def sample(self, n):
        n = min(n, len(self.samples))
        return random.sample(self.samples, n)
           

Memory類存儲了State（s），Action（a），Reward（r）和下一個State（s_）。

Agent

class Agent:
    steps = 
    epsilon = max_epsilon

    def __init__(self):
        self.brain = Brain()
        self.memory = Memory()

    def act(self, s):
        if random.random() < self.epsilon:
            return random.randint(, n_action-)
        else:
            return np.argmax(self.brain.predict(s))

    def observe(self, sample): # in (s, a, r, s_) format
        self.memory.add(sample)
        self.steps += 
        self.epsilon = min_epsilon + (max_epsilon - min_epsilon) * math.exp(-epsilon_decay * self.steps)

    def replay(self):
        batch = self.memory.sample(batch_size)

        no_state = np.zeros(n_state)

        states = np.array([ o[] for o in batch ], dtype=np.float32)
        states_ = np.array([ (no_state if o[] is None else o[]) for o in batch ], dtype=np.float32)

        p = self.brain.predict(states)
        p_ = self.brain.predict((states_))

        x = np.zeros((len(batch), n_state)).astype(np.float32)
        y = np.zeros((len(batch), n_action)).astype(np.float32)

        for i in range(len(batch)):
            s, a, r, s_ = batch[i]

            t = p[][i] # CNTK: [0] because of sequence dimension
            if s_ is None:
                t[a] = r
            else:
                t[a] = r + reward_discount * np.amax(p_[][i])

            x[i] = s
            y[i] = t

        self.brain.train(x, y)
           

Agent類實作了MDP的Agent。

方法act和observe實作了ε-greedy exploration算法。初始的時候，epsilon接近1，是以act方法的“andom.random() < self.epsilon”總是為True，是以Agent總是會随機選擇一個Action。随着observe執行的次數增加，step越來越大，epsilon越來越小，那麼随機選擇的機率越來越小，act方法會越來越多的選擇Brain預測的Reward最大的Action。這正是在上一章中描述的探索和開發困境的解決方法。

方法replay從Momory中随機選擇樣本，然後按照batch_size建構Brain的訓練資料集。這裡面有一個可能讓人困惑的地方是“t = p[0][i]”，為啥要先取0這個index。這個是因為predict方法調用的CNTK的eval方法，傳回的是一個list，而0這個index下面包含了shape為(batch_size, n_action)的數組。另外t[a]的指派，用到了上一章說的折扣的未來獎勵的概念。這裡我們還可以看到，Brain的訓練資料，是從Memory中取樣，并且使用了Brain自己的predict方法，為每一個采樣的(s, a)産生新的Reward，即最大的未來獎勵，然後再用這些(s, a)去訓練模型。

run

def run(agent):
    s = env.reset()
    R = 

    while True:
        env.render()

        a = agent.act(s.astype(np.float32))
        s_, r, done, info = env.step(a)

        if done:
            s_ = None

        agent.observe((s, a, r, s_))
        agent.replay()

        s = s_
        R += r

        if done:
            return R


agent = Agent()

epoch = 
reward_sum = 
while epoch < :
    reward = run(agent)
    reward_sum += reward
    epoch += 
    if epoch % epoch_baseline == :
        print('Epoch %d, average reward is %f, memory size is %d'
              % (epoch, reward_sum / epoch_baseline, len(agent.memory.samples)))

        if reward_sum / epoch_baseline > reward_target:
            print('Task solved in %d epoch' % epoch)
            break

        reward_sum = 
           

run方法運作一次遊戲。首先重置遊戲環境，然後不停的用act方法從agent中讀取一個動作，用step方法輸入到遊戲環境，拿到環境的回報，用observe和replay方法将回報更新到Agent并且訓練神經網絡模型，直到遊戲結束。這裡我們也可以看到上一章講的經曆重放：每一個Action之後，新的(s, a, r, s_)會被observe，即添加到Memory，然後會随機從Memory抽取batch_size個樣本進行模型訓練。

最後循環調用run方法，不停的玩遊戲，直到按照gym的定義，CartPole-v0問題被解決了。按照目前的算法，大概需要2500個epoch才能解決問題（在gym的官網，一個叫Svalorzen的使用者，用了42個epoch就解決了這個問題）。在調用的過程中，通過log我們可以發現，随着玩遊戲的局數增加，Brain的能力确實在增強。是以Brain确實是從不會玩遊戲，到玩2500次遊戲之後，就玩得很好了。第一個epoch的時候，動作完全是随機的，有戲可能很快就結束了，這樣的未來獎勵會很低。通過選擇高未來獎勵的Action，Brain玩遊戲的能力也越來越強。同樣未來獎勵也不斷在增加。

最後再強調一下（如果你還沒有意識到），我們在Agent的replay方法中，準備訓練資料y的時候，給出的值的含義，實際上是在目前State，采取Action之後，能得到的最大的未來獎勵Reward，參看上一章的貝爾曼方程。Brain的輸出，實際上是去逼近這個值。這樣本章開頭的問題應該都被解決了。