前言
总感觉强化学习公式真难学，也难表达心中所想，我还是白话强化学习吧。
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。
总结
-
为了避免正数陷阱，我们希望Actor的更新权重有正有负。因此，我们把Q值减去他们的均值V。有：Q(s,a)-V(s)
-
为了避免需要预估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
实现
文章来源: blog.csdn.net，作者：快了的程序猿小可哥，版权归原作者所有，如需转载，请联系作者。
原文链接：blog.csdn.net/qq_35914625/article/details/109908708