Deep Q-Network DQN

# Q-Learning 的表格方法与维度灾难

import numpy as np

class TableQLearning:
    def __init__(self, n_states, n_actions, alpha=0.1, gamma=0.99, epsilon=0.1):
        self.Q = np.zeros((n_states, n_actions))
        self.alpha = alpha
        self.gamma = gamma
        self.epsilon = epsilon

    def choose_action(self, state):
        if np.random.random() < self.epsilon:
            return np.random.randint(self.Q.shape[1])
        return np.argmax(self.Q[state])

    def update(self, s, a, r, s_next, done):
        target = r if done else r + self.gamma * np.max(self.Q[s_next])
        self.Q[s, a] += self.alpha * (target - self.Q[s, a])

# 维度灾难演示：4 维状态，每维离散化为 20 个区间
dims = [20, 20, 20, 20]
n_states = np.prod(dims)
print(f"离散化后状态数: {n_states:,}")  # 160,000

# 如果是 8 维状态呢？
dims_8d = [20] * 8
print(f"8维离散化状态数: {np.prod(dims_8d):,}")  # 25,600,000,000

# 线性函数近似 Q-Learning

class LinearQLearning:
    def __init__(self, n_features, n_actions, alpha=0.01, gamma=0.99):
        self.W = np.zeros((n_features, n_actions))
        self.alpha = alpha
        self.gamma = gamma

    def extract_features(self, state):
        return np.concatenate([state, state**2, np.sin(state)])

    def get_q_values(self, state):
        phi = self.extract_features(state)
        return phi @ self.W

    def choose_action(self, state, epsilon=0.1):
        if np.random.random() < epsilon:
            return np.random.randint(self.W.shape[1])
        return np.argmax(self.get_q_values(state))

    def update(self, state, action, reward, next_state, done):
        phi = self.extract_features(state)
        q_values = self.get_q_values(state)
        if done:
            target = reward
        else:
            next_q = np.max(self.get_q_values(next_state))
            target = reward + self.gamma * next_q
        td_error = target - q_values[action]
        self.W[:, action] += self.alpha * td_error * phi

# 线性模型的局限
agent = LinearQLearning(n_features=12, n_actions=4)
state = np.array([0.5, -0.3, 0.1, 0.2])
print(f"Q 值: {agent.get_q_values(state)}")
print("线性模型无法拟合复杂的 Q 函数曲面")

# DQN 网络架构（PyTorch 风格）

import numpy as np

class DQN:
    def __init__(self, input_shape=(4, 84, 84), n_actions=18):
        # 第一层卷积：32个8x8卷积核，步长4
        self.conv1_out_channels = 32
        self.conv1_kernel = 8
        self.conv1_stride = 4
        # 84 -> (84 - 8) / 4 + 1 = 20
        self.conv1_out_h = 20

        # 第二层卷积：64个4x4卷积核，步长2
        self.conv2_out_channels = 64
        self.conv2_kernel = 4
        self.conv2_stride = 2
        # 20 -> (20 - 4) / 2 + 1 = 9
        self.conv2_out_h = 9

        # 第三层卷积：64个3x3卷积核，步长1
        self.conv3_out_channels = 64
        self.conv3_kernel = 3
        self.conv3_stride = 1
        # 9 -> (9 - 3) / 1 + 1 = 7
        self.conv3_out_h = 7

        # 全连接层
        fc_input_size = self.conv3_out_channels * self.conv3_out_h * self.conv3_out_h
        self.fc_hidden_size = 512
        self.n_actions = n_actions

        # 初始化权重
        self._init_weights(fc_input_size)
        print(f"FC 输入维度: {fc_input_size}")
        print(f"FC 输出维度: {n_actions}")

    def _init_weights(self, fc_input_size):
        scale_conv = np.sqrt(2.0 / (self.conv1_kernel**2 * input_shape[0]))
        self.conv1 = np.random.randn(
            self.conv1_out_channels, input_shape[0],
            self.conv1_kernel, self.conv1_kernel
        ) * scale_conv
        self.fc1 = np.random.randn(fc_input_size, self.fc_hidden_size) * 0.01
        self.fc2 = np.random.randn(self.fc_hidden_size, self.n_actions) * 0.01

# 前向传播模拟（简化版）

import numpy as np

class SimplifiedDQN:
    def __init__(self, n_features=4, n_actions=2, hidden_size=128):
        self.W1 = np.random.randn(n_features, hidden_size) * np.sqrt(2.0 / n_features)
        self.b1 = np.zeros(hidden_size)
        self.W2 = np.random.randn(hidden_size, hidden_size) * np.sqrt(2.0 / hidden_size)
        self.b2 = np.zeros(hidden_size)
        self.W3 = np.random.randn(hidden_size, n_actions) * 0.01
        self.b3 = np.zeros(n_actions)

    def forward(self, state):
        h1 = np.maximum(0, state @ self.W1 + self.b1)  # ReLU
        h2 = np.maximum(0, h1 @ self.W2 + self.b2)     # ReLU
        q_values = h2 @ self.W3 + self.b3               # 线性输出
        return q_values

    def choose_action(self, state, epsilon=0.05):
        if np.random.random() < epsilon:
            return np.random.randint(self.W3.shape[1])
        q_values = self.forward(state)
        return np.argmax(q_values)

# CartPole 输入：4 维连续状态 -> 2 个动作 Q 值
dqn = SimplifiedDQN(n_features=4, n_actions=2)
state = np.array([0.1, -0.2, 0.05, 0.15])
q_values = dqn.forward(state)
print(f"Q 值: {q_values}")
print(f"选择动作: {np.argmax(q_values)}")

# 经验回放缓冲区

import numpy as np
from collections import deque
import random

class ReplayBuffer:
    def __init__(self, capacity=100000):
        self.buffer = deque(maxlen=capacity)

    def push(self, state, action, reward, next_state, done):
        self.buffer.append((state, action, reward, next_state, done))

    def sample(self, batch_size=32):
        batch = random.sample(self.buffer, batch_size)
        states = np.array([t[0] for t in batch], dtype=np.float32)
        actions = np.array([t[1] for t in batch], dtype=np.int64)
        rewards = np.array([t[2] for t in batch], dtype=np.float32)
        next_states = np.array([t[3] for t in batch], dtype=np.float32)
        dones = np.array([t[4] for t in batch], dtype=np.float32)
        return states, actions, rewards, next_states, dones

    def __len__(self):
        return len(self.buffer)

# 使用示例
buffer = ReplayBuffer(capacity=50000)
for i in range(1000):
    s = np.random.randn(4)
    a = np.random.randint(2)
    r = np.random.randn()
    s_next = np.random.randn(4)
    done = np.random.random() < 0.02
    buffer.push(s, a, r, s_next, done)

print(f"缓冲区大小: {len(buffer)}")
states, actions, rewards, next_states, dones = buffer.sample(batch_size=32)
print(f"采样 batch 形状: states={states.shape}, rewards={rewards.shape}")

# DQN 训练循环：经验回放 + 梯度下降

class DQNTrainer:
    def __init__(self, n_features, n_actions, lr=0.001, gamma=0.99,
                 buffer_capacity=100000, batch_size=32, learn_start=1000):
        self.gamma = gamma
        self.batch_size = batch_size
        self.learn_start = learn_start
        self.buffer = ReplayBuffer(capacity=buffer_capacity)
        self.q_network = SimplifiedDQN(n_features, n_actions)
        self.optimizer_lr = lr

    def compute_td_targets(self, states, actions, rewards, next_states, dones):
        next_q = self.q_network.forward(next_states)  # 简化：应该用目标网络
        max_next_q = np.max(next_q, axis=1)
        targets = rewards + self.gamma * max_next_q * (1 - dones)
        return targets

    def train_step(self):
        if len(self.buffer) < self.learn_start:
            return None
        states, actions, rewards, next_states, dones = self.buffer.sample(self.batch_size)
        targets = self.compute_td_targets(states, actions, rewards, next_states, dones)
        # 模拟梯度更新
        current_q = self.q_network.forward(states)
        q_predicted = current_q[np.arange(self.batch_size), actions]
        td_error = targets - q_predicted
        return {
            "mean_td_error": np.mean(np.abs(td_error)),
            "max_td_error": np.max(np.abs(td_error)),
            "mean_q": np.mean(q_predicted),
        }

trainer = DQNTrainer(n_features=4, n_actions=2)
for _ in range(1000):
    trainer.buffer.push(np.random.randn(4), np.random.randint(2),
                        np.random.randn(), np.random.randn(4), False)
result = trainer.train_step()
print(f"TD 误差: mean={result['mean_td_error']:.3f}, max={result['max_td_error']:.3f}")

# 目标网络实现

class DQNWithTargetNetwork:
    def __init__(self, n_features, n_actions, target_update_freq=1000):
        self.q_network = SimplifiedDQN(n_features, n_actions)
        self.target_network = SimplifiedDQN(n_features, n_actions)
        # 初始化时同步权重
        self._copy_weights()
        self.target_update_freq = target_update_freq
        self.step_count = 0

    def _copy_weights(self):
        self.target_network.W1 = self.q_network.W1.copy()
        self.target_network.b1 = self.q_network.b1.copy()
        self.target_network.W2 = self.q_network.W2.copy()
        self.target_network.b2 = self.q_network.b2.copy()
        self.target_network.W3 = self.q_network.W3.copy()
        self.target_network.b3 = self.q_network.b3.copy()

    def soft_update(self, tau=0.005):
        for q_w, t_w in [
            (self.q_network.W1, self.target_network.W1),
            (self.q_network.W2, self.target_network.W2),
            (self.q_network.W3, self.target_network.W3),
        ]:
            t_w[:] = tau * q_w + (1 - tau) * t_w

    def compute_targets(self, rewards, next_states, dones, gamma=0.99):
        next_q = self.target_network.forward(next_states)
        max_next_q = np.max(next_q, axis=1)
        return rewards + gamma * max_next_q * (1 - dones)

    def step(self, states, actions, rewards, next_states, dones):
        self.step_count += 1
        targets = self.compute_targets(rewards, next_states, dones)
        current_q = self.q_network.forward(states)
        q_predicted = current_q[np.arange(len(actions)), actions]
        td_error = targets - q_predicted

        if self.step_count % self.target_update_freq == 0:
            self._copy_weights()
            print(f"Step {self.step_count}: 目标网络已更新")

        return td_error

# 硬更新 vs 软更新的对比

def compare_target_update_strategies():
    np.random.seed(42)
    q_weights = np.random.randn(10)
    target_weights_hard = q_weights.copy()
    target_weights_soft = q_weights.copy()
    tau = 0.005

    history_hard = []
    history_soft = []

    for step in range(5000):
        q_weights += np.random.randn(10) * 0.01
        if (step + 1) % 100 == 0:
            target_weights_hard = q_weights.copy()
        target_weights_soft = tau * q_weights + (1 - tau) * target_weights_soft
        history_hard.append(np.linalg.norm(target_weights_hard - q_weights))
        history_soft.append(np.linalg.norm(target_weights_soft - q_weights))

    print(f"硬更新 - 最后同步时误差: {history_hard[99]:.4f}, 同步前误差: {history_hard[199]:.4f}")
    print(f"软更新 - 平均误差: {np.mean(history_soft):.4f}")
    print(f"软更新 - 最大误差: {np.max(history_soft):.4f}")

compare_target_update_strategies()

# DQN vs Double DQN 的 TD 目标对比

def dqn_target(reward, next_state, gamma, q_network, target_network, done):
    """标准 DQN: 目标网络选择+评估"""
    if done:
        return reward
    next_q = target_network.forward(next_state)
    return reward + gamma * np.max(next_q)

def double_dqn_target(reward, next_state, gamma, q_network, target_network, done):
    """Double DQN: 当前网络选择, 目标网络评估"""
    if done:
        return reward
    # 当前 Q 网络选择最佳动作
    next_q_current = q_network.forward(next_state)
    best_action = np.argmax(next_q_current)
    # 目标网络评估该动作的价值
    next_q_target = target_network.forward(next_state)
    return reward + gamma * next_q_target[best_action]

# 高估偏差的数值演示
def demonstrate_overestimation():
    np.random.seed(42)
    n_actions = 10
    true_q = np.zeros(n_actions)  # 真实 Q 值全为 0
    n_estimates = 1000
    estimates = np.random.randn(n_estimates, n_actions) * 0.5
    max_values = np.max(estimates, axis=1)
    print(f"真实最大 Q 值: {np.max(true_q):.3f}")
    print(f"估计最大 Q 值的均值: {np.mean(max_values):.3f}")
    print(f"高估偏差: {np.mean(max_values) - np.max(true_q):.3f}")

demonstrate_overestimation()

# Double DQN 训练器完整实现

class DoubleDQNTrainer:
    def __init__(self, n_features, n_actions, lr=0.001, gamma=0.99,
                 target_update_freq=1000, buffer_capacity=100000):
        self.gamma = gamma
        self.target_update_freq = target_update_freq
        self.q_network = SimplifiedDQN(n_features, n_actions)
        self.target_network = SimplifiedDQN(n_features, n_actions)
        self._copy_weights()
        self.buffer = ReplayBuffer(capacity=buffer_capacity)
        self.step_count = 0

    def _copy_weights(self):
        for attr in ["W1", "b1", "W2", "b2", "W3", "b3"]:
            setattr(self.target_network, attr,
                    getattr(self.q_network, attr).copy())

    def train_step(self, batch_size=32):
        if len(self.buffer) < 1000:
            return None
        self.step_count += 1
        states, actions, rewards, next_states, dones = self.buffer.sample(batch_size)
        # Double DQN 核心：解耦选择与评估
        next_q_current = self.q_network.forward(next_states)
        best_actions = np.argmax(next_q_current, axis=1)
        next_q_target = self.target_network.forward(next_states)
        next_q_values = next_q_target[np.arange(batch_size), best_actions]
        targets = rewards + self.gamma * next_q_values * (1 - dones)
        current_q = self.q_network.forward(states)
        q_predicted = current_q[np.arange(batch_size), actions]
        td_error = targets - q_predicted

        if self.step_count % self.target_update_freq == 0:
            self._copy_weights()
        return np.mean(np.abs(td_error))

# Dueling DQN 网络架构

class DuelingDQN:
    def __init__(self, n_features, n_actions, hidden_size=128):
        # 共享特征提取层
        self.W_shared = np.random.randn(n_features, hidden_size) * np.sqrt(2.0 / n_features)
        self.b_shared = np.zeros(hidden_size)
        # 状态价值头 V(s)
        self.W_value = np.random.randn(hidden_size, 1) * 0.01
        self.b_value = np.zeros(1)
        # 动作优势头 A(s,a)
        self.W_advantage = np.random.randn(hidden_size, n_actions) * 0.01
        self.b_advantage = np.zeros(n_actions)
        self.n_actions = n_actions

    def forward(self, state):
        shared = np.maximum(0, state @ self.W_shared + self.b_shared)
        value = shared @ self.W_value + self.b_value
        advantage = shared @ self.W_advantage + self.b_advantage
        # Q(s,a) = V(s) + (A(s,a) - mean(A(s)))
        q_values = value + (advantage - np.mean(advantage))
        return q_values.squeeze()

    def choose_action(self, state, epsilon=0.05):
        if np.random.random() < epsilon:
            return np.random.randint(self.n_actions)
        q_values = self.forward(state)
        return np.argmax(q_values)

# 对比标准 DQN
dueling = DuelingDQN(n_features=4, n_actions=2)
state = np.array([0.1, -0.2, 0.05, 0.15])
print(f"Dueling Q 值: {dueling.forward(state)}")

# Dueling 架构的优势可视化

def visualize_dueling_decomposition():
    np.random.seed(42)
    dueling = DuelingDQN(n_features=4, n_actions=4, hidden_size=64)
    states = np.array([
        [0.0, 0.0, 0.0, 0.0],    # 中性状态
        [1.0, 0.0, 0.0, 0.0],    # 关键状态
        [-1.0, 0.0, 0.0, 0.0],   # 另一关键状态
    ])

    for i, state in enumerate(states):
        shared = np.maximum(0, state @ dueling.W_shared + dueling.b_shared)
        value = float(shared @ dueling.W_value + dueling.b_value)
        advantage = (shared @ dueling.W_advantage + dueling.b_advantage).squeeze()
        q_values = dueling.forward(state)

        print(f"状态 {i}:")
        print(f"  V(s) = {value:.4f}")
        print(f"  A(s,a) = {advantage}")
        print(f"  Q(s,a) = {q_values}")
        print(f"  动作优势范围: {np.max(advantage) - np.min(advantage):.4f}")
        print()

visualize_dueling_decomposition()

# CartPole + DQN 完整训练

import numpy as np
import gymnasium as gym

class CartPoleDQN:
    def __init__(self, n_features=4, n_actions=2, hidden_size=128,
                 lr=0.001, gamma=0.99, buffer_capacity=50000,
                 batch_size=32, target_update_freq=200, learn_start=1000):
        self.gamma = gamma
        self.batch_size = batch_size
        self.target_update_freq = target_update_freq
        self.learn_start = learn_start
        self.epsilon_start = 1.0
        self.epsilon_end = 0.01
        self.epsilon_decay = 0.995
        self.epsilon = self.epsilon_start
        self.step_count = 0

        self.q_network = SimplifiedDQN(n_features, n_actions, hidden_size)
        self.target_network = SimplifiedDQN(n_features, n_actions, hidden_size)
        self._copy_weights()
        self.buffer = ReplayBuffer(capacity=buffer_capacity)

    def _copy_weights(self):
        for attr in ["W1", "b1", "W2", "b2", "W3", "b3"]:
            setattr(self.target_network, attr,
                    getattr(self.q_network, attr).copy())

    def choose_action(self, state):
        if np.random.random() < self.epsilon:
            return np.random.randint(self.q_network.W3.shape[1])
        q = self.q_network.forward(state)
        return np.argmax(q)

    def store(self, s, a, r, s_next, done):
        self.buffer.push(s, a, r, s_next, done)

    def learn(self):
        if len(self.buffer) < self.learn_start:
            return
        self.step_count += 1
        states, actions, rewards, next_states, dones = self.buffer.sample(self.batch_size)
        next_q_target = self.target_network.forward(next_states)
        targets = rewards + self.gamma * np.max(next_q_target, axis=1) * (1 - dones)
        current_q = self.q_network.forward(states)
        td_error = targets - current_q[np.arange(self.batch_size), actions]
        if self.step_count % self.target_update_freq == 0:
            self._copy_weights()
        self.epsilon = max(self.epsilon_end, self.epsilon * self.epsilon_decay)

    def train(self, episodes=500):
        env = gym.make("CartPole-v1")
        history = []
        for ep in range(episodes):
            state, _ = env.reset()
            total = 0
            while True:
                action = self.choose_action(state)
                next_state, reward, term, trunc, _ = env.step(action)
                done = term or trunc
                self.store(state, action, reward, next_state, done)
                self.learn()
                total += reward
                state = next_state
                if done:
                    break
            history.append(total)
            if ep % 50 == 0 and ep > 0:
                avg = np.mean(history[-50:])
                print(f"Ep {ep:4d} | reward={total:4d} | avg_50={avg:6.1f} | eps={self.epsilon:.3f}")
            if np.mean(history[-100:]) >= 195:
                print(f"Solved at episode {ep}!")
                break
        env.close()
        return history

# Atari 训练框架（含帧预处理）

class AtariDQN:
    def __init__(self, n_actions=18, lr=0.00025, gamma=0.99,
                 buffer_capacity=1000000, batch_size=32):
        self.gamma = gamma
        self.batch_size = batch_size
        self.epsilon = 1.0
        self.epsilon_final = 0.01
        self.epsilon_decay_steps = 1000000
        self.step_count = 0
        self.buffer = ReplayBuffer(capacity=buffer_capacity)
        # 简化 CNN Q 网络（实际需要 PyTorch/JAX 实现）
        self.n_actions = n_actions

    def preprocess_frame(self, frame):
        gray = 0.2989 * frame[:,:,0] + 0.5870 * frame[:,:,1] + 0.1140 * frame[:,:,2]
        resized = gray[::2, ::2]  # 简单缩放
        normalized = resized / 255.0
        return normalized

    def get_epsilon(self):
        return max(self.epsilon_final,
                   1.0 - self.step_count / self.epsilon_decay_steps)

    def train_atari(self, env_name="BreakoutNoFrameskip-v4",
                    total_steps=10000000):
        env = gym.make(env_name)
        state, _ = env.reset()
        state = self.preprocess_frame(state)

        for step in range(total_steps):
            self.step_count += 1
            epsilon = self.get_epsilon()
            action = np.random.randint(self.n_actions) if np.random.random() < epsilon else 0

            next_frame, reward, term, trunc, _ = env.step(action)
            next_state = self.preprocess_frame(next_frame)
            done = term or trunc

            self.buffer.push(state.flatten(), action, reward, next_state.flatten(), done)
            state = next_state
            if done:
                state, _ = env.reset()
                state = self.preprocess_frame(state)

            if step % 100000 == 0:
                print(f"Step {step:,} | epsilon = {epsilon:.3f}")

        env.close()