強化學習之CartPole

0x00 任務

  通過強化學習算法完成倒立擺任務，控制倒立擺在一定范圍內擺動。

0x01 設置jupyter登錄密碼

jupyter notebook --generate-config

jupyter notebook password （會輸入兩次密碼，用來驗證）

jupyter notebook 登錄

0x02 創建python note

0x03 代碼

# 聲明包
import numpy as np
import matplotlib.pyplot as plt
%matplotlib inline
import gym

# 聲明繪圖功能
from JSAnimation.IPython_display import display_animation
from matplotlib import animation
from IPython.display import display
def display_frames_as_gif(frames):
    plt.figure(figsize=(frames[0].shape[1]/72.0,frames[0].shape[0]/72.0),dpi=72)
    patch=plt.imshow(frames[0])
    plt.axis("off")
    
    def animate(i):
        patch.set_data(frames[i])
    anim=animation.FuncAnimation(plt.gcf(),animate,frames=len(frames),interval=50)
    anim.save('move_carpole.mp4') # 保存動畫
    display(display_animation(anim,default_mode='loop'))

# 隨機移動CartPole
frames=[]
env=gym.make('CartPole-v0')
observation=env.reset() # 重置環境
for step in range(0,200):
    frames.append(env.render(mode='rgb_array')) # 加載各個時刻圖像到幀
    action=np.random.choice(2) # 隨機返回： 0 小車向左，1 小車向右
    gym.logger.set_level(40)
    observation,reward,done,info=env.step(action) # 執行動作

運行后

移動 Caprpole的代碼並不重要，重要的是最后一行observation,reward,done,info=env.step(action)
reward 是 即時獎勵，若執行了action后，小車位置在+-2.4范圍之內而且桿的傾斜成都沒有超過20.9°，則設置獎勵為1.相反，若小車移出+-2.4范圍或者桿傾斜超過了20.9°的話，則獎勵為0。退出時 done是一個變量。若為結束狀態 則為true
這里代碼忽略了done, info變量保存調試信息。

最后使用display_frames_as_gif(frames) 函數去保存我們的gif

# 保存並繪制視頻
display_frames_as_gif(frames)

可正常保存視頻

CartPole的狀態

之前討論的迷宮問題中，狀態指的是每個格子的編號，由單個變量表示，0~8，然而倒立擺具有更復雜的狀態定義。

CartPole的狀態存儲在observation中,變量observarion是4個變量組成的列表，每個變量的內容如
小車位置 -2.4~2.4
小車速度 -∞~+∞
桿的角度 -41.8°~+41.8°
桿的角速度 -∞~+∞

因為變量是連續值，如果想要通過表格的形式來表達Q函數，就需要將他們進行離散化
比如使用0~5來標記變量的連續值
-2.4~-1.6=0
-1.6~-0.8=1
依次類推
則總共有6的4次方總組合 1296種類型 數字 表示 CartPole的狀態
而這個時候小車的方向只有向左和向右
所以，可以用1296行x2列的表格來表示Q函數

算法實現

• 變量設置

import numpy as np
import matplotlib.pyplot as plt
%matplotlib inline
import gym
# 變量設定
ENV='CartPole-v0' # 設置任務名
NUM_DIZITIZED=6 # 設置離散值個數
# 嘗試運行 CartPole
env=gym.make(ENV) # 設置要執行的任務
observarion=env.reset() # 環境初始化

• 求取用於離散化的闕值

# 求取用於離散化的闕值
def bins(clip_min,clip_max,num):
    return np.linespace(clip_min,clip_max,num+1)[1:-1] # 返回[-1.6,-0.8,0,0,0.8,1.6]
    

-∞~-1.6=0 -1.6~0.8=1
依次類推

• 創建函數 根據獲得的闕值對連續變量進行離散化

def digitize_state(observation):
    cart_pos,cart_v,pole_angle,pole_v=observation
    digitized=[
        np.digitize(cart_pos,bins=bins(-2.4,2.4,NUM_DIZITIZED)),
        np.digitize(cart_v,bins=bins(-3.0,3.0,NUM_DIZITIZED)),
        np.digitize(pole_angle,bins=bins(-0.5,0.5,NUM_DIZITIZED)),
        np.digitize(pole_v,bins=bins(-2.0,2.0,NUM_DIZITIZED))
    ]
    return sum([x*(NUM_DIZITIZED)**i) for i,x in enumerate(digitized)])

以6進制進行計算 如果 存在一個離散值(1,2,3,4) 則求得當前狀態值為 16^0+2*61+36^2+4*63=985

• Q學習實現

這里需要定義實現類，主要有三個類 Agent Brain 和 Environmet

Agent類表示小推車對象，主要有2個函數，更新Q函數，和確定下一步動作函數
Agent中有一個Brain類的對象作為成員變量。

Brain類可認為是Agent的大腦，通過Q表來實現Q學習，主要有4個函數 bin digitize_state 用來離散化Agent觀察到的observation
函數update_Q_table來更新Q表
函數decision_action 來確定來自Q表的動作。
為什么需要將Agent和Brain類分開》？ 因為如果使用深度強化學習，將表格型Q改成深度強化學習時只需要改變Brain類就行了。

Environment類是OpenAI Gym的執行環境，執行CartPole環境的是run函數

• start

首先我們需要決定要執行的值動作，所以 Agent將當前狀態 observation_t傳給Brain ，Brain 離散化狀態再根據Q表來確定動作，並將確定的動作返回給Agent，
之后是動作的實際執行環境步驟，Agent將動作action_t傳遞給Environment，Environment執行動作action_t並將執行后的狀態observation_t+1和即時獎勵 reward+1 返回給Agent
再更新Q 表， Agent將當前狀態observation_t 執行動作 action_t 和執行動作后的observation_t+1 即時獎勵reward_t+1傳回給Brain，Brain更新Q表，這4個變量綜合起來被稱為transition
之后 重復該過程就行了，因為獲得最大價值的方式只有一種，所以通過Q學習不斷擬合，最后會形成唯一解。

完整代碼

#!/usr/bin/env python
# coding: utf-8

# In[ ]:


# 聲明包
import numpy as np
import matplotlib.pyplot as plt
get_ipython().run_line_magic('matplotlib', 'inline')
import gym


# In[17]:


# 聲明繪圖功能
from JSAnimation.IPython_display import display_animation
from matplotlib import animation
from IPython.display import display
def display_frames_as_gif(frames):
    plt.figure(figsize=(frames[0].shape[1]/72.0,frames[0].shape[0]/72.0),dpi=72)
    patch=plt.imshow(frames[0])
    plt.axis("off")
    
    def animate(i):
        patch.set_data(frames[i])
    anim=animation.FuncAnimation(plt.gcf(),animate,frames=len(frames),interval=50)
    anim.save('move_carpole.mp4') # 保存動畫
    display(display_animation(anim,default_mode='loop'))


# In[18]:


# 變量設定
ENV='CartPole-v0' # 設置任務名
NUM_DIZITIZED=6 # 設置離散值個數
GAMMA=0.99 # 時間折扣率
ETA=0.5 # 學習率
MAX_STEPS=200 # 一次實驗中的步數
NUM_EPISODES=1000 # 最大實驗次數

# 嘗試運行 CartPole
env=gym.make(ENV) # 設置要執行的任務
observarion=env.reset() # 環境初始化


# In[19]:


# 定義Agent類
class Agent:
    def __init__(self,num_states,num_actions):
        self.brain=Brain(num_states,num_actions) # 創建了Brain類的對象，構造方法中有當前狀態和動作
        # 為智能體創建大腦以做出決策
    def update_Q_function(self,observation,action,reward,observation_next):
        self.brain.update_Q_table(observation,action,reward,observation_next)
        # 更新Brain類中的Q函數
    def get_action(self,observation,step):
        action=self.brain.decide_action(observation,step)
        return action
        # 動作的確定
    


# In[20]:


# 定義Brain類
class Brain:
    def __init__(self,num_states,num_actions):
        self.num_actions=num_actions # CartPole的動作 向左或者向右
        self.q_table=np.random.uniform(low=0,high=1,size=(NUM_DIZITIZED**num_states,num_actions))
        # 初始化Q表，行為將狀態轉換成數字得到的分割數，列為動作數
    def bins(self,clip_min,clip_max,num):
        return np.linspace(clip_min,clip_max,num+1)[1:-1] # 返回[-1.6,-0.8,0,0,0.8,1.6]
        # 求取用於離散化的闕值
    def digitize_state(self,observation):
        cart_pos,cart_v,pole_angle,pole_v=observation
        digitized=[
            np.digitize(cart_pos,bins=self.bins(-2.4,2.4,NUM_DIZITIZED)),
            np.digitize(cart_v,bins=self.bins(-3.0,3.0,NUM_DIZITIZED)),
            np.digitize(pole_angle,bins=self.bins(-0.5,0.5,NUM_DIZITIZED)),
            np.digitize(pole_v,bins=self.bins(-2.0,2.0,NUM_DIZITIZED))
        ]
        return sum([x*(NUM_DIZITIZED**i) for i,x in enumerate(digitized)])
        # 將連續值轉成離散值
    def update_Q_table(self,observation,action,reward,observation_next):
        # 更新Q表
        state=self.digitize_state(observation) # 狀態離散化
        state_next=self.digitize_state(observation_next) # 下一個狀態離散化
        Max_Q_next=max(self.q_table[state_next][:]) # 求Q表中下一個狀態的最大值 即向右或向左方向的狀態值
        self.q_table[state,action]=self.q_table[state,action]+ETA*(reward+GAMMA*Max_Q_next-self.q_table[state,action])
        # 若要保證Q表幾乎不變，則下一跳中狀態最大值*時間折扣率+0的獎勵需要幾乎等於當前狀態。
        # 也就是說在不斷擬合一個定值，最終收斂於該值。
        # 為什么會擬合該值？因為該范圍值是這個物體運動的大范圍值的子集，也就是說會一直重復這個范圍值，如果次數夠多，並且每次都能通過算法將在這個小范圍值之外的區間縮小，
        # 那么可通過這種方法進行擬合，訓練量足夠多，則擬合的范圍值越接近我們想要的范圍值。
        print(self.q_table[state,action])
    def decide_action(self,observation,episode):
        state=self.digitize_state(observation)
        epsilon=0.5*(1/(episode+1))
        if epsilon<=np.random.uniform(0,1):
            action=np.argmax(self.q_table[state][:]) # 獲取對應狀態最大值索引。
        else:
            action=np.random.choice(self.num_actions) # 隨機返回0,1動作
        return action
    
        


# In[21]:


# 執行環境類
class Environment:
    def __init__(self):
        self.env=gym.make(ENV) # 設置要執行的任務
        num_states=self.env.observation_space.shape[0] # 獲取任務狀態個數
        num_actions=self.env.action_space.n # 獲取CartPole的動作數 為 2
        self.agent=Agent(num_states,num_actions) # 創建在環境中行動的Agent
    def run(self):
        complete_episodes=0 # 持續195步或者更多
        is_episode_final=False # 最終實驗標志
        frames=[] # 用來存儲視頻圖象的變量
        for episode in range(NUM_EPISODES):
            observation=self.env.reset() # 環境初始化
            for step in range(MAX_STEPS): # 每個回合的循環
                if is_episode_final is True: # 將最終實驗各個時刻圖像添加到幀
                    frames.append(self.env.render(mode='rgb_array'))
                # 求動作
                action=self.agent.get_action(observation,episode)
                # 根據執行動作找到 s_T ,r_t
                observation_next,_,done,_=self.env.step(action) # 不用regain  info
                # 給予獎勵
                if done:
                    if step<195:
                        reward=-1 # 半途摔倒給予-1作為懲罰
                        complete_episodes=0 # 站立超過195 重置
                    else:
                        reward=1 # 完成給予獎勵
                        complete_episodes+=1 # 更新
                else:
                    reward=0 # 途中獎勵為0
                # 使用step_1 的狀態 observation_next更新Q函數
                self.agent.update_Q_function(observation,action,reward,observation_next)
                # observation 更新
                observation=observation_next
                # 結束時處理
                if done:
                    print('{0} Episode: Finished after {1} time steps:'.format(episode,step+1))
                    break
            if is_episode_final is True: # 最后一次實驗保存
                display_frames_as_gif(frames)
                break
            if complete_episodes>=10:
                print('10回合連續成功')
                is_episode_final=True
            


# In[ ]:


# 主函數調用
if __name__=="__main__":
    cartpole_env=Environment()
    cartpole_env.run()


# In[ ]:





# In[ ]:

運行結果

0x04 強化學習之Q學習的原理(重點)

  觀察代碼 self.q_table[state,action]=self.q_table[state,action]+ETA(reward+GAMMAMax_Q_next-self.q_table[state,action]) ，剛開始看時，一臉懵逼，直到想通了某個點。
首先我們需要清楚在大量的數據面前，能夠滿足我們想要的最好的策略 只有一條，有些時候我們可以自己求得該策略，比如迷宮問題，我們可以輕松做到，這里的倒立擺問題我們也能輕松做到，但是小球消方塊呢？ 我們人類幾乎不能在很短時間做出判斷，然后消除掉所有的方塊，但是機器能。為什么？

滿足最優策略只有一條，大數據訓練只是為了讓我們的策略最終擬合成為最優策略

還是拿這段代碼來說 self.q_table[state,action]=self.q_table[state,action]+ETA(reward+GAMMAMax_Q_next-self.q_table[state,action]) Q表的更新是當前Q表+變化值。
所以Q表的更新量其實就是 ETA*(reward+GAMMA*Max_Q_next-self.q_table[state,action])
ETA是學習率。 reward是獎勵，這里可以認為是0， GAMMA是時間折扣率為0.99 接近為1 self.q_table[state,action] 為當前Q表，記錄了當前狀態和當前的方向。
當我們設置變化值為很小時，最終實現Q表幾乎不變，此時Q表代表了我們的最優策略。 但是為什么，為什么它就能擬合到最優，而不是別的？每次擬合的過程是什么？

概率問題。

用epsilon貪婪法先進行隨機運動，目的是防止智能體一開始就一直朝一個方向進行擬合，比如全部向右，這樣小車就飛出去了。之后通過逐步減小epsilon值讓小車朝獲得最大價值的方向運動，如果成功朝向了最大價值方向運動，則向該方向運動的概率變大，宏觀上表現為變化量減小。也就是到達目標的步數減少，而步數的減少使得數據范圍縮小，而我們目標數據范圍不變，所以相當於我們目標占比變大了，而目標占比變大，使得智能體朝該方向運動概率也變大，周而復始，最終達到幾乎完全擬合，變化量忽略不計。