深度强化学习(DRL 5) - 优先回放DQN(Prioritized experience replay)

雨点打透心脏的1/2处 2024-02-24 03:56 27阅读 0赞

#### 目录 ####

*   *  [一、优先回放][Link 1]
     *  [二、代码][Link 2]
     *  [参考][Link 3]

全部代码

https://github.com/ColinFred/Reinforce\_Learning\_Pytorch/tree/main/RL/DQN

### 一、优先回放 ###

在经验回放中是利用均匀分布采样，而这种方式看上去并不高效，对于智能体而言，这些数据的重要程度并不一样，因此提出优先回放（Prioritized Replay）的方法。优先回放的基本思想就是打破均匀采样，赋予学习效率高的样本以更大的采样权重。

一个理想的标准是智能体学习的效率越高，权重越大。符合该标准的一个选择是TD偏差δ。TD偏差越大，说明该状态处的值函数与TD目标的差距越大，智能体的更新量越大，因此该处的学习效率越高。

简而言之，就是在原来的replay buffer中给每个Transition增加了抽样的优先级（priority）

优先回放DQN主要有三点改变：

1， 为了方便优先回放存储与及采样，采用sumTree树来存储；

原文有两种方法计算样本抽样概率：proportional priority和rank-based priority。proportional priority就是样本被sample到的概率是正比于TD偏差的priority；rank-based priority就是概率正比于Transition priority的排序(rank)。这里考虑proportional priority，Transition被抽到的概率与TD偏差成正比。

并且，为保证每一个存入的Transition都能被sample到，新Transition会被赋予一个很大的priority。

2, 目标函数在计算时根据样本的TD偏差添加了权重（权重和TD偏差有关，偏差越大，权重越大）：  
1 m ∑ j = 1 m w j ( y j − Q ( s j , a j , w ) ) 2 \\frac\{1\}\{m\}\\sum\\limits\_\{j=1\}^m w\_j (y\_j-Q(s\_j, a\_j, w))^2 m1j=1∑mwj(yj−Q(sj,aj,w))2

3，每次更新Q网络参数时，都需要重新计算TD误差 δ j = y j − Q ( s j , a j , w ) \\delta\_j = y\_j- Q(s\_j, a\_j, w) δj=yj−Q(sj,aj,w)

### 二、代码 ###

Prioritized experience replay 结合之前的 Double DQN 和 Dueling DQN

SumTree和ReplayMemory\_Per

SumTree主要实现：add()添加experience；get()按priority抽样；update()更新某个Transition的priority。

ReplayMemory\_Per主要实现：push()插入新experience；sample()按priority抽样Transition；update()更新已有经验的priority

class SumTree:
        write = 0
    
        def __init__(self, capacity):
            self.capacity = capacity
            self.tree = np.zeros(2 * capacity - 1)
            self.data = np.zeros(capacity, dtype=object)
            self.n_entries = 0
    
        # update to the root node
        def _propagate(self, idx, change):
            parent = (idx - 1) // 2
    
            self.tree[parent] += change
    
            if parent != 0:
                self._propagate(parent, change)
    
        # find sample on leaf node
        def _retrieve(self, idx, s):
            left = 2 * idx + 1
            right = left + 1
    
            if left >= len(self.tree):
                return idx
    
            if s <= self.tree[left]:
                return self._retrieve(left, s)
            else:
                return self._retrieve(right, s - self.tree[left])
    
        def total(self):
            return self.tree[0]
    
        # store priority and sample
        def add(self, p, data):
            idx = self.write + self.capacity - 1
    
            self.data[self.write] = data
            self.update(idx, p)
    
            self.write += 1
            if self.write >= self.capacity:
                self.write = 0
    
            if self.n_entries < self.capacity:
                self.n_entries += 1
    
        # update priority
        def update(self, idx, p):
            change = p - self.tree[idx]
    
            self.tree[idx] = p
            self._propagate(idx, change)
    
        # get priority and sample
        def get(self, s):
            idx = self._retrieve(0, s)
            dataIdx = idx - self.capacity + 1
    
            return (idx, self.tree[idx], self.data[dataIdx])
    
    
    class ReplayMemory_Per(object):
        # stored as ( s, a, r, s_ ) in SumTree
        def __init__(self, capacity=1000, a=0.6, e=0.01):
            self.tree = SumTree(capacity)
            self.memory_size = capacity
            self.prio_max = 0.1
            self.a = a
            self.e = e
    
        def push(self, *args):
            data = Transition(*args)
            p = (np.abs(self.prio_max) + self.e) ** self.a  # proportional priority
            self.tree.add(p, data)
    
        def sample(self, batch_size):
            idxs = []
            segment = self.tree.total() / batch_size
            sample_datas = []
    
            for i in range(batch_size):
                a = segment * i
                b = segment * (i + 1)
                s = uniform(a, b)
                idx, p, data = self.tree.get(s)
    
                sample_datas.append(data)
                idxs.append(idx)
            return idxs, sample_datas
    
        def update(self, idxs, errors):
            self.prio_max = max(self.prio_max, max(np.abs(errors)))
            for i, idx in enumerate(idxs):
                p = (np.abs(errors[i]) + self.e) ** self.a
                self.tree.update(idx, p)
    
        def size(self):
            return self.tree.n_entries

每次更新Q网络参数时，都需要重新计算TD误差，并且更新SumTree。

关于目标函数在计算时根据样本的TD偏差添加了权重这一点并未采用

class PerDQN:
        def __init__(self, n_action, n_state, learning_rate):
    
            self.n_action = n_action
            self.n_state = n_state
    
            self.memory = ReplayMemory_Per(capacity=100)
            self.memory_counter = 0
    
            self.model_policy = DNN(self.n_state, self.n_action)
            self.model_target = DNN(self.n_state, self.n_action)
            self.model_target.load_state_dict(self.model_policy.state_dict())
            self.model_target.eval()
    
            self.optimizer = optim.Adam(self.model_policy.parameters(), lr=learning_rate)
    
        def store_transition(self, s, a, r, s_):
            state = torch.FloatTensor([s])
            action = torch.LongTensor([a])
            reward = torch.FloatTensor([r])
            next_state = torch.FloatTensor([s_])
            self.memory.push(state, action, next_state, reward)
    
        def choose_action(self, state):
            state = torch.FloatTensor(state)
            if np.random.randn() <= EPISILO:  # greedy policy
                with torch.no_grad():
                    q_value = self.model_policy(state)
                    action = q_value.max(0)[1].view(1, 1).item()
            else:  # random policy
                action = torch.tensor([randrange(self.n_action)], dtype=torch.long).item()
    
            return action
    
        def learn(self):
            if self.memory.size() < BATCH_SIZE:
                return
            idxs, transitions = self.memory.sample(BATCH_SIZE)
            batch = Transition(*zip(*transitions))
    
            state_batch = torch.cat(batch.state)
            action_batch = torch.cat(batch.action).unsqueeze(1)
            reward_batch = torch.cat(batch.reward)
            next_state_batch = torch.cat(batch.next_state)
    
            state_action_values = self.model_policy(state_batch).gather(1, action_batch)
    
            next_action_batch = torch.unsqueeze(self.model_policy(next_state_batch).max(1)[1], 1)
            next_state_values = self.model_target(next_state_batch).gather(1, next_action_batch)
            expected_state_action_values = (next_state_values * GAMMA) + reward_batch.unsqueeze(1)
    
            td_errors = (state_action_values - expected_state_action_values).detach().squeeze().tolist()
            self.memory.update(idxs, td_errors)  # update td error
            loss = F.mse_loss(state_action_values, expected_state_action_values)
    
            self.optimizer.zero_grad()
            loss.backward()
            for param in self.model_policy.parameters():
                param.grad.data.clamp_(-1, 1)
            self.optimizer.step()
    
        def update_target_network(self):
            self.model_target.load_state_dict(self.model_policy.state_dict())

[Link 1]: http://www.voidme.com/content/vmmvwsoulx#_7
[Link 2]: http://www.voidme.com/content/vmmvwsoulx#_30
[Link 3]: http://www.voidme.com/content/vmmvwsoulx#_217