🧠 DQN Fundamentals
🎯 What is DQN?
DQN (Deep Q-Network), introduced by DeepMind in 2013/2015, is a landmark algorithm that combined deep learning with reinforcement learning. It learned to play Atari games from raw pixels at superhuman level, sparking the modern deep RL revolution.
Why DQN Matters
End-to-end learning: From pixels to actions. Human-level: Superhuman on 49 Atari games. Deep RL: Bridged deep learning and RL. Foundation: Basis for modern algorithms. Practical: Works on many problems. Influential: Sparked the deep RL era.
📌 Deep Learning📌 Q-Learning📌 Atari📌 Experience Replay
# DQN — the big picture
import torch
import torch.nn as nn
class DQN(nn.Module):
"""Deep Q-Network: maps state to Q-values."""
def __init__(self, state_dim, action_dim, hidden=128):
super().__init__()
self.net = nn.Sequential(
nn.Linear(state_dim, hidden),
nn.ReLU(),
nn.Linear(hidden, hidden),
nn.ReLU(),
nn.Linear(hidden, action_dim)
)
def forward(self, state):
"""Return Q-values for all actions."""
return self.net(state)
# DQN's three key innovations:
# 1. Deep neural network as Q-function approximator
# 2. Experience replay buffer to break correlations
# 3. Target network to stabilize training
# The DQN loss:
# L(θ) = E[(r + γ max_a' Q(s', a'; θ⁻) - Q(s, a; θ))²]
#
# Where:
# - θ: online network parameters (updated every step)
# - θ⁻: target network parameters (updated periodically)
# - (s, a, r, s'): transition from replay buffer
# Comparison with classical RL:
comparison = {
'Tabular Q-Learning': {
'state_space': 'Discrete, small',
'function': 'Lookup table',
'generalization': 'None',
'examples': 'GridWorld, small MDPs'
},
'DQN': {
'state_space': 'Continuous, high-dim',
'function': 'Neural network',
'generalization': 'Yes',
'examples': 'Atari, control tasks'
}
}
for method, info in comparison.items():
print(f"\n{method}:")
for k, v in info.items():
print(f" {k}: {v}")📊 Q-Learning Review
🎯 The Foundation
Q-Learning (Watkins, 1989) is a model-free, off-policy RL algorithm. It learns the optimal action-value function Q*(s,a) directly from experience using the Bellman optimality equation.
The Bellman Equation
Q*(s,a) = E[r + γ max_a' Q*(s', a')]. Update rule: Q(s,a) ← Q(s,a) + α[r + γ max_a' Q(s',a') - Q(s,a)]. Off-policy: Learns optimal policy regardless of behavior. Bootstrapping: Updates using its own estimates. Model-free: No environment model needed.
📐 Bellman🎯 Off-policy🔄 TD Learning📊 Q-Values
# Tabular Q-Learning from scratch
import numpy as np
class TabularQLearning:
"""Classical Q-learning with lookup table."""
def __init__(self, n_states, n_actions, alpha=0.1, gamma=0.99, epsilon=0.1):
self.n_states = n_states
self.n_actions = n_actions
self.alpha = alpha # Learning rate
self.gamma = gamma # Discount factor
self.epsilon = epsilon # Exploration rate
# Q-table: [n_states, n_actions]
self.Q = np.zeros((n_states, n_actions))
def select_action(self, state):
"""ε-greedy action selection."""
if np.random.random() < self.epsilon:
return np.random.randint(self.n_actions)
return np.argmax(self.Q[state])
def update(self, state, action, reward, next_state, done):
"""Q-learning update."""
# Current Q-value
current_q = self.Q[state, action]
# TD target
if done:
target = reward
else:
target = reward + self.gamma * np.max(self.Q[next_state])
# TD error
td_error = target - current_q
# Update
self.Q[state, action] += self.alpha * td_error
return td_error
def train(self, env, episodes=1000):
"""Train on environment."""
rewards_history = []
for episode in range(episodes):
state = env.reset()
if isinstance(state, tuple):
state = state[0]
total_reward = 0
done = False
while not done:
action = self.select_action(state)
next_state, reward, done, truncated, _ = env.step(action)
if isinstance(done, bool) and truncated:
done = True
self.update(state, action, reward, next_state, done)
state = next_state
total_reward += reward
rewards_history.append(total_reward)
return rewards_history
# THE PROBLEM WITH TABULAR Q-LEARNING
print("=== Tabular Q-Learning Limitations ===")
print("""
For small state spaces: Works perfectly.
For large state spaces:
- GridWorld 10×10: 100 states — manageable
- GridWorld 100×100: 10,000 states — getting large
- Chess: ~10^47 states — impossible
- Go: ~10^170 states — impossible
- Atari (210×160 pixels): 256^(210×160) states — astronomically large
Problems:
1. Memory: Table can't fit
2. Sample efficiency: Must visit every state-action pair
3. Generalization: No transfer between similar states
4. Continuous states: Infinite states
""")
# THE SOLUTION: FUNCTION APPROXIMATION
print("=== Solution: Function Approximation ===")
print("""
Instead of Q-table: Q: S × A → ℝ
Use a function: Q_θ: S × A → ℝ
The function is parameterized by θ (neural network weights).
Benefits:
✓ Generalization: Similar states → similar Q-values
✓ Memory: Fixed size regardless of state space
✓ Continuous: Handles any input
✓ Scaling: Works for complex problems
The challenge:
- Training instability
- Moving target problem
- Correlated samples
DQN solves these!
""")
# FROM TABULAR TO DEEP
print("=== Tabular to Deep ===")
print("""
Tabular Q-Learning:
Q(s, a) is a table entry
Update: Q(s,a) ← Q(s,a) + α[r + γ max Q(s',a') - Q(s,a)]
DQN:
Q(s, a; θ) is a neural network output
Update: θ ← θ - α ∇_θ L(θ)
Loss: L(θ) = E[(r + γ max Q(s',a'; θ⁻) - Q(s,a; θ))²]
Key changes:
1. Replace table with neural network
2. Replace direct update with gradient descent
3. Add experience replay
4. Add target network
""")
# KEY CONCEPTS
print("=== Key RL Concepts in DQN ===")
print("""
Off-policy:
- Behavior policy: ε-greedy (exploration)
- Target policy: greedy (exploitation)
- Can learn from old data
Bootstrapping:
- Update using Q-estimates
- Not just raw returns
- Faster learning
TD Learning:
- Temporal difference
- Learn from incomplete episodes
- Balance bias-variance
Model-free:
- No environment model
- Learn from experience
- Versatile
""")🎯 Function Approximation
🎯 Neural Networks as Q-Functions
Instead of a Q-table, DQN uses a neural network Q(s,a;θ) to approximate the Q-function. This allows generalization across similar states and handles high-dimensional inputs like images.
Network Design
Input: State (or state-action pair). Output: Q-value per action. Architecture: MLP for low-dim, CNN for images. Loss: MSE between predicted and target Q. Optimization: Gradient descent (Adam, RMSprop). Key: One forward pass gives all Q-values.
🧠 Neural Net🖼️ CNN📊 Q-values⚡ Efficient
# Function approximation with neural networks
import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np
# 1. MLP FOR LOW-DIM STATE
class MLP_QNetwork(nn.Module):
"""Simple MLP for vector observations."""
def __init__(self, state_dim, action_dim, hidden=128):
super().__init__()
self.net = nn.Sequential(
nn.Linear(state_dim, hidden),
nn.ReLU(),
nn.Linear(hidden, hidden),
nn.ReLU(),
nn.Linear(hidden, action_dim)
)
def forward(self, state):
return self.net(state)
# 2. CNN FOR IMAGE INPUT
class CNN_QNetwork(nn.Module):
"""CNN for Atari-style image input."""
def __init__(self, n_frames, action_dim, hidden=512):
super().__init__()
# Convolutional feature extractor
self.conv = nn.Sequential(
nn.Conv2d(n_frames, 32, kernel_size=8, stride=4),
nn.ReLU(),
nn.Conv2d(32, 64, kernel_size=4, stride=2),
nn.ReLU(),
nn.Conv2d(64, 64, kernel_size=3, stride=1),
nn.ReLU(),
)
# Compute conv output size
# Input: (batch, n_frames, 84, 84)
# After convs: (batch, 64, 7, 7)
self.fc = nn.Sequential(
nn.Linear(64 * 7 * 7, hidden),
nn.ReLU(),
nn.Linear(hidden, action_dim)
)
def forward(self, state):
"""state: (batch, n_frames, 84, 84)"""
features = self.conv(state)
features = features.view(features.size(0), -1)
return self.fc(features)
# 3. DUELING ARCHITECTURE (preview)
class DuelingQNetwork(nn.Module):
"""Dueling architecture: separate V and A streams."""
def __init__(self, state_dim, action_dim, hidden=128):
super().__init__()
# Shared feature extractor
self.features = nn.Sequential(
nn.Linear(state_dim, hidden),
nn.ReLU(),
nn.Linear(hidden, hidden),
nn.ReLU()
)
# Value stream: V(s)
self.value_stream = nn.Sequential(
nn.Linear(hidden, hidden),
nn.ReLU(),
nn.Linear(hidden, 1)
)
# Advantage stream: A(s, a)
self.advantage_stream = nn.Sequential(
nn.Linear(hidden, hidden),
nn.ReLU(),
nn.Linear(hidden, action_dim)
)
def forward(self, state):
features = self.features(state)
value = self.value_stream(features)
advantage = self.advantage_stream(features)
# Combine: Q(s,a) = V(s) + A(s,a) - mean_a A(s,a)
q_values = value + advantage - advantage.mean(dim=-1, keepdim=True)
return q_values
# 4. ARCHITECTURE COMPARISON
print("=== Network Architecture Guide ===")
print("""
Input Type Architecture Example
---------- ------------ -------
Vector (< 100) MLP CartPole, LunarLander
Vector (100-1000) MLP with more layers Custom control
Image CNN Atari, video
Sequence LSTM/GRU Partially observable
Graph GNN Molecules, traffic
Multi-modal Transformer Language + vision
Layer sizes:
- Small tasks: 64-128 hidden units
- Medium: 128-256
- Large: 256-512
- Very large: 512-1024
Activation:
- ReLU: Standard, fast
- Tanh: Bounded, stable
- LeakyReLU: Avoid dead neurons
- ELU: Smooth, no dead neurons
Output:
- One head per action (standard)
- Dueling: separate V and A heads
- Distributional: distribution over returns
""")
# 5. EXAMPLE: CartPole NETWORK
print("=== Example: CartPole DQN ===")
class CartPoleDQN(nn.Module):
def __init__(self, state_dim=4, action_dim=2):
super().__init__()
self.net = nn.Sequential(
nn.Linear(state_dim, 128),
nn.ReLU(),
nn.Linear(128, 128),
nn.ReLU(),
nn.Linear(128, action_dim)
)
def forward(self, state):
return self.net(state)
# Create network
model = CartPoleDQN()
state = torch.randn(1, 4)
q_values = model(state)
print(f"State: {state.numpy()}")
print(f"Q-values: {q_values.detach().numpy()}")
print(f"Best action: {q_values.argmax().item()}")
# 6. NUMBER OF PARAMETERS
print(f"\nCartPole DQN parameters: {sum(p.numel() for p in model.parameters())}")
# 7. TARGET OUTPUT
print("\n=== Network Output Interpretation ===")
print("""
For state s, network outputs Q(s, a) for all a:
Q(s, 0) = Expected return if we take action 0
Q(s, 1) = Expected return if we take action 1
...
Policy: π(s) = argmax_a Q(s, a)
Q-values are in the same units as rewards.
Higher Q = better action.
Important: Q-values are NOT probabilities!
They can be any real number (usually related to reward scale).
""")
# 8. INITIALIZATION
print("=== Weight Initialization ===")
print("""
Good initialization matters for training stability.
Common choices:
1. Xavier/Glorot: Good for tanh
- Used in many DQN implementations
2. He/Kaiming: Good for ReLU
- Better for deep networks
3. Orthogonal: Stable training
- Used in some implementations
4. Small random: Safe default
- Ensures similar initial Q-values
DQN original paper:
- Used small random initialization
- Uniform [-0.05, 0.05]
Modern implementations:
- He initialization for ReLU
- Orthogonal for RL
Rule of thumb:
- Small initial Q-values → stable training
- Large → can cause divergence
""")
# 9. WHY THIS WORKS
print("=== Why Function Approximation Works ===")
print("""
1. GENERALIZATION
Similar states → similar Q-values
Can handle unseen states
2. COMPRESSION
Fixed parameters vs. huge table
Efficient memory usage
3. FEATURE LEARNING
Learns useful representations
End-to-end training
4. SCALING
Works with any input size
Handles images, sequences, etc.
5. SMOOTHNESS
Continuous function
No lookup needed
""")💾 Experience Replay
🎯 Breaking Correlations
Experience Replay is DQN's first key innovation. It stores transitions (s, a, r, s', done) in a buffer and samples random minibatches for training. This breaks temporal correlations and improves sample efficiency.
Why Experience Replay?
Break correlations: Sequential data is highly correlated. Sample efficiency: Reuse each transition multiple times. Smooth distribution: Random samples approximate i.i.d. Stability: Less variance in gradients. Off-policy: Learn from old data. Buffer size: Balance memory and diversity.
💾 Buffer🎲 Random Sample📊 I.I.D.🔄 Reuse
# Experience Replay Buffer
import numpy as np
import torch
from collections import deque
import random
class ReplayBuffer:
"""Fixed-size experience replay buffer."""
def __init__(self, capacity=100000):
self.buffer = deque(maxlen=capacity)
self.capacity = capacity
def push(self, state, action, reward, next_state, done):
"""Add transition to buffer."""
self.buffer.append((state, action, reward, next_state, done))
def sample(self, batch_size):
"""Sample random minibatch."""
batch = random.sample(self.buffer, batch_size)
states, actions, rewards, next_states, dones = zip(*batch)
return (
np.array(states, dtype=np.float32),
np.array(actions, dtype=np.int64),
np.array(rewards, dtype=np.float32),
np.array(next_states, dtype=np.float32),
np.array(dones, dtype=np.float32)
)
def __len__(self):
return len(self.buffer)
# TORCH VERSION (for GPU training)
class TorchReplayBuffer:
"""Efficient replay buffer for PyTorch."""
def __init__(self, capacity, state_dim, device='cpu'):
self.capacity = capacity
self.device = device
# Pre-allocate arrays
self.states = np.zeros((capacity, state_dim), dtype=np.float32)
self.next_states = np.zeros((capacity, state_dim), dtype=np.float32)
self.actions = np.zeros(capacity, dtype=np.int64)
self.rewards = np.zeros(capacity, dtype=np.float32)
self.dones = np.zeros(capacity, dtype=np.float32)
self.position = 0
self.size = 0
def push(self, state, action, reward, next_state, done):
"""Add transition."""
idx = self.position
self.states[idx] = state
self.actions[idx] = action
self.rewards[idx] = reward
self.next_states[idx] = next_state
self.dones[idx] = done
self.position = (self.position + 1) % self.capacity
self.size = min(self.size + 1, self.capacity)
def sample(self, batch_size):
"""Sample random minibatch."""
indices = np.random.randint(0, self.size, size=batch_size)
return (
torch.FloatTensor(self.states[indices]).to(self.device),
torch.LongTensor(self.actions[indices]).to(self.device),
torch.FloatTensor(self.rewards[indices]).to(self.device),
torch.FloatTensor(self.next_states[indices]).to(self.device),
torch.FloatTensor(self.dones[indices]).to(self.device)
)
def __len__(self):
return self.size
# PRIORITIZED EXPERIENCE REPLAY (improvement)
class PrioritizedReplayBuffer:
"""Sample important transitions more frequently."""
def __init__(self, capacity, alpha=0.6, beta=0.4):
self.capacity = capacity
self.alpha = alpha # Priority exponent
self.beta = beta # Importance sampling exponent
self.buffer = []
self.priorities = np.zeros(capacity, dtype=np.float32)
self.position = 0
self.size = 0
def push(self, state, action, reward, next_state, done, td_error=None):
"""Add transition with priority."""
# Max priority for new experiences
priority = self.priorities[:self.size].max() if self.size > 0 else 1.0
if len(self.buffer) < self.capacity:
self.buffer.append((state, action, reward, next_state, done))
else:
self.buffer[self.position] = (state, action, reward, next_state, done)
self.priorities[self.position] = priority
self.position = (self.position + 1) % self.capacity
self.size = min(self.size + 1, self.capacity)
def sample(self, batch_size):
"""Sample based on priorities."""
# Compute sampling probabilities
priorities = self.priorities[:self.size]
probs = priorities ** self.alpha
probs /= probs.sum()
# Sample indices
indices = np.random.choice(self.size, batch_size, p=probs)
# Importance sampling weights
weights = (self.size * probs[indices]) ** (-self.beta)
weights /= weights.max()
# Get batch
batch = [self.buffer[idx] for idx in indices]
states, actions, rewards, next_states, dones = zip(*batch)
return (
np.array(states, dtype=np.float32),
np.array(actions, dtype=np.int64),
np.array(rewards, dtype=np.float32),
np.array(next_states, dtype=np.float32),
np.array(dones, dtype=np.float32),
indices,
np.array(weights, dtype=np.float32)
)
def update_priorities(self, indices, td_errors):
"""Update priorities based on TD errors."""
for idx, td_error in zip(indices, td_errors):
self.priorities[idx] = abs(td_error) + 1e-6
# USAGE EXAMPLE
print("=== Experience Replay Usage ===")
buffer = ReplayBuffer(capacity=10000)
# During training:
# 1. Interact with environment
# 2. Store transition
state = np.random.randn(4)
action = 0
reward = 1.0
next_state = np.random.randn(4)
done = False
buffer.push(state, action, reward, next_state, done)
# 3. Sample random minibatch
if len(buffer) >= 32:
states, actions, rewards, next_states, dones = buffer.sample(32)
print(f"Sampled batch shape: {states.shape}")
# WHY IT WORKS
print("\n=== Why Experience Replay Works ===")
print("""
Without replay:
Train on sequential data
Highly correlated samples
Gradients have high variance
Can diverge
With replay:
Train on random minibatches
Samples ~ i.i.d.
Lower variance gradients
Stable training
Benefits:
1. Sample efficiency: 1 transition → many updates
2. Stability: Decorrelated samples
3. Generalization: Diverse experience
4. Off-policy: Learn from old data
5. Break feedback loops: Not just recent data
""")
# BUFFER SIZE CONSIDERATIONS
print("=== Buffer Size ===")
print("""
Too small:
✗ Samples too correlated
✗ Doesn't help with decorrelation
✗ Less sample reuse
Too large:
✗ Old policies contribute
✗ Slower sampling (memory)
✗ Off-policyness grows
Typical sizes:
- CartPole: 10,000 - 100,000
- Atari: 1,000,000
- Complex tasks: 1,000,000+
- Continuous control: 100,000 - 1,000,000
Rule of thumb:
10-100x larger than the longest episode
Enough to hold diverse experience
""")
# FRAME STACKING
print("=== Frame Stacking ===")
print("""
For Atari, single frames don't provide motion info.
Solution: Stack 4 consecutive frames.
State: (4, 84, 84) — channels = frames
This gives the network:
- Velocity information
- Motion direction
- Temporal context
Implementation:
frame_buffer = deque(maxlen=4)
state = np.stack(frame_buffer, axis=0)
""")
# N-STEP RETURNS
print("=== N-Step Returns ===")
print("""
Instead of 1-step TD:
target = r + γ max Q(s', a')
Use n-step:
target = r + γr' + γ²r'' + ... + γ^n max Q(s_{t+n}, a')
Benefits:
- Faster propagation of rewards
- Better credit assignment
- Trade-off with bias
DQN implementations:
- Standard DQN: 1-step
- Rainbow: 3-step (n=3)
- Ape-X: variable n-step
For most tasks:
- n=3 or n=5 works well
""")🎯 Target Network
🎯 Stabilizing the Moving Target
Target Network is DQN's second key innovation. It's a copy of the Q-network that's held fixed for a period, providing stable targets during training. This prevents the "moving target" problem.
The Moving Target Problem
Problem: Target uses Q(s',a';θ) — same θ being updated. Result: Chasing a moving target — unstable. Solution: Use separate network Q(s',a';θ⁻). Update: Copy θ to θ⁻ every C steps. Effect: Stable targets, better convergence. Trade-off: Slower adaptation but more stable.
🎯 Target🔒 Frozen🔄 Periodic📊 Stable
# Target Network — the key to DQN stability
import torch
import torch.nn as nn
import numpy as np
class DQNAgentWithTarget:
"""DQN agent with target network."""
def __init__(
self,
state_dim,
action_dim,
hidden=128,
lr=1e-3,
gamma=0.99,
tau=0.005, # Soft update rate
target_update=1000, # Hard update frequency
use_soft_update=False,
device='cpu'
):
self.device = device
self.gamma = gamma
self.tau = tau
self.target_update = target_update
self.use_soft_update = use_soft_update
# Online network: Q(s, a; θ)
self.online_net = self._build_network(state_dim, action_dim, hidden).to(device)
# Target network: Q(s, a; θ⁻)
self.target_net = self._build_network(state_dim, action_dim, hidden).to(device)
# Initialize target = online
self.target_net.load_state_dict(self.online_net.state_dict())
self.target_net.eval() # Never train target directly
# Optimizer
self.optimizer = torch.optim.Adam(self.online_net.parameters(), lr=lr)
self.steps = 0
def _build_network(self, state_dim, action_dim, hidden):
return nn.Sequential(
nn.Linear(state_dim, hidden),
nn.ReLU(),
nn.Linear(hidden, hidden),
nn.ReLU(),
nn.Linear(hidden, action_dim)
)
def select_action(self, state, epsilon=0.1):
"""ε-greedy action selection."""
if np.random.random() < epsilon:
return np.random.randint(self.online_net[-1].out_features)
with torch.no_grad():
state = torch.FloatTensor(state).unsqueeze(0).to(self.device)
q_values = self.online_net(state)
return q_values.argmax().item()
def compute_loss(self, batch):
"""Compute DQN loss with target network."""
states, actions, rewards, next_states, dones = batch
# Current Q-values: Q(s, a; θ)
q_values = self.online_net(states)
q_values = q_values.gather(1, actions.unsqueeze(1)).squeeze(1)
# Target Q-values: r + γ max_a' Q(s', a'; θ⁻)
with torch.no_grad():
next_q_values = self.target_net(next_states)
max_next_q = next_q_values.max(dim=1)[0]
targets = rewards + self.gamma * max_next_q * (1 - dones)
# Loss
loss = nn.MSELoss()(q_values, targets)
return loss
def update(self, batch):
"""One training step."""
loss = self.compute_loss(batch)
self.optimizer.zero_grad()
loss.backward()
# Gradient clipping for stability
torch.nn.utils.clip_grad_norm_(self.online_net.parameters(), 10.0)
self.optimizer.step()
# Update target network
self.steps += 1
if self.use_soft_update:
# Soft update: θ⁻ ← τθ + (1-τ)θ⁻
self.soft_update()
else:
# Hard update: θ⁻ ← θ every C steps
if self.steps % self.target_update == 0:
self.hard_update()
return loss.item()
def soft_update(self):
"""Polyak averaging: θ⁻ ← τθ + (1-τ)θ⁻"""
for target_param, online_param in zip(
self.target_net.parameters(), self.online_net.parameters()
):
target_param.data.copy_(
self.tau * online_param.data +
(1 - self.tau) * target_param.data
)
def hard_update(self):
"""Copy: θ⁻ ← θ"""
self.target_net.load_state_dict(self.online_net.state_dict())
# WHY TARGET NETWORK WORKS
print("=== The Moving Target Problem ===")
print("""
WITHOUT target network:
Target: r + γ max Q(s', a'; θ)
Update: θ ← θ - α ∇ (Q(s,a;θ) - target)²
Problem: Target depends on θ
When θ changes, target changes
Like chasing your own tail
Result: Unstable, can diverge
WITH target network:
Target: r + γ max Q(s', a'; θ⁻)
Update: θ ← θ - α ∇ (Q(s,a;θ) - target)²
Update: θ⁻ ← θ periodically
θ⁻ changes slowly or periodically
Target is stable during updates
Like supervised learning
Result: Stable, converges
""")
# UPDATE STRATEGIES
print("=== Update Strategies ===")
print("""
1. HARD UPDATE (original DQN):
Every C steps: θ⁻ ← θ
C = 1000-10000 steps typically
Pros: Simple, stable
Cons: Sudden changes, tuning C
2. SOFT UPDATE (Polyak averaging):
Every step: θ⁻ ← τθ + (1-τ)θ⁻
τ = 0.001-0.01 typically
Pros: Smooth, no sudden changes
Cons: More hyperparameters
3. EXPONENTIAL MOVING AVERAGE:
Similar to soft update
θ⁻ = (1-α)θ⁻ + αθ
Recommendations:
- Small tasks: Hard update (C=1000)
- Large tasks: Soft update (τ=0.005)
- Continuous control: Soft update
- Atari: Hard update (C=10000)
""")
# UPDATE FREQUENCY
print("=== Update Frequency ===")
print("""
Hard update frequency C:
Too small: Target changes too fast → unstable
Too large: Target too stale → slow learning
Typical values:
Atari: 10,000 steps
CartPole: 100-500 steps
General: 1,000 steps
Soft update rate τ:
Too small: Target too slow
Too large: Target too fast
Typical: 0.001 - 0.01
Rule of thumb:
Hard update: Every 1,000-10,000 steps
Soft update: Every step with τ=0.005
""")
# COMMON MISTAKES
print("=== Common Mistakes ===")
print("""
❌ Forgetting to freeze target network
Fix: Set target_net.eval() and use no_grad()
❌ Using same network for online and target
Fix: Create separate networks
❌ Not updating target network
Fix: Update every C steps or soft update
❌ Updating target with optimizer
Fix: Only copy parameters, don't optimize
❌ Initializing target randomly
Fix: target = online at start
❌ Wrong sign in target calculation
Fix: target = r + γ max Q(s', a') * (1 - done)
""")📉 Loss Function (Huber)
🎯 Stable Learning
DQN uses the Huber loss (smooth L1) instead of MSE for training. This makes learning more robust to outliers and large TD errors, which are common in RL.
Why Huber Loss?
MSE: Quadratic, sensitive to outliers. Huber: Linear for large errors, quadratic for small. Benefit: Clips large gradients. RL context: TD errors can be huge early. Gradient clipping: Built-in. Stability: Better convergence.
📉 MSE📊 Huber✂️ Clip🎯 Stable
# Loss functions for DQN
import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np
# 1. MEAN SQUARED ERROR (MSE)
def mse_loss(q_values, targets):
"""Standard MSE loss: (Q - target)²"""
return F.mse_loss(q_values, targets)
# 2. HUBER LOSS (Smooth L1) — DQN's choice
def huber_loss(q_values, targets, delta=1.0):
"""
Huber loss: quadratic for |x| < δ, linear for |x| > δ
L(x) = 0.5 * x² if |x| < δ
= δ * (|x| - 0.5*δ) otherwise
"""
return F.smooth_l1_loss(q_values, targets, beta=delta)
# 3. CLIPPED MSE (alternative)
def clipped_mse_loss(q_values, targets, clip_value=1.0):
"""Clip TD errors to [-c, c] before squaring."""
errors = targets - q_values
clipped = torch.clamp(errors, -clip_value, clip_value)
return (clipped ** 2).mean()
# COMPARISON
print("=== Loss Function Comparison ===")
print("""
MSE: L(x) = x²
✓ Smooth everywhere
✓ Simple
✗ Sensitive to outliers
✗ Large gradients
✗ Can diverge
Huber: L(x) = x² if |x|<δ, δ|x| otherwise
✓ Robust to outliers
✓ Bounded gradients
✓ Stable
✗ Slightly more complex
Clipped MSE: L(x) = clip(x)²
✓ Simple
✓ Bounded
✗ Non-smooth at clip point
""")
# VISUALIZE DIFFERENCES
errors = np.linspace(-5, 5, 100)
mse_losses = errors ** 2
huber_losses = np.where(
np.abs(errors) < 1,
0.5 * errors ** 2,
np.abs(errors) - 0.5
)
print("\nError values and losses:")
print(f"{'Error':<10} {'MSE':<12} {'Huber':<12}")
print("-" * 36)
for e in [-5, -2, -1, -0.5, 0, 0.5, 1, 2, 5]:
mse = e ** 2
huber = 0.5 * e**2 if abs(e) < 1 else abs(e) - 0.5
print(f"{e:<10} {mse:<12.4f} {huber:<12.4f}")
# WHY THIS MATTERS
print("\n=== Why This Matters in RL ===")
print("""
In RL:
- Rewards can be large
- TD errors start large
- Q-values are unknown scale
With MSE:
- Large TD errors → huge gradients
- Training unstable
- Can diverge
With Huber:
- Large TD errors → clipped gradients
- Stable training
- Still accurate for small errors
The clipping threshold δ:
- δ=1: Standard in DQN
- Smaller: More robust, less accurate
- Larger: Closer to MSE
""")
# IMPLEMENTATION IN DQN
class DQNWithHuberLoss(nn.Module):
"""DQN with Huber loss."""
def __init__(self, state_dim, action_dim, hidden=128):
super().__init__()
self.net = nn.Sequential(
nn.Linear(state_dim, hidden),
nn.ReLU(),
nn.Linear(hidden, hidden),
nn.ReLU(),
nn.Linear(hidden, action_dim)
)
def forward(self, state):
return self.net(state)
def compute_loss(self, states, actions, rewards, next_states, dones,
target_net, gamma=0.99, delta=1.0):
"""Compute Huber loss."""
# Current Q-values
q_values = self.forward(states)
q_values = q_values.gather(1, actions.unsqueeze(1)).squeeze(1)
# Target Q-values
with torch.no_grad():
next_q_values = target_net(next_states)
max_next_q = next_q_values.max(dim=1)[0]
targets = rewards + gamma * max_next_q * (1 - dones)
# Huber loss
loss = F.smooth_l1_loss(q_values, targets, beta=delta)
return loss
# DQN ORIGINAL LOSS (with specific implementation detail)
print("=== Original DQN Loss ===")
print("""
From Mnih et al. (2015):
L(θ) = E[(y - Q(s, a; θ))²]
where y = r + γ max_a' Q(s', a'; θ⁻)
They clip TD errors:
|y - Q(s, a; θ)| ≤ 1
This is equivalent to Huber loss with δ=1.
Actually, they used a slightly different formulation:
- Gradient clipping at [-1, 1]
- Prevents exploding gradients
- Stable training
Modern implementations:
- Use smooth_l1_loss (Huber)
- Or clip_grad_norm_ for gradients
""")
# PRACTICAL TIPS
print("=== Practical Tips ===")
print("""
1. Use Huber loss (smooth_l1) with δ=1
- Standard for DQN
- Robust and stable
2. Clip gradients as well
- torch.nn.utils.clip_grad_norm_(params, 10.0)
- Prevents rare large updates
3. Normalize rewards
- Divide by running std
- Scale to ~[-1, 1]
- Much more stable
4. Watch for Q-value explosion
- Q-values should be bounded
- If growing → problem
- Check learning rate
5. Consider reward scaling
- Similar to normalization
- Per-environment
- Important for multi-task
""")
# REWARD NORMALIZATION
print("=== Reward Normalization ===")
class RewardNormalizer:
"""Normalize rewards using running statistics."""
def __init__(self, epsilon=1e-8, clip=10.0):
self.mean = 0.0
self.var = 1.0
self.count = 1e-4
self.epsilon = epsilon
self.clip = clip
def __call__(self, reward):
"""Normalize a single reward."""
# Update running stats
self.count += 1
delta = reward - self.mean
self.mean += delta / self.count
self.var += delta * (reward - self.mean)
# Normalize
std = np.sqrt(self.var / self.count) + self.epsilon
normalized = (reward - self.mean) / std
# Clip for safety
return np.clip(normalized, -self.clip, self.clip)
# Usage
normalizer = RewardNormalizer()
rewards = [1, 2, 3, 100, 5, 6, 7, 8, 9, 10]
print("\nReward normalization:")
for r in rewards:
normalized = normalizer(r)
print(f" {r} → {normalized:.4f}")
# ADVANCED: DISTRIBUTIONAL DQN
print("\n=== Distributional DQN (C51) ===")
print("""
Instead of single Q-value, learn distribution:
Q(s,a) = E[Z(s,a)]
where Z(s,a) is a random variable.
C51 (Categorical DQN):
- 51 atoms for distribution
- Bellman update on distribution
- Cross-entropy loss
Benefits:
- Better learning signal
- Handles uncertainty
- State-of-the-art performance
Loss: KL divergence between predicted and target distributions
""")🏗️ Network Architecture
🎯 Designing the Q-Network
The architecture of the Q-network is crucial for DQN's success. For low-dimensional states, MLPs work well. For images (like Atari), CNNs are essential. Modern improvements include dueling and distributional heads.
Architecture Choices
MLP: For vector states (CartPole, MuJoCo). CNN: For image states (Atari, pixels). Dueling: Separate V and A. Distributional: Learn distribution. Recurrent: For POMDPs. Transformer: For sequences.
🧠 MLP🖼️ CNN🎭 Dueling📊 Distributional
# Network architectures for DQN
import torch
import torch.nn as nn
import torch.nn.functional as F
# 1. MLP FOR LOW-DIM STATE
class MLP_DQN(nn.Module):
"""Standard MLP for vector observations."""
def __init__(self, state_dim, action_dim, hidden=128):
super().__init__()
self.net = nn.Sequential(
nn.Linear(state_dim, hidden),
nn.ReLU(),
nn.Linear(hidden, hidden),
nn.ReLU(),
nn.Linear(hidden, action_dim)
)
def forward(self, state):
return self.net(state)
# 2. CNN FOR IMAGE STATE (Atari)
class CNN_DQN(nn.Module):
"""
DQN CNN for Atari.
Input: 4 frames of 84×84 grayscale
"""
def __init__(self, n_frames=4, action_dim=18, hidden=512):
super().__init__()
# Feature extractor
self.conv = nn.Sequential(
nn.Conv2d(n_frames, 32, kernel_size=8, stride=4),
nn.ReLU(),
nn.Conv2d(32, 64, kernel_size=4, stride=2),
nn.ReLU(),
nn.Conv2d(64, 64, kernel_size=3, stride=1),
nn.ReLU()
)
# Compute conv output size: 64 × 7 × 7 = 3136
self.fc = nn.Sequential(
nn.Linear(64 * 7 * 7, hidden),
nn.ReLU(),
nn.Linear(hidden, action_dim)
)
def forward(self, state):
"""state: (batch, n_frames, 84, 84)"""
features = self.conv(state)
features = features.flatten(start_dim=1)
return self.fc(features)
# 3. DUELING ARCHITECTURE
class DuelingDQN(nn.Module):
"""
Dueling DQN: Separate V(s) and A(s,a) streams.
Q(s,a) = V(s) + A(s,a) - mean_a A(s,a)
"""
def __init__(self, state_dim, action_dim, hidden=128):
super().__init__()
# Shared feature extractor
self.features = nn.Sequential(
nn.Linear(state_dim, hidden),
nn.ReLU()
)
# Value stream: V(s)
self.value_stream = nn.Sequential(
nn.Linear(hidden, hidden),
nn.ReLU(),
nn.Linear(hidden, 1)
)
# Advantage stream: A(s,a)
self.advantage_stream = nn.Sequential(
nn.Linear(hidden, hidden),
nn.ReLU(),
nn.Linear(hidden, action_dim)
)
def forward(self, state):
features = self.features(state)
value = self.value_stream(features)
advantage = self.advantage_stream(features)
# Combine streams
q_values = value + advantage - advantage.mean(dim=-1, keepdim=True)
return q_values
# 4. NOISY NETWORK (for exploration)
class NoisyLinear(nn.Module):
"""Noisy linear layer for exploration."""
def __init__(self, in_features, out_features, sigma_init=0.5):
super().__init__()
self.in_features = in_features
self.out_features = out_features
self.sigma_init = sigma_init
# Learnable parameters
self.weight_mu = nn.Parameter(torch.empty(out_features, in_features))
self.weight_sigma = nn.Parameter(torch.empty(out_features, in_features))
self.bias_mu = nn.Parameter(torch.empty(out_features))
self.bias_sigma = nn.Parameter(torch.empty(out_features))
self.reset_parameters()
def reset_parameters(self):
mu_range = 1 / (self.in_features ** 0.5)
self.weight_mu.data.uniform_(-mu_range, mu_range)
self.weight_sigma.data.fill_(self.sigma_init / (self.in_features ** 0.5))
self.bias_mu.data.uniform_(-mu_range, mu_range)
self.bias_sigma.data.fill_(self.sigma_init / (self.out_features ** 0.5))
def forward(self, x):
# Sample noise
if self.training:
weight_epsilon = torch.randn_like(self.weight_sigma)
bias_epsilon = torch.randn_like(self.bias_sigma)
weight = self.weight_mu + self.weight_sigma * weight_epsilon
bias = self.bias_mu + self.bias_sigma * bias_epsilon
else:
weight = self.weight_mu
bias = self.bias_mu
return F.linear(x, weight, bias)
# 5. RECURRENT DQN (for POMDPs)
class RecurrentDQN(nn.Module):
"""DQN with LSTM for partial observability."""
def __init__(self, state_dim, action_dim, hidden=128):
super().__init__()
self.features = nn.Sequential(
nn.Linear(state_dim, hidden),
nn.ReLU()
)
self.lstm = nn.LSTM(hidden, hidden, batch_first=True)
self.head = nn.Sequential(
nn.Linear(hidden, hidden),
nn.ReLU(),
nn.Linear(hidden, action_dim)
)
def forward(self, state, hidden=None):
"""
state: (batch, seq_len, state_dim) for sequences
"""
features = self.features(state)
lstm_out, hidden = self.lstm(features, hidden)
q_values = self.head(lstm_out)
return q_values, hidden
# 6. DISTRIBUTIONAL DQN (C51)
class C51_DQN(nn.Module):
"""
Distributional DQN with 51 atoms.
Instead of Q(s,a), outputs distribution over returns.
"""
def __init__(self, state_dim, action_dim, num_atoms=51, v_min=-10, v_max=10):
super().__init__()
self.action_dim = action_dim
self.num_atoms = num_atoms
self.v_min = v_min
self.v_max = v_max
# Support: discrete values
self.register_buffer(
'support',
torch.linspace(v_min, v_max, num_atoms)
)
self.net = nn.Sequential(
nn.Linear(state_dim, 128),
nn.ReLU(),
nn.Linear(128, 128),
nn.ReLU(),
nn.Linear(128, action_dim * num_atoms)
)
def forward(self, state):
"""Return distribution over returns for each action."""
logits = self.net(state)
logits = logits.view(-1, self.action_dim, self.num_atoms)
return F.softmax(logits, dim=-1)
def get_q_values(self, state):
"""Compute expected Q-values from distribution."""
probs = self.forward(state)
q_values = (probs * self.support).sum(dim=-1)
return q_values
# COMPARISON TABLE
print("=== Architecture Comparison ===")
print("""
Architecture Input Type Use Case Advantages
------------ ---------- -------- ----------
MLP Vector Control tasks Simple, fast
CNN Image Atari, pixels Image features
Dueling Any General improvement Better credit assignment
Noisy Any Exploration Learned exploration
Recurrent Sequence POMDPs Memory
C51 Any General improvement Uncertainty
Transformer Sequence Complex tasks Attention
When to use what:
- Vector state: MLP with 2-3 hidden layers
- Image state: CNN (like original DQN)
- Partial observability: Recurrent (LSTM/GRU)
- Better performance: Dueling + C51 + Noisy
- Best default: Dueling DQN with some tricks
""")
# MODERN DQN ARCHITECTURE
print("=== Modern DQN Architecture ===")
print("""
For most tasks, this works well:
Input (state_dim)
↓
Linear(state_dim, 256) + ReLU
↓
Linear(256, 256) + ReLU
↓
┌─────────────────────┐
│ Dueling Head │
│ Value: Linear(256, 1)
│ Advantage: Linear(256, action_dim)
│ Q = V + A - mean(A)
└─────────────────────┘
↓
Output (action_dim Q-values)
Improvements to add:
1. Layer normalization after each layer
2. NoisyLinear for exploration
3. C51 for distributional RL
4. Prioritized experience replay
5. N-step returns
""")
# PRACTICAL TIPS
print("=== Architecture Tips ===")
print("""
1. START SIMPLE
- 2-3 hidden layers
- 128-256 units each
- ReLU activation
2. FOR IMAGES
- CNN with 3 conv layers
- Followed by 1-2 FC layers
- Standard Atari architecture works
3. FOR CONTINUOUS CONTROL
- Deeper networks
- Layer normalization
- Careful initialization
4. NORMALIZATION
- Layer norm helps with distribution shift
- Batch norm can cause issues (batch correlations)
- Layer norm is safer
5. ACTIVATION FUNCTIONS
- ReLU: Standard, fast
- Tanh: Older, still works
- LeakyReLU: Avoid dead neurons
6. INITIALIZATION
- He/Kaiming for ReLU
- Small random for stability
7. OUTPUT
- One Q-value per action
- No activation on output (unbounded)
- Initialize to small values
""")🎲 Exploration (ε-greedy)
🎯 Balancing Exploration and Exploitation
Exploration is crucial for RL. DQN uses ε-greedy: with probability ε, take a random action; otherwise, take the greedy action. ε typically decays over time.
Exploration Strategies
ε-greedy: Random action with probability ε. Linear decay: ε from 1.0 to 0.1. Exponential: Faster decay initially. Adaptive: Adjust based on performance. Noisy nets: Learned exploration. Boltzmann: Sample from softmax over Q.
🎲 ε-greedy📉 Decay🔊 Noisy📊 Softmax
# Exploration strategies for DQN
import numpy as np
import torch
# 1. ε-GREEDY EXPLORATION
class EpsilonGreedy:
"""ε-greedy exploration with decay."""
def __init__(self, start=1.0, end=0.01, decay=0.995, decay_type='exponential'):
self.epsilon = start
self.start = start
self.end = end
self.decay = decay
self.decay_type = decay_type
def select_action(self, q_values, n_actions):
"""Select action using ε-greedy."""
if np.random.random() < self.epsilon:
return np.random.randint(n_actions)
return np.argmax(q_values)
def update(self):
"""Decay epsilon."""
if self.decay_type == 'exponential':
self.epsilon = max(self.end, self.epsilon * self.decay)
elif self.decay_type == 'linear':
self.epsilon = max(self.end, self.epsilon - self.decay)
def set_epsilon(self, value):
self.epsilon = value
# 2. EXPONENTIAL DECAY
def exponential_decay(step, start=1.0, end=0.01, decay_steps=10000):
"""Exponential decay over steps."""
if step >= decay_steps:
return end
fraction = step / decay_steps
return start * (end / start) ** fraction
# 3. LINEAR DECAY
def linear_decay(step, start=1.0, end=0.01, decay_steps=100000):
"""Linear decay over steps."""
if step >= decay_steps:
return end
return start - (start - end) * (step / decay_steps)
# 4. ADAPTIVE ε-GREEDY
class AdaptiveEpsilon:
"""
Adjust epsilon based on performance.
Increase if performance drops.
"""
def __init__(self, initial=1.0, min_eps=0.01, max_eps=1.0):
self.epsilon = initial
self.min_eps = min_eps
self.max_eps = max_eps
self.recent_rewards = []
def update(self, reward):
self.recent_rewards.append(reward)
if len(self.recent_rewards) > 100:
self.recent_rewards.pop(0)
# Simple heuristic: increase exploration if rewards are low
if len(self.recent_rewards) >= 50:
avg_reward = np.mean(self.recent_rewards[-50:])
if avg_reward < 0:
self.epsilon = min(self.max_eps, self.epsilon + 0.01)
else:
self.epsilon = max(self.min_eps, self.epsilon - 0.001)
# 5. BOLTZMANN EXPLORATION (softmax)
def boltzmann_exploration(q_values, temperature=1.0):
"""
Sample action from softmax over Q-values.
P(a) = exp(Q(s,a)/τ) / Σ exp(Q(s,a')/τ)
"""
q_values = np.array(q_values)
# Subtract max for numerical stability
q_values = q_values - q_values.max()
# Softmax with temperature
probs = np.exp(q_values / temperature)
probs /= probs.sum()
return np.random.choice(len(q_values), p=probs)
# 6. ENTROPY REGULARIZATION
def entropy_bonus(q_values, temperature=1.0):
"""
Add entropy bonus to encourage exploration.
L = -Q(s,a) - β * H(π)
"""
q_values = np.array(q_values)
q_values = q_values - q_values.max()
probs = np.exp(q_values / temperature)
probs /= probs.sum()
# Entropy
entropy = -np.sum(probs * np.log(probs + 1e-8))
return entropy
# COMPARISON OF STRATEGIES
print("=== Exploration Strategies Comparison ===")
print("""
Strategy Pros Cons
-------- ---- ----
ε-greedy Simple, works well Random actions in bad states
Linear decay Predictable Needs tuning of steps
Exponential decay Fast initial learning Sensitive to decay rate
Boltzmann Smooth exploration Temperature tuning
Noisy networks Learned exploration More complex
Thompson sampling Optimal exploration Complex, slow
UCB Efficient Hard in DQN
""")
# DECAY SCHEDULES
print("=== Decay Schedules ===")
print("""
DQN original (Atari):
- ε: 1.0 → 0.1 over 1,000,000 steps
- Linear decay
- Then constant at 0.1
Typical for CartPole:
- ε: 1.0 → 0.01 over 10,000-50,000 steps
- Exponential decay (0.995 per step)
For continuous control:
- ε: 1.0 → 0.1
- Or use Gaussian noise (for continuous actions)
Rule of thumb:
- Start: 1.0 (fully random)
- End: 0.01-0.1 (mostly greedy)
- Decay steps: 10-20% of total training
""")
# SCHEDULE COMPARISON
print("\n=== Schedule Comparison (per step) ===")
print(f"{'Step':<10} {'Linear':<12} {'Exponential':<12}")
print("-" * 36)
for step in [0, 1000, 5000, 10000, 50000]:
lin = linear_decay(step, decay_steps=100000)
exp = exponential_decay(step, decay_steps=10000)
print(f"{step:<10} {lin:<12.4f} {exp:<12.4f}")
# PRACTICAL TIPS
print("\n=== Practical Tips ===")
print("""
1. ALWAYS DECAY ε
- Constant ε = poor performance
- Decay to ~0.01-0.1
- Balance exploration vs exploitation
2. USE EXPONENTIAL FOR SHORT TASKS
- Reaches low ε faster
- Good for CartPole, LunarLander
3. USE LINEAR FOR LONG TASKS
- More predictable
- Standard for Atari
4. EVALUATE WITHOUT EXPLORATION
- ε=0 for evaluation
- Clean performance measurement
5. CONSIDER NOISY NETWORKS
- Learned exploration
- State-dependent noise
- Better than ε-greedy
6. FOR CONTINUOUS ACTIONS
- Gaussian noise on actions
- Or Ornstein-Uhlenbeck noise
- No ε-greedy (actions are continuous)
7. ADAPTIVE METHODS
- Rarely used in practice
- Can help in specific cases
- Simpler decay is usually fine
""")
# COMMON MISTAKES
print("=== Common Mistakes ===")
print("""
❌ Not decaying ε
→ Policy stays random forever
❌ Decaying too fast
→ Premature convergence, poor exploration
❌ Decaying too slow
→ Wastes time on random actions
❌ Using ε-greedy in evaluation
→ Wrong performance measurement
❌ Not resetting ε for continued training
→ Should start high again
❌ Same ε for all problems
→ Tune per task
""")🎓 Training Loop
🎯 Putting It All Together
The DQN training loop combines all components: interaction with environment, storing transitions, sampling batches, computing loss, and updating networks. Getting the loop right is essential for success.
The Training Loop
1. Interact: Take action, observe reward, next state. 2. Store: Push transition to replay buffer. 3. Sample: Get random minibatch. 4. Compute: Loss = Huber(Q - target). 5. Update: Gradient descent step. 6. Sync: Periodically copy to target. 7. Repeat.
🎮 Interact💾 Store🎲 Sample📉 Update
# Complete DQN training loop
import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim
import random
from collections import deque
import gym
class DQNAgent:
"""Complete DQN agent."""
def __init__(
self,
state_dim,
action_dim,
hidden=128,
lr=1e-3,
gamma=0.99,
buffer_size=100000,
batch_size=64,
target_update=1000,
epsilon_start=1.0,
epsilon_end=0.01,
epsilon_decay=0.995,
device='cpu'
):
self.state_dim = state_dim
self.action_dim = action_dim
self.gamma = gamma
self.batch_size = batch_size
self.target_update = target_update
self.device = device
# Networks
self.online_net = self._build_network(state_dim, action_dim, hidden).to(device)
self.target_net = self._build_network(state_dim, action_dim, hidden).to(device)
self.target_net.load_state_dict(self.online_net.state_dict())
self.target_net.eval()
# Optimizer
self.optimizer = optim.Adam(self.online_net.parameters(), lr=lr)
# Replay buffer
self.replay_buffer = deque(maxlen=buffer_size)
# Exploration
self.epsilon = epsilon_start
self.epsilon_end = epsilon_end
self.epsilon_decay = epsilon_decay
# Counters
self.steps = 0
self.episodes = 0
def _build_network(self, state_dim, action_dim, hidden):
return nn.Sequential(
nn.Linear(state_dim, hidden),
nn.ReLU(),
nn.Linear(hidden, hidden),
nn.ReLU(),
nn.Linear(hidden, action_dim)
)
def select_action(self, state, evaluate=False):
"""Select action using ε-greedy."""
if evaluate or np.random.random() > self.epsilon:
with torch.no_grad():
state_tensor = torch.FloatTensor(state).unsqueeze(0).to(self.device)
q_values = self.online_net(state_tensor)
return q_values.argmax().item()
else:
return np.random.randint(self.action_dim)
def store_transition(self, state, action, reward, next_state, done):
"""Store transition in replay buffer."""
self.replay_buffer.append((state, action, reward, next_state, done))
def sample_batch(self):
"""Sample random minibatch."""
batch = random.sample(self.replay_buffer, self.batch_size)
states, actions, rewards, next_states, dones = zip(*batch)
return (
torch.FloatTensor(np.array(states)).to(self.device),
torch.LongTensor(actions).to(self.device),
torch.FloatTensor(rewards).to(self.device),
torch.FloatTensor(np.array(next_states)).to(self.device),
torch.FloatTensor(dones).to(self.device)
)
def update(self):
"""One training step."""
if len(self.replay_buffer) < self.batch_size:
return None
states, actions, rewards, next_states, dones = self.sample_batch()
# Current Q-values
q_values = self.online_net(states)
q_values = q_values.gather(1, actions.unsqueeze(1)).squeeze(1)
# Target Q-values
with torch.no_grad():
next_q_values = self.target_net(next_states)
max_next_q = next_q_values.max(dim=1)[0]
targets = rewards + self.gamma * max_next_q * (1 - dones)
# Loss (Huber)
loss = nn.SmoothL1Loss()(q_values, targets)
# Update
self.optimizer.zero_grad()
loss.backward()
torch.nn.utils.clip_grad_norm_(self.online_net.parameters(), 10.0)
self.optimizer.step()
# Update target network
self.steps += 1
if self.steps % self.target_update == 0:
self.target_net.load_state_dict(self.online_net.state_dict())
return loss.item()
def decay_epsilon(self):
"""Decay exploration rate."""
self.epsilon = max(self.epsilon_end, self.epsilon * self.epsilon_decay)
# TRAINING LOOP
def train_dqn(env_name='CartPole-v1', n_episodes=500):
"""Complete training loop."""
env = gym.make(env_name)
state_dim = env.observation_space.shape[0]
action_dim = env.action_space.n
agent = DQNAgent(state_dim, action_dim)
rewards_history = []
for episode in range(n_episodes):
state = env.reset()
if isinstance(state, tuple):
state = state[0]
episode_reward = 0
done = False
while not done:
# Select and execute action
action = agent.select_action(state)
next_state, reward, done, truncated, _ = env.step(action)
if isinstance(done, bool) and truncated:
done = True
# Store transition
agent.store_transition(state, action, reward, next_state, done)
# Update network
loss = agent.update()
# Move to next state
state = next_state
episode_reward += reward
# Episode complete
agent.episodes += 1
agent.decay_epsilon()
rewards_history.append(episode_reward)
# Logging
if episode % 20 == 0:
avg_reward = np.mean(rewards_history[-20:])
print(f"Episode {episode}, "
f"Reward: {episode_reward:.1f}, "
f"Avg (last 20): {avg_reward:.1f}, "
f"ε: {agent.epsilon:.3f}")
if avg_reward > 475:
print(f"Solved in {episode} episodes!")
break
env.close()
return agent, rewards_history
# RUN TRAINING
if __name__ == "__main__":
try:
agent, rewards = train_dqn()
except Exception as e:
print(f"Training demo: {e}")
# KEY TRAINING CONSIDERATIONS
print("\n=== Training Best Practices ===")
print("""
1. WARMUP
- Don't train until buffer has enough samples
- Fill buffer with random actions first
- Usually 1,000-10,000 steps
2. UPDATE FREQUENCY
- Update every step (after warmup)
- Or every N steps
- More updates = faster learning
3. BATCH SIZE
- 32-256 typically
- Larger = more stable
- Smaller = faster, noisier
4. TARGET UPDATE
- Hard: every 1,000-10,000 steps
- Soft: every step with τ=0.005
- Choose based on task
5. GRADIENT CLIPPING
- Clip norm to 10.0
- Prevents exploding gradients
- Essential for stability
6. REWARD SCALING
- Normalize rewards to [-1, 1]
- Or divide by running std
- Important for stability
7. EVALUATION
- Separate eval loop
- ε=0 (no exploration)
- Every N episodes
8. MONITORING
- Track: reward, loss, Q-values, ε
- Q-values should be stable
- Loss should decrease
""")
# DEBUGGING CHECKLIST
print("=== Debugging DQN ===")
print("""
If DQN isn't working:
1. CHECK THE BASICS
□ Replay buffer working?
□ Target network being updated?
□ ε decaying properly?
□ Rewards being stored correctly?
2. CHECK THE NETWORK
□ Q-values reasonable scale?
□ Loss decreasing?
□ Gradients not exploding?
□ Network output shape correct?
3. CHECK HYPERPARAMETERS
□ Learning rate not too high?
□ Batch size reasonable?
□ Target update frequency?
□ γ appropriate for task?
4. CHECK THE ENVIRONMENT
□ Rewards shaped well?
□ Episode length reasonable?
□ Action space correct?
5. COMMON ISSUES
- Q-values exploding → reduce lr, clip gradients
- Policy not improving → increase training
- Unstable training → use target network properly
- Slow learning → check batch size, lr
""")🚀 DQN Improvements
🎯 Beyond Vanilla DQN
Vanilla DQN works but has limitations. A series of improvements make it more stable, sample-efficient, and performant. These range from small tweaks to major architectural changes.
Key Improvements
Double DQN: Reduces overestimation. Dueling DQN: Better credit assignment. Prioritized replay: Sample important transitions. Multi-step: Faster reward propagation. Noisy nets: Learned exploration. Distributional: Learn uncertainty. Rainbow: All combined.
🔀 Double🎭 Dueling📊 Prioritized🌈 Rainbow
# DQN improvements — overview and code
import torch
import torch.nn as nn
import numpy as np
# 1. DOUBLE DQN (reduces overestimation)
class DoubleDQN(nn.Module):
"""
Fixes overestimation bias in vanilla DQN.
Vanilla: target = r + γ max_a Q(s', a; θ⁻)
Double: target = r + γ Q(s', argmax_a Q(s', a; θ); θ⁻)
"""
def __init__(self, state_dim, action_dim, hidden=128):
super().__init__()
self.net = nn.Sequential(
nn.Linear(state_dim, hidden),
nn.ReLU(),
nn.Linear(hidden, hidden),
nn.ReLU(),
nn.Linear(hidden, action_dim)
)
def forward(self, state):
return self.net(state)
def compute_target(self, next_states, rewards, dones, target_net, gamma=0.99):
with torch.no_grad():
# Select action using ONLINE network
next_q_online = self.forward(next_states)
next_actions = next_q_online.argmax(dim=1)
# Evaluate using TARGET network
next_q_target = target_net(next_states)
next_q = next_q_target.gather(1, next_actions.unsqueeze(1)).squeeze(1)
targets = rewards + gamma * next_q * (1 - dones)
return targets
# 2. DUELING DQN (better credit assignment)
class DuelingDQN(nn.Module):
"""
Separates V(s) and A(s,a).
Q(s,a) = V(s) + A(s,a) - mean_a A(s,a)
"""
def __init__(self, state_dim, action_dim, hidden=128):
super().__init__()
self.features = nn.Sequential(
nn.Linear(state_dim, hidden),
nn.ReLU()
)
self.value = nn.Sequential(
nn.Linear(hidden, hidden),
nn.ReLU(),
nn.Linear(hidden, 1)
)
self.advantage = nn.Sequential(
nn.Linear(hidden, hidden),
nn.ReLU(),
nn.Linear(hidden, action_dim)
)
def forward(self, state):
features = self.features(state)
value = self.value(features)
advantage = self.advantage(features)
q_values = value + advantage - advantage.mean(dim=1, keepdim=True)
return q_values
# 3. PRIORITIZED EXPERIENCE REPLAY
class PrioritizedReplayBuffer:
"""Sample transitions based on TD error magnitude."""
def __init__(self, capacity, alpha=0.6, beta_start=0.4, beta_frames=100000):
self.capacity = capacity
self.alpha = alpha
self.beta = beta_start
self.beta_start = beta_start
self.beta_frames = beta_frames
self.frame = 0
self.buffer = []
self.priorities = np.zeros(capacity)
self.position = 0
def push(self, transition, td_error=None):
# New experiences have max priority
if td_error is None:
priority = self.priorities.max() if len(self.buffer) > 0 else 1.0
else:
priority = abs(td_error) + 1e-6
if len(self.buffer) < self.capacity:
self.buffer.append(transition)
else:
self.buffer[self.position] = transition
self.priorities[self.position] = priority
self.position = (self.position + 1) % self.capacity
def sample(self, batch_size):
# Update beta for importance sampling
self.frame += 1
self.beta = min(1.0, self.beta_start + (1 - self.beta_start) * self.frame / self.beta_frames)
# Compute sampling probabilities
priorities = self.priorities[:len(self.buffer)] ** self.alpha
probs = priorities / priorities.sum()
# Sample
indices = np.random.choice(len(self.buffer), batch_size, p=probs)
samples = [self.buffer[i] for i in indices]
# Importance sampling weights
weights = (len(self.buffer) * probs[indices]) ** (-self.beta)
weights = weights / weights.max()
return samples, indices, weights
def update_priorities(self, indices, td_errors):
for idx, td_error in zip(indices, td_errors):
self.priorities[idx] = abs(td_error) + 1e-6
# 4. N-STEP RETURNS
def compute_n_step_target(rewards, next_states, dones, target_net, gamma=0.99, n=3):
"""
n-step return:
R^(n) = r_t + γr_{t+1} + ... + γ^(n-1)r_{t+n-1} + γ^n max Q(s_{t+n})
"""
n_step_rewards = []
current_reward = 0
discount = 1.0
for i in range(n):
if i < len(rewards):
current_reward += discount * rewards[i]
discount *= gamma
if dones[i]:
break
# Bootstrap from n-th state
with torch.no_grad():
next_q = target_net(next_states[-1]).max(dim=1)[0]
target = current_reward + discount * next_q
return target
# 5. NOISY NETWORKS (learned exploration)
class NoisyLinear(nn.Module):
"""Noisy linear layer for exploration."""
def __init__(self, in_features, out_features, sigma_init=0.017):
super().__init__()
self.in_features = in_features
self.out_features = out_features
self.weight_mu = nn.Parameter(torch.empty(out_features, in_features))
self.weight_sigma = nn.Parameter(torch.empty(out_features, in_features))
self.bias_mu = nn.Parameter(torch.empty(out_features))
self.bias_sigma = nn.Parameter(torch.empty(out_features))
self.register_buffer('weight_epsilon', torch.empty(out_features, in_features))
self.register_buffer('bias_epsilon', torch.empty(out_features))
# Initialize
self.weight_mu.data.uniform_(-0.1, 0.1)
self.weight_sigma.data.fill_(sigma_init)
self.bias_mu.data.uniform_(-0.1, 0.1)
self.bias_sigma.data.fill_(sigma_init)
def forward(self, x):
if self.training:
self.weight_epsilon.normal_()
self.bias_epsilon.normal_()
weight = self.weight_mu + self.weight_sigma * self.weight_epsilon
bias = self.bias_mu + self.bias_sigma * self.bias_epsilon
else:
weight = self.weight_mu
bias = self.bias_mu
return nn.functional.linear(x, weight, bias)
# 6. CATEGORICAL DQN (C51 — distributional)
class CategoricalDQN(nn.Module):
"""Distributional DQN with 51 atoms."""
def __init__(self, state_dim, action_dim, num_atoms=51, v_min=-10, v_max=10):
super().__init__()
self.action_dim = action_dim
self.num_atoms = num_atoms
self.v_min = v_min
self.v_max = v_max
self.register_buffer('support', torch.linspace(v_min, v_max, num_atoms))
self.net = nn.Sequential(
nn.Linear(state_dim, 128),
nn.ReLU(),
nn.Linear(128, 128),
nn.ReLU(),
nn.Linear(128, action_dim * num_atoms)
)
def forward(self, state):
logits = self.net(state)
logits = logits.view(-1, self.action_dim, self.num_atoms)
return nn.functional.softmax(logits, dim=-1)
# COMBINING IMPROVEMENTS: RAINBOW
print("=== Rainbow DQN ===")
print("""
Rainbow (Hessel et al., 2018) combines 6 improvements:
1. Double DQN
- Reduces overestimation
2. Dueling DQN
- Better credit assignment
3. Prioritized Experience Replay
- Sample important transitions
4. Multi-step Returns
- Faster reward propagation
5. Distributional DQN (C51)
- Learn distribution over returns
6. Noisy Networks
- Learned exploration
Ablation study showed:
- All 6 contribute
- Biggest gains: prioritized replay + multi-step
- Smaller gains: dueling + distributional
Rainbow vs. individual:
- Rainbow > sum of parts
- More stable training
- Better final performance
""")
# IMPROVEMENT EFFECTS
print("\n=== Improvement Effects (Atari) ===")
print(f"{'Method':<25} {'Median Score':<15} {'Notes':<30}")
print("-" * 70)
print(f"{'DQN (baseline)':<25} {'~200%':<15} {'Human-level':<30}")
print(f"{'+Double DQN':<25} {'~300%':<15} {'Reduced overestimation':<30}")
print(f"{'+Dueling':<25} {'~400%':<15} {'Better credit assignment':<30}")
print(f"{'+Prioritized':<25} {'~500%':<15} {'Sample efficiency':<30}")
print(f"{'+Multi-step':<25} {'~600%':<15} {'Faster propagation':<30}")
print(f"{'+Noisy Nets':<25} {'~650%':<15} {'Learned exploration':<30}")
print(f"{'+Distributional':<25} {'~750%':<15} {'Uncertainty aware':<30}")
print(f"{'Rainbow (all)':<25} {'~1000%':<15} {'State-of-the-art':<30}")
# IMPLEMENTATION ORDER
print("\n=== Implementation Order ===")
print("""
If you're implementing from scratch:
1. Start with VANILLA DQN
- Get it working first
- Basic replay buffer
- Target network
- Huber loss
2. Add DOUBLE DQN
- Simple change to target computation
- Immediate improvement
3. Add DUELING DQN
- Change network architecture
- Small code change
4. Add PRIORITIZED REPLAY
- More complex
- Bigger gains
5. Add MULTI-STEP RETURNS
- Small change
- Good gains
6. Add NOISY NETS
- Replace linear layers
- Remove ε-greedy
7. Add DISTRIBUTIONAL
- Major change
- Largest single gain
Or: Just use RAINBOW
- All improvements
- Proven to work
- More complex to implement
""")🔀 Double DQN
🎯 Reducing Overestimation
Double DQN (van Hasselt et al., 2016) fixes the overestimation bias in vanilla DQN. It's a simple change with significant impact: use online network to select actions, target network to evaluate them.
The Overestimation Problem
Vanilla DQN: target = r + γ max_a Q(s',a;θ⁻). Problem: max of noisy estimates is biased upward. Result: Q-values inflated, suboptimal policy. Double DQN: Decouple selection and evaluation. Formula: target = r + γ Q(s', argmax_a Q(s',a;θ); θ⁻). Impact: More accurate Q-values.
📈 Overestimation🔀 Decoupling🎯 Accuracy✅ Simple
# Double DQN — fixing overestimation bias
import torch
import torch.nn as nn
import numpy as np
# THE OVERESTIMATION PROBLEM
print("=== The Overestimation Problem ===")
print("""
Why does vanilla DQN overestimate?
Consider a state where all actions have true Q = 0.
But our estimates are noisy: Q_est(a) ~ N(0, ε²)
Vanilla target: max_a Q_est(a)
Expected value: E[max_a Q_est(a)] > max_a E[Q_est(a)] = 0
The max operation picks the action with the highest noise,
so the estimate is biased upward!
This bias:
- Grows with the number of actions
- Compounds over iterations
- Leads to suboptimal policies
Double DQN fixes this by decoupling:
- Action selection: online network
- Action evaluation: target network
""")
# DEMONSTRATION
np.random.seed(42)
n_actions = 10
n_samples = 10000
true_q = np.zeros(n_actions)
noisy_estimates = np.random.randn(n_samples, n_actions) * 0.5
# Vanilla: max over noisy estimates
vanilla_max = noisy_estimates.max(axis=1).mean()
# Double: pick action with online, evaluate with target (independent noise)
online_estimates = np.random.randn(n_samples, n_actions) * 0.5
target_estimates = np.random.randn(n_samples, n_actions) * 0.5
best_actions = online_estimates.argmax(axis=1)
double_estimates = target_estimates[np.arange(n_samples), best_actions]
double_max = double_estimates.mean()
print(f"True Q-value: 0.0")
print(f"Vanilla DQN: {vanilla_max:.3f} (overestimate)")
print(f"Double DQN: {double_max:.3f} (unbiased!)")
# DOUBLE DQN IMPLEMENTATION
class DoubleDQNAgent:
"""Double DQN agent."""
def __init__(
self,
state_dim,
action_dim,
hidden=128,
lr=1e-3,
gamma=0.99
):
self.gamma = gamma
# Online network
self.online_net = self._build_net(state_dim, action_dim, hidden)
# Target network
self.target_net = self._build_net(state_dim, action_dim, hidden)
self.target_net.load_state_dict(self.online_net.state_dict())
self.target_net.eval()
self.optimizer = torch.optim.Adam(self.online_net.parameters(), lr=lr)
def _build_net(self, state_dim, action_dim, hidden):
return nn.Sequential(
nn.Linear(state_dim, hidden),
nn.ReLU(),
nn.Linear(hidden, hidden),
nn.ReLU(),
nn.Linear(hidden, action_dim)
)
def compute_target(self, next_states, rewards, dones):
"""
Compute Double DQN target:
target = r + γ Q(s', argmax_a Q(s', a; θ); θ⁻)
"""
with torch.no_grad():
# Step 1: Select action using ONLINE network
next_q_online = self.online_net(next_states)
next_actions = next_q_online.argmax(dim=1)
# Step 2: Evaluate action using TARGET network
next_q_target = self.target_net(next_states)
next_q = next_q_target.gather(
1, next_actions.unsqueeze(1)
).squeeze(1)
# Step 3: Compute target
targets = rewards + self.gamma * next_q * (1 - dones)
return targets
def compute_loss(self, batch):
"""Compute Double DQN loss."""
states, actions, rewards, next_states, dones = batch
# Current Q-values
q_values = self.online_net(states)
q_values = q_values.gather(1, actions.unsqueeze(1)).squeeze(1)
# Double DQN target
targets = self.compute_target(next_states, rewards, dones)
# Huber loss
loss = nn.SmoothL1Loss()(q_values, targets)
return loss
def update(self, batch, target_update_freq=1000, step=0):
"""Update networks."""
loss = self.compute_loss(batch)
self.optimizer.zero_grad()
loss.backward()
torch.nn.utils.clip_grad_norm_(self.online_net.parameters(), 10.0)
self.optimizer.step()
# Update target
if step % target_update_freq == 0:
self.target_net.load_state_dict(self.online_net.state_dict())
return loss.item()
# COMPARISON
print("\n=== Comparison ===")
print("""
Vanilla DQN:
Selection: max_a Q(s', a; θ⁻)
Evaluation: Q(s', max_action; θ⁻)
Same network for both → coupled → biased
Double DQN:
Selection: argmax_a Q(s', a; θ)
Evaluation: Q(s', selected_action; θ⁻)
Different networks → decoupled → unbiased
Implementation difference:
VANILLA:
next_q = target_net(next_states).max(dim=1)[0]
DOUBLE:
next_actions = online_net(next_states).argmax(dim=1)
next_q = target_net(next_states).gather(1, next_actions.unsqueeze(1))
""")
# WHY IT WORKS
print("\n=== Why Double DQN Works ===")
print("""
Mathematical intuition:
Let Q_est(s,a) = Q_true(s,a) + ε_a
where ε_a is noise (mean 0, variance σ²)
Vanilla:
max_a Q_est(s,a) = max_a [Q_true(s,a) + ε_a]
The max picks the largest ε (on average)
→ overestimation bias of order σ√(2 log n)
Double:
a* = argmax_a Q_online(s,a)
Q_target(s,a*)
Since ε_online and ε_target are independent:
E[Q_target(s,a*)] = Q_true(s,a*) (unbiased!)
The key insight:
Noise in argmax doesn't matter if evaluation is independent.
""")
# RESULTS
print("\n=== Empirical Results ===")
print("""
Atari results (median human-normalized score):
DQN: ~200%
Double DQN: ~300% (50% improvement!)
Benefits:
- More accurate Q-values
- Better policies
- More stable learning
- Simple change (few lines of code)
- Works with other improvements
When it helps most:
- Large action spaces
- Noisy environments
- Long training runs
Cost:
- Same as vanilla DQN
- No extra parameters
- No extra computation
- Almost free improvement
""")
# DOUBLE DQN IN PRACTICE
print("\n=== Double DQN in Practice ===")
print("""
USE DOUBLE DQN ALWAYS!
- No downside
- Simple to implement
- Improves performance
- Standard in modern DQN
Combined with:
- Dueling architecture
- Prioritized replay
- Multi-step returns
- Distributional RL
Part of Rainbow:
- One of 6 key components
- Small but important gain
Real-world impact:
- Better Q-value estimates
- More accurate value predictions
- Better decisions
- Used in production systems
""")🏛️ Dueling DQN
🎯 Better Credit Assignment
Dueling DQN (Wang et al., 2016) separates the Q-function into a state value V(s) and an advantage A(s,a). This helps the network learn which states are valuable without having to learn the effect of each action.
The Dueling Architecture
Two streams: Value V(s) and Advantage A(s,a). Combine: Q(s,a) = V(s) + A(s,a) - mean_a A(s,a). Why subtract mean: Identifiability — forces unique solution. Benefit: Better credit assignment. Result: Learns faster, more accurate. Intuition: Some states are good regardless of action.
💎 Value V(s)🎯 Advantage A(s,a)➕ Combine📊 Unique
# Dueling DQN — separating value and advantage
import torch
import torch.nn as nn
import torch.nn.functional as F
# THE DUELING ARCHITECTURE
class DuelingDQN(nn.Module):
"""
Dueling DQN: separate value and advantage streams.
Q(s,a) = V(s) + A(s,a) - mean_a A(s,a)
"""
def __init__(self, state_dim, action_dim, hidden=128):
super().__init__()
# Shared feature extractor
self.features = nn.Sequential(
nn.Linear(state_dim, hidden),
nn.ReLU(),
nn.Linear(hidden, hidden),
nn.ReLU()
)
# Value stream: V(s)
self.value_stream = nn.Sequential(
nn.Linear(hidden, hidden),
nn.ReLU(),
nn.Linear(hidden, 1)
)
# Advantage stream: A(s,a)
self.advantage_stream = nn.Sequential(
nn.Linear(hidden, hidden),
nn.ReLU(),
nn.Linear(hidden, action_dim)
)
def forward(self, state):
"""Compute Q-values using dueling architecture."""
features = self.features(state)
# Value V(s)
value = self.value_stream(features)
# Advantage A(s,a)
advantage = self.advantage_stream(features)
# Combine: Q(s,a) = V(s) + A(s,a) - mean_a A(s,a)
q_values = value + advantage - advantage.mean(dim=-1, keepdim=True)
return q_values
def get_value_and_advantage(self, state):
"""Return V(s) and A(s,a) separately for analysis."""
features = self.features(state)
value = self.value_stream(features)
advantage = self.advantage_stream(features)
return value, advantage
# WHY THE MEAN SUBTRACTION?
print("=== Why Subtract the Mean? ===")
print("""
Without mean subtraction:
Q(s,a) = V(s) + A(s,a)
Problem: Not identifiable!
V(s) + A(s,a) = V(s) + c + A(s,a) - c
Multiple solutions give the same Q-values
With mean subtraction:
Q(s,a) = V(s) + A(s,a) - mean_a A(s,a)
This forces: mean_a A(s,a) = 0
Result:
- Unique decomposition
- V(s) = mean_a Q(s,a) (average value)
- A(s,a) = Q(s,a) - V(s) (relative advantage)
Alternative (max subtraction):
Q(s,a) = V(s) + A(s,a) - max_a A(s,a)
This forces: max_a A(s,a) = 0
- Q(s, best_action) = V(s)
- V(s) = max value over actions
Both work. Mean is more common.
""")
# VISUALIZING THE STREAMS
print("\n=== Visualizing Value and Advantage ===")
print("""
Example: Playing a video game
Consider two states:
State A: You're about to win
V(s_A) = HIGH (this state is good)
A(s_A, each action) = similar (any action wins)
State B: You're in a tight spot
V(s_B) = LOW (this state is bad)
A(s_B, actions) = vary a lot (some actions help, some hurt)
Dueling DQN learns this structure:
- Value stream: How good is this state?
- Advantage stream: How much better is each action?
This decomposition helps:
1. Learn V(s) even without action info
2. Better generalize across similar states
3. Faster learning on state-only rewards
""")
# COMPARISON WITH STANDARD DQN
print("=== Dueling vs Standard DQN ===")
print("""
Standard DQN:
features → Q(s,a) directly
Network must learn Q(s,a) for each action
Dueling DQN:
features → V(s) and A(s,a) separately
Combine into Q(s,a)
Advantages of Dueling:
1. Better credit assignment
2. Learns state values efficiently
3. More robust to action-space changes
4. Can learn from state-only rewards
Results:
- Improved performance on Atari
- Better on tasks with many actions
- Faster learning
- Part of Rainbow
""")
# COMPARISON CODE
def compare_architectures():
"""Compare standard and dueling DQN."""
class StandardDQN(nn.Module):
def __init__(self, state_dim, action_dim, hidden=128):
super().__init__()
self.net = nn.Sequential(
nn.Linear(state_dim, hidden),
nn.ReLU(),
nn.Linear(hidden, hidden),
nn.ReLU(),
nn.Linear(hidden, action_dim)
)
def forward(self, state):
return self.net(state)
state_dim = 8
action_dim = 4
standard = StandardDQN(state_dim, action_dim)
dueling = DuelingDQN(state_dim, action_dim)
# Count parameters
standard_params = sum(p.numel() for p in standard.parameters())
dueling_params = sum(p.numel() for p in dueling.parameters())
print(f"\nParameter comparison:")
print(f" Standard DQN: {standard_params:,}")
print(f" Dueling DQN: {dueling_params:,}")
print(f" Difference: {dueling_params - standard_params:,}")
compare_architectures()
# DUELING WITH PRIORITIZED REPLAY
class DuelingAgent:
"""Complete dueling DQN agent."""
def __init__(self, state_dim, action_dim, hidden=128, lr=1e-3):
self.online_net = DuelingDQN(state_dim, action_dim, hidden)
self.target_net = DuelingDQN(state_dim, action_dim, hidden)
self.target_net.load_state_dict(self.online_net.state_dict())
self.target_net.eval()
self.optimizer = torch.optim.Adam(self.online_net.parameters(), lr=lr)
def compute_loss(self, batch, gamma=0.99):
"""Compute loss with dueling network."""
states, actions, rewards, next_states, dones = batch
# Current Q-values (using dueling forward)
q_values = self.online_net(states)
q_values = q_values.gather(1, actions.unsqueeze(1)).squeeze(1)
# Target using Double DQN trick
with torch.no_grad():
next_actions = self.online_net(next_states).argmax(dim=1)
next_q = self.target_net(next_states).gather(
1, next_actions.unsqueeze(1)
).squeeze(1)
targets = rewards + gamma * next_q * (1 - dones)
loss = nn.SmoothL1Loss()(q_values, targets)
return loss
# ADVANCED: DUELING WITH ATTENTION
class DuelingWithAttention(nn.Module):
"""Dueling DQN with attention over state features."""
def __init__(self, state_dim, action_dim, hidden=128):
super().__init__()
# Feature extractor
self.features = nn.Sequential(
nn.Linear(state_dim, hidden),
nn.ReLU()
)
# Attention
self.attention = nn.Sequential(
nn.Linear(hidden, hidden),
nn.ReLU(),
nn.Linear(hidden, 1),
nn.Sigmoid()
)
# Value stream
self.value_stream = nn.Sequential(
nn.Linear(hidden, hidden),
nn.ReLU(),
nn.Linear(hidden, 1)
)
# Advantage stream
self.advantage_stream = nn.Sequential(
nn.Linear(hidden, hidden),
nn.ReLU(),
nn.Linear(hidden, action_dim)
)
def forward(self, state):
features = self.features(state)
attention_weights = self.attention(features)
weighted_features = features * attention_weights
value = self.value_stream(weighted_features)
advantage = self.advantage_stream(weighted_features)
q_values = value + advantage - advantage.mean(dim=-1, keepdim=True)
return q_values
# RESULTS AND IMPACT
print("\n=== Dueling DQN Results ===")
print("""
Atari results:
- Improved median score by ~100% over vanilla DQN
- Better on games with many actions
- More stable training
Specific games where dueling shines:
- Seaquest (multi-action underwater)
- Asterix (complex action space)
- Zaxxon (3D isometric)
- Beam Rider (multi-action)
Why it helps:
1. Separates state value from action choice
2. Better credit assignment
3. More efficient learning
Part of Rainbow:
- One of 6 improvements
- Combined with others gives best results
""")
# PRACTICAL TIPS
print("\n=== Practical Tips ===")
print("""
1. USE DUELING BY DEFAULT
- No significant downside
- Improves performance
- Small code change
2. COMBINE WITH OTHER IMPROVEMENTS
- Double DQN: use both
- Prioritized replay: works well
- Multi-step returns: good
- Noisy nets: complementary
3. AGGREGATION FUNCTION
- Mean: Standard, works well
- Max: Alternative, sometimes better
- Softmax: Rarely used
4. NETWORK SIZE
- Dueling doesn't need bigger network
- Same architecture size
- Just different structure
5. FOR CONTINUOUS ACTIONS
- Dueling less critical
- SAC has similar ideas
- Focus on other techniques
6. DEBUGGING
- Check V and A ranges
- A should center near 0
- V should be similar to Q mean
""")
# FULL DUELING ARCHITECTURE WITH IMPROVEMENTS
print("\n=== Full Dueling DQN Architecture ===")
print("""
Input: state (state_dim)
↓
Linear(state_dim, 256) + ReLU
↓
Linear(256, 256) + ReLU
↓
┌─────────────────────────────────────┐
│ │
▼ ▼
Value Stream Advantage Stream
Linear(256, 256) + ReLU Linear(256, 256) + ReLU
Linear(256, 1) Linear(256, action_dim)
↓ ↓
V(s) A(s,a)
│ │
└─────────────┬─────────────────────┘
↓
Q = V + A - mean(A)
↓
Q-values
Additions for better performance:
- NoisyLinear instead of Linear (exploration)
- LayerNorm after linear (stability)
- Distributional head (uncertainty)
- Shared features before streams (efficiency)
""")🌈 Rainbow DQN
🎯 All the Improvements Combined
Rainbow (Hessel et al., 2018) combines 6 key DQN improvements into one algorithm: Double DQN, Dueling DQN, Prioritized Replay, Multi-step Returns, Distributional DQN, and Noisy Nets.
The 6 Components
Double DQN: Reduce overestimation. Dueling DQN: Separate V and A. Prioritized: Important transitions. Multi-step: n-step returns. Distributional: C51. Noisy: Learned exploration. Result: State-of-the-art.
🔀 Double🏛️ Dueling📊 Prioritized🎯 C51
# Rainbow DQN — combining all improvements
import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np
# 1. NOISY LINEAR (for exploration)
class NoisyLinear(nn.Module):
def __init__(self, in_features, out_features, sigma_init=0.5):
super().__init__()
self.in_features = in_features
self.out_features = out_features
self.sigma_init = sigma_init
# Learnable parameters
self.weight_mu = nn.Parameter(torch.empty(out_features, in_features))
self.weight_sigma = nn.Parameter(torch.empty(out_features, in_features))
self.bias_mu = nn.Parameter(torch.empty(out_features))
self.bias_sigma = nn.Parameter(torch.empty(out_features))
self.reset_parameters()
def reset_parameters(self):
mu_range = 1 / (self.in_features ** 0.5)
self.weight_mu.data.uniform_(-mu_range, mu_range)
self.weight_sigma.data.fill_(self.sigma_init / (self.in_features ** 0.5))
self.bias_mu.data.uniform_(-mu_range, mu_range)
self.bias_sigma.data.fill_(self.sigma_init / (self.out_features ** 0.5))
def forward(self, x):
if self.training:
weight_epsilon = torch.randn_like(self.weight_sigma)
bias_epsilon = torch.randn_like(self.bias_sigma)
weight = self.weight_mu + self.weight_sigma * weight_epsilon
bias = self.bias_mu + self.bias_sigma * bias_epsilon
else:
weight = self.weight_mu
bias = self.bias_mu
return F.linear(x, weight, bias)
# 2. RAINBOW NETWORK (Distributional + Dueling + Noisy)
class RainbowNetwork(nn.Module):
"""
Rainbow network combining:
- Dueling architecture
- Distributional RL (C51)
- Noisy networks
"""
def __init__(
self,
state_dim,
action_dim,
num_atoms=51,
v_min=-10,
v_max=10,
hidden=256,
noisy_std=0.5
):
super().__init__()
self.action_dim = action_dim
self.num_atoms = num_atoms
self.v_min = v_min
self.v_max = v_max
# Register support
self.register_buffer(
'support',
torch.linspace(v_min, v_max, num_atoms)
)
# Shared feature layers (NoisyLinear)
self.features = nn.Sequential(
NoisyLinear(state_dim, hidden, noisy_std),
nn.ReLU(),
NoisyLinear(hidden, hidden, noisy_std),
nn.ReLU()
)
# Dueling: Value stream
self.value_stream = nn.Sequential(
NoisyLinear(hidden, hidden, noisy_std),
nn.ReLU(),
NoisyLinear(hidden, num_atoms, noisy_std)
)
# Dueling: Advantage stream
self.advantage_stream = nn.Sequential(
NoisyLinear(hidden, hidden, noisy_std),
nn.ReLU(),
NoisyLinear(hidden, action_dim * num_atoms, noisy_std)
)
def forward(self, state):
"""
Forward pass returns probability distribution over atoms.
Returns:
probs: (batch, action_dim, num_atoms) — distribution
"""
batch_size = state.size(0)
features = self.features(state)
# Value stream: (batch, num_atoms)
value = self.value_stream(features)
value = value.view(batch_size, 1, self.num_atoms)
# Advantage stream: (batch, action_dim, num_atoms)
advantage = self.advantage_stream(features)
advantage = advantage.view(batch_size, self.action_dim, self.num_atoms)
# Combine: Q = V + A - mean(A)
q_atoms = value + advantage - advantage.mean(dim=1, keepdim=True)
# Convert to probabilities
probs = F.softmax(q_atoms, dim=-1)
probs = probs.clamp(min=1e-3) # Avoid zero probabilities
return probs
def get_q_values(self, state):
"""Compute expected Q-values from distribution."""
probs = self.forward(state)
q_values = (probs * self.support).sum(dim=-1)
return q_values
# 3. PRIORITIZED REPLAY BUFFER
class PrioritizedReplayBuffer:
"""Prioritized experience replay."""
def __init__(self, capacity, alpha=0.6, beta_start=0.4, beta_frames=100000):
self.capacity = capacity
self.alpha = alpha
self.beta = beta_start
self.beta_start = beta_start
self.beta_frames = beta_frames
self.frame = 0
self.buffer = []
self.priorities = np.zeros(capacity)
self.position = 0
def push(self, transition, td_error=None):
if td_error is None:
priority = self.priorities.max() if len(self.buffer) > 0 else 1.0
else:
priority = abs(td_error) + 1e-6
if len(self.buffer) < self.capacity:
self.buffer.append(transition)
else:
self.buffer[self.position] = transition
self.priorities[self.position] = priority
self.position = (self.position + 1) % self.capacity
def sample(self, batch_size):
self.frame += 1
self.beta = min(1.0, self.beta_start +
(1 - self.beta_start) * self.frame / self.beta_frames)
priorities = self.priorities[:len(self.buffer)] ** self.alpha
probs = priorities / priorities.sum()
indices = np.random.choice(len(self.buffer), batch_size, p=probs)
samples = [self.buffer[i] for i in indices]
weights = (len(self.buffer) * probs[indices]) ** (-self.beta)
weights = weights / weights.max()
return samples, indices, weights
def update_priorities(self, indices, td_errors):
for idx, td_error in zip(indices, td_errors):
self.priorities[idx] = abs(td_error) + 1e-6
# 4. RAINBOW AGENT
class RainbowAgent:
"""Complete Rainbow DQN agent."""
def __init__(
self,
state_dim,
action_dim,
num_atoms=51,
v_min=-10,
v_max=10,
lr=6.25e-5,
gamma=0.99,
n_step=3,
buffer_size=1000000,
batch_size=32,
target_update=8000
):
self.action_dim = action_dim
self.num_atoms = num_atoms
self.v_min = v_min
self.v_max = v_max
self.gamma = gamma
self.n_step = n_step
self.batch_size = batch_size
self.target_update = target_update
# Networks
self.online_net = RainbowNetwork(
state_dim, action_dim, num_atoms, v_min, v_max
)
self.target_net = RainbowNetwork(
state_dim, action_dim, num_atoms, v_min, v_max
)
self.target_net.load_state_dict(self.online_net.state_dict())
self.target_net.eval()
# Optimizer
self.optimizer = torch.optim.Adam(self.online_net.parameters(), lr=lr)
# Replay buffer
self.replay_buffer = PrioritizedReplayBuffer(buffer_size)
# N-step buffer
self.n_step_buffer = []
self.steps = 0
def store_transition(self, state, action, reward, next_state, done):
"""Store transition with n-step returns."""
self.n_step_buffer.append((state, action, reward, next_state, done))
if len(self.n_step_buffer) < self.n_step:
return
# Compute n-step return
R = 0
for i, (_, _, r, _, d) in enumerate(self.n_step_buffer):
R += (self.gamma ** i) * r
if d:
break
# Get first state and action
first_state = self.n_step_buffer[0][0]
first_action = self.n_step_buffer[0][1]
# Get last next_state
last_next_state = self.n_step_buffer[-1][3]
last_done = self.n_step_buffer[-1][4]
# Store n-step transition
self.replay_buffer.push(
(first_state, first_action, R, last_next_state, last_done)
)
# Remove first element
self.n_step_buffer.pop(0)
def select_action(self, state, evaluate=False):
"""Select action (uses noisy nets in training)."""
with torch.no_grad():
state_tensor = torch.FloatTensor(state).unsqueeze(0)
q_values = self.online_net.get_q_values(state_tensor)
return q_values.argmax().item()
def update(self):
"""One training step."""
if len(self.replay_buffer.buffer) < self.batch_size:
return None
# Sample prioritized batch
batch, indices, weights = self.replay_buffer.sample(self.batch_size)
weights = torch.FloatTensor(weights)
states, actions, rewards, next_states, dones = zip(*batch)
states = torch.FloatTensor(np.array(states))
actions = torch.LongTensor(actions)
rewards = torch.FloatTensor(rewards)
next_states = torch.FloatTensor(np.array(next_states))
dones = torch.FloatTensor(dones)
# Forward pass
dist = self.online_net(states) # (batch, action_dim, num_atoms)
dist = dist[range(len(actions)), actions] # (batch, num_atoms)
# Compute target distribution (Distributional + Double DQN)
with torch.no_grad():
# Double DQN: select action with online net
next_q_online = self.online_net.get_q_values(next_states)
next_actions = next_q_online.argmax(dim=1)
# Evaluate with target net
next_dist = self.target_net(next_states)
next_dist = next_dist[range(len(next_actions)), next_actions]
# Project distribution
target_dist = self.project_distribution(
next_dist, rewards, dones
)
# Loss: KL divergence
loss = -(target_dist * torch.log(dist + 1e-8)).sum(dim=1)
loss = (loss * weights).mean()
# Update
self.optimizer.zero_grad()
loss.backward()
torch.nn.utils.clip_grad_norm_(self.online_net.parameters(), 10.0)
self.optimizer.step()
# Update priorities
with torch.no_grad():
td_errors = (target_dist - dist).abs().sum(dim=1).numpy()
self.replay_buffer.update_priorities(indices, td_errors)
# Update target network
self.steps += 1
if self.steps % self.target_update == 0:
self.target_net.load_state_dict(self.online_net.state_dict())
return loss.item()
def project_distribution(self, next_dist, rewards, dones):
"""
Project the target distribution onto the support.
This is the C51 projection step.
"""
batch_size = rewards.size(0)
support = self.online_net.support
num_atoms = self.num_atoms
# Compute target support: r + γ * support
target_support = rewards.unsqueeze(1) + \
self.gamma * support.unsqueeze(0) * \
(1 - dones.unsqueeze(1))
target_support = target_support.clamp(self.v_min, self.v_max)
# Compute bin width
delta_z = (self.v_max - self.v_min) / (num_atoms - 1)
# Project
b = (target_support - self.v_min) / delta_z
l = b.floor().long()
u = b.ceil().long()
# Handle edge cases
l = l.clamp(0, num_atoms - 1)
u = u.clamp(0, num_atoms - 1)
# Distribute probability
target_dist = torch.zeros_like(next_dist)
offset = torch.arange(batch_size).unsqueeze(1) * num_atoms
target_dist.view(-1).index_add_(
0, (l + offset).view(-1),
(next_dist * (u.float() - b)).view(-1)
)
target_dist.view(-1).index_add_(
0, (u + offset).view(-1),
(next_dist * (b - l.float())).view(-1)
)
return target_dist
# RAINBOW COMPONENTS SUMMARY
print("=== Rainbow Components ===")
print("""
1. Double DQN
- Reduces overestimation
- Online selects, target evaluates
2. Dueling DQN
- Separate V and A
- Q = V + A - mean(A)
3. Prioritized Replay
- Sample by TD error
- Importance sampling weights
4. Multi-step Returns (n=3)
- R = Σ γ^i r_i
- Faster propagation
5. Distributional (C51, n_atoms=51)
- Learn distribution
- KL loss
6. Noisy Networks
- Learned exploration
- Replaces ε-greedy
Result: State-of-the-art DQN
""")
# RAINBOW RESULTS
print("=== Rainbow Results ===")
print("""
Atari benchmark (57 games):
Method Median Score Mean Score
------------- ------------ ----------
DQN ~200% ~230%
Double DQN ~300% ~330%
Prioritized DQN ~400% ~430%
Dueling DQN ~400% ~470%
A3C ~500% ~620%
Rainbow ~1000% ~1200%
Rainbow is dramatically better
- State-of-the-art at publication
- Still competitive today
- Reference for DQN improvements
Ablation study showed:
- All 6 components contribute
- Prioritized replay: biggest individual gain
- Multi-step: second biggest
- Distributional: third
- Dueling: moderate
- Double: small but consistent
- Noisy: good for final performance
""")
# IMPLEMENTATION TIPS
print("\n=== Implementation Tips ===")
print("""
1. START WITH VANILLA DQN
- Make sure it works first
- Then add improvements
2. ADD IN THIS ORDER
- Double DQN (easy)
- Dueling (easy)
- Multi-step (easy)
- Prioritized (medium)
- Noisy (easy)
- Distributional (hardest)
3. HYPERPARAMETERS
- n_step = 3
- num_atoms = 51
- v_min = -10, v_max = 10
- lr = 6.25e-5
- Adam optimizer
- target_update = 8000
- batch_size = 32
- buffer_size = 1M
- alpha = 0.5 (prioritization)
- beta = 0.4 → 1.0
4. COMMON ISSUES
- Distributional projection bugs
- Numerical stability
- Priority explosion
- Memory usage
5. MONITORING
- Q-value distribution
- Loss evolution
- Priority distribution
- TD errors
""")
# CONCLUSION
print("\n=== Conclusion ===")
print("""
Rainbow is the culmination of DQN research.
It demonstrates:
- Careful engineering matters
- Multiple improvements compound
- Theory-informed design works
- Benchmark results are reproducible
Learning path:
1. Understand vanilla DQN
2. Learn each improvement
3. See how they combine
4. Implement Rainbow
5. Adapt to your problem
Beyond Rainbow:
- R2D2 (recurrent + distributed)
- Agent57 (all improvements + more)
- Efficient Rainbow variants
- Modern algorithms (SAC, Dreamer)
But the principles from Rainbow remain:
- Double DQN: decouple selection/evaluation
- Dueling: separate V and A
- Prioritized: focus on important transitions
- Multi-step: bootstrap faster
- Distributional: model uncertainty
- Noisy: learned exploration
""")🚀 Future: Beyond DQN
🎯 The Next Generation
DQN was just the beginning. Modern value-based methods like R2D2, Agent57, and offline RL algorithms have pushed performance further. And policy-based methods now dominate many domains.
Future Directions
R2D2: Recurrent + distributed. Agent57: Beyond human on all Atari. Offline RL: Learn from fixed data. Transformer RL: Sequence modeling. Model-based: DreamerV3. Foundation models: Gato, RT-2. Real-world: Deployment at scale.
🔄 R2D2🤖 Agent57📊 Offline🧠 Transformer
# Beyond DQN — modern algorithms
import numpy as np
# 1. R2D2 (Recurrent Replay Distributed DQN)
print("=== R2D2 ===")
print("""
R2D2 (Kapturowski et al., 2019):
- Combines DQN with LSTM
- Distributed training
- Prioritized replay
- n-step returns
- Double DQN
Key innovations:
1. Recurrent network (LSTM) for memory
2. Distributed experience collection
3. Stored hidden states in replay
4. Burn-in for sequence initialization
5. Prioritized replay with sequence
Results:
- State-of-the-art on Atari
- Beyond human on 52/57 games
- Handles partially observable tasks
Architecture:
Input → Conv → LSTM → Dueling heads
Training:
- Parallel actors
- Central learner
- Replay buffer of sequences
""")
# R2D2 pseudocode
print("--- R2D2 Architecture ---")
print("""
class R2D2Network(nn.Module):
def __init__(self):
self.conv = CNN()
self.lstm = LSTM(hidden_size=512)
self.value = Linear(512, 1)
self.advantage = Linear(512, action_dim)
def forward(self, state, hidden):
features = self.conv(state)
lstm_out, new_hidden = self.lstm(features, hidden)
value = self.value(lstm_out)
advantage = self.advantage(lstm_out)
q = value + advantage - advantage.mean()
return q, new_hidden
# Store sequences in replay
# Sample sequences, not individual transitions
# Use burn-in for LSTM initialization
""")
# 2. AGENT57
print("\n=== Agent57 ===")
print("""
Agent57 (Badia et al., 2020):
- First algorithm to beat humans on ALL 57 Atari games
- Combines many techniques:
* R2D2 architecture
* Never Give Up (NGU) exploration
* Intrinsic rewards
* Meta-controller
* Population of agents
Key insight:
- Different games need different strategies
- Adapt exploration per task
- Population-based training
Complexity:
- 57 separate policies
- Meta-controller selects per episode
- Training takes millions of steps
Result:
- Superhuman on all Atari games
- Breakthrough result
- Shows value of exploration
""")
# 3. OFFLINE RL
print("=== Offline RL ===")
print("""
Learn from fixed dataset without interaction:
Methods:
1. BCQ (Batch-Constrained Q-Learning)
2. CQL (Conservative Q-Learning)
3. IQL (Implicit Q-Learning)
4. TD3+BC
5. Decision Transformer
6. AWAC
Key challenges:
- Distribution shift
- Extrapolation error
- Uncertainty
Applications:
- Healthcare (patient data)
- Robotics (demonstrations)
- Recommendation
- Autonomous driving
- Finance
Online vs Offline:
Online: Interact, learn, explore
Offline: Fixed data, no exploration
Offline: Safer, but harder
""")
# 4. TRANSFORMER RL
print("=== Transformer RL ===")
print("""
Decision Transformer (Chen et al., 2021):
- Treat RL as sequence modeling
- Input: (R, s, a, R, s, a, ...)
- Output: next action
- Train with supervised learning
Benefits:
- No TD learning
- No bootstrapping
- Simple, stable
- Long context
Extensions:
- Trajectory Transformer
- Multi-game Decision Transformer
- Gato (multi-modal)
- RT-2 (robotics)
Power:
- Leverages transformer scaling
- Combines with LLM capabilities
- Foundation for general agents
""")
# Decision Transformer pseudocode
print("--- Decision Transformer ---")
print("""
# Input: trajectory as sequence
# (R_0, s_0, a_0, R_1, s_1, a_1, ..., R_T, s_T, a_T)
# Goal: predict a_t given (R_t, s_t, history)
def decision_transformer_forward(trajectory):
# Embed each token
R_emb = embed_return(R)
s_emb = embed_state(s)
a_emb = embed_action(a)
# Interleave tokens
tokens = interleave(R_emb, s_emb, a_emb)
# Transformer
hidden = transformer(tokens)
# Predict actions
action_pred = action_head(hidden)
return action_pred
# Train with cross-entropy on actions
# At test time, condition on target return
""")
# 5. MODEL-BASED RL
print("=== Model-Based RL ===")
print("""
Learn environment model, plan with it:
DreamerV3 (Hafner et al., 2023):
- Learns latent world model
- Trains actor-critic in imagination
- Sample efficient
- General across domains
Architecture:
1. Encoder: state → latent
2. Dynamics: (latent, action) → next latent
3. Reward: latent → reward
4. Decoder: latent → state
5. Actor-Critic: trained in imagination
Benefits:
- 10-100x more sample efficient
- Planning capability
- Transfer learning
Results:
- Minecraft diamond from scratch
- 150+ Atari games
- Various robotics tasks
- General algorithm
""")
# DreamerV3 pseudocode
print("--- DreamerV3 ---")
print("""
# Learn world model
for each iteration:
# Collect data with current policy
data = collect_episodes(env, policy)
# Update world model
for _ in range(wm_updates):
batch = sample(data)
update_encoder_decoder(batch)
update_dynamics(batch)
update_reward(batch)
# Train actor-critic in imagination
for _ in range(ac_updates):
# Imagine trajectories
latent = sample_state()
for t in range(horizon):
action = actor(latent)
latent = dynamics(latent, action)
reward = reward_model(latent)
# Update actor and critic using imagined returns
update_actor_critic(imagined_trajectories)
""")
# 6. FOUNDATION MODELS FOR RL
print("=== Foundation Models for RL ===")
print("""
Gato (Reed et al., 2022):
- Single model for many tasks
- Text, images, robot control
- Same transformer architecture
- Trained on diverse data
RT-2 (Google DeepMind):
- Vision-Language-Action model
- Robot control via LLM
- Leverages internet data
- Generalization to new tasks
Foundation policies:
- Pretrain on diverse tasks
- Fine-tune for specific
- Transfer learning
- Generalization
Applications:
- Robotics (RT-1, RT-2, Octo)
- Games (multi-game DT)
- Manipulation
- Navigation
- Dialogue
Future:
- Unified RL + LLM
- Language-conditioned policies
- Embodied AI
- General agents
""")
# 7. KEY CHALLENGES
print("=== Open Challenges ===")
print("""
1. SAMPLE EFFICIENCY
- Real-world data expensive
- Need faster learning
- Model-based methods help
2. GENERALIZATION
- New tasks, environments
- Distribution shift
- Foundation models address
3. SAFETY
- Constrained optimization
- Provable guarantees
- Human oversight
4. SCALABILITY
- Large state/action spaces
- Long horizons
- Multi-agent systems
5. STABILITY
- Hyperparameter sensitivity
- Non-stationarity
- Training dynamics
6. INTERPRETABILITY
- Understanding policies
- Explaining decisions
- Building trust
7. EVALUATION
- Off-policy evaluation
- Safety metrics
- Real-world impact
""")
# 8. WHERE TO GO FROM HERE
print("=== Where to Go From Here ===")
print("""
If you've mastered DQN:
1. LEARN POLICY GRADIENT METHODS
- REINFORCE, PPO, SAC
- Better for continuous control
- Foundation of modern RL
2. EXPLORE ADVANCED VARIANTS
- R2D2, Agent57
- IQN, FQF (distributional)
- QR-DQN
3. TRY MODEL-BASED
- DreamerV3, MuZero
- Sample efficiency
- Planning capability
4. STUDY OFFLINE RL
- CQL, IQL, TD3+BC
- Real-world applications
- Safety guarantees
5. EXPLORE TRANSFORMER RL
- Decision Transformer
- Gato, RT-2
- Foundation models
6. APPLY TO REAL PROBLEMS
- Games
- Robotics
- Control
- Recommendation
7. CONTRIBUTE TO RESEARCH
- Open problems
- New algorithms
- Benchmarks
Recommended resources:
- Sutton & Barto (textbook)
- Spinning Up (OpenAI)
- CleanRL (implementations)
- Deep RL course (Berkeley)
- Papers with Code
""")
# CONCLUSION
print("\n=== Conclusion ===")
print("""
DQN was a breakthrough but not the end.
Evolution of value-based methods:
2013: DQN
2015: Nature DQN (with target network)
2016: Double DQN, Dueling DQN
2017: Prioritized Replay, Distributional
2018: Rainbow
2019: R2D2
2020: Agent57
2021: Decision Transformer
2022: Gato, RT-1
2023: DreamerV3, RT-2
2024+: Foundation models for RL
Each generation:
- Better performance
- More general
- More sample efficient
- More aligned with human values
The future of RL:
- Multi-modal foundation models
- Real-world deployment
- Safe, aligned agents
- Human-AI collaboration
DQN taught us:
- Deep learning + RL works
- Stability requires careful design
- Sample efficiency matters
- Benchmarks drive progress
Now go build the next breakthrough!
""")