如何用令人拍案稱奇的強化學習玩小遊戲？

知識 03-01

如何用令人拍案稱奇的強化學習玩小遊戲？

打開今日頭條，查看更多圖片

作者 | 井森堡

責編 | 郭芮

光陰似箭日月如梭，轉眼已經9012年啦，現在哪怕隨便在大街上拉個路人，都聽過AlphaGo, AlphaGo Zero。

2016年3月，谷歌公布的AlphaGo打敗了人類頂尖圍棋棋手李世乭，吃瓜群眾們紛紛瑟瑟發抖，生怕人工智慧要打敗人類，佔領世界。之後谷歌又推出了AlphaGo Zero，之前的AlphaGo跟它一比簡直是個弟弟。

這兩個軟體的核心演算法都是強化學習，這篇文章想要幫助大家粗淺地理一理，一窺強化學習的皮毛。本文會介紹強化學習的基本概念以及最基本入門的演算法policy gradient，同時用這個演算法去訓練一個能玩小遊戲的模型。我們會通過細緻的講解讓大家了解強化學習，同時附上可讀性極高的代碼。

如何用令人拍案稱奇的強化學習玩小遊戲？

強化學習基本知識

強化學習是一種演算法，讓機器從剛開始時什麼都不會，通過不斷學習到最後達到某一目標。學習過程中，機器一直不斷地試錯與學習，環境一直會根據機器的做法給予懲罰或是激勵，機器根據環境給它的反饋不斷摸索出正確的實現目標的方法。

舉個例子，就好比老師教小朋友算術，小朋友剛開始什麼都不會，老師不斷讓小朋友做算術題，小朋友答的好，老師就給他糖，小朋友答得差，老師就嚴厲地斥責他。但在這個過程中，老師一直沒有直接告訴小朋友怎樣去答題，小朋友只能根據老師的反應來學習算術，久而久之，小朋友就記住了如何做才能得到糖同時規避斥責，最後達到學會算術的目的。

強化學習的幾大因素：agent, environment, state, action, reward。在上面的例子中，小朋友就是agent, 老師就是environment, 小朋友做題的情況就是state，小朋友採取的做題的行為就是action，老師給糖還是斥責就是reward。

再回到我們今天要讓電腦學會玩的小遊戲上，我們今天要玩一個打乒乓球的遊戲，遊戲環境可以在下面這個網站找到：https://gym.openai.com/envs/#classic_control。下圖是一張遊戲畫面。

如何用令人拍案稱奇的強化學習玩小遊戲？

遊戲里的agent是右邊的木板，agent可以採取的action是向上、向下以及不動。最終目標是想讓右邊的木板在與左邊的木板進行乒乓球比賽時能夠獲勝，observation指的是遊戲畫面。evironment是遊戲環境，每次agent要根據state做出action，然後envrionment會根據當前state和action給出reward和經過action後的state。

更加詳細的信息可以查看上文中提到的遊戲環境網站。

代碼實現

對state進行前處理，state是一張RGB圖片，前處理會將圖片轉化為灰度圖片，能加快收斂速度。同時前處理會裁剪圖片中無關部分，讓訓練過程更加穩定。

def prepro(o,image_size=[105,80]):
y = 0.2126 * o[:, :, 0] + 0.7152 * o[:, :, 1] + 0.0722 * o[:, :, 2] #gray scale
resized = skimage.transform.resize(y, image_size)[17:-8,:] #delete score board
return np.expand_dims(resized.astype(np.float32),axis=2) #shape (height, wodth) -> (1, height, wodth)

搭建agent模型，這裡的agent用兩層全連接的神經網路來實現，當然也可以用卷積神經網路，但是訓練結束時全連接的神經網路和卷積神經網路模型的性能差不多。出於搭建方便與加速收斂的目的，我們採用了全連接神經網路。

self.model = nn.Sequential(
nn.Linear(80*80*1, 256),
nn.ReLU(),
nn.Linear(256, 1),
nn.Sigmoid()

初始化環境：

def init_game_setting(self):
self.last_frame = None
pass

policy gradient的演算法實作過程如下：

如何用令人拍案稱奇的強化學習玩小遊戲？

代碼實現如下，看到這麼一長串代碼130多行的代碼不必驚慌，我們在整理GitHub代碼的時候，每一個關鍵變數都給了必要的注釋與說明，花上一點時間仔細研讀下，相信你們一定能讀懂的。在參數更新的時候採取了variance reduction這一竅門，加速了收斂速度。

for episode in range(self.hyper_param["episode"]):
o = self.env.reset()
self.last_frame = prepro(o)
# initial the first frame, otherwise self.last_frame is None
action = self.env.action_space.sample()
# random sample an action for initialization
o, _, _, _ = self.env.step(action)
episode_r = []
while True:
# make_action return the action to do and its probability.
# done means the game is over or not, if it"s over, then we update
# the parameters of model.
# When reward != 0, we calculate the advantage function using
# discount factor and baseline. And put it to "cumulated_r" according
# to each action.
o = prepro(o)
residual_state = o - self.last_frame
self.last_frame = o
action, p = self.make_action(residual_state, test = False)
o, reward, done, _ = self.env.step(action)
p_list.append(p) #left for BCELoss
action_list.append(action-2) #left for BCELoss
if reward != 0:
# When someone gets point, reset discount parameter,
# and record the experience
# cumulated_r means experience until someone getting point
# also calculate mean and var to do reward normalization
# this flag is used to compare with normal policy grad
if not self.hyper_param["base"]:
T = len(cumulated_r)
for i in range(T):
cumulated_r[i] += (gamma**(T-i))*reward
cumulated_r.append(reward)
reward_mean += sum(cumulated_r)
n_reward += len(cumulated_r)
for i in cumulated_r:
reward_var += i**2
episode_r.append(cumulated_r[-1])
total_reward.extend(cumulated_r)
cumulated_r = []
else:
T = len(cumulated_r)
cumulated_r = [reward]*(T+1)
episode_r.append(cumulated_r[-1])
total_reward.extend(cumulated_r)
cumulated_r = []
else:
cumulated_r.append(reward)
if done and n_b != batch:
n_b +=1
self.init_game_setting()
if len(lastest_r) < 30:
lastest_r.append(np.sum(episode_r))
elif len(lastest_r) == 30:
lastest_r.pop(0)
lastest_r.append(np.sum(episode_r))
self.training_curve.append(np.mean(lastest_r))
#print("Episode : %d Mean : %4f" % (episode, np.mean(lastest_r)), end = " ")
break
elif done and n_b == batch:
self.init_game_setting()
if len(lastest_r) < 30:
lastest_r.append(np.sum(episode_r))
elif len(lastest_r) == 30:
lastest_r.pop(0)
lastest_r.append(np.sum(episode_r))
self.training_curve.append(np.mean(lastest_r))
print("Episode : %d Mean : %4f" % (episode, np.mean(lastest_r)), end = " ")
#update per episode
self.optimizer.zero_grad()
loss = 0
if not self.hyper_param["base"]:
reward_mean = reward_mean/n_reward
reward_stddev = (reward_var/n_reward - (reward_mean)**2)**0.5
total_reward = torch.Tensor(total_reward).to(device)
total_reward = (total_reward - reward_mean) / reward_stddev
else:
total_reward = torch.Tensor(total_reward).to(device)
action_list = torch.Tensor(action_list).to(device)
p_list = torch.stack(p_list).view(-1)
loss = F.binary_cross_entropy(input = p_list,
target = action_list,
weight = total_reward,
reduction = "sum")
loss /= n_b
loss.backward()
self.optimizer.step()
#reset all record
n_b = 1
n_reward = 0
reward_mean = 0
reward_var = 0
reward_stddev = 0
total_reward = []
cumulated_r = []
p_list = []
action_list = []
print("Loss : %4f" % (loss.item()), end = "
")
break
if episode % 500 == 0 and episode != 0:
#Testing per 100 episode
print("Testing")
test_env = Environment("Pong-v0", self.argument, test=True)
result = draw_curve(agent = self, env=test_env)
if result >= best_result:
best_result = result
torch.save(self.model, "./log/"+self.hyper_param["model_name"]+".pkl"+str(episode))
torch.save(self.optimizer.state_dict(),
"./log/"+self.hyper_param["model_name"]+".optim")
np.save("training_curve_"+str(episode)+".npy", np.array(self.training_curve))