本文原文来自知乎： https://zhuanlan.zhihu.com/p/103605702

高性能的ReplayBuffer应该满足以下三点：

随机采样 random sample 的速度要快，尽可能加快读取速度（最为重要）
减少保存的数据量，增加吞吐效率（对分布式而言重要）
保存能简化计算的变量（对特定算法而言重要）
为了达成以上要求，我建议做出以下修改：

把Replay Buffer 的数据都放在连续的内存里，加快读取速度
按trajectory的顺序保存 env transition，避免重复保存next state，减少数据量
分开保存 state 与其他数据，减少数据量
将off-policy 的数据一直保存在显存内
保存 mask = gamma if done else 0 用于计算Q值，而不是保存 done
为on-policy 的PPO 算法保存 noise 用于计算新旧策略的熵

本片文章的目录就是的1_{3点注意事项，以及1}6点解决方案

本文的重点：ReplayBuffer的数据要放在连续内存里，实验结果如下：

我们使用Numpy库在内存里、使用PyTorch库在显存里创建了一整块连续的空间，对比了List 和 Tuple 的方案。结果：连续存储空间的明显更节省时间。因此，DRL库的ReplayBuffer 有必要围绕着连续内存空间来设计。

代码见 github Yonv1943 ReplayBufferComparison.py ，是一份基于Python Numpy 的实现，代码在保证高性能的同时做到了优雅。我根据这一页的结论还写出来一个DRL库「小雅：ElegantRL」

更新日志

2020-06-17 初版匹配了文字与图片，修改语法错误
2020-12-10 第二版在@UncleDrew 的提醒下修复了失效链接
原标题 DRL的经验回放(Experiment Replay Buffer ) 用Numpy实现让它更快一点，改正为Experience Replay
2021-03-24 修改标题 DRL的经验回放(Experience Replay Buffer)的三点高性能修改建议
新增 3. 将state 与其他数据分开保存

0. 简单介绍「经验回放」

简单介绍「经验回放」(Experimen Replay)：当DQN（Deep Q Network）[1]使用一个神经网络用于拟合一个值函数时，Q-learning完成了从离散状态空间到连续状态空间的跨越。为了更好地使用这个叫Q Network的神经网络，我们需要用符合独立同分布（independent and identically distributed，i.i.d.）的数据分批次地训练它。

而DQN使用了1997年就存在的 Experience Replay [2]技术：梯度下降要求用于批次训练的数据符合 i.i.d.，然而与环境交互后产生的数据不能直接用于训练。因此，我们先把贝尔曼公式（Bellman Equation）需要的数据保存起来，当缓存中的数据足够多时，随机抽样得到的数据就能接近i.i.d.。

而高性能的ReplayBuffer应该满足以下三点：

1. 随机采样 random sample 的速度要快，尽可能加快读取速度（重要）

需要 ReplayBuffer 完成的任务有两个，会降低读写速度的方案不应该采用：

+写入：actor与环境交互，得到 environment transition元组 (state, action, reward, done, next state)，或者得到一整条trajectory 写入其中

读取：使用ReplayBuffer内部的数据训练网络的时候，需要从中 random sample 出许多批次的数据用于随机梯度下降（Stochastic Gradient Descent）
2. 减少保存的数据量，增加吞吐效率（对分布式而言重要）
深度学习（Deep Learning）中的有监督训练，训练数据一开始就存放在数据集中，因此可以进行数据预处理，预加载等操作。除去offline RL，大部分强化学习任务做不到：数据需要智能体在环境中探索，一点点存入ReplayBuffer中。如果是同策略（on-policy），那么在探索前还需要清空数据，再重新收集。

因此，ReplayBuffer需要减少需要保存的数据量，增加吞吐效率。

这对多进程、分布式的通信很重要，直接影响通信效率。
网络训练时，数据一定会从产生地（在内存中的env）传输到使用地（在显存中的优化器 optimizer），数据量越小，吞吐效率越高。

3. 保存能简化计算的变量（对特定算法而言重要）

这是编程常用技巧，有一点动态规划Dynamic Programming 的思想：已经算好的中间变量如果以后有用，就试着保存下来，简化后面的计算。例如：不要保存一个done 这个bool，而是保存为一个mask float，简化TD-errors的计算：

state, action, reward, done, next_state, gamma
mask = 0.0 if done else gamma

bellman equation
原本的: y = r + (1-done) * gamma * next_q_value
改变为: y = r + mask * next_q_value

有一些代码保存了 undone，明显就是有“简化计算”的意识，却没有贯彻到底
为了达成以上要求，我提出的修改方案主要有以下6点：

1. 把Replay Buffer 的数据都放在连续的内存里

从对比实验得出：把数据都放在连续的内存/显存里是最快的。

实验结果如下，可以看到使用了 Numpy将数据放在连续内存中的方法最快，用了13秒（比其他方法快了 50%），而不管在何时传入GPU，对总用时的影响都很小：实验结果如下，可以看到使用了 Numpy将数据放在连续内存中的方法最快，用了13秒（比其他方法快了 50%），而不管在何时传入GPU，对总用时的影响都很小：

Method       Used Times (second)   Detail
List         24                    list()
NamedTuple   20                    collections.namedtuple
Array(CPU)   13                    numpy.array/torch.tensor(CPU)
Array(GPU)   13                    torch.tensor (GPU)

2. 按trajectory的顺序保存 env transition，避免重复保存next state

一般我们会说 env transition 是一个元组 (state, action, reward, next_state) ，我们保存了这些状态转移元组，来描述环境与策略。如果我们按 trajectory的顺序保存trainsition，那么不必重复保存 next_state。因为它可以在下一行找到。

优点：如果state 的数据量是action 的2倍以上，那么数据量将降低至 60% ~ 50%。
缺点：random sample 的时候，需要使用random id 寻址两次而不是一次

ReplayBuffer (in order)

state1, reward1, mask1, action1, state2, ...
state2, reward2, mask2, action2, state3, ...
...

数据量降低带来的提升非常明显，还收顺便节省显存、内存。而寻址两次并不慢，可以承受。值得一提的是，按顺序保存，给分布式下的ReplayBuffer提出了更高的要求，不同workers 收集到的 env transition 在汇集的时候，一定要以trajectory 为单位。对于无终止状态的环境，无法以 trajectory为单位。此时我们需要给每一个 workers 对应配置一个ReplayBuffer，而不是直接把它们都存到同一个内。

class ReplayBuffer:

class ReplayBufferMP:
    def __init__(...):
        ...
        self.buffers = [ReplayBuffer(...) for _ in range(rollout_num)]
        ...

3. 分开保存 state 与其他数据，减少数据量

在以图像作为state 的任务中（Atari Game），很有必要分开保存state 与其他数据。图片的格式是 uint8 (0₂₅₅)，而其他数据是 float32。结合上面第二点：“不重复保存 next state”，ReplayBuffer 的数据量将降低至 30% ₁₂.5% 。分开保存之后，如果state 是图片，它也可以直接保存成 (n, height, width, channel) 的格式，减少 flatten 的麻烦。

byte:    =      # 一个字节
uint8:   =
float16: ==
float32: ====

此外，结合上面第二点“不重复保存 next state”，我们可以直接在state 上寻址两次。来获取state 以及 next state。

buf_state.shape == (sum_of_trajectory_len, state_dim) 是按trajectory顺序保存的 state

self.buf_state[indices],
self.buf_state[indices + 1]

此外，随机抽样时需要产生 indices。如果允许在同一批次中产生重复的抽样，那么算法速度更快。很意外的是：重复抽样竟然效果更好。（我猜测这是因为：这种方式丰富了 batch 数据的多样性，因此更不容易过拟合）

# indices = rd.choice(self.memo_len, batch_size, replace=False) # why perform worse?
# indices = rd.choice(self.memo_len, batch_size, replace=True) # why perform better?
# same as:
indices = rd.randint(self.now_len, size=batch_size)

buffer[indices]

4. 将off-policy 的数据一直保存在显存内

异策略 off-policy：可以使用与“被更新的策略”相异的策略收集到的ReplayBuffer数据用于更新的算法。同策略 on-policy 则不能。因此异策略的ReplayBuffer中，有很多数据在达到最大容量前能被保留。因此有必要将 off-policy 的数据一直保存在显存内，减少数据吞吐量。

5. 保存mask = gamma if done else 0 用于计算Q值，而不是保存 done

不要保存一个done 这个bool，而是保存为一个mask float，简化TD-errors的计算：

state, action, reward, done, next_state, gamma
mask = 0.0 if done else gamma

bellman equation
原本的: y = r + (1-done) * gamma * next_q_value
改变为: y = r + mask             * next_q_value
有一些代码保存了 undone，明显就是有“简化计算”的意识，却没有贯彻到底

6. 为on-policy 的PPO 算法保存 noise 用于计算新旧策略的熵

在随机策略中，动作由高斯噪声产生。PPO为了计算旧策略的熵需要知道noise的值，PPO算法计算新旧策略的熵 logprob 的代码如下：

# in AgentZoo.py AgentPPO.update_net()

action = action_avg + action_std * noise
noise = (action_avg - action)/(action_std)
a_std_log = np.log(action_std)
sqrt_2pi_log = np.log(np.sqrt(2*pi))

logprob = -(noise.pow(2).__mul__(0.5) + a_std_log + sqrt_2pi_log).sum(1)
# same as below
logprob = -((action_avg - action) ** 2 /(2 * (action_std ** 2)) + a_std_log + sqrt_2pi_log).sum(1)

1. 附录：用于对比的其他ReplayBuffer方案

使用List：维护一个长度可变的List，到达最大容量时，删除最早加入的记忆，这个方案最简单。此列表的每个元素为这样的一个元组（state, action, reward, mask, next_state）。但是随机抽样的速度较慢，将位于不同内存空间的元素组成批次数据时较慢。
```
self.buffer.append(memory_tuple)
del self.buffer[:del_len]
```

使用NamedTuple：具体实现与List相同。维护一个长度可变的NamedTuple。（其实这也是一个List，只是使用了Python内建的 NamedTuple为 state, action, reward, mask, next_state 命名）。代码十分简洁优雅。但是随机抽样的速度较慢。PyTorch 官网上的RL入门教程就使用了这种方法。

from collections import namedtuple
self.transition = namedtuple(
    'Transition', ('reward', 'mask', 'state', 'action', 'next_state',)
)

self.buffer.append(self.transition(*args))

'''convert tuple into array'''
arrays = self.transition(*zip(*[self.buffer[i] for i in indices]))

使用Numpy.ndarray：直接划出一整块连续的内存空间。设置更新的指针。到达最大容量时，依照指针让新的记忆覆盖旧的记忆。抽样时使用Numpy自带的方法。速度较快，且代码优雅。


self.buffer = np.empty((memo_max_len, memo_dim), dtype=np.float32)

self.buffer[self.next_idx] = np.hstack(memo_tuple)

tensor = torch.tensor(buffer, device=device)

'''convert array into torch.tensor'''
tensors = (
    tensor[:, 0:1],  # rewards
    tensor[:, 1:2],  # masks, mark == (1-float(done)) * gamma
    tensor[:, 2: ],  # states
    tensor[:,  : ],  # actions
    tensor[:,  : ],  # next_states
)