DQN

Learning objectives

Compare state representations in a Q-table and a neural network on the same Queue problem, and evaluate the learned policy against rule-based policies.

Execution conditions

Run the baseline Queue example with NumPy. Compare policies on seed=100…104, which was not used for training.

PyTorch is required. This is a short synthetic experiment; its configuration does not guarantee learning performance.

Public environment activities use gymnasium and stable-baselines3. gymnasium[classic-control] is required only for visualization.

If the required package is missing, run %pip install control==0.10.2 gymnasium==1.3.0 stable-baselines3==2.9.2 in a Colab or Jupyter code cell.

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.

Design energy storage operation

A research design case distinguishing the roles of price, load, and stored energy before connecting measured data.

Observation and state: SOC, load, price, and time. Observations must exclude future information.

Action: charging/discharging power. First choose discrete levels or a continuous range.

Reward and constraints: electricity and degradation costs. Specify rules for handling SOC and power limits.

Comparison baseline: time-of-day rules and, if possible, optimization using the same information.

Metrics to check: total cost, constraint violations, and maximum grid power draw. Split train/test data by date.

Connect one key change to actual policy metrics instead of adding more full retraining runs.

DQN on the same small problem

Starting conditions

• Environment: Queue.
• Network: $1\to32\to3$.
• Optimizer: Adam, lr=.001.
• Discount Factor: $\gamma=.95$.
• Replay: 3000. Batch: 32.
• Training budget: 1800 transitions.

Compare how Q-tables and neural networks represent actions for each state, and evaluate the learned policy.

Procedure

1. Check the performance of Queue's rule-based policies first, then run the provided training function.
2. Plot training loss and selected actions for $q=0$…20.
3. Evaluate return, queue length, cost, and dropped items on seed=100…104, which was not used for training.

Results to retain

• Loss and actions by state plots, and an evaluation table comparing with the rule-based policy.

"""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

policy,losses=train_dqn(0,variant='dqn')
print(evaluate(policy).mean(0))

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?

Move to a public DQN implementation

Starting conditions

SB3 DQN, CartPole-v1, and 2048 transitions. Compare with random and rule-based policies in the same environment, without directly comparing scores with the earlier Queue results.

Procedure

1. Map the public DQN's policy, replay size, target update interval, and learning start to the preceding implementation.
2. Run the provided short training and evaluate deterministically on seed=100…102.
3. Compare episode lengths and returns of the learned policy, Random, and Rule, and state the evidence for judging whether the training budget is sufficient.

Results to retain

• An implementation configuration mapping table and a common evaluation 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.

model=public_model('DQN')
for name,policy in [('random',None),('rule',cartpole_rule),('dqn',lambda o:int(model.predict(o,deterministic=True)[0]))]:
    rows,data=collect(policy=policy);print(name,rows[:,1:3].mean(0),rows[:,1:3].std(0))
# TODO: Locate and connect replay and target components in the public implementation and the preceding code.

Record results and self-check

Policy:

Training transition count:

Evaluation return mean and standard deviation:

Mean length:

Points to check

• Did you avoid assuming SB3 DQN includes Double, Dueling, and PER?
• Did you avoid claiming superiority based on Queue and CartPole scores from different environments?

Apply this to your research problem

Starting conditions

Based on the preceding experiments, choose Queue, Motor, or one of your research problems and design an experiment for real application.

Procedure

1. Define state, observation, action, reward, termination, and time limit in one table. Record each variable's units and the information you can actually obtain.
2. Write 3 expected transitions numerically and check whether rewards for good and bad actions match your intent.
3. Choose one rule-based comparison baseline, one evaluation condition, and one failure case of greatest concern. Also list the required measurements or simulator.

Results to retain

• MDP design table, 3 transition examples, and comparative experiment plan.
• If real research data are unavailable, design using synthetic conditions from an application case.

You may continue with the research problem and design table selected earlier. Revisit existing entries and transition examples using the concepts learned in this activity, and update changed entries or reasons for retaining them.

Record results and self-check

Item:

Concrete definition and units:

Observability and verification method:

Points to check

• Did you exclude future information and evaluation labels from observations?
• Did you define domain-specific metrics rather than judging the research contribution solely by training return?

Official examples and data

• SB3 · DQN
• CartPole · environment definition