DQN Variants and Replay Buffer Sampling

English translation

Learning objectives

Identify which problem each network modification addresses and validate one needed change, rather than applying every improvement.

Execution conditions

Uses NumPy, PyTorch, and Gymnasium. In the final activity, read an external data card and analyze transitions you collected.

Applications

Throughput and cost of equipment and communication resources

An average of 2 items arrive at the synthetic queue each time step. Increasing service capacity reduces queue length but increases energy cost.

Observation and state: Queue length $q=0$ …20. Observation $q/20$ .

Action: choose processing capacity 0, 2, or 4.

Reward and constraints: −(next queue length + 0.2×capacity² + 10×dropped-item count). Time limit at 100 steps.

Comparison baselines: Constant capacity of 2, and a threshold policy based on queue length.

Metrics to check: Evaluation return, mean queue length, mean capacity², and dropped items.

Choose equipment inspection times

Frame the research problem as balancing the cost of inspecting every time against the cost of missed failures.

Observation and state: Recent sensor readings, trends, and time since the last inspection. Distinguish the actual fault state from observations.

Actions: Continue operating, inspect, or replace.

Reward and constraints: Operating benefit − inspection, replacement, and failure costs.

Comparison baselines: Periodic inspection and threshold alarms.

Metrics to check: availability, failure count, and total cost. Check for failure label leakage.

Distinguish selection, evaluation, representation, and sampling

Starting conditions

$Q_{\mathrm{online}}=[2,4,3]$ , $Q_{\mathrm{target}}=[5,1,2]$ .
$r=1$ , $\gamma=.95$ .
Dueling: $V=2$ , $A=[1,2,6]$ .
TD Error: $[0.1,1,5]$ .

Procedure

Calculate DQN and Double DQN targets by hand and mark the parts of the equations that differ.
Use small arrays to inspect Dueling's mean correction and PER sampling probabilities for $\alpha=0$ /.6.
In one line each, connect the three methods to the possible failures they target.

Results to retain

Calculation table distinguishing targets, representations, and sampling probabilities.

import numpy as np
online=np.array([2.,4.,3.]);target=np.array([5.,1.,2.])
print('DQN',1+.95*target.max(),'Double',1+.95*target[online.argmax()])
advantage=np.array([1.,2.,6.]);print('Dueling Q',2+advantage-advantage.mean())
error=np.array([.1,1.,5.])
for alpha in [0,.6]:
    probability=(error+1e-5)**alpha;probability/=probability.sum();print(alpha,probability)

Record results and self-check

Technique:

What changes:

Calculation example:

Target problem:

Points to check

Did you distinguish the networks used for action selection and evaluation?
Did you distinguish sampling probabilities from loss weights?

Validate only one needed improvement

Starting conditions

On Queue, compare dqn with just one of double, dueling, or per. Use 1800 transitions for each, the same seed=0, and the same evaluation set.

Procedure

Find where the selected method changes the code and first state why it may change the results.
Run dqn and the selected method with the same training budget and evaluation set.
Record which of return, queue length, and cost actually changed, and judge whether the method is needed based on these results.

Results to retain

Selection rationale and an evaluation table for both configurations. If needed, leave rerunning seed=1 as an extension.

"""Synthetic service queue. An action chooses capacity, including its energy cost."""
import numpy as np

class Queue:
    def reset(self, seed=0):
        self.rng = np.random.default_rng(seed)
        self.q, self.t = 0, 0
        return self.q

    def step(self, action):
        capacity = (0, 2, 4)[int(action)]
        arriving = int(self.rng.poisson(2))
        raw = max(0, self.q - capacity) + arriving
        dropped = max(0, raw - 20)
        self.q = min(20, raw)
        self.t += 1
        reward = -(self.q + .2 * capacity**2 + 10 * dropped)
        return self.q, reward, False, self.t >= 100, {'queue': self.q, 'capacity': capacity, 'dropped': dropped}

def evaluate(policy, seeds=range(100, 105)):
    rows = []
    for seed in seeds:
        env = Queue(); q = env.reset(seed)
        total = waiting = energy = dropped = 0
        while True:
            q, r, terminated, truncated, info = env.step(policy(q))
            total += r; waiting += q; energy += info['capacity']**2; dropped += info['dropped']
            if terminated or truncated: break
        rows.append([total, waiting / env.t, energy / env.t, dropped])
    return np.array(rows)

policies = {'constant': lambda q: 1, 'threshold': lambda q: 0 if q == 0 else (1 if q < 5 else 2)}
for name, policy in policies.items():
    print(name, evaluate(policy).mean(axis=0))

# Requires the Queue and evaluate definitions in the preceding cell.
import random
from collections import deque
import torch
from torch import nn

def train_dqn(seed=0, variant='dqn', steps=1800):
    torch.manual_seed(seed); rng = random.Random(seed)
    class Network(nn.Module):
        def __init__(self):
            super().__init__()
            self.shared = nn.Sequential(nn.Linear(1, 32), nn.ReLU())
            self.values = nn.Linear(32, 1 if variant == 'dueling' else 3)
            self.advantage = nn.Linear(32, 3)
        def forward(self, x):
            h = self.shared(x); v = self.values(h)
            if variant != 'dueling': return v
            a = self.advantage(h)
            return v + a - a.mean(dim=-1, keepdim=True)
    online = Network(); target = Network(); target.load_state_dict(online.state_dict())
    optimizer = torch.optim.Adam(online.parameters(), lr=.001)
    buffer = deque(maxlen=3000); priorities = deque(maxlen=3000)
    env = Queue(); q = env.reset(seed); losses = []; sample_counts = [0, 0]
    def tensor(states): return torch.tensor(states, dtype=torch.float32).reshape(-1, 1) / 20
    for step in range(steps):
        epsilon = max(.1, 1 - step / 1000)
        action = rng.randrange(3) if rng.random() < epsilon else int(online(tensor([q])).argmax())
        nq, reward, terminated, truncated, _ = env.step(action)
        buffer.append((q, action, reward / 20, nq, terminated))
        priorities.append(max(priorities, default=1.0))
        q = env.reset(seed + step + 1) if terminated or truncated else nq
        if len(buffer) < 64: continue
        probabilities = np.array(priorities, dtype=float)**.6
        probabilities /= probabilities.sum()
        if variant == 'per':
            # Sampling with replacement; importance weighting uses the same probabilities.
            ids = rng.choices(range(len(buffer)), weights=probabilities, k=32)
            weights = (len(buffer) * probabilities[ids])**(-.4)
            weights /= weights.max()
        else:
            ids = rng.sample(range(len(buffer)), 32); weights = np.ones(32)
        bq, ba, br, bn, bt = zip(*(buffer[i] for i in ids))
        predicted = online(tensor(bq)).gather(1, torch.tensor(ba).reshape(-1, 1)).flatten()
        with torch.no_grad():
            if variant == 'double':
                selected = online(tensor(bn)).argmax(dim=1, keepdim=True)
                next_value = target(tensor(bn)).gather(1, selected).flatten()
            else: next_value = target(tensor(bn)).max(dim=1).values
            expected = torch.tensor(br, dtype=torch.float32) + .95 * (~torch.tensor(bt)) * next_value
        td = expected - predicted
        loss = (torch.tensor(weights, dtype=torch.float32) * td.square()).mean()
        optimizer.zero_grad(); loss.backward(); nn.utils.clip_grad_norm_(online.parameters(), 5); optimizer.step()
        for i, delta in zip(ids, td.detach().numpy()): priorities[i] = abs(float(delta)) + 1e-5
        losses.append(float(loss.detach()))
        if step % 100 == 0: target.load_state_dict(online.state_dict())
    def policy(q):
        with torch.no_grad(): return int(online(tensor([q])).argmax())
    return policy, losses

for variant in ['dqn','double']:
    policy,loss=train_dqn(0,variant=variant);print(variant,evaluate(policy).mean(0))
# TODO: Choose one method to validate instead of double.

Record results and self-check

Training seed:

Training transition count:

Return mean and standard deviation:

Mean queue length:

Capacity²:

Dropped items:

Points to check

Did you avoid interpreting a decrease in loss as an immediate improvement in policy performance?
Did you record results that fell below the baseline in a short training run as observed?

Examine the distribution of transition data

Starting conditions

Compare transitions collected by Random and Rule in the public CartPole environment. Replay cannot create experiences outside the collection policies' visited range.

Procedure

Compare angle, angular velocity, and action distributions and episode lengths for both policies with the same seed.
Reload transitions saved as NPZ, and explain how drawing one transition repeatedly differs from experiencing a new state.
From the Minari Fourrooms-Random data card, find the collection policy, observation format, environment ID, and license, and tabulate the fields needed for your research data. Leave data downloading and training as extensions.

Results to retain

1 plot of the visited-state distribution and a data card summary table.

"""Small public-environment benchmark; evaluation traces are ordinary numeric arrays."""
import json
from importlib.metadata import version
from pathlib import Path
import numpy as np
import gymnasium as gym

def collect(env_id='CartPole-v1', policy=None, seeds=range(100,103)):
    env=gym.make(env_id); transitions=[]; summary=[]
    for seed in seeds:
        obs,info=env.reset(seed=seed);env.action_space.seed(seed);total=0.;length=0
        while True:
            action=env.action_space.sample() if policy is None else policy(obs)
            nxt,reward,terminated,truncated,info=env.step(action)
            transitions.append((obs.copy(),np.asarray(action).copy(),reward,nxt.copy(),terminated,truncated,seed))
            obs=nxt;total+=reward;length+=1
            if terminated or truncated:break
        summary.append([seed,total,length,int(terminated),int(truncated)])
    env.close()
    columns=list(zip(*transitions))
    data={k:np.asarray(v) for k,v in zip(['observations','actions','rewards','next_observations','terminated','truncated','episode_seed'],columns)}
    return np.asarray(summary),data

def cartpole_rule(obs):return int(obs[2]+.25*obs[3]>0)
def zero_torque(obs):return np.zeros(1,dtype=np.float32)

def save_trace(path, env_id, data, policy_name, seeds):
    # Contains numeric arrays and JSON text only; never requires pickle.
    metadata=dict(env_id=env_id,policy=policy_name,seeds=list(seeds),gymnasium=version('gymnasium'))
    np.savez_compressed(path,**data,metadata_json=np.array(json.dumps(metadata)))
    with np.load(path,allow_pickle=False) as saved:
        assert len(saved['observations'])==len(saved['rewards'])
        print(json.loads(str(saved['metadata_json'])))

def public_model(algorithm='DQN',env_id='CartPole-v1',seed=7,steps=2048,**changes):
    from stable_baselines3 import DQN,PPO,SAC
    options=dict(policy_kwargs=dict(net_arch=[32,32]),seed=seed,verbose=0,device='cpu')
    if algorithm=='DQN':options.update(buffer_size=5000,learning_starts=100,target_update_interval=100,batch_size=32)
    if algorithm=='PPO':options.update(n_steps=128,batch_size=64,n_epochs=3)
    if algorithm=='SAC':options.update(buffer_size=5000,learning_starts=100,batch_size=64)
    options.update(changes)
    env=gym.make(env_id)
    model={'DQN':DQN,'PPO':PPO,'SAC':SAC}[algorithm]('MlpPolicy',env,**options)
    model.learn(total_timesteps=steps)
    env.close()
    return model

print('Gymnasium',version('gymnasium'))
env=gym.make('CartPole-v1')
print('Observation:',env.observation_space,'Action:',env.action_space,'Limit:',env.spec.max_episode_steps)
env.close()
# For visual inspection, create a separate env with render_mode='rgb_array'.
# Rendering requires gymnasium[classic-control]; do not render every training step.

import matplotlib.pyplot as plt
for name,policy in [('random',None),('rule',cartpole_rule)]:
    rows,data=collect(policy=policy)
    plt.hist(data['observations'][:,2],bins=20,alpha=.5,density=True,label=name)
    save_trace(name+'_replay.npz','CartPole-v1',data,name,range(100,103))
plt.xlabel('Pole angle [rad]');plt.ylabel('Density');plt.legend()
# TODO: Compare the action distribution with the dataset card entries.

Record results and self-check

Materials:

Collection policy:

Environment and version:

Observation and action format:

Distribution and terms of use:

Points to check

Did you avoid treating randomly shuffled transitions as a continuous episode?
Did you distinguish data collected directly from a public environment from an external public dataset?

DQN Variants and Replay Buffer Sampling

Learning objectives

Execution conditions

Applications

Throughput and cost of equipment and communication resources

Choose equipment inspection times

Distinguish selection, evaluation, representation, and sampling

Starting conditions

Procedure

Results to retain

Record results and self-check

Points to check

Validate only one needed improvement

Starting conditions

Procedure

Results to retain

Record results and self-check

Points to check

Examine the distribution of transition data

Starting conditions

Procedure

Results to retain

Record results and self-check

Points to check

Official examples and data