浅谈PPO算法-玩转月球登陆_ppo critic 通俗理解

前言

总感觉强化学习公式真难学，也难表达心中所想，我还是白话强化学习吧。

github

https://github.com/yanjingke/PPO-PyTorch

什么是Actor-Critic？

Actor-Critic，其实是用了两个网络：两个网络有一个共同点，输入状态S: 一个输出策略，负责选择动作，我们把这个网络成为Actor； 一个负责计算每个动作的分数，我们把这个网络成为Critic。
大家可以形象地想象为，Actor是舞台上的舞者，Critic是台下的评委。Actor在台上跳舞，一开始舞姿并不好看，Critic根据Actor的舞姿打分。Actor通过Critic给出的分数，去学习：如果Critic给的分数高，那么Actor会调整这个动作的输出概率；相反，如果Critic给的分数低，那么就减少这个动作输出的概率。
在AC中中的Critic，估算的是V值也就是一个评分，因为在强化学习中，往往没有足够的时间让我们去和环境互动。在Critic中如果预测动作评价，假设我们用Critic网络，预估到S状态下三个动作A1，A2，A3的Q值分别为1,2,10。但在开始的时候，我们采用平均策略，于是随机到A1。于是我们用策略梯度的带权重方法更新策略，这里的权重就是Q值。于是策略会更倾向于选择A1，意味着更大概率选择A1。结果A1的概率就持续升高…
这就掉进了正数陷阱。我们明明希望A3能够获得更多的机会，最后却是A1获得最多的机会。这是为什么呢？
这是因为Q值用于是一个正数，如果权重是一个正数，那么我们相当于提高对应动作的选择的概率。权重越大，我们调整的幅度将会越大。其实当我们有足够的迭代次数，这个是不用担心这个问题的。因为总会有机会抽中到权重更大的动作，因为权重比较大，抽中一次就能提高很高的概率。

但在强化学习中，往往没有足够的时间让我们去和环境互动。这就会出现由于运气不好，使得一个很好的动作没有被采样到的情况发生。要解决这个问题，我们可以通过减去一个baseline，令到权重有正有负。而通常这个baseline，我们选取的是权重的平均值。减去平均值之后，值就变成有正有负了。

所以我们可以得到更新的权重：Q(s,a)-V(s)。然而在每次的奖励中会进行得分的衰减gamma。
总结

1. 为了避免正数陷阱，我们希望Actor的更新权重有正有负。因此，我们把Q值减去他们的均值V。有：Q(s,a)-V(s)

2. 为了避免需要预估V值和Q值，我们希望把Q和V统一；由于Q(s,a) = gamma * V(s’) + r - V(s)。所以我们得到TD-error公式： TD-error = gamma * V(s’) + r - V(s)

Actor-Critic代码

class ActorCritic(nn.Module):
    def __init__(self, state_dim, action_dim, n_latent_var):
        super(ActorCritic, self).__init__()

        # actor
        self.action_layer = nn.Sequential(
                nn.Linear(state_dim, n_latent_var),
                nn.Tanh(),
                nn.Linear(n_latent_var, n_latent_var),
                nn.Tanh(),
                nn.Linear(n_latent_var, action_dim),
                nn.Softmax(dim=-1)
                )
        
        # critic
        self.value_layer = nn.Sequential(
                nn.Linear(state_dim, n_latent_var),
                nn.Tanh(),
                nn.Linear(n_latent_var, n_latent_var),
                nn.Tanh(),
                nn.Linear(n_latent_var, 1)
                )
        
    def forward(self):
        raise NotImplementedError
        
    def act(self, state, memory):
        state = torch.from_numpy(state).float().to(device) 
        action_probs = self.action_layer(state)
        dist = Categorical(action_probs)#按照给定的概率分布来进行采样
        action = dist.sample()
        
        memory.states.append(state)
        memory.actions.append(action)
        memory.logprobs.append(dist.log_prob(action))
        
        return action.item()
    
    def evaluate(self, state, action):
        action_probs = self.action_layer(state)
        # Categorical代表随机策略
        dist = Categorical(action_probs)
        
        action_logprobs = dist.log_prob(action)
        dist_entropy = dist.entropy()
        #cricle对state评价
        state_value = self.value_layer(state)
        
        return action_logprobs, torch.squeeze(state_value), dist_entropy
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

ppo算法

在On Policy，在线策略AC产生的数据，只能进行1次更新，更新完就只能丢掉，等待下一次跑游戏的数据。在数据就是生命的时代，这可是天大的浪费呀，尤其在强化学习中，数据更是弥足珍贵呀。而·Off Policy，离线策略，可以通过产生的数据多次更新，这样说有点难以理解，我们举个例子：如果我们在智能体和环境进行互动时产生的数据打上一个标记。标记这是第几版本的策略产生的数据,例如 1， 2… 10，现在我们的智能体用的策略 10，需要更新到 11。如果算法只能用 10版本的产生的数据来更新，那么这个就是在线策略；如果算法允许用其他版本的数据来更新，那么就是离线策略。
我们来看看这个例子。

假设，我们已知在同一个环境下，有两个动作可以选择。现在两个策略，分别是P和B：P: [0.5,0.5] B: [0.1,0.9]
现在我们按照两个策略，进行采样；也就是分别按照这两个策略，以S状态下出发，与环境进行10次互动。获得上图数据。那么，我们可以用B策略下获得的数据，更新P吗？
如果用行动策略B[0.1,0.9]产出的数据，对目标策略P进行更新，动作1会被更新1次，而动作2会更新9次。虽然动作A的TD-error比较大，但由于动作2更新的次数更多，最终动作2的概率会比动作1的要大。这自然不是我们期望看到的更新结果，因为动作1的TD-error比动作2要大，我们希望选择概率动作1的能更多呀。从这个例子，大家可以大致明白，为什么我们在更新策略的时候，不能用其他策略产生的数据了。
那么，PPO是怎样做到离线更新策略的呢？答案是Important-sampling，重要性采样技术。
如果我们想用策略B抽样出来的数据，来更新策略P也不是不可以。但我们要把td-error乘以一个重要性权重（IW：importance weight）。
重要性权重：IW = P（a）/ B（a）
应用在PPO，就是目标策略出现动作a的概率 除以 行为策略出现a的概率。回到我们之前的例子，我们可以计算出，每个动作的 重要性权重，P: [0.5,0.5] B: [0.1,0.9]

我们以a1为例，计算重要性权重IW = P / B = ( 1 / 0.5 ) / ( 1 / 0.1 ) = 5

对应论文：

在PPO论文中提出了截断代理

其中，Et代表随机策略，At代表优势函数在时间步长为 t 时的估计
如图，实际上我们只需要计算最后的V(s’)，根据这个估算的V(s’), 我们反推经过的所有state的V值，用网络估算。用网络估计的方式现在其实相当常见，我们最著名的围棋AI Alpha Zero，也有用到类似的技术。

代码：

class PPO:
    def __init__(self, state_dim, action_dim, n_latent_var, lr, betas, gamma, K_epochs, eps_clip):
        self.lr = lr
        self.betas = betas
        self.gamma = gamma
        self.eps_clip = eps_clip
        self.K_epochs = K_epochs
        
        self.policy = ActorCritic(state_dim, action_dim, n_latent_var).to(device)
        print(self.policy.parameters())
        self.optimizer = torch.optim.Adam(self.policy.parameters(), lr=lr, betas=betas)
        self.policy_old = ActorCritic(state_dim, action_dim, n_latent_var).to(device)
        self.policy_old.load_state_dict(self.policy.state_dict())
        
        self.MseLoss = nn.MSELoss()
    
    def update(self, memory):   
        # Monte Carlo estimate of state rewards:
        rewards = []
        discounted_reward = 0
        for reward, is_terminal in zip(reversed(memory.rewards), reversed(memory.is_terminals)):
            if is_terminal:
                discounted_reward = 0
            #每一步得分衰减
            discounted_reward = reward + (self.gamma * discounted_reward)
            #插入每一步得分
            rewards.insert(0, discounted_reward)
        
        # Normalizing the rewards:
        rewards = torch.tensor(rewards, dtype=torch.float32).to(device)
        rewards = (rewards - rewards.mean()) / (rewards.std() + 1e-5)
        
        # convert list to tensor
        old_states = torch.stack(memory.states).to(device).detach()
        old_actions = torch.stack(memory.actions).to(device).detach()
        old_logprobs = torch.stack(memory.logprobs).to(device).detach()
        
        # Optimize policy for K epochs:
        for _ in range(self.K_epochs):
            # Evaluating old actions and values :
            logprobs, state_values, dist_entropy = self.policy.evaluate(old_states, old_actions)
            
            # Finding the ratio (pi_theta / pi_theta__old):
            ratios = torch.exp(logprobs - old_logprobs.detach())
                
            # Finding Surrogate Loss:
            advantages = rewards - state_values.detach()
            surr1 = ratios * advantages
            surr2 = torch.clamp(ratios, 1-self.eps_clip, 1+self.eps_clip) * advantages
            loss = -torch.min(surr1, surr2) + 0.5*self.MseLoss(state_values, rewards) - 0.01*dist_entropy
            
            # take gradient step
            self.optimizer.zero_grad()
            loss.mean().backward()
            self.optimizer.step()
        
        # Copy new weights into old policy:
        self.policy_old.load_state_dict(self.policy.state_dict())
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

实现