什么是lighting？
Lightning是一个最近发布的Pythorch库，它可以清晰地抽象和自动化ML模型所附带的所有日常样板代码，允许您专注于实际的ML部分（这些也往往是最有趣的部分）。
除了自动化样板代码外，Lightning还可以作为一种样式指南，用于构建干净且可复制的ML系统。
这非常吸引人，原因如下：
- 通过抽象出样板工程代码，可以更容易地识别和理解ML代码。
- Lightning的统一结构使得在现有项目的基础上进行构建和理解变得非常容易。
- Lightning 自动化的代码是用经过全面测试、定期维护并遵循ML最佳实践的高质量代码构建的。
DQN
在我们进入代码之前，让我们快速回顾一下DQN的功能。DQN通过学习在特定状态下执行每个操作的值来学习给定环境的最佳策略。这些值称为Q值。
最初，智能体对其环境的理解非常差，因为它没有太多的经验。因此，它的Q值将非常不准确。然而，随着时间的推移，当智能体探索其环境时，它会学习到更精确的Q值，然后可以做出正确的决策。这允许它进一步改进，直到它最终收敛到一个最优策略（理想情况下）。
我们感兴趣的大多数环境，如现代电子游戏和模拟环境，都过于复杂和庞大，无法存储每个状态/动作对的值。这就是为什么我们使用深度神经网络来近似这些值。
智能体的一般生命周期如下所述：
- 智能体获取环境的当前状态并将其通过网络进行运算。然后，网络输出给定状态的每个动作的Q值。
- 接下来，我们决定是使用由网络给出智能体所认为的最优操作，还是采取随机操作，以便进一步探索。
- 这个动作被传递到环境中并得到反馈，告诉智能体它处于的下一个状态是什么，在上一个状态中执行上一个动作所得到的奖励，以及该步骤中的事件是否完成。
- 我们以元组（状态, 行为, 奖励, 下一状态, 已经完成的事件）的形式获取在最后一步中获得的经验，并将其存储在智能体内存中。
- 最后，我们从智能体内存中抽取一小批重复经验，并使用这些过去的经验计算智能体的损失。
这是DQN功能的一个高度概述。
轻量化DQN
启蒙时代是一场支配思想世界的智力和哲学运动,让我们看看构成我们的DQN的组成部分
模型：用来逼近Q值的神经网络
重播缓冲区：这是我们智能体的内存，用于存储以前的经验
智能体：智能体本身就是与环境和重播缓冲区交互的东西
Lightning模块：处理智能体的所有训练
模型
对于这个例子，我们可以使用一个非常简单的多层感知器（MLP）。这意味着我们没有使用任何花哨的东西，像卷积层或递归层，只是正常的线性层。这样做的原因是由于卡倒立摆环境的简单性，任何比这更复杂的东西都是过度复杂的。
class DQN(nn.Module): """ Simple MLP network Args: obs_size: observation/state size of the environment n_actions: number of discrete actions available in the environment hidden_size: size of hidden layers """ def __init__(self, obs_size: int, n_actions: int, hidden_size: int = 128): super(DQN, self).__init__() self.net = nn.Sequential( nn.Linear(obs_size, hidden_size), nn.ReLU(), nn.Linear(hidden_size, n_actions) ) def forward(self, x): return self.net(x.float())
重播缓冲区
重播缓冲区的构建相当直接，我们只需要某种类型的数据结构来存储元组。我们需要能够对这些元组进行采样并添加新的元组。本例中的缓冲区基于Lapins重播缓冲区，因为它是迄今为止我发现的最简洁并且最快的实现。代码如下
# Named tuple for storing experience steps gathered in training Experience = collections.namedtuple( 'Experience', field_names=['state', 'action', 'reward', 'done', 'new_state']) class ReplayBuffer: """ Replay Buffer for storing past experiences allowing the agent to learn from them Args: capacity: size of the buffer """ def __init__(self, capacity: int) -> None: self.buffer = collections.deque(maxlen=capacity) def __len__(self) -> None: return len(self.buffer) def append(self, experience: Experience) -> None: """ Add experience to the buffer Args: experience: tuple (state, action, reward, done, new_state) """ self.buffer.append(experience) def sample(self, batch_size: int) -> Tuple: indices = np.random.choice(len(self.buffer), batch_size, replace=False) states, actions, rewards, dones, next_states = zip(*[self.buffer[idx] for idx in indices]) return (np.array(states), np.array(actions), np.array(rewards, dtype=np.float32), np.array(dones, dtype=np.bool), np.array(next_states))
但我们还没有完成。如果您在知道它的结构是基于创建数据加载器的思想创建的，然后使用它将小批量的经验传递给每个训练步骤这些原理之前使用过Lightning；那么对于大多数ML系统（如监督模型），这一切如何生效的是很清楚的。但是当我们在生成数据集时，它又是如何生效的呢？
我们需要创建自己的可迭代数据集，它使用不断更新的重播缓冲区来采样以前的经验。然后，我们有一小批经验被传递到训练步骤中用于计算我们的损失，就像其他任何模型一样。除了包含输入和标签之外，我们的小批量包含（状态, 行为, 奖励, 下一状态, 已经完成的事件）
class RLDataset(IterableDataset): """ Iterable Dataset containing the ReplayBuffer which will be updated with new experiences during training Args: buffer: replay buffer sample_size: number of experiences to sample at a time """ def __init__(self, buffer: ReplayBuffer, sample_size: int = 200) -> None: self.buffer = buffer self.sample_size = sample_size def __iter__(self) -> Tuple: states, actions, rewards, dones, new_states = self.buffer.sample(self.sample_size) for i in range(len(dones)): yield states[i], actions[i], rewards[i], dones[i], new_states[i]
您可以看到，在创建数据集时，我们传入重播缓冲区，然后可以从中采样，以允许数据加载器将批处理传递给Lightning模块。
