表格 Q-learning为每个状态、动作保存一个数。状态很多、观察是图像时，这张表难以直接列全；DQN 用神经网络根据观察估计每个离散动作的 Q 值，再沿 TD 误差更新参数。

把 Q 表换成网络还不够：相邻经验高度相关，当前网络又参与生成自己的学习目标，训练可能变成追逐不断移动的数值。经验回放复用并打散样本，目标网络延迟学习目标的变化。 它们是稳定化设计，不是任何环境都保证收敛的证明。

Q 值不是动作概率

网络输入是状态或观察的表示，输出维度等于动作数。Q(s,a) 估计采取动作 a 后的折扣回报，可以为负，也不要求各动作值之和为 1。不要给输出加 Softmax，误把动作价值当作概率。

动作可按 ε-greedy选择；训练时只对实际执行动作的输出计算这一条经验的损失。共享隐藏层更新仍可能影响其他动作的输出，不等于“未选动作永远不变”。

方法 网络或表输出 学习信号
表格 Q-learning 每个状态、动作的独立数值 直接更新这个表项
DQN 所有离散动作的 Q 值 固定 TD 目标下的回归误差
策略梯度 动作分布的参数 用回报或优势更新策略

这里的“神经”指函数近似器，不规定一定用卷积。CNN适合某些具有空间结构的输入；本文用只有一个输入数的微型多层感知机，让参数与梯度可以逐项手算。

经验回放与目标网络分别解决什么

Mnih 等人的原始论文描述了容量有限的经验回放和独立目标网络。回放保存旧策略采集的转移并抽取小批次；目标网络定期复制在线参数，在两次复制之间固定。

组件 本文实现 不要误解成
回放池 保存最近 N 条转移，均匀无放回抽样 采样后完全独立，或自动覆盖从未探索的状态
在线网络 θ 当前 Q 预测与动作选择；梯度更新它 每条经验有自己独立的一套参数
目标网络 θ⁻ 生成下一状态的目标值，固定到下次同步 另一个会同时接受同一损失梯度的网络
同步周期 本文按优化器更新次数计 忘记区分环境步、批次更新与 episode

回放数据来自旧的行为策略，所以 DQN 是 off-policy 方法；并不意味着任意过期、非平稳数据都同样合适。

  1. 抽取已经发生的转移

    状态、动作、奖励、下一状态与终止标志必须匹配同一步。

  2. 先用目标网络冻结学习目标

    本次梯度只走在线预测路径，不穿过下一状态的最大值。

  3. 更新在线参数，再按计划同步

    深复制参数，不能让两个网络共享同一组可变列表。

一个训练步的数字从哪里来

设在线网络为 1→2→2：输入 x=1，隐藏层权重为 1、2，偏置均 0，经 ReLU 得隐藏值 [1,2]。两个动作的输出权重分别为 [1,0.5]、[−1,0]，偏置 0，所以 Q=[2,−1]。

本条经验采取动作 0，奖励 1，下一状态输入仍为 1。目标网络的输出偏置为 [1,2]，其余参数与初始在线网络相同，因此下一状态 Q⁻=[3,1]。取 γ=0.9、未终止，目标 y=1+0.9×3=3.7。

量 计算 结果
实际动作的预测 Q(s,0) 2
固定目标 1+0.9×max(3,1) 3.7
本例半平方损失 0.5×(2−3.7)² 1.445
对预测的梯度 2−3.7 −1.7
动作 0 输出权重梯度 −1.7×[1,2] [−1.7,−3.4]
动作 1 输出层参数的直接梯度 没有作为本条经验的预测 0
在线预测2与冻结目标3.7构成损失1.445，梯度只流向在线网络，目标网络由延迟复制更新
原创合成训练步示意。网络和转移均为教学设定，不是 Atari 训练或性能实测。查看原图

小批次损失取半平方误差的平均值，因此每条预测的梯度是 (Q−y)/B，不能漏掉批次大小 B。目标 y 在这个训练步内被当作常数，这通常称为半梯度更新。

真正终止与外部截断不能混成一个 done

目标为 **y=r+γ×(1−terminated)×max_a Q⁻(s′,a)**。真正终止不再自举；外部截断使采样停止，但不必意味着 MDP 已进入终止状态。Gymnasium 的时间限制说明明确区分这两种情况。

当前转移 本例目标 原因
未终止 3.7 保留后续价值
真正终止 1 没有后续回报
只有外部截断 3.7 本例继续自举，使用截断时的最终观察
任务本身的有限时域结束 按任务的终止定义处理 不能仅看它也叫“时间上限”；状态还应表达所需剩余时间

外部截断后的下一状态不能偷换成 reset 返回的新回合初始观察。实际环境或自动重置包装器需要核对最终观察的提供方式；本文只使用明确给出的合成转移。

纯 Python 神经网络与回放实验

依赖 Python 3.10 或更新版本的标准库，固定一个输入、两个隐藏单元、两个离散动作。实现前向、手写反向、小批次平均、有界回放与硬同步；使用半平方误差和简单 SGD，不实现原论文的误差裁剪、图像预处理或 Atari 交互训练。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
from collections import deque
from copy import deepcopy
from dataclasses import dataclass
import json
import math
import random


def finite(value):
    if type(value) not in (int, float) or not math.isfinite(value):
        raise ValueError("invalid number")


@dataclass(frozen=True)
class Transition:
    state: tuple
    action: int
    reward: float
    next_state: tuple
    terminated: bool
    truncated: bool = False


def validate(t):
    if not isinstance(t, Transition):
        raise ValueError("invalid transition")
    for state in (t.state, t.next_state):
        if type(state) is not tuple or len(state) != 1:
            raise ValueError("expected one-feature tuple")
        finite(state[0])
    if type(t.action) is not int or not 0 <= t.action < 2:
        raise ValueError("invalid action")
    finite(t.reward)
    if type(t.terminated) is not bool or type(t.truncated) is not bool:
        raise ValueError("invalid stop flags")


def make_net():
    return {"w1": [[1.0], [2.0]], "b1": [0.0, 0.0],
            "w2": [[1.0, 0.5], [-1.0, 0.0]], "b2": [0.0, 0.0]}


def forward(net, state):
    if len(state) != 1:
        raise ValueError("expected one feature")
    finite(state[0])
    z = [net["w1"][j][0] * state[0] + net["b1"][j] for j in range(2)]
    hidden = [max(0.0, value) for value in z]
    q = [sum(net["w2"][a][j] * hidden[j] for j in range(2)) + net["b2"][a]
         for a in range(2)]
    return q, (state[0], z, hidden)


def targets(target, batch, gamma=0.9):
    finite(gamma)
    if not 0 <= gamma <= 1:
        raise ValueError("invalid discount")
    values = []
    for t in batch:
        validate(t)
        values.append(t.reward if t.terminated else
                      t.reward + gamma * max(forward(target, t.next_state)[0]))
    return values


def loss_grad(online, batch, values):
    if not batch or len(batch) != len(values):
        raise ValueError("invalid batch")
    grad = {"w1": [[0.0], [0.0]], "b1": [0.0, 0.0],
            "w2": [[0.0, 0.0], [0.0, 0.0]], "b2": [0.0, 0.0]}
    loss = 0.0
    for t, y in zip(batch, values):
        validate(t)
        finite(y)
        q, (x, z, hidden) = forward(online, t.state)
        error = q[t.action] - y
        loss += 0.5 * error * error / len(batch)
        dq = error / len(batch)
        grad["b2"][t.action] += dq
        for j in range(2):
            grad["w2"][t.action][j] += dq * hidden[j]
            dz = dq * online["w2"][t.action][j] if z[j] > 0 else 0.0
            grad["w1"][j][0] += dz * x
            grad["b1"][j] += dz
    return loss, grad


def train_step(online, target, batch, *, gamma=0.9, lr=0.01):
    finite(lr)
    if lr <= 0:
        raise ValueError("invalid learning rate")
    values = targets(target, batch, gamma)
    loss, grad = loss_grad(online, batch, values)
    for name, parameter in online.items():
        for i in range(len(parameter)):
            if isinstance(parameter[i], list):
                for j in range(len(parameter[i])):
                    parameter[i][j] -= lr * grad[name][i][j]
            else:
                parameter[i] -= lr * grad[name][i]
    return loss


class Replay:
    def __init__(self, capacity):
        if type(capacity) is not int or capacity < 1:
            raise ValueError("invalid capacity")
        self.data = deque(maxlen=capacity)

    def add(self, transition):
        validate(transition)
        self.data.append(transition)

    def sample(self, size, rng):
        if type(size) is not int or not 1 <= size <= len(self.data):
            raise ValueError("invalid sample size")
        return rng.sample(list(self.data), size)


if __name__ == "__main__":
    online = make_net()
    target = deepcopy(online)
    target["b2"] = [1.0, 2.0]
    before_target = deepcopy(target)
    continuing = Transition((1.0,), 0, 1.0, (1.0,), False)
    terminal = Transition((1.0,), 0, 1.0, (1.0,), True)
    truncated = Transition((1.0,), 0, 1.0, (1.0,), False, True)
    memory = Replay(2)
    for item in (continuing, terminal, truncated):
        memory.add(item)
    sample = memory.sample(2, random.Random(7))
    frozen = targets(target, [continuing])
    before = train_step(online, target, [continuing])
    after = loss_grad(online, [continuing], frozen)[0]
    print(json.dumps({"targets": targets(target, [continuing, terminal, truncated]),
                      "before": before, "after": after,
                      "target_unchanged": target == before_target,
                      "sample_size": len(sample)}, indent=2))
    target = deepcopy(online)  # Hard synchronization, when the chosen update period is reached.

运行可核对 targets 为 [3.7,1,3.7]、首次损失 1.445、固定目标下更新后的损失下降，且同步前目标参数不变。回放容量为 2，第三条加入后最旧一条被替换。这个结果只检验合成训练步，不证明学会了一个环境的最优策略。

对批次大小 B，这个固定小网络的训练时间 O(B)、目标列表占 O(B)，回放容量 N 的存储 O(N)。本例抽样会把 deque 转成列表，因此一次 sample 还需要 O(N) 时间与临时空间；不能把它当作大型训练回放池的最优实现。

训练和评估的边界

易错点 检查方式
目标和在线网络浅复制 修改在线参数后，目标参数必须保持不变
对目标值反向求导 冻结目标后再做数值差分，不把目标也随在线参数一起变化
漏掉实际动作索引 本条经验只直接回归所执行动作的 Q 输出
把截断视为终止 对照三种目标值，并检查最终观察来源
用 Softmax 输出 Q Q 是价值，不是概率，允许负数
缓冲池不足批大小 先完成采样条件检查，不用无效空批次更新
训练损失降低 另做环境回报评估，多随机种子，不能替代策略质量指标
隐藏单元恰在 ReLU 零点 代码采用零导数约定，中心差分在不可微点不一定一致
自检：因外部时间上限停止这一回合，就一定要把下一状态 Q 值清零吗？

不一定。本例的外部截断不等于真正终止，仍使用最终观察自举。如果时间上限本来就是任务定义的一部分，则按该 MDP 的终止与状态定义处理。

项目 tests/verify-dqn-principles-article.cjs 用冻结目标的中心差分检查所有在线参数梯度，核验批次平均、未选动作的直接梯度、回放覆盖与采样，以及目标深复制。继续比较策略梯度与 Actor-Critic，可以分清“先回归动作价值再选动作”和“直接优化动作分布”的差别。