DQN：把 Q-learning 接上神經網路

state 0 → [1,0,0]
state 1 → [0,1,0]
state 2 → [0,0,1]

one-hot 3 → Linear(3,8) → ReLU → Linear(8,2)
輸出：[Q(s,L), Q(s,R)]

第一層：3×8 weights + 8 biases = 32
第二層：8×2 weights + 2 biases = 18
合計：50 個可訓練 scalar parameters

狀態 inputs                  [16,3]
online(inputs)               [16,2]
已選 actions                 [16]，int64
挑出的 chosen_Q              [16]
rewards、terminal、labels     [16]

next_max = max_a Q_target(next_state,a)
label = reward + gamma × (1 − terminal) × next_max
loss = mean((chosen_Q − label)²)

−0.1 + 0.9×0.3433555501 ≈ 0.2090199951

torch==2.3.1

python3 tiny_dqn.py > stdout.json

"""Original one-hot MDP DQN mechanics, not an Atari CNN or DQN benchmark."""
import collections
import copy
import hashlib
import json
import math
import platform
import random

import torch
from torch import nn


SEED, REPLAY_SEED = 7, 7007
EPISODES, MAX_STEPS = 600, 20
GAMMA, EPSILON, LEARNING_RATE = .9, .2, .01
BATCH_SIZE, CAPACITY, TARGET_PERIOD = 16, 512, 50
GOAL, TRAP = 3, 4
DTYPE = torch.float64


def environment_step(state, action):
    table = {(0, 0): (TRAP, -1., True), (0, 1): (1, -.1, False),
             (1, 0): (0, -.1, False), (1, 1): (2, -.1, False),
             (2, 0): (1, -.1, False), (2, 1): (GOAL, 1., True)}
    return table[state, action]


def encode(states):
    # Terminal input is never used for choosing actions. Its zero vector is masked below.
    return torch.tensor([[float(state == k) for k in range(3)] for state in states], dtype=DTYPE)


def make_network():
    return nn.Sequential(nn.Linear(3, 8), nn.ReLU(), nn.Linear(8, 2)).to(dtype=DTYPE, device="cpu")


def parameter_digest(network):
    payload = json.dumps([p.detach().tolist() for p in network.parameters()], separators=(",", ":"))
    return hashlib.sha256(payload.encode("ascii")).hexdigest()


def batch_tensors(transitions):
    states, actions, rewards, next_states, terminated = zip(*transitions)
    return (encode(states), torch.tensor(actions, dtype=torch.int64),
            torch.tensor(rewards, dtype=DTYPE), encode(next_states),
            torch.tensor(terminated, dtype=torch.bool))


def targets_for_batch(target, rewards, next_inputs, terminated):
    with torch.no_grad():
        next_max = target(next_inputs).max(dim=1).values
        values = rewards + GAMMA * (~terminated).to(DTYPE) * next_max
    return values, next_max


def choose(online, state, rng):
    if rng.random() < EPSILON:
        return rng.randrange(2)
    with torch.no_grad():
        # torch.argmax returns the first index in a tie.
        return int(online(encode([state]))[0].argmax())


def model_oracle():
    q = [[0., 0.] for _ in range(3)]
    for sweep in range(1, 1001):
        updated = []
        for state in range(3):
            row = []
            for action in range(2):
                next_state, reward, terminal = environment_step(state, action)
                row.append(reward if terminal else reward + GAMMA * max(q[next_state]))
            updated.append(row)
        delta = max(abs(updated[s][a] - q[s][a]) for s in range(3) for a in range(2))
        q = updated
        if delta < 1e-14:
            return q, sweep
    raise RuntimeError("Oracle did not stabilize")


def greedy_evaluate(network):
    state, trajectory = 0, []
    with torch.no_grad():
        for _ in range(MAX_STEPS):
            action = int(network(encode([state]))[0].argmax())
            next_state, reward, terminal = environment_step(state, action)
            trajectory.append({"state": state, "action": action, "reward": reward,
                               "next_state": next_state, "terminated": terminal})
            state = next_state
            if terminal:
                break
    return {"trajectory": trajectory,
        "undiscounted_return": math.fsum(x["reward"] for x in trajectory),
        "discounted_return": math.fsum(GAMMA**t * x["reward"] for t, x in enumerate(trajectory)),
        "terminated": trajectory[-1]["terminated"], "truncated": not trajectory[-1]["terminated"]}


def main():
    torch.set_num_threads(1)
    torch.manual_seed(SEED)
    behavior_rng, replay_rng = random.Random(SEED), random.Random(REPLAY_SEED)
    online = make_network()
    initial = copy.deepcopy(online)
    target = copy.deepcopy(online)
    target.requires_grad_(False)
    target.eval()
    optimizer = torch.optim.Adam(online.parameters(), lr=LEARNING_RATE, foreach=False)
    replay = collections.deque(maxlen=CAPACITY)
    visits = [[0, 0] for _ in range(3)]
    endings = {"goal": 0, "trap": 0, "truncated": 0}
    updates, decisions, terminal_batch_entries, nonterminal_batch_entries = 0, 0, 0, 0
    copies, losses, samples = [], [], {"terminal": [], "nonterminal": []}
    first_batch = None
    target_unchanged_during_optimizer = True
    target_grad_always_none = True
    first_episode = []
    for episode in range(EPISODES):
        state = 0
        for step in range(1, MAX_STEPS + 1):
            action = choose(online, state, behavior_rng)
            next_state, reward, terminal = environment_step(state, action)
            truncated = step == MAX_STEPS and not terminal
            transition = (state, action, reward, next_state, terminal)
            replay.append(transition)
            visits[state][action] += 1
            decisions += 1
            if episode == 0:
                first_episode.append(list(transition))
            if len(replay) >= BATCH_SIZE:
                batch = replay_rng.sample(list(replay), BATCH_SIZE)  # without replacement within batch
                inputs, actions, rewards, next_inputs, terminated = batch_tensors(batch)
                labels, next_max = targets_for_batch(target, rewards, next_inputs, terminated)
                selected = online(inputs).gather(1, actions[:, None]).squeeze(1)
                loss = ((selected - labels)**2).mean()
                if first_batch is None:
                    first_batch = {"transitions": batch, "labels": labels.detach().clone(),
                                   "initial_selected_predictions": selected.detach().tolist()}
                target_before = parameter_digest(target)
                optimizer.zero_grad(set_to_none=True)
                loss.backward()
                target_grad_always_none &= all(p.grad is None for p in target.parameters())
                optimizer.step()
                updates += 1
                target_unchanged_during_optimizer &= parameter_digest(target) == target_before
                losses.append(float(loss.detach()))
                terminal_batch_entries += int(terminated.sum())
                nonterminal_batch_entries += int((~terminated).sum())
                for j, trans in enumerate(batch):
                    kind = "terminal" if trans[4] else "nonterminal"
                    if len(samples[kind]) < 3:
                        samples[kind].append({"optimizer_update": updates,
                            "transition": list(trans), "target_network_next_max": float(next_max[j]),
                            "continuation_multiplier": 0 if trans[4] else 1,
                            "label": float(labels[j]), "selected_q_before_step": float(selected.detach()[j]),
                            "label_requires_grad": labels.requires_grad})
                if updates % TARGET_PERIOD == 0:
                    target.load_state_dict(online.state_dict())
                    copies.append({"optimizer_update": updates, "environment_decisions": decisions,
                        "online_digest": parameter_digest(online), "target_digest": parameter_digest(target),
                        "equal_after_copy": parameter_digest(target) == parameter_digest(online)})
            if terminal:
                endings["goal" if next_state == GOAL else "trap"] += 1
                break
            if truncated:
                endings["truncated"] += 1
                break
            state = next_state
    with torch.no_grad():
        final_q = online(encode(range(3))).tolist()
        initial_q = initial(encode(range(3))).tolist()
        inputs, _, _, _, _ = batch_tensors(first_batch["transitions"])
        first_actions = torch.tensor([t[1] for t in first_batch["transitions"]], dtype=torch.int64)
        final_on_first = online(inputs).gather(1, first_actions[:, None]).squeeze(1)
        frozen_first_label_mse_after = float(((final_on_first - first_batch["labels"])**2).mean())
    oracle_q, sweeps = model_oracle()  # Evaluator only, never training labels.
    oracle_error = max(abs(final_q[s][a] - oracle_q[s][a]) for s in range(3) for a in range(2))
    bellman_residual = []
    for state in range(3):
        for action in range(2):
            ns, reward, terminal = environment_step(state, action)
            label = reward if terminal else reward + GAMMA * max(final_q[ns])
            bellman_residual.append(abs(final_q[state][action] - label))
    # Controlled frozen-target/sync probe, separate from and not modifying trained networks.
    probe_online, probe_target = copy.deepcopy(online), copy.deepcopy(target)
    probe_before = parameter_digest(probe_target)
    with torch.no_grad():
        next(probe_online.parameters()).add_(.123)
    unchanged_before_copy = parameter_digest(probe_target) == probe_before
    different_before_copy = parameter_digest(probe_target) != parameter_digest(probe_online)
    probe_target.load_state_dict(probe_online.state_dict())
    equal_after_probe_copy = parameter_digest(probe_target) == parameter_digest(probe_online)
    changed_tensors = sum(not torch.equal(a, b) for a, b in zip(initial.parameters(), online.parameters()))
    assert sum(endings.values()) == EPISODES
    assert sum(map(sum, visits)) == decisions
    assert updates == decisions - BATCH_SIZE + 1
    assert len(copies) == updates // TARGET_PERIOD
    assert target_unchanged_during_optimizer and target_grad_always_none
    assert all(x["equal_after_copy"] for x in copies)
    assert unchanged_before_copy and different_before_copy and equal_after_probe_copy
    print(json.dumps({
        "scope": "Original single-seed deterministic one-hot MDP neural Q-learning with replay and hard targets; "
                 "not Atari, a CNN benchmark, a multi-seed study or a convergence guarantee.",
        "versions": {"python": platform.python_version(), "torch": torch.__version__,
                     "device": "cpu", "dtype": "float64", "threads": torch.get_num_threads()},
        "settings": {"episodes": EPISODES, "max_steps": MAX_STEPS, "gamma": GAMMA, "epsilon": EPSILON,
            "torch_and_behavior_seed": SEED, "separate_replay_seed": REPLAY_SEED,
            "network": "onehot3 -> Linear3x8+bias -> ReLU -> Linear8x2+bias;50parameters",
            "optimizer": "Adam lr.01 foreach=False, no weight decay", "batch_size": BATCH_SIZE,
            "capacity": CAPACITY, "target_copy_every_optimizer_updates": TARGET_PERIOD,
            "warmup": "first15observations collected without optimization; first update at decision16",
            "sampling": "uniform without replacement within each minibatch, repeated reuse across updates",
            "update_frequency": "one optimizer update per new environment decision once replay size>=16",
            "loss": "mean squared error on chosen action Q only; target from frozen target network, no gradient",
            "terminal": "r only for true goal/trap; zero-input terminal Q is masked and has no semantic state value",
            "truncation": "collector limit only, retains bootstrap on observed nonterminal next state"},
        "actual_training": {"decisions": decisions, "optimizer_updates": updates,
            "target_copies_excluding_initial_clone": len(copies), "episode_endings": endings,
            "visits_L_R": visits, "final_replay_size": len(replay),
            "terminal_batch_entries": terminal_batch_entries,
            "nonterminal_batch_entries": nonterminal_batch_entries,
            "first_episode_observed_transitions": first_episode,
            "first_update_loss": losses[0], "last_update_loss": losses[-1],
            "mean_first_50_update_losses": math.fsum(losses[:50]) / len(losses[:50]),
            "mean_last_50_update_losses": math.fsum(losses[-50:]) / len(losses[-50:]),
            "loss_scope": "different sampled batches and changing targets; not a fixed validation learning curve",
            "changed_parameter_tensors": changed_tensors,
            "parameter_count_online_only": sum(p.numel() for p in online.parameters()),
            "target_unchanged_across_every_optimizer_step": target_unchanged_during_optimizer,
            "target_grad_always_none": target_grad_always_none},
        "initial_online_q_L_R": initial_q, "final_online_q_L_R": final_q,
        "oracle_after_training_only": {"q_L_R": oracle_q, "sweeps": sweeps,
            "max_online_abs_error": oracle_error, "max_online_Bellman_residual": max(bellman_residual)},
        "greedy_evaluation_frozen_online": greedy_evaluate(online),
        "saved_training_target_examples": samples, "actual_target_copy_log": copies,
        "fixed_first_minibatch_old_target_diagnostic": {
            "description": "Same first minibatch and its original labels, NOT new Bellman targets or validation",
            "initial_mse": losses[0], "final_mse_against_old_labels": frozen_first_label_mse_after,
            "old_labels": first_batch["labels"].tolist()},
        "controlled_copy_probe_not_training": {"target_unchanged_after_perturbing_probe_online": unchanged_before_copy,
            "probe_networks_different_before_copy": different_before_copy,
            "probe_networks_equal_after_copy": equal_after_probe_copy},
        "parameter_records": {"initial_online": {k: v.tolist() for k, v in initial.state_dict().items()},
            "final_online": {k: v.tolist() for k, v in online.state_dict().items()},
            "final_target": {k: v.tolist() for k, v in target.state_dict().items()}},
        "limits": ["600episodes and allhyperparameters fixed before this one run; no retuning to oracle error.",
            "Oracle and allstate diagnostics are computed after learning, never used for replay or labels.",
            "Replay reduces serial coupling but is finite/policy-generated, not a proof of iid coverage.",
            "Hardtarget copies are counted by optimizer updates, not episodes or environment-only warmup.",
            "No-gradient is enforced for labels and target params; only online params are in optimizer.",
            "MLP shares parameters: unchosen outputs can change indirectly despite chosen-only loss.",
            "ReLU/onehot/Adam/MSE/fixedepsilon/.9gamma differ from original Atari CNN/RMSProp/preprocessing.",
            "2013paper has replay;2015algorithm explicitly uses separate periodically copied target network.",
            "Initial/final changingbatch losses and oracle error are not generalization/online-performance proof.",
            "CPUfloat64 only;installedCUDA-taggedwheel does not mean GPU was used or tested."]}, indent=2))


if __name__ == "__main__":
    main()

Source: “Figura 03 arquitectura dqn.svg” / “Compoñentes dun axente DQN” by Juanciño, https://commons.wikimedia.org/wiki/File:Figura_03_arquitectura_dqn.svg, licensed CC BY-SA 4.0, https://creativecommons.org/licenses/by-sa/4.0/. Changes: rendered the full SVG as PNG; Figure 02 is a crop of two complete boxes and their connecting arrow with 6 px white padding; the cover is a proportional full-figure fit on a white canvas. These adaptations are provided under CC BY-SA 4.0. This later explanatory schematic is not a Mnih et al. paper figure.

Source: PyTorch Tutorials, “Reinforcement Learning (DQN) Tutorial,” by Adam Paszke and Mark Towers, https://docs.pytorch.org/tutorials/intermediate/reinforcement_q_learning.html. The data-flow diagram comes from the tutorial repository at commit 5672cd98a7671eda4d9122b10f8eec580f74d03e; the Huber equation is a screenshot of the page’s KaTeX-rendered piecewise definition only (the adjacent batch mean is outside the crop); the duration chart is the tutorial’s CartPole-v1 output. Changes: data-flow image reproduced as provided; Huber screenshot cropped only to remove white margins; result chart reproduced whole. The diagram shows the tutorial’s soft target update, and the result is CartPole, not Atari or this project’s run. The tutorial repository’s BSD 3-Clause license is reproduced below.

BSD 3-Clause License (verbatim from pinned PyTorch Tutorials repository)
BSD 3-Clause License

Copyright (c) 2017-2022, Pytorch contributors
All rights reserved.

Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are met:

* Redistributions of source code must retain the above copyright notice, this
  list of conditions and the following disclaimer.

* Redistributions in binary form must reproduce the above copyright notice,
  this list of conditions and the following disclaimer in the documentation
  and/or other materials provided with the distribution.

* Neither the name of the copyright holder nor the names of its
  contributors may be used to endorse or promote products derived from
  this software without specific prior written permission.

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

KaTeX renderer and fonts used in the formula screenshot
The live page reports KaTeX 0.16.10; the font files were served from the katex@0.10.0-beta package URL. The screenshot contains rendered glyphs only and does not redistribute the font binaries. Their license notices are reproduced below.

KaTeX code MIT License (verbatim from KaTeX v0.10.0-beta)
The MIT License (MIT)

Copyright (c) 2015 Khan Academy

This software also uses portions of the underscore.js project, which is
MIT licensed with the following copyright:

Copyright (c) 2009-2015 Jeremy Ashkenas, DocumentCloud and Investigative
Reporters & Editors

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

KaTeX font package MIT License (verbatim from KaTeX fonts repository)
The MIT License (MIT)

Copyright (c) 2018 Khan Academy

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.