Zhangzhe's Blog

The projection of my life.

0%

强化学习(6)——Actor-Critic系列算法

Posted on Edited on In Valine:

URL

TL;DR

  • Actor-Critic 算法是一种结合了策略梯度和值函数的方法，Actor 负责输出策略，Critic 负责输出状态价值函数。和 REINFORCE 算法相比，Actor-Critic 算法多引入了一个 Critic 网络，用于评估 Actor 的策略。
  • A2C 是对 Actor-Critic 算法的改进，引入了 Advantage 项来评估当前状态的优势，优化了 Actor 的损失计算。
  • A3C 是在 A2C 的基础上引入了多个并行的 Actor-Critic 网络，用于加速训练。
  • PPO 也是 Actor-Critic 架构下的一种算法，引入了 Clipped Surrogate Objective 和 Generalized Advantage Estimation (GAE)，用于提高策略梯度的稳定性和效率。

Algorithms

1. Actor-Critic Algorithms

公式表述

  • 这是一篇 1999 年的强化学习论文，是对于 1992 年 REINFORCE 算法论文 Simple Statistical Gradient-Following Algorithms for Connectionist Reinforcement Learning 的扩展。
  • REINFORCE 算法是一种纯粹的策略梯度算法，也就是只有 Actor 网络，没有 Critic 网络。
  • Actor-Critic 算法在 REINFORCE 算法的基础上引入了一个 Critic 网络，用于评估当前状态的价值（或当前状态-空间的价值）。
  • Actor 的输入是 state，输出是 action 的分布均值和标准差，在均值和标准差组成的正态分布中采样得到 action。
  • Critic 的输入是 state 或 state-action，输出是当前状态的价值，即 V(s) 或 Q(s, a)。
  • Actor 的损失:

Loss=−∑t=1Tlog⁡πθ(at∣st)⋅G(st,at)Loss = -\sum_{t=1}^T \log \pi_{\theta}(a_t|s_t) \cdot G(s_t,a_t)

  • 其中，G(st,at)G(s_t,a_t) 是当前状态的实际累积回报
  • Critic 的损失:

Loss=∑t=1TMSE(G(st,at)−Vθπ(st))Loss=\sum_{t=1}^T MSE(G(s_t,a_t)- V_{\theta}^{\pi}(s_t))

代码表述

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
94
95
96
class ActorCriticNetwork(nn.Module):
    """Actor-Critic 网络，包含策略网络和价值网络。"""
    def __init__(self, obs_space_dims: int, action_space_dims: int):
        """初始化 Actor-Critic 网络。
        参数：
            obs_space_dims: 观测空间维度
            action_space_dims: 动作空间维度
        """
        super().__init__()
        hidden_space1 = 16
        hidden_space2 = 32
        # 共享网络
        self.shared_net = nn.Sequential(
            nn.Linear(obs_space_dims, hidden_space1),
            nn.Tanh(),
            nn.Linear(hidden_space1, hidden_space2),
            nn.Tanh(),
        )
        # 策略网络输出动作的均值和标准差
        self.policy_mean_net = nn.Linear(hidden_space2, action_space_dims)
        self.policy_stddev_net = nn.Linear(hidden_space2, action_space_dims)
        # 价值网络输出状态价值估计
        self.value_net = nn.Linear(hidden_space2, 1)
    def forward(self, x: torch.Tensor):
        """前向传播，输出动作参数和状态价值。
        参数：
            x: 环境的观测值
        返回：
            action_means: 动作均值
            action_stddevs: 动作标准差
            state_values: 状态价值估计
        """
        shared_features = self.shared_net(x.float())
        action_means = self.policy_mean_net(shared_features)
        action_stddevs = torch.log(
            1 + torch.exp(self.policy_stddev_net(shared_features))
        )
        state_values = self.value_net(shared_features)
        return action_means, action_stddevs, state_values
class ActorCriticAgent:
    """Actor-Critic 算法的实现。"""
    def __init__(self, obs_space_dims: int, action_space_dims: int):
        """初始化 Agent。
        参数：
            obs_space_dims: 观测空间维度
            action_space_dims: 动作空间维度
        """
        self.learning_rate = 1e-4
        self.gamma = 0.99
        self.eps = 1e-6
        self.log_probs = []
        self.state_values = []
        self.rewards = []
        self.net = ActorCriticNetwork(obs_space_dims, action_space_dims)
        self.optimizer = torch.optim.AdamW(self.net.parameters(), lr=self.learning_rate)
    def sample_action(self, state: np.ndarray):
        """根据策略采样动作。
        参数：
            state: 环境的观测值
        返回：
            action: 采样的动作
        """
        state = torch.tensor(np.array([state]))
        action_means, action_stddevs, state_value = self.net(state)
        distrib = Normal(action_means[0] + self.eps, action_stddevs[0] + self.eps)
        action = distrib.sample()
        log_prob = distrib.log_prob(action)
        action = action.numpy()
        self.log_probs.append(log_prob)
        self.state_values.append(state_value)
        return action
    def update(self):
        """更新策略网络和价值网络。"""
        # 将列表转换为张量
        log_probs = torch.stack(self.log_probs)
        state_values = torch.cat(self.state_values).squeeze()
        rewards = self.rewards
        # 计算折扣回报
        returns = []
        G = 0
        for R in rewards[::-1]:
            G = R + self.gamma * G
            returns.insert(0, G)
        returns = torch.tensor(returns)
        # 计算损失
        policy_loss = -(log_probs * returns.detach()).sum()
        value_loss = nn.functional.mse_loss(state_values, returns)
        loss = policy_loss + value_loss
        # 更新网络参数
        self.optimizer.zero_grad()
        loss.backward()
        self.optimizer.step()
        # 清空存储的值
        self.log_probs = []
        self.state_values = []
        self.rewards = []
  • 实现上和 REINFORCE 算法的代码相比，有以下几点不同：
    • 引入了 Critic 网络，用于评估当前状态的价值（可以理解为训练一个可微的环境仿真器）。
    • 损失函数中包含了 Critic 的损失，用于更新 Critic 网络。

2. Advantage Actor-Critic (A2C)

公式表述

  • A2C 的全称是 Advantage Actor-Critic，是对于 Actor-Critic 算法的改进，引入了 Advantage 项，用于评估当前状态的优势。
  • 具体来说：在计算 Actor 的损失时，使用的是 Advantage 项，而不是直接使用回报，即：

Loss=−∑t=1Tlog⁡πθ(at∣st)⋅A(st,at)A(st,at)=G(st,at)−V(st)Loss = -\sum_{t=1}^T \log \pi_{\theta}(a_t|s_t) \cdot A(s_t,a_t)\\A(s_t,a_t)=G(s_t,a_t) - V(s_t)

  • 其中，A(st,at)A(s_t,a_t) 是当前状态的优势，G(st,at)G(s_t,a_t) 是当前状态的实际回报，V(st)V(s_t) 是当前状态的价值估计。
  • 其他部分和 Actor-Critic 算法基本一致。

代码表述

1
2
3
4
5
# Actor-Critic 算法的实现
actor_loss = -(log_probs * returns.detach()).sum()
# Advantage Actor-Critic 算法的实现
advantages = returns - state_values
actor_loss = -(log_probs * advantages.detach()).sum()

3. Asynchronous Advantage Actor-Critic (A3C)

  • A3C 是在 A2C 的基础上引入了多个并行的 Actor-Critic 网络，用于加速训练。
  • 由于强化学习的数据来自于和环境交互，因此如果可以让多个 worker 并行地和环境交互，然后将数据汇总到一个 learner 网络中，就可以加速训练。A3C 就是这样一种算法。
  • 更偏工程化一些，主要涉及到多线程/多进程的并行计算，以及梯度的更新和参数同步等。

4. Proximal Policy Optimization (PPO)

公式表述

  • PPO 是一种 Actor-Critic 算法，核心思想是：在更新策略参数时，限制新策略和旧策略之间的差异，以此来保证策略更新的稳定性。
  • PPO 相较于 A2C：
    1. 限制更新幅度：引入了 Clipped Surrogate Objective，限制新策略和旧策略之间的差异，即：

    LCLIP(θ)=Et[min⁡(rt(θ)⋅At,clip(rt(θ),1−ϵ,1+ϵ)⋅At)]rt(θ)=πθ(at∣st)πθold(at∣st)L^{CLIP}(\theta) = \mathbb{E}_t[\min(r_t(\theta) \cdot A_t, \text{clip}(r_t(\theta), 1-\epsilon, 1+\epsilon) \cdot A_t)]\\r_t(\theta)=\frac{\pi_{\theta}(a_t|s_t)}{\pi_{\theta_{old}}(a_t|s_t)}

    • 其中:
      • LCLIP(θ)L^{CLIP}(\theta) 是 Clipped Surrogate Objective
      • rt(θ)r_t(\theta) 是新旧策略在状态 sts_t 采取动作 ata_t 的概率比
      • AtA_t 是优势函数
      • ϵ\epsilon 是区间宽度超参数。
    1. 优势函数估计：引入了广义优势估计 Generalized Advantage Estimation (GAE)，用于提高策略梯度估计的准确性和平滑性。

    At=∑l=0∞(γλ)lδt+lδt=rt+γV(st+1)−V(st)A_t = \sum_{l=0}^{\infty}(\gamma \lambda)^l \delta_{t+l}\\\delta_t=r_t+\gamma V(s_{t+1}) - V(s_t)

    • 其中:
      • AtA_t 是广义优势估计
      • δt\delta_t 是时序差分 (TD Error)
      • λ\lambda 是折扣因子
      • γ\gamma 是优势函数的折扣因子
      • V(st)V(s_t) 是状态价值估计
    1. 多次采样和小批量更新：引入了记忆、多次采样和小批量更新，用于提高数据的利用效率，即：存储一个或多个 episode 所有轨迹，然后每次更新时从中随机抽取数据分 batch 训练
    2. 一阶优化算法：使用一阶优化算法，如 Adam / SGD 等，而不是 TRPO 等算法用到的二阶优化算法，训练复杂度低效率高。
    3. 联合优化：同时优化策略和价值网络，即：

    L(θ)=LCLIP(θ)+c1⋅E[(Vθ(st)−Vttarget)2]+c2⋅H(πθ)L(\theta)= L^{CLIP}(\theta) + c_1 \cdot \mathbb E[(V_{\theta}(s_t) - V_t^{target})^2] + c_2 \cdot H(\pi_\theta)

    • 其中：
      • LCLIP(θ)L^{CLIP}(\theta) 是 Clipped Surrogate Objective，用于优化策略
      • 第二项是价值网络的损失，用于优化价值网络
      • 第三项是熵正则项，用于提高策略的探索性

代码表述

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
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
class PPOActorCriticNetwork(nn.Module):
    """ PPO Actor-Critic 网络，包含策略网络和价值网络。 """
    def __init__(self, obs_space_dims: int, action_space_dims: int):
        super().__init__()
        hidden_space1 = 16
        hidden_space2 = 32
        # 共享网络
        self.shared_net = nn.Sequential(
            nn.Linear(obs_space_dims, hidden_space1),
            nn.Tanh(),
            nn.Linear(hidden_space1, hidden_space2),
            nn.Tanh(),
        )
        # 策略网络输出动作的均值和标准差
        self.policy_mean_net = nn.Linear(hidden_space2, action_space_dims)
        self.policy_stddev_net = nn.Linear(hidden_space2, action_space_dims)
        # 价值网络输出状态价值估计
        self.value_net = nn.Linear(hidden_space2, 1)
    def forward(self, x: torch.Tensor):
        """前向传播，输出动作参数和状态价值。"""
        shared_features = self.shared_net(x.float())
        action_means = self.policy_mean_net(shared_features)
        action_stddevs = torch.nn.functional.softplus(
            self.policy_stddev_net(shared_features)
        )
        action_stddevs = torch.clamp(action_stddevs, min=1e-3)  # 标准差裁剪
        state_values = self.value_net(shared_features)
        return action_means, action_stddevs, state_values
class PPOAgent:
    """ PPO 算法的实现。 """
    def __init__(self, obs_space_dims: int, action_space_dims: int):
        self.learning_rate = 1e-5
        self.gamma = 0.99
        self.lam = 0.95
        self.eps_clip = 0.1
        self.entropy_coeff = 0.01
        self.epochs = 10
        self.batch_size = 64
        self.net = PPOActorCriticNetwork(obs_space_dims, action_space_dims)
        self.optimizer = torch.optim.Adam(self.net.parameters(), lr=self.learning_rate)
        self.memory = []
    def sample_action(self, state: np.ndarray):
        """根据策略采样动作。"""
        state = torch.tensor(np.array([state]))
        action_means, action_stddevs, state_value = self.net(state)
        distrib = Normal(action_means[0], action_stddevs[0] + 1e-3)
        action = distrib.sample()
        action = torch.clamp(action, -3.0, 3.0)
        log_prob = distrib.log_prob(action).sum()
        return action.numpy(), log_prob, state_value
    def store_transition(self, transition):
        """存储轨迹数据。"""
        self.memory.append(transition)
    def process_batch(self):
        """处理并计算优势函数和目标回报值。"""
        states, actions, log_probs, rewards, dones, state_values = zip(*self.memory)
        states = torch.tensor(states, dtype=torch.float32)
        actions = torch.tensor(actions, dtype=torch.float32)
        log_probs = torch.tensor(log_probs, dtype=torch.float32)
        state_values = torch.tensor(state_values, dtype=torch.float32).squeeze()
        rewards = torch.tensor(rewards, dtype=torch.float32)
        dones = torch.tensor(dones, dtype=torch.float32)
        # 在 state_values 末尾添加一个 0，以表示最后一个状态的下一个状态价值
        next_state_values = torch.cat([state_values[1:], torch.tensor([0.0])]) * (
            1 - dones
        )
        # 计算 delta
        deltas = rewards + self.gamma * next_state_values * (1 - dones) - state_values
        # 计算优势函数
        advantages = torch.zeros_like(rewards)
        running_adv = 0
        for t in reversed(range(len(rewards))):
            running_adv = deltas[t] + self.gamma * self.lam * running_adv * (
                1 - dones[t]
            )
            advantages[t] = running_adv
        # 计算目标回报值
        returns = advantages + state_values
        # 标准化优势函数
        advantages = (advantages - advantages.mean()) / (advantages.std() + 1e-8)
        # 清空记忆
        self.memory = []
        return states, actions, log_probs, returns, advantages
    def update(self):
        """更新策略和价值网络。"""
        states, actions, old_log_probs, returns, advantages = self.process_batch()
        dataset = torch.utils.data.TensorDataset(
            states, actions, old_log_probs, returns, advantages
        )
        loader = torch.utils.data.DataLoader(
            dataset, batch_size=self.batch_size, shuffle=True
        )
        for _ in range(self.epochs):
            for (
                batch_states,
                batch_actions,
                batch_old_log_probs,
                batch_returns,
                batch_advantages,
            ) in loader:
                action_means, action_stddevs, state_values = self.net(batch_states)
                distrib = Normal(action_means, action_stddevs + 1e-3)
                log_probs = distrib.log_prob(batch_actions).sum(axis=-1)
                entropy = distrib.entropy().sum(axis=-1)
                # 计算 PPO 损失
                # 1. 计算裁剪的策略损失
                ratios = torch.exp(log_probs - batch_old_log_probs)
                surr1 = ratios * batch_advantages
                surr2 = (
                    torch.clamp(ratios, 1 - self.eps_clip, 1 + self.eps_clip)
                    * batch_advantages
                )
                policy_loss = -torch.min(surr1, surr2).mean()
                # 2. 计算价值损失
                value_loss = nn.functional.mse_loss(
                    state_values.squeeze(), batch_returns
                )
                # 3. 计算熵损失
                entropy_loss = -entropy.mean()
                # 4. 加权得到总损失
                loss = (
                    policy_loss + 0.5 * value_loss + self.entropy_coeff * entropy_loss
                )
                # 更新网络参数
                self.optimizer.zero_grad()
                loss.backward()
                torch.nn.utils.clip_grad_norm_(self.net.parameters(), max_norm=0.1)
                self.optimizer.step()

Thoughts

  • Actor-Critic 算法是一种结合了策略梯度和值函数的方法，多用于连续动作空间的强化学习任务。
  • 结合多种改进算法，Actor-Critic 算法占据了强化学习领域的重要地位。