强化学习：策略梯度定理与 REINFORCE 算法

策略梯度定理

揭示了目标函数的梯度等价于以下公式：

证明过程

• 首先将梯度提进去，然后提出一个 P(τ;θ)，拼凑一个 log f(x) 求导的公式

• 根据期望定义将上面的公式重新还原为期望

我们带入 P(τ;θ) 的定义，然后老老实实求梯度，发现除了 π(at|st) 之外，所有项不包含 θ，因此直接消去。

对于这个期望，我们可以利用大数定理对其采样求解其均值。

蒙特卡洛 MC Reinforce 算法

我们得到了目标函数的梯度，然后运行梯度上升来最优化我们的策略函数。

一般来说收集多个轨迹来计算平均梯度。

策略梯度方法流程：

• 经历一次完整的动作序列后才能开始更新。

对于好的动作序列，增加其动作选择的概率，对于不好的动作，降低其动作选择的概率。

实现

我们需要的是： ∑t ∇θ log πθ(at(i)|st(i)) R(τ(i)) 然后用其进行梯度上升。 这转换为利用 torch 对以下函数进行优化 (会自动求梯度并且执行梯度下降)

−∑t log πθ(at(i)|st(i)) R(τ(i))

Reinforce 算法的改进：使用 Gt 替代准确的 R(τ)

• 公平性：R(τ) 表示某轨迹的累积折扣奖励，对于所有的状态的动作都给相同的 R(τ) 是不公平的，因为前面的动作更加重要，因此采用 Gt 为每个动作概率根据执行顺序赋予不同的权重进行优化。
• 方差改进：R(τ) 偏差为 0，所以方差必然很大。Gt 作为当前动作的一种估计，以提高偏差为代价降低方差，进而增强稳定性。

我们最终需要优化的方程为： ∑t log πθ(at(i)|st(i)) Gt

参数化策略代码

class Policy(nn.Module):
    def __init__(self, s_size, a_size, h_size):
        super(Policy, self).__init__()
        self.fc1 = nn.Linear(s_size, h_size)
        self.fc2 = nn.Linear(h_size, a_size)

    def forward(self, x):
        x = F.relu(self.fc1(x))
        x = self.fc2(x)
        return F.softmax(x, dim=1)

    def act(self, state):
        state = torch.from_numpy(state).float().unsqueeze(0).to(device)
        probs = self.forward(state).cpu()
        m = Categorical(probs)  # torch.distribution 的对象
        action = m.sample()  # idx, log_prob[idx]
        return action.item(), m.log_prob(action)

Reinforce 训练代码

def reinforce(policy, optimizer, n_training_episodes, max_t, gamma, print_every):
    # Help us to calculate the score during the training
    scores_deque = deque(maxlen=100)
    scores = []
    
    # Line 3 of pseudocode
    for i_episode in range(1, n_training_episodes + 1):
        saved_log_probs = []
        rewards = []
        state = env.reset()
        
        # Line 4 of pseudocode
        for t in range(max_t):
            action, log_prob = policy.act(state)
            saved_log_probs.append(log_prob)
            state, reward, done, _ = env.step(action)
            rewards.append(reward)
            if done:
                break
        
        scores_deque.append(sum(rewards))
        scores.append(sum(rewards))
        
        # Line 6 of pseudocode: calculate the return
        returns = deque(maxlen=max_t)
        n_steps = len(rewards)
        
        # Compute the discounted returns at each timestep,
        # as the sum of the gamma-discounted return at time t (G_t) + the reward at time t
        # In O(N) time, where N is the number of time steps
        # (this definition of the discounted return G_t follows the definition of this quantity 
        # shown at page 44 of Sutton&Barto 2017 2nd draft)
        # G_t = r_(t+1) + r_(t+2) + ...
        # Given this formulation, the returns at each timestep t can be computed 
        # by re-using the computed future returns G_(t+1) to compute the current return G_t
        # G_t = r_(t+1) + gamma*G_(t+1)
        # G_(t-1) = r_t + gamma* G_t
        # (this follows a dynamic programming approach, with which we memorize solutions in order 
        # to avoid computing them multiple times)
        # This is correct since the above is equivalent to (see also page 46 of Sutton&Barto 2017 2nd draft)
        # G_(t-1) = r_t + gamma*r_(t+1) + gamma*gamma*r_(t+2) + ...
        
        # Given the above, we calculate the returns at timestep t as: 
        # gamma[t] * return[t] + reward[t]
        
        # We compute this starting from the last timestep to the first, in order
        # to employ the formula presented above and avoid redundant computations that would be needed 
        # if we were to do it from first to last.
        
        # Hence, the queue "returns" will hold the returns in chronological order, from t=0 to t=n_steps
        # thanks to the appendleft() function which allows to append to the position 0 in constant time O(1)
        # a normal python list would instead require O(N) to do this.
        for t in range(n_steps)[::-1]:  # inverse order
            disc_return_t = (returns[0] if len(returns) > 0 else 0)
            returns.appendleft(gamma * disc_return_t + rewards[t])
        
        # standardization of the returns is employed to make training more stable
        eps = np.finfo(np.float32).eps.item()
        # eps is the smallest representable float, which is 
        # added to the standard deviation of the returns to avoid numerical instabilities
        returns = torch.tensor(returns)
        returns = (returns - returns.mean()) / (returns.std() + eps)
        
        # Line 7: policy_loss = []
        policy_loss = []
        for log_prob, disc_return in zip(saved_log_probs, returns):
            policy_loss.append(-log_prob * disc_return)  # G(tau)
        
        policy_loss = torch.cat(policy_loss).sum()
        
        # Line 8: PyTorch prefers gradient descent
        optimizer.zero_grad()
        policy_loss.backward()
        optimizer.step()
        
        if i_episode % print_every == 0:
            print('Episode {}\tAverage Score: {:.2f}'.format(i_episode, np.mean(scores_deque)))
    return scores