【人工智能】使用DQN进行CartPole游戏的经典强化学习训练

使用DQN进行CartPole游戏的经典强化学习训练

一、无经验回放

先放一个没有经验池(经验回放)的代码

Q-learning 的直观理解:从走迷宫到贝尔曼方程

在深入代码之前,我们先用一个简单的例子来建立对 Q-learning 的直觉。这是理解 DQN 中最重要的一步——因为如果不知道 Q 函数到底在做什么,后面的经验回放、ε-greedy 探索就只是一堆"看起来有用"的技巧。

想象你站在一个 5×5 的迷宫格子里,当前坐标是 (3, 4),目标是走到出口。你面前有四个选择:向上、向下、向左、向右。问题是——你该怎么选?

Q-learning 给出的答案很直接:哪个动作的 Q 值 最大,就选哪个。那么 Q 值到底是什么?

Q(s, a) 的含义是:在状态 s 下执行动作 a,之后一直按照最优策略走,未来能获得的累积奖励总和。

也就是说,Q( (3,4), “向上” ) 代表:你站在格子 (3,4),先向上走一步,然后每一步都以最佳方式决策——这样一路走到终点的总得分。它不是只看眼前这一步,而是把"未来的可能性"都打包进了一个数字里。

那怎么计算 Q 值呢?这里就引出了强化学习中最重要的公式——贝尔曼方程(Bellman Equation):

Q(s, a) = r  +  γ · max_a' Q(s', a')

这个公式的理解其实非常朴素。想象你在银行存钱:

今天你存入 100 块(即时奖励 r),年利率是 5%。一年后你的总钱 = 100 + 100×0.05 ≈ 105。贝尔曼方程做的事完全一样——当前价值 = 即时奖励 + 打了折扣的未来最大可能价值

其中 γ(gamma,折扣因子) 的作用是给未来的奖励"打折"。γ=0.99 意味着:1 步之后的奖励只值现在的 99%,2 步之后值 99%×99%≈98%。这样做有两个好处:

  1. 数学上保证无限序列的 Q 值不会发散到无穷大
  2. 直觉上鼓励智能体尽早拿奖励——拖延没有好处

好,那问题来了:迷宫只有 25 个格子,我们可以画一张 25×4=100 格的表格,每个格子存一个 Q 值(这叫 Q-Table)。但 CartPole 的状态是连续的——小车位置可以是 2.4 也可以是 2.41,杆子角度可以是 1.57° 也可以是 1.58°,你不可能为无限种状态各建一行。

这就是 DQN(Deep Q-Network)的动机:用一个神经网络来近似 Q 函数。输入是 4 维状态向量 [位置, 速度, 角度, 角速度],输出是 2 个 Q 值(向左推、向右推)。网络通过梯度下降学习——看到很多"状态→最佳Q值"的例子后,它就能对没见过的新状态也给出合理的 Q 估计。这本质上就是把那张放不下的 Q-Table “压缩"成了神经网络的权重。

import gymnasium as gym
import random
import torch
import torch.optim as optim
import torch.nn.functional as F
import torch.nn as nn
import time
 
# 创建环境
#env = gym.make("CartPole-v1", render_mode="human")  # human模式会用pyglet显示窗口
env = gym.make('CartPole-v1')
# 重置环境
observation, info = env.reset()
print("初始观察值:", observation)
 
#记住DQN训练的是Q*,输出的也是Q*,而不是动作,动作要根据Q*判断并反馈
class DQN(nn.Module):
    def __init__(self, state_size, action_size):
        super(DQN, self).__init__()
        self.l1 = nn.Linear(state_size, 128)
        self.l3 = nn.Linear(128, 64)
        self.l4 = nn.Linear(64, 32)
        self.l5 = nn.Linear(32, action_size)
 
    def forward(self, x):
        x = F.relu(self.l1(x))
        x = F.relu(self.l3(x))
        x = F.relu(self.l4(x))
        x = self.l5(x)
        return x
 
 
#定义参数
epsilon = 0.95
state_size = env.observation_space.shape[0]
action_size = env.action_space.n
gamma = 0.99
lrs = 0.005
epsilon_decay = 0.995
epsilon_min = 0.01
batch_size = 64
#memory = deque(maxlen=10000)
num_episodes = 500
 
#初始化
ez = DQN(state_size, action_size)
optimizer = optim.Adam(ez.parameters(), lr = lrs)
criterion = nn.MSELoss()
 
#训练
for i in range(500):
    state, _ = env.reset()   # gym >=0.26 返回 (obs, info)
    total_reward = 0.0
    done = False
 
    while not done:
        # ε-greedy 选择动作
        if random.random() < epsilon:
            action = env.action_space.sample()
        else:
            action = ez(torch.tensor(state, dtype=torch.float32)).argmax().item()
 
        # 与环境交互
        next_state, reward, terminated, truncated, _ = env.step(action)
        done = terminated or truncated
        total_reward += reward
 
        # 计算 Q(s, a)
        q_values = ez(torch.tensor(state, dtype=torch.float32))
        now_Q = q_values[action]
 
        # 计算 target
        next_q_value = ez(torch.tensor(next_state, dtype=torch.float32)).max().item()
        target_Q = reward + gamma * next_q_value * (0 if done else 1)
 
        # 计算损失
        target_Q = torch.tensor(target_Q, dtype=torch.float32)
        loss = criterion(now_Q, target_Q)
 
        # 更新网络
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
 
        # 状态更新
        state = next_state
 
    print(f"Episode {i}, total_reward = {total_reward}")
 
# 测试函数
def test_agent(env, model, episodes=10):
    total_rewards = []
    for ep in range(episodes):
        state, _ = env.reset()
        done = False
        total_reward = 0.0
        while not done:
            # 关闭梯度计算,加速
            with torch.no_grad():
                action = model(torch.tensor(state, dtype=torch.float32)).argmax().item()
            next_state, reward, terminated, truncated, _ = env.step(action)
            done = terminated or truncated
            state = next_state
            total_reward += reward
        total_rewards.append(total_reward)
        print(f"Test Episode {ep+1}: reward = {total_reward}")
 
    avg_reward = sum(total_rewards) / episodes
    print(f"Average reward over {episodes} episodes: {avg_reward}")
 
# 调用测试
test_agent(env, ez, episodes=10)
 
#最佳数据:平均150

这里很明显可以看到,测试的平均reward只有150,学习了,但是学习得很差

为什么没有经验回放就学不好?

一个关键问题藏在"状态之间的相关性"里。CartPole 的相邻两帧状态几乎一模一样——上一帧小车在位置 0.12,下一帧在位置 0.13,杆子角度也只变了零点几度。如果你拿这些高度相似的状态按顺序训练神经网络,会发生什么?

神经网络会对最近看到的数据产生"偏好”(灾难性遗忘)。它刚学会在位置 0.5 时向左推,下一秒看到的全是位置 -0.3 附近的数据,就把之前学的东西覆盖掉了。更糟的是,Bellman 方程的 target 值会根据网络自己的输出计算——如果网络因为连续相似数据波动了,target 也跟着波动,于是波动被放大,形成震荡甚至发散。

这就是时间相关性(temporal correlation) 带来的问题。

经验回放:像抽卡片一样复习

DQN 的解决方案优雅而直观——经验回放(Experience Replay)

  1. 存储:智能体每走一步,就把这次经历打包成一个元组 (状态, 动作, 奖励, 下一状态, 是否结束),扔进一个叫做"记忆库"的缓冲区
  2. 采样:训练时,不从记忆库里按顺序读,而是随机抽取一小批(比如 64 条)出来训练
  3. 重复:每条经历可能被抽到多次,让网络反复咀嚼重要的经验

你可以把这个过程想象成期末复习抽卡片:你把一学期的笔记都写在小卡片上,扔进盒子里。复习的时候,你随手一抓就是一张——可能是第一课的,也可能是最后一课的,还有可能是中间的。这样复习出来的知识不会偏科,因为它打破了学习发生的顺序。

为什么随机抽就能解决问题?

从数学上看,监督学习成功的一个重要前提是训练数据满足 i.i.d(独立同分布)——每个训练样本之间没有依赖关系,且来自同一个分布。但强化学习的原始数据天然不满足 i.i.d:这一帧和上一帧几乎一样,前后强依赖。

经验回放做的事情,就是把时间上相关的序列"拆散",通过随机抽样让每个 batch 内的数据近似独立。虽然不是严格的 i.i.d(毕竟同一个 episode 内部的样本仍然来自类似的状态分布),但这种"打破时间相关性"(breaking temporal correlation)的做法已经足够让神经网络稳定训练了。

deque 的 maxlen:记忆库的"遗忘"机制

代码中使用 deque(maxlen=1000000) 作为记忆库,这里有一个容易忽视的细节:maxlen 决定了记忆库最多存多少条经历。一旦存满,新经历进来时,最旧的那条会被自动丢弃。

这不是 bug,而是刻意设计的"遗忘"。智能体初期的策略很差(大量随机探索),那些早期的经历——“在杆子快倒时还乱推”——对后期训练没有帮助。maxlen 让记忆库保持"新鲜",始终存的是相对近期、来自更好策略的经历。这也解释了为什么代码第二部分的记忆库(maxlen=1000000)比第一部分注释掉的(maxlen=10000)大得多——更大的记忆库意味着更多样化的经历被保留,随机抽样的效果更好。

二、有经验回放

这里放有经验回放的代码,精髓是deque和随机batch抽取

ε-greedy 与探索-利用困境:新城市的饭馆该怎么选?

假设你第一次去一座新城市出差,中午饿了要找吃的。你面前有 20 家饭馆,但你一家都没吃过。你该怎么办?

  • 策略 A(纯探索):每次都去没吃过的店。你能尝遍全城,但可能踩雷不断,永远吃不到最好的
  • 策略 B(纯利用):第一天偶然走进一家还不错的面馆,之后就每天去同一家。你永远不会踩雷,但可能错过了隔壁的米其林三星

这就是强化学习中的探索-利用困境(Exploration-Exploitation Dilemma),也是 ε-greedy 算法要解决的核心问题:

  • 探索(Exploration):以概率 ε 随机选择一个动作。这是"去试试没做过的选择",目的是发现可能有更高回报的动作
  • 利用(Exploitation):以概率 1-ε 选择当前 Q 值最高的动作。这是"去我觉得最好的地方",目的是最大化当前收益

ε 的调度策略是成败关键。对比两个版本:

版本 ε 策略 效果
第一部分 ε = 0.95,固定不变 95% 的时间都在随机探索,只有 5% 的时间在做最优决策。训练 500 轮也没学到多少东西
第二部分 ε = 1.0 → 衰减 → ε_min = 0.01 前期充分探索各种动作组合,后期稳定利用已学到的知识,但始终保留 1% 的"好奇心"

为什么第二部分的设计更合理?

  1. ε 从 1.0 开始:智能体一开始什么都不懂,不如让它完全随机探索,把各种"状态→动作→结果"的组合都经历一遍,塞进记忆库
  2. 逐步衰减(decay = 0.995):随着训练推进,网络开始学到靠谱的 Q 值,这时候应该更多地信任网络的判断。每轮衰减 0.5%,大概 200 轮后 ε 就降到约 0.37,460 轮后降到 0.1 左右
  3. 保留下限 ε_min = 0.01:永远不要完全关闭探索。强化学习的环境可能会骗人——一个动作在训练早期看起来很糟糕(因为当时策略太差,后续操作全错),但随着策略改善,它可能变成一个好动作。保留 1% 的随机探索就是给网络"发现新大陆"的机会

这就像你在新城市生活了一年:第一个月到处尝试(高 ε),三个月后有了心水的几家店(中 ε),半年后大部分时间去常去的店,但每个月还是会尝试一家新店(ε_min)——万一有新开的好馆子呢?

[修改说明结束]

from collections import deque
import numpy as np
import random
import torch
import torch.optim as optim
import torch.nn as nn
import gymnasium as gym
import torch.nn.functional as F
 
 
class DQN(nn.Module):
    def __init__(self, state_size, action_size):
        super(DQN, self).__init__()
        self.l1 = nn.Linear(state_size, 128)
        self.l2 = nn.Linear(128, 32)
        self.l4 = nn.Linear(32, action_size)
 
    def forward(self, x):
        x = F.relu(self.l1(x))
        x = F.relu(self.l2(x))
        x = self.l4(x)
        return x
 
lr = 0.001
gamma = 0.99
epsilon = 1.0
memory = deque(maxlen=1000000)
batch_size = 64
epsilon_decay = 0.995
epsilon_min = 0.01
 
env = gym.make('CartPole-v1')
state_size = env.observation_space.shape[0]
action_size = env.action_space.n
 
 
ez = DQN(state_size, action_size)
optimizer = optim.Adam(ez.parameters(), lr=lr)
criterion = nn.MSELoss()
 
def replay():
    if(len(memory) < batch_size):
        return
 
    #随机选一组64个训练
    batch = random.sample(memory, batch_size)
    states, actions, rewards, next_states, dones = zip(*batch)
    states = torch.from_numpy(np.stack(states))  # 直接把 batch 堆成 array 再转 tensor
    next_states = torch.from_numpy(np.stack(next_states))
    rewards = torch.from_numpy(np.array(rewards, dtype=np.float32))
    actions = torch.from_numpy(np.array(actions, dtype=np.int64))
    dones = torch.from_numpy(np.array(dones, dtype=np.float32))
 
    q_values = ez(states).gather(1, actions.unsqueeze(1)).squeeze()
    next_q_values = ez(next_states).max(1)[0]
    target = rewards + gamma * next_q_values * (1 - dones)
 
    loss = criterion(q_values, target.detach())
    optimizer.zero_grad()
    loss.backward()
    optimizer.step()
 
for e in range(350):
    state, _ = env.reset()
    done = False
    total_reward = 0
    while not done:
        if random.random() < epsilon:
            action = env.action_space.sample()
        else:
            action = ez(torch.tensor(state, dtype=torch.float32)).argmax().item()
 
        next_state, reward, terminated, truncated, _ = env.step(action)
        done = terminated or truncated
        # 在存储到 memory 时:
        memory.append((np.array(state, dtype=np.float32),
                       action,
                       reward,
                       np.array(next_state, dtype=np.float32),
                       done))
        state = next_state
        total_reward += reward
        replay()
 
    epsilon = max(epsilon * epsilon_decay, epsilon_min)
    print(f"Episode {e + 1}: Reward = {total_reward}")
 
torch.save(ez.state_dict(), "cartpole_dqn.pth")
print("Model saved to cartpole_dqn.pth")
 
# 测试函数
def test_agent(env, model, episodes=10):
    env = gym.make('CartPole-v1', render_mode='human')
    total_rewards = []
    for ep in range(episodes):
        state, _ = env.reset()
        done = False
        total_reward = 0.0
        while not done:
            # 关闭梯度计算,加速
            with torch.no_grad():
                action = model(torch.tensor(state, dtype=torch.float32)).argmax().item()
            next_state, reward, terminated, truncated, _ = env.step(action)
            done = terminated or truncated
            state = next_state
            total_reward += reward
        total_rewards.append(total_reward)
        print(f"Test Episode {ep+1}: reward = {total_reward}")
 
    avg_reward = sum(total_rewards) / episodes
    print(f"Average reward over {episodes} episodes: {avg_reward}")
 
# 调用测试
test_agent(env, ez, episodes=10)

训练结果如下

dc7e4c08100287899297ebd2ff9ebdfc.png

模型经过训练可以拿到全部500的奖励

不过这个版本也是有问题的,没有抑制过拟合(添加噪声等),也没有使用DDQN(目标网络w-)

三、总结

DQN 的后续改进方向:从经典到现代

本文实现的 DQN 是 2013 年 DeepMind 提出的经典版本,它在 CartPole 上已经能取得不错的效果。但它有几个已知的缺陷,学术界和工业界随后提出了一系列改进。这里介绍三个最有影响力的方向,也是后续深入学习强化学习的自然路线:


1. DDQN(Double DQN):不要把 Q 值估得太乐观

经典 DQN 中有一个微妙的"自我欺骗"问题。回顾 target 的计算公式:

target = r + γ · max_a' Q(s', a')

注意:同一个网络既负责选出最佳动作 a',又负责估计这个动作的 Q 值。问题来了——神经网络对每个动作的 Q 值估计都有误差,当你用 max 在所有动作中取最大值时,你倾向于选到那个被高估了的动作。这就像考完试对着答案估分,你总觉得自己能考得很好——因为你会不自觉地挑那些"可能对了"的题往高处算。

DDQN 的解决方法很巧妙:用两个网络把"选动作"和"评估动作"分开

  • 当前网络(online network)负责选择哪个动作的 Q 值最大
  • 目标网络(target network)负责评估这个被选中的动作到底值多少

公式变成:target = r + γ · Q_target(s', argmax_a' Q_online(s', a'))

这一步改动几乎不增加计算量,但能显著缓解 Q 值的系统性高估,让训练更稳定。


2. Dueling DQN:拆开"局面好坏"和"动作好坏"

经典 DQN 直接输出每个动作的 Q 值。但有些时候,一个状态本身就很糟糕——比如 CartPole 中杆子已经倾斜到快倒的角度,无论向左推还是向右推都很难救回来。网络需要同时理解两件事:

  • V(s)——状态价值(Value):这个局面本身有多好/多危险?即使什么都不做,在这个局面下最终能拿多少分?
  • A(s, a)——动作优势(Advantage):在这个局面下,选这个动作比平均好多少?

Dueling DQN 在网络结构上做了一个巧妙的改动:把最后几层拆成两条分支——一条输出 V(s),一条输出 A(s, a),然后把它们合起来得到 Q(s, a) = V(s) + A(s, a)。这样做的好处是,网络可以学到"杆子快倒了,这个局面的 V 值就是低",而不需要为每个动作都单独学一遍这个结论。对于某些状态(比如 CartPole 中杆子接近直立),动作选择其实不太重要——什么动作都能维持平衡——此时 V(s) 主导了 Q 值,网络可以更清晰地做出判断。


3. 目标网络(Target Network):别让训练目标变成移动靶

当前本文第二部分的代码中,target 的计算直接使用了正在训练的当前网络:

next_q_values = ez(next_states).max(1)[0]  # ez 正在被优化!
target = rewards + gamma * next_q_values * (1 - dones)

这意味着:你拿着靶子在练射击,但靶子自己也在动——每次网络更新,下一轮的 target 也跟着变了。这在理论上可能导致不收敛,因为固定的 Q 值才是正确的优化目标。

目标网络的方案:额外维护一个和当前网络结构完全一样的"旧网络",它的参数每隔 N 步才从当前网络同步一次。计算 target 时使用这个固定住的目标网络:

# 每 C 步执行一次:
# target_net.load_state_dict(online_net.state_dict())
next_q_values = target_net(next_states).max(1)[0]  # 用目标网络,不是当前网络!
target = rewards + gamma * next_q_values * (1 - dones)

这样训练目标在一段时间内是稳定的,网络可以踏实地往正确的方向优化。通常 C 取 100-1000 步。


这三种改进可以叠加使用——事实上,DeepMind 2015 年在 Nature 发表的经典 DQN 论文就同时使用了经验回放和目标网络,而 DDQN + Dueling + Prioritized Replay(优先经验回放)的组合至今仍是很多强化学习任务的强基线。如果你对本文的代码感到顺手,下一步很自然地就是把这些改进一个一个加进去,感受每项改进带来的性能提升。

很明显可以感觉到,DQN是十分随机的,很多时候都会有各种情况导致训练无法收敛/过拟合,使用技巧和方法去优化DQN是极为重要的