强化学习 8 —— DQN 代码 Tensorflow 2.0 实现详解

一、引言：DQN在强化学习中的地位

作为深度强化学习（Deep Reinforcement Learning, DRL）的里程碑式算法，Deep Q-Network（DQN）通过将深度神经网络与Q-learning结合，首次实现了在复杂环境（如Atari游戏）中通过原始像素输入直接学习策略的能力。其核心突破在于解决了传统Q-learning在状态空间爆炸时的维度灾难问题，并通过经验回放（Experience Replay）和目标网络（Target Network）两大创新机制，显著提升了训练的稳定性。本文将基于Tensorflow 2.0框架，从算法原理到代码实现进行系统性解析，帮助读者构建可运行的DQN系统。

二、DQN算法核心原理

1. Q-learning的深度化延伸

DQN的核心思想是用深度神经网络（通常为CNN）近似Q函数，即通过输入状态（如游戏画面）输出各动作的Q值。其优化目标是最小化TD误差：
[
L(\theta) = \mathbb{E}{(s,a,r,s’)} \left[ \left( r + \gamma \max{a’} Q(s’,a’;\theta^-) - Q(s,a;\theta) \right)^2 \right]
]
其中，(\theta)为当前网络参数，(\theta^-)为目标网络参数，(\gamma)为折扣因子。

2. 经验回放机制

传统Q-learning采用在线更新，导致样本相关性高、方差大。DQN引入经验回放缓冲区（Replay Buffer），存储转移样本((s,a,r,s’,\text{done}))，训练时随机采样小批量数据，打破时间相关性，提升数据利用率。

3. 目标网络分离

为解决目标Q值依赖当前网络参数导致的振荡问题，DQN维护一个目标网络（参数为(\theta^-)），每隔(N)步同步当前网络参数。目标Q值的计算改为：
[
yj = \begin{cases}
r_j & \text{if episode terminated at } s{j+1} \
rj + \gamma \max{a’} Q(s_{j+1},a’;\theta^-) & \text{otherwise}
\end{cases}
]

三、Tensorflow 2.0实现关键步骤

1. 网络架构设计

以Atari游戏为例，输入为84x84灰度图像（4帧堆叠），输出为动作空间大小（如18个有效动作）。典型CNN结构如下：

import tensorflow as tf
from tensorflow.keras import layers
def build_dqn(input_shape, num_actions):
    model = tf.keras.Sequential([
        layers.Conv2D(32, kernel_size=8, strides=4, activation='relu', input_shape=input_shape),
        layers.Conv2D(64, kernel_size=4, strides=2, activation='relu'),
        layers.Conv2D(64, kernel_size=3, strides=1, activation='relu'),
        layers.Flatten(),
        layers.Dense(512, activation='relu'),
        layers.Dense(num_actions)
    ])
    return model

关键点：

• 使用tf.keras构建模型，兼容Eager Execution模式
• 输出层无激活函数，直接预测Q值
• 输入形状需匹配预处理后的状态（如(84,84,4)）

2. 经验回放实现

import numpy as np
import random
from collections import deque
class ReplayBuffer:
    def __init__(self, capacity):
        self.buffer = deque(maxlen=capacity)
    def store(self, state, action, reward, next_state, done):
        self.buffer.append((state, action, reward, next_state, done))
    def sample(self, batch_size):
        batch = random.sample(self.buffer, batch_size)
        states, actions, rewards, next_states, dones = map(np.array, zip(*batch))
        return states, actions, rewards, next_states, dones
    def __len__(self):
        return len(self.buffer)

优化建议：

• 使用deque实现固定大小缓冲区
• 采样时直接解压为NumPy数组，提升效率
• 初始时填充一定量数据再开始训练（避免冷启动）

3. 目标网络更新策略

class DQNAgent:
    def __init__(self, state_shape, num_actions):
        self.q_network = build_dqn(state_shape, num_actions)
        self.target_network = build_dqn(state_shape, num_actions)
        self.update_target()  # 初始同步
        self.optimizer = tf.keras.optimizers.Adam(learning_rate=1e-4)
    def update_target(self):
        self.target_network.set_weights(self.q_network.get_weights())
    def train_step(self, states, actions, rewards, next_states, dones, gamma=0.99):
        with tf.GradientTape() as tape:
            # 当前Q值
            q_values = self.q_network(states, training=True)
            selected_q = tf.reduce_sum(q_values * tf.one_hot(actions, self.num_actions), axis=1)
            # 目标Q值
            next_q = tf.reduce_max(self.target_network(next_states), axis=1)
            target_q = rewards + gamma * (1 - dones) * next_q
            # 计算损失
            loss = tf.reduce_mean(tf.square(target_q - selected_q))
        # 反向传播
        grads = tape.gradient(loss, self.q_network.trainable_variables)
        self.optimizer.apply_gradients(zip(grads, self.q_network.trainable_variables))
        return loss

关键机制：

• 使用tf.GradientTape自动微分
• 目标网络参数通过set_weights同步
• 损失函数为MSE，优化器推荐Adam

4. 完整训练流程

import gym
from collections import deque
# 参数配置
env = gym.make('CartPole-v1')  # 示例环境，实际可替换为Atari
state_shape = env.observation_space.shape
num_actions = env.action_space.n
buffer_capacity = 10000
batch_size = 32
target_update_freq = 1000
# 初始化
agent = DQNAgent(state_shape, num_actions)
buffer = ReplayBuffer(buffer_capacity)
epsilon = 1.0
epsilon_min = 0.01
epsilon_decay = 0.995
# 训练循环
for episode in range(1000):
    state = env.reset()
    done = False
    episode_reward = 0
    while not done:
        # ε-贪婪策略选择动作
        if np.random.rand() < epsilon:
            action = env.action_space.sample()
        else:
            state_tensor = tf.expand_dims(tf.convert_to_tensor(state), 0)
            q_values = agent.q_network(state_tensor)
            action = tf.argmax(q_values[0]).numpy()
        # 执行动作
        next_state, reward, done, _ = env.step(action)
        buffer.store(state, action, reward, next_state, done)
        episode_reward += reward
        state = next_state
        # 经验回放训练
        if len(buffer) >= batch_size:
            states, actions, rewards, next_states, dones = buffer.sample(batch_size)
            loss = agent.train_step(states, actions, rewards, next_states, dones)
        # 定期更新目标网络
        if episode % target_update_freq == 0:
            agent.update_target()
        # 衰减ε
        epsilon = max(epsilon_min, epsilon * epsilon_decay)
    print(f"Episode {episode}, Reward: {episode_reward}, Epsilon: {epsilon:.2f}")

四、实践中的优化技巧

1. 超参数调优

• 学习率：初始值建议1e-4，可尝试自适应优化器（如RMSprop）
• 折扣因子γ：通常设为0.99，长期回报任务可适当增大
• 经验回放大小：Atari环境建议1e6，简单任务可减小至1e4

2. 改进型DQN

• Double DQN：解决过高估计问题，修改目标Q值计算为：
  [
  yj = r_j + \gamma Q(s{j+1}, \arg\max{a’} Q(s{j+1},a’;\theta);\theta^-)
  ]
• Dueling DQN：将Q网络拆分为状态价值函数和优势函数，提升稀疏奖励任务表现
• Prioritized Experience Replay：根据TD误差优先级采样，加速收敛

3. 调试与可视化

• 使用TensorBoard记录损失、奖励曲线
• 监控Q值分布，避免梯度消失/爆炸
• 定期测试模型在环境中的表现（无探索噪声）

五、总结与展望

本文通过Tensorflow 2.0实现了标准DQN算法，覆盖了从理论到代码的全流程。实际项目中，建议从简单环境（如CartPole）开始验证，再逐步迁移到复杂任务。未来方向可探索：

1. 结合分布式框架（如Ray）实现大规模并行训练
2. 集成其他DRL算法（如PPO、SAC）形成混合架构
3. 应用到机器人控制、自动驾驶等真实场景

DQN作为DRL的基石算法，其设计思想（如经验回放、目标网络）已被后续研究广泛采用。掌握其实现细节，将为深入理解强化学习领域的其他高级算法奠定坚实基础。