用pytorch简单实现DQN

本文内容参考 《Deep Reinforcement Learning Hands-On》第六章 这篇博客默认读者已经熟悉Q-learning。

DQN算法

1. 初始化Q网络和target_Q网络, ϵ = 1 \epsilon = 1 ϵ=1。
2. 以 ϵ \epsilon ϵ 的概率随机选取动作 a a a, 否则 a a a 由Q网络选取，即 a = a r g m a x a Q s , a a = argmax_aQ_{s,a} a=argmaxa​Qs,a​。
3. 执行动作 a a a, 获取奖励reward r r r 和下一个状态 s ′ s' s′。
4. 储存（ s , a , r , s ′ s,a,r,s' s,a,r,s′）到经验池中。每一组（ s , a , r , s ′ s,a,r,s' s,a,r,s′）被称为transition， 即转变。
5. 在经验池中选取一组transition样本集（minibatch）。
6. 对于每一个transition， 有 y = r + γ m a x a ′ t a r g e t _ Q ( s ′ , a ′ ) y = r + \gamma max_a'target\_Q(s',a') y=r+γmaxa′​target_Q(s′,a′)。也就是说，使用target_Q网络根据Bellman方程计算Q值。由于这里的 y , r y,r y,r是相对于这个transition的 s , a s,a s,a而言。因此，y实际上的意义是等于 Q ( s , a ) Q(s,a) Q(s,a)，也就是我们训练Q网络要逼近的目标。若s为最后状态，则 y = r y=r y=r (没有s’了）。 注：
   • 状态s的价值 V ( s ) = m a x a ∈ A Q s , a V(s) = max_{a\in A}Q_{s,a} V(s)=maxa∈A​Qs,a​，即以所有动作中价值最高者为准。
   • Q s , a = r s , a + γ m a x a ′ ∈ A Q ( s ′ , a ′ ) Q_{s,a} = r_{s,a} + \gamma max_{a'\in A}Q(s',a') Qs,a​=rs,a​+γmaxa′∈A​Q(s′,a′) 其实很好理解，Q代表状态 s s s下动作 a a a的价值， V则以所有动作的最大值代表这个状态的价值。
7. 计算损失函数： L = ( Q s , a − y ) 2 L = (Q_{s,a}-y)^2 L=(Qs,a​−y)2。 即实际Q值y（担当标签的作用）和网络输出Q值的差距。
8. 使用SGD更新Q网络以降低 L L L。
9. 每 N N N步后， 令target_Q = Q，即同步两个网络。
10. 从步骤2开始重复直至收敛。

pytorch的实现

所有代码可见于：https://github.com/PacktPublishing/Deep-Reinforcement-Learning-Hands-On 这里介绍最重要的几个组件：

Env

代码使用了gym里的env环境，然而自定义的话也不难，只需实现下列:

• env.step(a):
  1. 执行输入的动作a
  2. 获取新状态下新的observation （在代码里observation一般就当做state用）
  3. 获得reward
  4. 得知episode是否结束。
• env.reset(): 重置环境。
• env.observation_space: 告知observation的维度信息之类，并不重要
• env.action_space: action空间，即能进行的动作范畴。

ExperienceBuffer 经验池

经验池的作用就是把每次进行的游戏回合transition（episode，step）记录下来存储起来。 在训练的时候则是在经验池中随机取一组transition batch对Q网络进行优化。同时，也需要及时丢掉过老的记录，及时更新。 Experience = collections.namedtuple('Experience', field_names=['state', 'action', 'reward', 'done', 'new_state']) 首先，定义了一个名为Experience的namedtuple。包括内容除了上面算法中提到的（ s , a , r , s ′ s,a,r,s' s,a,r,s′），还有结束的标识’done’。

   def __init__(self, capacity):
        self.buffer = collections.deque(maxlen=capacity)

使用deque，可以使得当队列满了之后自动将最老的transition踢出。

def append(self, experience):
       self.buffer.append(experience)

def sample(self, batch_size):
      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.uint8), np.array(next_states)

这段代码也非常易懂：使用append增添新的transition进入经验池， 使用sample从经验池中随意取batch_size个样本transition。

Agent

除了简单的init和reset，Agent只有一个最重要的函数，play_step(): - 根据设定的概率 ϵ \epsilon ϵ 随机选取动作或按照Q网络选取动作。 - 调用env.step（a）执行动作，并记录 （ s , a , r , s ′ s,a,r,s' s,a,r,s′）送入经验池。 因此，Agent主要是与环境进行交互，累计经验值。 最后, 若s’ = Done， 即游戏结束时，agent也负责累计total_reward作为整个策略的评估。

Calc_loss（）

这个函数负责计算损失函数， 为什么单独搬出来说是因为别的优化部分已经不需要太过多说。

	state_action_values = net(states_v).gather(1, actions_v.unsqueeze(-1).long()).squeeze(-1)
    next_state_values = tgt_net(next_states_v).max(1)[0]
    next_state_values[done_mask] = 0.0
    next_state_values = next_state_values.detach()
    expected_state_action_values = next_state_values * GAMMA + rewards_v

这段代码的第一句是根据Q网络得到的Q(s,a)值。 s和a都来自于经验池。 由于net(states_v)输出的是6个动作（在这个pong游戏中）的各自Q值，但我们只需要Q(s,a)，也就是说经验池中那个a所对应的Q，因此使用gather函数，将6个动作中的那一个取出。 后面则是通过target_Q网络由bellman方程计算出的Q值，即前文所提到的 Q s , a = r s , a + γ m a x a ′ ∈ A Q ( s ′ , a ′ ) Q_{s,a} = r_{s,a} + \gamma max_{a'\in A}Q(s',a') Qs,a​=rs,a​+γmaxa′∈A​Q(s′,a′) 作为实际Q值，然后缩小两者的差距。next_state_values[done_mask] = 0.0负责当s为最终状态时的特殊情况( Q s , a = r s , a Q_{s,a} = r_{s,a} Qs,a​=rs,a​)。detach用于冻结梯度，防止对target_Q进行更新。

其余代码均不值一提。熟悉这个项目的人会发现这篇攻略写的非常赞。可惜应该没几个人会看或看懂，主要还是为了自己以后翻阅。 此致。