训练强化学习的经验回放策略：experience replay

经验回放：Experience Replay（训练DQN的一种策略）

优点：可以重复利用离线经验数据；连续的经验具有相关性，经验回放可以在离线经验BUFFER随机抽样，减少相关性；

超参数：Replay Buffer的长度；
$$\begin{array}{rl}& \bullet \text{ Find w by minimizing }L(\mathbf{w})=\frac{1}{T}\sum _{t=1}^T\frac{{\delta }_t^2}{2}.\\ & \bullet \text{ Stochastic gradient descent (SGD):}\\ & \bullet \text{ Randomly sample a transition, }(s_i,a_i,r_i,s_{i+1}),\text{from the buffer}\\ & \bullet \text{ Compute TD error, }{\delta }_i.\\ & \bullet {\text{ Stochastic gradient: g}}_i=\frac{\partial {\delta }_i^2/2}{\partial \mathbf{w}}={\delta }_i\cdot \frac{\partial Q(s_i,a_i;\mathbf{w})}{\partial \mathbf{w}}\\ & \bullet \text{ SGD: w}\leftarrow \mathbf{w}-\alpha \cdot {\mathbf{g}}_i.\end{array}$$

注：实践中通常使用minibatch SGD，每次抽取多个经验，计算小批量随机梯度；
Replay Buffer代码实现如下：

@dataclass
class ReplayBuffer:maxsize: intsize: int = 0state: list = field(default_factory=list)action: list = field(default_factory=list)next_state: list = field(default_factory=list)reward: list = field(default_factory=list)done: list = field(default_factory=list)def push(self, state, action, reward, done, next_state):""":param state: 状态:param action: 动作:param reward: 奖励:param done::param next_state:下一个状态:return:"""if self.size < self.maxsize:self.state.append(state)self.action.append(action)self.reward.append(reward)self.done.append(done)self.next_state.append(next_state)else:position = self.size % self.maxsizeself.state[position] = stateself.action[position] = actionself.reward[position] = rewardself.done[position] = doneself.next_state[position] = next_stateself.size += 1def sample(self, n):total_number = self.size if self.size < self.maxsize else self.maxsizeindices = np.random.randint(total_number, size=n)state = [self.state[i] for i in indices]action = [self.action[i] for i in indices]reward = [self.reward[i] for i in indices]done = [self.done[i] for i in indices]next_state = [self.next_state[i] for i in indices]return state, action, reward, done, next_state

训练时的代码如下：

离线数据放到BUFFER里面：

#动作、状态、奖励、结束标志、下一状态
replay_buffer.push(state, action, reward, done, next_state)

训练时采样然后计算损失

bs, ba, br, bd, bns = replay_buffer.sample(n=args.batch_size)
bs = torch.tensor(bs, dtype=torch.float32)
ba = torch.tensor(ba, dtype=torch.long)
br = torch.tensor(br, dtype=torch.float32)
bd = torch.tensor(bd, dtype=torch.float32)
bns = torch.tensor(bns, dtype=torch.float32)loss = agent.compute_loss(bs, ba, br, bd, bns)
loss.backward()
optimizer.step()
optimizer.zero_grad()