强化学习-价值学习算法

Sarsa

理论解释

Sarsa是基于时序差分算法的，它的公式非常简单且易理解，不像策略梯度算法那样需要复杂的推导过程。

Sarsa的核心函数是$Q(s,a)$，它的含义是在状态$s$下执行$a$，在后续轨迹中获取的期望总奖励。时序差分算法的核心思想，就是用当前获得的奖励加上下一个状态的价值估计来作为当前状态的价值估计，在Sarsa算法中体现为如下公式：
$$Q(s_t,a_t)\leftarrow r_t+\gamma \cdot Q(s_{t+1},a_{t+1})$$

在这里，我们使用神经网络来拟合$Q(s,a)$，在选取动作时采用$ϵ$-greedy策略，即有$ϵ$的概率随机选取一个动作，$1-ϵ$的概率选取$Q(s,a)$最大的动作。

按照此策略我们在状态$s_t$时选取动作$a_t$，此时环境会返回状态$s_{t+1}$，则再按照此策略选取动作$a_{t+1}$，然后按照上述的公式来更新$Q(s,a)$参数。由于这里我们使用神经网络来拟合参数，所以我们更新的方式是计算loss值，然后进行梯度下降。如下面所示，其中$los{s}_{fn}$是指根据现有值和目标值来计算loss值的函数，在代码中采取的MSE均方误差函数。
$$q_{value}=Q(s,a)$$
$$q_{target}=r_t+\gamma \cdot Q(s_{t+1},a_{t+1})$$
$$loss=los{s}_{fn}(q_{value},q_{target})$$

代码

环境为python3.12，各依赖包均为最新版。

import random
import gymnasium as gym
import torch
import torch.nn as nn
from torch import tensorclass QNet(torch.nn.Module):def __init__(self, action_state_dim, hidden_dim):"""网络的输入由action和state连接而成，网络的输出是长度为1的向量，代表 q value。action用one-hot向量表示，例如动作空间为A = {0, 1, 2}时，向量(1, 0, 0)和(0, 1, 0)分别代表动作a = 0和动作a = 1。"""super(QNet, self).__init__()# 一个线性层 + 激活函数 + 一个线性层self.network = nn.Sequential(nn.Linear(action_state_dim, hidden_dim),nn.ReLU(),nn.Linear(hidden_dim, 1),)def forward(self, x):x = self.network(x)return xclass Agent:def __init__(self, state_dim, hidden_dim, action_dim, learning_rate, gamma, device, epsilon):# 策略网络self.action_value_net = QNet(state_dim + action_dim, hidden_dim).to(device)# 创建优化器，优化器的作用是根据每个参数的梯度来更新参数self.optimizer = torch.optim.Adam(self.action_value_net.parameters(), lr=learning_rate)# 折扣因子self.gamma = gamma# 进行神经网络计算的设备self.device = device# 探索策略，有epsilon的概率随机选取动作self.epsilon = epsilon# 状态维度self.state_dim = state_dim# 动作维度self.action_dim = action_dim# 损失函数，根据当前值和目标值来计算得出损失值self.loss_fn = nn.MSELoss()def take_action(self, state):# 随机探索if random.random() < self.epsilon:return random.choice(range(self.action_dim))# 生成一个对角线矩阵，矩阵的每一行元素代表一个动作actions = torch.eye(self.action_dim).to(self.device)# 对state进行复制，actions中有多少个动作，就state复制为多少行state = tensor(state, dtype=torch.float).to(self.device)states = state.unsqueeze(0).repeat(actions.shape[0], 1)# 连接actions和states矩阵，得到的action_states可以看做是一个batch的动作状态向量action_states = torch.cat((actions, states), dim=1)# 将一个batch的动作状态向量输入到Q网络中，得到一组Q值# 注意q_values的形状是(batch_size, 1)，我们将它转换成一维向量q_values = self.action_value_net(action_states).view(-1)# 获取最大Q值对应的下标，下标的值就是采取的最优动作max_value, max_index = torch.max(q_values, dim=0)return max_index.item()def update(self, transition):# 取出相关数据reward = torch.tensor(transition['reward']).to(self.device)state = torch.tensor(transition['state']).to(self.device)next_state = torch.tensor(transition['next_state']).to(self.device)terminated = transition['terminated']# 将数字action转换成one-hot action向量action = torch.zeros(self.action_dim, dtype=torch.float).to(self.device)action[transition['action']] = 1.# 将数字next_action转换成one-hot next_action向量next_action = torch.zeros(self.action_dim, dtype=torch.float).to(self.device)next_action[transition['next_action']] = 1.# 连接action和state向量action_state = torch.cat((action, state), dim=0)next_action_state = torch.cat((next_action, next_state), dim=0)# 获取Q值q_value = self.action_value_net(action_state)[0]# 计算目标Q值。一定要注意如果terminated为true，说明执行action后游戏就终止了# 那么next_state和next_action是无意义的，它们的Q值应该为0# 通过将Q值乘以(1. - float(terminated))的方式，来使其在终止时为0q_target = reward + self.action_value_net(next_action_state)[0] * self.gamma \* (1. - float(terminated))# 计算损失值，第一个参数为当前Q值，第二个参数为目标Q值loss = self.loss_fn(q_value, q_target)# 更新参数self.optimizer.zero_grad()loss.backward()self.optimizer.step()if __name__ == '__main__':# 更新网络参数的学习率learning_rate = 1e-3# 训练轮次num_episodes = 1000# 隐藏层神经元数量hidden_dim = 128# 计算累计奖励时的折扣率gamma = 0.98epsilon = 0.2# 如果存在cuda就用cuda，否则用cpudevice = torch.device("cuda") if torch.cuda.is_available() else torch.device("cpu")env = gym.make('CartPole-v1')# 获取状态维度，为4state_dim = env.observation_space.shape[0]# 获取离散动作数量，为2action_dim = env.action_space.n# 强化学习智能体agent = Agent(state_dim, hidden_dim, action_dim, learning_rate, gamma, device, epsilon)for episode in range(num_episodes):# transition含义是，在state执行action后，环境返回reward、next_state、terminated# 根据next_state，继续采取next_action作为下一动作transition = {'state': None,'action': None,'next_state': None,'next_action': None,'reward': None,'terminated': None}# 统计信息，游戏结束时获得的总奖励sum_reward = 0# reset返回的是一个元组，第一个元素是初始state值，第二个元素是一个字典state = env.reset()[0]# 游戏终止信号terminated = Falseaction = agent.take_action(state)while not terminated:next_state, reward, terminated, _, _ = env.step(action)next_action = agent.take_action(next_state)# 为transition中添加当前的状态、动作等信息transition['state'] = statetransition['action'] = actiontransition['next_state'] = next_statetransition['reward'] = rewardtransition['next_action'] = next_actiontransition['terminated'] = terminated# 一定确保这里会学习到terminated为true的那一步agent.update(transition)sum_reward += reward# 进入下一状态state = next_stateaction = next_action# 每10轮打印一次统计信息if episode % 10 == 0:print(f"Episode: {episode}, Reward: {sum_reward}")

DQN

理论解释

DQN全程Deep Q Learning，与Sarsa算法十分类似，依然是使用时序差分算法来优化$Q(s,a)$函数。不过DQN的$Q(s,a)$函数含义和优化方式与Sarsa略有不同。

DQN中$Q(s,a)$的含义是在状态$s$执行动作$a$后，在后续的轨迹中所能获得的最大累积奖励，为了作区分也有人把DQN的$Q(s,a)$表示为$Q^{\star }(s,a)$，本文就不在作区分表示了。

DQN中$Q(s,a)$的时序差分优化过程如下，其中$A$是动作空间：
$$Q(s_t,a_t)\leftarrow r_t+\gamma \cdot \underset{a^{\prime }\in A}{\max }Q(s_{t+1},a^{\prime })$$

使用神经网络来拟合$Q(s,a)$，在选取动作时依然采用$ϵ$-greedy策略。按照此策略我们在状态$s_t$时选取动作$a_t$，此时环境会返回状态$s_{t+1}$，然后遍历所有的动作，选取$Q(s_{t+1},a^{\prime })$最大的动作$a^{\prime }$，然后计算loss值。
$$q_{value}=Q(s,a)$$
$$q_{target}=r_t+\gamma \cdot \underset{a^{\prime }\in A}{\max }Q(s_{t+1},a^{\prime })$$
$$loss=los{s}_{fn}(q_{value},q_{target})$$

与Sarsa相同，损失函数的计算方式依然选择MSE均方误差。

代码

环境为python3.12，各依赖包均为最新版。
实现代码与Sarsa基本相同，仅有两处做了修改，修改位置已在代码中注释。

import random
import gymnasium as gym
import torch
import torch.nn as nn
from torch import tensorclass QNet(torch.nn.Module):def __init__(self, action_state_dim, hidden_dim):"""网络的输入由action和state连接而成，网络的输出是长度为1的向量，代表 q value。action用one-hot向量表示，例如动作空间为A = {0, 1, 2}时，向量(1, 0, 0)和(0, 1, 0)分别代表动作a = 0和动作a = 1。"""super(QNet, self).__init__()# 一个线性层 + 激活函数 + 一个线性层self.network = nn.Sequential(nn.Linear(action_state_dim, hidden_dim),nn.ReLU(),nn.Linear(hidden_dim, 1),)def forward(self, x):x = self.network(x)return xclass Agent:def __init__(self, state_dim, hidden_dim, action_dim, learning_rate, gamma, device, epsilon):# 策略网络self.action_value_net = QNet(state_dim + action_dim, hidden_dim).to(device)# 创建优化器，优化器的作用是根据每个参数的梯度来更新参数self.optimizer = torch.optim.Adam(self.action_value_net.parameters(), lr=learning_rate)# 折扣因子self.gamma = gamma# 进行神经网络计算的设备self.device = device# 探索策略，有epsilon的概率随机选取动作self.epsilon = epsilon# 状态维度self.state_dim = state_dim# 动作维度self.action_dim = action_dim# 损失函数，根据当前值和目标值来计算得出损失值self.loss_fn = nn.MSELoss()def take_action(self, state):# 随机探索if random.random() < self.epsilon:return random.choice(range(self.action_dim))# 生成一个对角线矩阵，矩阵的每一行元素代表一个动作actions = torch.eye(self.action_dim).to(self.device)# 对state进行复制，actions中有多少个动作，就state复制为多少行state = tensor(state, dtype=torch.float).to(self.device)states = state.unsqueeze(0).repeat(actions.shape[0], 1)# 连接actions和states矩阵，得到的action_states可以看做是一个batch的动作状态向量action_states = torch.cat((actions, states), dim=1)# 将一个batch的动作状态向量输入到Q网络中，得到一组Q值# 注意q_values的形状是(batch_size, 1)，我们将它转换成一维向量q_values = self.action_value_net(action_states).view(-1)# 获取最大Q值对应的下标，下标的值就是采取的最优动作max_value, max_index = torch.max(q_values, dim=0)return max_index.item()def update(self, transition):# 取出相关数据reward = torch.tensor(transition['reward']).to(self.device)state = torch.tensor(transition['state']).to(self.device)next_state = torch.tensor(transition['next_state']).to(self.device)terminated = transition['terminated']# 将数字action转换成one-hot action向量action = torch.zeros(self.action_dim, dtype=torch.float).to(self.device)action[transition['action']] = 1.# 连接action和state向量action_state = torch.cat((action, state), dim=0)# 获取Q值q_value = self.action_value_net(action_state)[0]"""与Sarsa算法主要不同的地方，在于q_target的计算方式：类似于take_action函数中的内容，这里需要把所有动作都进行one-hot操作，与状态连接并输入到网络中，获取所有动作的q_value中最大的值，作为计算q_target的一部分。"""next_actions = torch.eye(self.action_dim).to(self.device)next_states = next_state.unsqueeze(0).repeat(next_actions.shape[0], 1)next_action_states = torch.cat((next_actions, next_states), dim=1)q_target = reward + torch.max(self.action_value_net(next_action_states)) \* self.gamma * (1. - float(terminated))# 计算损失值，第一个参数为当前Q值，第二个参数为目标Q值loss = self.loss_fn(q_value, q_target)# 更新参数self.optimizer.zero_grad()loss.backward()self.optimizer.step()if __name__ == '__main__':# 更新网络参数的学习率learning_rate = 1e-3# 训练轮次num_episodes = 1000# 隐藏层神经元数量hidden_dim = 128# 计算累计奖励时的折扣率gamma = 0.98epsilon = 0.2# 如果存在cuda就用cuda，否则用cpudevice = torch.device("cuda") if torch.cuda.is_available() else torch.device("cpu")env = gym.make('CartPole-v1')# 获取状态维度，为4state_dim = env.observation_space.shape[0]# 获取离散动作数量，为2action_dim = env.action_space.n# 强化学习智能体agent = Agent(state_dim, hidden_dim, action_dim, learning_rate, gamma, device, epsilon)for episode in range(num_episodes):# transition含义是，在state执行action后，环境返回reward、next_state、terminated# 根据next_state，继续采取next_action作为下一动作transition = {'state': None,'action': None,'next_state': None,'next_action': None,'reward': None,'terminated': None}# 统计信息，游戏结束时获得的总奖励sum_reward = 0# reset返回的是一个元组，第一个元素是初始state值，第二个元素是一个字典state = env.reset()[0]# 游戏终止信号terminated = Falsewhile not terminated:"""与Sarsa算法略有不同的地方，这里不需要再获取next_action"""action = agent.take_action(state)next_state, reward, terminated, _, _ = env.step(action)# 为transition中添加当前的状态、动作等信息transition['state'] = statetransition['action'] = actiontransition['next_state'] = next_statetransition['reward'] = rewardtransition['terminated'] = terminated# 一定确保这里会学习到terminated为true的那一步agent.update(transition)sum_reward += reward# 进入下一状态state = next_state# 每10轮打印一次统计信息if episode % 10 == 0:print(f"Episode: {episode}, Reward: {sum_reward}")