强化学习算法实战：一个例子搞懂sarsa、dqn、ddqn、qac、a2c及其区别

简介

在学习强化学习算法：sarsa、dqn、ddqn、qac、a2c、trpo、ppo时，由于有大量数学公式的推导，觉得十分晦涩，且听过就忘记了。
但是当把算法应用于实战时，代码的实现要比数学推导直观很多。
接下来通过不同的算法实现gym中的CartPole-v1游戏。

游戏介绍

CartPole（推车倒立摆） 是强化学习中经典的基准测试任务，因为其直观可视、方便调试、状态和动作空间小等特性，常用于入门教学和算法验证。它的目标是训练一个智能体（agent）通过左右移动小车，使车顶的杆子尽可能长时间保持竖直不倒。

• 环境：小车（cart）可以在水平轨道上左右移动，顶部通过关节连接一根自由摆动的杆子（pole）。
• 目标：通过左右移动小车，使杆子的倾斜角度不超出阈值（±12°或±15°），同时小车不超出轨道范围（如轨道长度的±2.4单位）。简单理解为，就是杆子不会倒下里，小车不会飞出屏幕。
• 状态：状态空间包含4个连续变量，分别是小车位置（x），小车速度（v），杆子角度（θ），杆子角速度（ω）
• 动作：动作空间只有2个离线动作，分别是0（向左移动）或1（向右移动）
  奖励机制：每成功保持杆子不倒+1分，目前是让奖励最大化，即杆子永远不倒

DQN&DDQN&Dueling DQN&Sarsa

首先我们通过DQN算法介绍完整代码，剩下的算法只需要在此基础上进行少量修改。

DQN

构建价值网络

DQN需要一个主网 络和目标网络，用来评估执行的动作得到的动作价值。这两个网络使用的是同一个网络结构，通常使用神经网络来实现。
输入的维度是状态空间的4个变量，分别是小车位置（x），小车速度（v），杆子角度（θ），杆子角速度（ω）。输出的维度是动作空间的维度，分别表示向左、向右移动的动作价值。
代码如下：

class QNetWork(nn.Module):def __init__(self, state_dim, action_dim):super().__init__()self.fc = nn.Sequential(nn.Linear(state_dim, 64),nn.ReLU(),nn.Linear(64, 64),nn.ReLU(),nn.Linear(64, action_dim))def forward(self,x):return self.fc(x)

构建DQN智能体

DQN智能体具备如下功能：

1. 选择动作
2. 存储经验
3. 训练上面的神经网络：目的是返回给定状态下尽可能接近真实的动作价值
4. 模型保存
5. 评估价值网络

初始化参数

初始化q网络、目标网络、设置优化器、设置经验回放使用的缓存大小、训练的batch_size
dqn的折扣因子、探索率、更新目标网络的频率、智能体步数计数器初始化、记录最佳网络分数值参数初始化、评估轮数设置

    def __init__(self,state_dim, action_dim):# 神经网络网络相关self.q_net = QNetWork(state_dim,action_dim)self.target_net = QNetWork(state_dim,action_dim)self.target_net.load_state_dict(self.q_net.state_dict())self.optimizer = optim.Adam(self.q_net.parameters(), lr=1e-3)self.replay_buffer = deque(maxlen=10000)self.batch_size = 64# DQN相关self.gamma = 0.99self.epsilon = 0.1self.update_target_freq = 100self.step_count=0self.best_avg_reward = 0self.eval_episodes=5

动作选择

在DQN算法中，会采用epsilon参数来增加智能体选择动作的探索性，因此，动作选择的代码逻辑为：

• 以epsilon的概率随机选择一个动作
• 以1-epsilon的概率来选择价值网络返回结果中动作价值更大的动作

    def choose_action(self, state):if np.random.rand() < self.epsilon:return np.random.randint(0,2) //随机选择动作else:state_tensor = torch.FloatTensor(state)q_values = self.q_net(state_tensor) //调用价值网络选择动作return q_values.cpu().detach().numpy().argmax()

存储经验

DQN中，使用经验回放，那么我们需要预留缓冲区来存放历史的轨迹数据，方便后续取出用来训练网络
存储当前的状态、选择的动作、获得的奖励、下一个状态、游戏是否结束（即杆子是不是倒下）

    def store_experience(self,state, action, reward, next_state, done):self.replay_buffer.append((state, action, reward, next_state, done))

训练神经网络⭐️

最终要的部分，不同的算法，基本上也就是这一部分存在差异。
代码流程如下：

1. 缓冲区中采样一个batch的数据
2. 计算当前q值和目标q值（不同的dqn算法，主要是这两步计算的方式不同）
3. 计算损失，除了后面策略网络需要自己构造损失函数，其他的基本都是用MSELoss，也就是当前q值和目标q值的平方差，
4. 梯度下降&更新网络，这两部都有现成的库来完成，基本上也是固定代码，

    def train(self):# 判断是否有足够经验用例用来学习if len(self.replay_buffer) < self.batch_size:return# 从缓冲区随机采样batch = random.sample(self.replay_buffer, self.batch_size)states, actions, rewards, next_states, dones = zip(*batch)states_tensor = torch.FloatTensor(np.array(states))actions_tensor = torch.LongTensor(actions)rewards_tensor = torch.FloatTensor(rewards)next_states_tensor = torch.FloatTensor(np.array(next_states))dones_tensor = torch.FloatTensor(dones)# 计算当前q值current_q = self.q_net(states_tensor).gather(1, actions_tensor.unsqueeze(1)).squeeze()# 计算目标q值with torch.no_grad():# 使用目标网络计算下一状态的所有动作价值，选择动作价值最大的动作# 这里蕴含了两步：使用目标网络计算价值+动作选择next_q = self.target_net(next_states_tensor).max(1)[0]target_q = rewards_tensor + self.gamma * next_q * (1 - dones_tensor)# 构建损失函数，两者的平方差loss = nn.MSELoss()(current_q,target_q)# 梯度下降，网络更新，固定代码self.optimizer.zero_grad()loss.backward()self.optimizer.step()# 计算步数self.step_count += 1# 每隔一定步数更新目标网络，即复制主网络（q_net)的全部参数至目标网络中if self.step_count % self.update_target_freq == 0:self.target_net.load_state_dict({k: v.clone() for k, v in self.q_net.state_dict().items()})

保存模型

    def save_model(self,path="./output/best_model_bak.pth"):torch.save(self.q_net.state_dict(),path)print(f"Model saved to {path}")

评估模型

这一步用来评估当前网络的好坏，我们评估得分更高的网络进行保存。
评估的逻辑如下：

1. 从初始状态开始，使用我们训练的q网络选择动作控制杆子
2. 累加轨迹下每个步骤获得的reward作为当前网络的得分
3. 一直到杆子倒下（说明网络效果不够好）
4. 或者分数足够高（说明网络可以控制杆子很长时间保持平衡，网络效果效果很好），再跳出循环
5. 评估一定轮数后，取分数均值作为评估结果，分数越高越好

代码如下：

    def evaluate(self, env):# 入参是游戏环境，需要一个新的环境进行评估# 由于模型评估不需要随机探索，因此记录当前的epsilon，并设置智能体epsilon=0origin_epsilon = self.epsilonself.epsilon = 0total_rewards = []# self.eval_episodes表示评估的轮数for _ in range(self.eval_episodes):state = env.reset()[0]episode_reward = 0while True:# 使用智能体选择动作action = self.choose_action(state)# 与环境交互next_state, reward, done, _, _ = env.step(action)# 得到reward和下一状态episode_reward += rewardstate = next_state# 判断杆子是否倒下、分数是否足够高if done or episode_reward>2e4:breaktotal_rewards.append(episode_reward)# 网络需要回到评估前的状态继续训练，因此恢复epsilon的值self.epsilon = origin_epsilon# 返回平均分数return np.mean(total_rewards)

主流程

1. 初始化游戏环境：游戏环境用于选择动作后交互，并获取reward和下一状态，初始化后获得初始状态
2. 初始化智能体：构建智能体对象，设置epsilon=1，因为初始q网络效果不好，所以设置很大的epsilon让智能体自由探索环境，设置训练的episode数目
3. 对于每个episode：
   • 对于episode中的每一步
     • 使用智能体选择动作
     • 与环境交互，并获得下一状态next_state，该动作的奖励值reward，杆子是否倒下done，存储本次经验
     • 智能体训练
   • 当一个episode结束后，更新一次epsilon参数，使epsilon慢慢衰减，这是因为随着训练，q网络效果变好，这时我们开始慢慢相信q网络给我们做出的选择
   • 每10个episode我们对q网络做一次评估，存储最佳的q网络

代码如下：

if __name__ == '__main__':env = gym.make('CartPole-v1')state_dim = env.observation_space.shape[0]action_dim = env.action_space.nagent = DQNAgent(state_dim, action_dim)config = {"episode": 600,"epsilon_start": 1.0,"epsilon_end": 0.01,"epsilon_decay": 0.995,}agent.epsilon = config["epsilon_start"]for episode in range(config["episode"]):state = env.reset()[0]total_reward = 0while True:action = agent.choose_action(state)next_state, reward, done, _, _ = env.step(action)agent.store_experience(state, action, reward, next_state, done)agent.train()total_reward += rewardstate=next_stateif done or total_reward > 2e4:breakagent.epsilon = max(config["epsilon_end"],agent.epsilon*config["epsilon_decay"])if episode % 10 == 0:eval_env = gym.make('CartPole-v1')avg_reward = agent.evaluate(eval_env)eval_env.close()if avg_reward > agent.best_avg_reward:agent.best_avg_reward = avg_rewardagent.save_model(path=f"output/best_model.pth")print(f"new best model saved with average reward: {avg_reward}")print(f"Episode: {episode},Train Reward: {total_reward},Best Eval Avg Reward: {agent.best_avg_reward}")

DDQN

DDQN与DQN的区别是：

• DQN使用目标网络选择动作并计算q值
• DDQN使用主网络（q_net)选择动作，使用目标网络计算q值

因此DDQN和DQN的代码区别仅有一行代码：

        # DQNcurrent_q = self.q_net(states_tensor).gather(1, actions_tensor.unsqueeze(1)).squeeze()with torch.no_grad():# 根据目标网络计算出来的所有Q值，选择了Q值最大的动作next_q = self.target_net(next_states_tensor).max(1)[0]target_q = rewards_tensor + self.gamma * next_q * (1 - dones_tensor)# DDQNcurrent_q = self.q_net(states_tensor).gather(1, actions_tensor.unsqueeze(1)).squeeze()with torch.no_grad():# 使用主网络选择下一状态要执行的动作next_actions = torch.LongTensor(torch.argmax(self.q_net(next_states_tensor),dim=1))# 使用目标网络计算执行这个动作后的q值next_q=self.target_net(next_states_tensor).gather(1,next_actions.unsqueeze(1)).squeeze()target_q = rewards_tensor + self.gamma * next_q * (1 - dones_tensor)

Dueling DQN

Dueling DQN与DQN的区别主要是神经网络结构不同

1. DQN是简单的神经网络，输入状态，输出不同动作的价值
2. Dueling DQN的网络内有两个分支
   • 价值流：记录当前状态的状态价值，为标量
   • 优势流：表示该动作的优势，与动作价值的作用相同，都是评估动作的好坏，是一个动作维度的向量

直接看代码，Dueling DQN的网络结构如下：

class QNetwork(nn.Module):def __init__(self, state_dim, action_dim):super(QNetwork, self).__init__()self.feature = nn.Sequential(nn.Linear(state_dim, 128),nn.ReLU())self.advantage = nn.Sequential(nn.Linear(128, 128),nn.ReLU(),nn.Linear(128, action_dim))self.value = nn.Sequential(nn.Linear(128, 128),nn.ReLU(),nn.Linear(128, 1))def forward(self, x):x = self.feature(x)# 优势流advantage = self.advantage(x)# 价值流value = self.value(x)# advantage - advantage.mean()消除优势函数基线影响return value + advantage - advantage.mean()

Sarsa

对比一下DQN和Sarsa：

• DQN在更新 Q 函数时使用了下一个状态 $s^{\prime }$ 的最大动作价值：${\max }_{a^{\prime }}{Q}_t(s^{\prime },a^{\prime })$。
  • 这意味着 DQN并不关心智能体实际采取的下一个动作，而是假设智能体将在下一个状态采取具有最大动作价值的动作。
  • 这使得 DQN能够学习到最优策略，而不受当前策略的影响。
  • 但是它可能会过于乐观地估计动作价值
• Sarsa在更新q_net时，它使用了智能体实际采取的下一个动作 $a^{\prime }$ 的动作价值：$Q_t(s^{\prime },a^{\prime })$。
  • 这意味着 SARSA 在学习过程中会考虑智能体当前的策略。
  • 因此，SARSA 学到的策略与智能体在训练过程中实际执行的策略密切相关。

由此，DQN和Sarsa的核心区别是：采取下一个动作的方式，在代码上的区别也是一行代码：

		# 使用DQNcurrent_q = self.q_net(states_tensor).gather(1, actions_tensor.unsqueeze(1)).squeeze()with torch.no_grad():# 使用目标网络选择Q值最大的动作next_q = self.target_net(next_states_tensor).max(1)[0]target_q = rewards_tensor + self.gamma * next_q * (1 - dones_tensor)# 使用Sarsacurrent_q = self.q_net(states_tensor).gather(1, actions_tensor.unsqueeze(1)).squeeze()with torch.no_grad():# 使用当前策略选出下一个动作next_actions_tensor = self.choose_actions(next_states)# 根绝当前策略的下一个动作计算q值next_q=self.target_net(next_states_tensor).gather(1,next_actions_tensor.unsqueeze(1)).squeeze()target_q = rewards_tensor + self.gamma * next_q * (1 - dones_tensor)

再对比一下DDQN和Sarsa：

		# 采用ddqn的方式current_q = self.q_net(states_tensor).gather(1, actions_tensor.unsqueeze(1)).squeeze()with torch.no_grad():next_actions = torch.LongTensor(torch.argmax(self.q_net(next_states_tensor),dim=1))next_q=self.target_net(next_states_tensor).gather(1,next_actions.unsqueeze(1)).squeeze()target_q = rewards_tensor + self.gamma * next_q * (1 - dones_tensor)# 使用Sarsacurrent_q = self.q_net(states_tensor).gather(1, actions_tensor.unsqueeze(1)).squeeze()with torch.no_grad():# 使用当前策略选出下一个动作next_actions = self.choose_actions(next_states)# 根绝当前策略的下一个动作计算q值next_q=self.target_net(next_states_tensor).gather(1,next_actions.unsqueeze(1)).squeeze()target_q = rewards_tensor + self.gamma * next_q * (1 - dones_tensor)

可以看到只有计算next_actions时不同，ddqn使用主网络来计算Q值，但是会选择Q值最大的动作；而Sarsa完全依赖当前策略做选择，意味着不一定选择了Q值最大的动作

QAC&A2C

对于actor-critic类算法，我们需要两个神经网络，分别代表运动员（PolicyNet）和裁判员（ValueNet）
运动员网络的作用是在游戏中根据策略做出动作
裁判员网络的作用是对运动员的动作进行打分
两个网络都需要不停地迭代，运动员的目标是得到更高的分数，裁判员的目标是更公正的评分。

actor-critic类算法与Q-Learning类算法相比，内部逻辑存在较大区别，不过我们仅需要修改智能体内部函数内的实现，整个智能体暴露出来的接口和DQN的接口结构仍然相同，因此在主函数里使用智能体的代码和DQN没有区别。
下面详细介绍actor-critic算法实现智能体的方式。

QAC

构建策略网络

输入的是当前的状态；输出一个动作维度的向量，代表执行不同动作的概率

class PolicyNet(nn.Module):def __init__(self, state_dim, action_dim):super(PolicyNet, self).__init__()self.fc1 = nn.Linear(state_dim, 128)self.fc2 = nn.Linear(128, action_dim)def forward(self, x):x = self.fc1(x)x = torch.relu(x)x = self.fc2(x)x = torch.softmax(x, dim=-1)return x

构建状态价值网络

输入的是当前的状态；输出一个标量，表示给定状态的状态价值

class ValueNet(nn.Module):def __init__(self, state_dim):super(ValueNet, self).__init__()self.fc1 = nn.Linear(state_dim, 128)self.fc2 = nn.Linear(128, 1)def forward(self, x):x = self.fc1(x)x = torch.relu(x)x = self.fc2(x)return x

智能体初始化参数

    def __init__(self,state_dim,action_dim):# 两个网络self.critic = ValueNet(state_dim)self.actor = PolicyNet(state_dim,action_dim)# 优化相关self.critic_optimizer = optim.Adam(self.critic.parameters(), lr=0.001)  # 优化器self.actor_optimizer = optim.Adam(self.actor.parameters(), lr=0.001)self.critic_criterion = nn.MSELoss()  # 损失函数# 最优相关self.gamma = 0.99  # 折损率self.best_avg_reward=0 #最佳网络的奖励self.eval_episodes = 5 #评估轮数

动作选择

    def choose_action(self,state):# 状态转为张量state_tensor = torch.FloatTensor(state)# 状态输入至神经网络得到每个动作的概率probs = self.actor(state_tensor)# 使用上面的概率分布构造抽样，抽样得到动作action = torch.multinomial(probs, 1).item()return action

这里的含义是，比如当前策略网络输出[0.7,0.3]，表示向左的概率是0.7，向右的概率是0.3，那么我们使用这个策略网络选择动作10次，可能有7次向左，3次向右。

训练过程

训练过程中需要对两个网络都进行优化：

1. critic更新

        # critic更新# 获取给定状态的状态价值value = self.critic(state_tensor)# 获取下一状态的状态价值next_value_tensor = self.critic(next_state_tensor)# 计算targettarget = rewards_tensor + self.gamma * next_value_tensor * (1 - done_tensor)# 使用td_error=target-value的平方差作为损失函数critic_loss = self.critic_criterion(value,target)# 梯度下降优化网络self.critic_optimizer.zero_grad()critic_loss.backward()self.critic_optimizer.step()

2. actor更新

		# actor更新# td_error = target - value# 重新通过actor获取action概率action_probs = self.actor(state_tensor)action_prob = action_probs.gather(0, action_tensor)# 关键一行🌟，构造actor损失函数actor_loss = (-torch.log(action_prob) * value.detach()).mean()# 梯度下降优化网络self.actor_optimizer.zero_grad()actor_loss.backward()self.actor_optimizer.step()

actor更新中，最关键的就是损失函数的构建，它代表的数学公式是
$$L_{\text{actor}}=-\mathbb{E}[\log {\pi }_{\theta }(a|s)\cdot V(s)]$$
实际上由于真实期望无法求得，采用了蒙特卡洛近似，也就是使用单次采样值：
$$L_{\text{actor}}=-\log {\pi }_{\theta }(a|s)\cdot V(s)$$
其中：

• ${\pi }_{\theta }(a|s)$ 是策略函数，表示在状态$s$下选择动作$a$的概率
• $V(s)$ 是状态价值函数，代码中用value表示
• $\log {\pi }_{\theta }(a|s)$ 是对策略概率取对数

直观理解：

• 这个损失函数的目标是增加那些价值高的状态-动作对的概率
• 也就是最大化$\log {\pi }_{\theta }(a|s)\cdot V(s)$，此时使用梯度上升优化即可
• 而通常的优化器是梯度下降算法，因此在前面增加负号

模型评估与模型保存

模型评估与DQN中并无区别

模型保存时需要保存两个网络：

    def save_model(self):if not os.path.exists("output"):os.makedirs("output")torch.save(self.critic.state_dict(),'./output/critic_best_model.pth')torch.save(self.actor.state_dict(), './output/actor_best_model.pth')

A2C

搞懂QAC之后，A2C很简单，它是在QAC的基础上进行优化。
QAC的actor网络损失函数是直接使用的价值函数与动作概率构建：
$$L_{\text{actor}}=-\mathbb{E}[\log {\pi }_{\theta }(a|s)\cdot V(s)]$$
A2C的actor网络损失函数是使用优势函数与动作概率构建：
$$L_{\text{actor}}=-\mathbb{E}[\log {\pi }_{\theta }(a|s)\cdot A(s,a)]$$
其中，$A(s,a)$是优势函数，通常计算为:
$$A(s,a)=Q(s,a)-V(s)\approx r+\gamma V(s^{\prime })-V(s)$$
这里：

• ${\pi }_{\theta }(a|s)$ 是策略网络，表示在状态$s$下选择动作$a$的概率
• $r$ 是即时奖励
• $\gamma$ 是折扣因子
• $V(s)$ 是当前状态$s$的价值估计
• $V(s^{\prime })$ 是下一状态$s^{\prime }$的价值估计

A2C使用优势函数$A(s,a)$，这能显著减少梯度估计的方差，提高训练稳定性和效率。

代码上的区别也仅仅两行：

		# qac actor更新# td_error = target - value# 重新通过actor获取action概率action_probs = self.actor(state_tensor)action_prob = action_probs.gather(0, action_tensor)# 这里是用value就是q值，此时是actor-criticactor_loss = (-torch.log(action_prob) * value.detach()).mean()# 梯度下降优化网络self.actor_optimizer.zero_grad()actor_loss.backward()self.actor_optimizer.step()# a2c actor更新# 计算优势函数，也是td_error，参考上面优势函数的计算，两者等价td_error = target - value# 重新通过actor获取action概率，以便建立梯度图action_probs = self.actor(state_tensor)action_prob = action_probs.gather(0, action_tensor)# 关键区别⭐️，损失函数的构建，这里使用的是td_error，而qac使用的valueactor_loss = (-torch.log(action_prob) * td_error.detach()).mean()self.actor_optimizer.zero_grad()actor_loss.backward()self.actor_optimizer.step()