强化学习:A2C求解MountainCar-v0小车上山问题

1.问题背景

小车上山问题的问题背景就不再赘述了，在实现过程中用到了python的gym库。导入该环境的过程代码如下:

import gym
# 环境类型
env = gym.make("MountainCar-v0")
env = env.unwrapped
print("初始状态{}".format(np.array(env.reset())))
1
2
3
4
5

而提前需要导入的库如下:

import sys
import numpy as np
import torch
# 导入torch的各种模块
import torch.nn as nn
from torch.nn import functional as F
from torch.distributions import Categorical
1
2
3
4
5
6
7

2.A2C原理

A2C的原理不过多赘述，只需要了解其策略网络$\pi (a|s;\theta )$的梯度为:
$$\begin{gathered} {\nabla }_{\theta }J(\theta )=E_{s_t,a_t\sim \pi (.|s_t;\theta )}[A(s_t,a_t;\omega ){\nabla }_{\theta }\ln \pi (a_t|s_t;\theta )] \\ \theta \leftarrow \theta +\alpha {\nabla }_{\theta }J(\theta ) \end{gathered}$$其中：$A(s_t,a_t)=Q(s_t,a_t)-v(s_t;\omega )\approx G_t-v(s_t;\omega )$为优势函数。
而对于每一个轨迹$\tau :s_0{a}_0{r}_0{s}_1,...s_{T-1}{a}_{T-1}{r}_{T-1}{s}_T$而言：
$${\nabla }_{\theta }J(\theta )=E_{\tau }[{\nabla }_{\theta }\sum _{i=0}^{T-1}\ln \pi (a_t|s_t;\theta )(R(\tau )-v(s_t;\omega ))]$$其中:$R(\tau )={\sum }_{i=0}^{\infty }{\gamma }^i{r}_t$。
对于$v(s;\omega )$其梯度采用MC算法计算:
$$\begin{gathered} {\nabla }_{\omega }J(\omega )=(G_t-v(s_t;\omega )){\nabla }_{\omega }v(s_t;\omega ) \\ \omega \leftarrow \omega -\beta {\nabla }_{\omega }J(\omega ) \end{gathered}$$其中:$G_t={\sum }_{i=t}^T{\gamma }^i{r}_i$。
将初始化函数__init__;网络的前馈函数forward;根据$\pi (a|s;\theta ),v(s;\omega )$采样动作及输出$\ln \pi (a_t|s_t;\theta ),G_t$的函数为:select_action;更新网络参数的函数为:update_policy;网络的训练函数train_network封装到policyNet类中。其代码如下:

class policyNet(nn.Module):
    def __init__(self,state_num,action_num,hidden_num=50,lr=0.01):
        super(policyNet,self).__init__()
        self.fc1 = nn.Linear(state_num,hidden_num)
        self.action_fc = nn.Linear(hidden_num,action_num)
        self.state_fc = nn.Linear(hidden_num,1)
        self.optimzer = torch.optim.Adam(self.parameters(),lr=lr)
        # 动作与回报的缓存
        self.saved_states = []
        self.saved_log_pi = []
        self.saved_values = []
        self.saved_actions = []
        self.saved_rewards = []
    # 前馈
    def forward(self,state):
        hidden_output = self.fc1(state)
        x = F.relu(hidden_output)
        action_prob = self.action_fc(x)
        action_prob = F.softmax(action_prob,dim=-1)
        state_values = self.state_fc(x)
        return action_prob,state_values
    # 选择动作
    def select_action(self,state):
        # 输入state为numpy数据类型，要先进行数据转化
        # 输出lnpi(a|s),v(s),r
        state = torch.from_numpy(state).float()
        probs,state_value = self.forward(state)
        m = Categorical(probs)
        action = m.sample()
        # 存储数据
        self.saved_states.append(state)
        self.saved_log_pi.append(m.log_prob(action))
        self.saved_values.append(state_value)
        self.saved_actions.append(action)
        # 返回action
        return action.item()
    # 实现A2C策略梯度更新
    def update_policy(self,gamma=1.0):
        # 先计算Gt
        R = 0
        G = np.zeros_like(self.saved_rewards)
        for t in reversed(range(len(self.saved_rewards))):
            R = self.saved_rewards[t] + gamma*R
            G[t] = R
        # 对G标准化
        G = (G - np.mean(G))/np.std(G)
        G = torch.tensor(G)
        # 下面开始求policy_loss,value_loss
        policy_loss = []
        value_loss = []
        for t in range(len(self.saved_rewards)):
            log_prob = self.saved_log_pi[t]
            value = self.saved_values[t]
            Gt = G[t]
            advantage = Gt - value
            # 计算actor的loss
            policy_loss.append(-advantage*log_prob)
            value_loss.append(F.smooth_l1_loss(value,Gt))
        # 下面开始优化
        self.optimzer.zero_grad()
        loss = torch.stack(policy_loss).sum() + torch.stack(value_loss).sum()
        loss.backward()
        self.optimzer.step()
        # 更新完后重置缓存
        del self.saved_actions[:]
        del self.saved_log_pi[:]
        del self.saved_rewards[:]
        del self.saved_states[:]
        del self.saved_values[:]
    # 实现训练
    def train_network(self,env,epsiodes=1000,gamma=1.0):
        # 设置数组用以存储奖励
        epsiode_rewards_array = []
        rewards_mean_array = []
        # env是环境
        for epsiode in range(epsiodes):
            # 初始化环境状态
            state = np.array(env.reset())
            epsiode_reward = 0
            while True:
                action = self.select_action(state)
                next_state,reward,done,_ = env.step(action)
                self.saved_rewards.append(reward)
                epsiode_reward += reward
                if done:
                    #print("第{}次迭代的奖励值为{}".format(epsiode,epsiode_reward))
                    break
                state = next_state
            self.update_policy(gamma=gamma)
            epsiode_rewards_array.append(epsiode_reward)
            rewards_mean_array.append(np.mean(epsiode_rewards_array))
            print("第{}次迭代的奖励值为{},累计奖励{}".format(epsiode,epsiode_reward,int(rewards_mean_array[-1])))
        return epsiode_rewards_array,rewards_mean_array
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93

3.训练网络

主函数调用如下:

policyNetWork = policyNet(2,env.action_space.n)
rewards_array,rewards_mean_array = policyNetWork.train_network(env,epsiodes=500)
# 下面是画图函数
import matplotlib.pyplot as plt
plt.plot(rewards_array)
plt.plot(rewards_mean_array)
plt.show()
1
2
3
4
5
6
7

由于算力有限，仅仅给出收敛次数epsiodes=50时的结果: