ML-For-Beginners 强化学习实战：基于 OpenAI Gym 的 CartPole Q-Learning 训练全流程

本篇基于 ML-For-Beginners 仓库第 8 模块（Reinforcement Learning）的第 2 课 "CartPole Skating"（原文档为 8-Reinforcement/2-Gym/translations/README.it.md，英文版为 8-Reinforcement/2-Gym/README.md）。课程将前一课 Q-Learning 基础 中学到的 Q-Table 方法，从离散状态空间迁移到由 OpenAI Gym 提供的连续状态环境 CartPole-v1 上。读完后，你将掌握 Gym 环境的初始化与 step 循环、两种连续观测的离散化方法、字典型 Q-Table 的构建、完整的训练/评估代码，以及通过移动平均分析训练曲线、调节超参数的实战技巧。

从离散棋盘到连续状态：本课要解决的问题

前一课（1-QLearning）中，我们解决的"Peter 找苹果"问题看似玩具问题，但与国际象棋、围棋等真实世界问题同属一类：给定规则 + 离散状态。本课则把 Q-Learning 应用于连续状态问题——状态由一个或多个实数给出：

问题设定：Peter 想从狼手中逃脱，就必须跑得更快。我们来看看 Peter 如何用 Q-Learning 学会"滑旱冰"，更具体地说，学会保持平衡。

教学采用经典简化平衡问题 CartPole：一个可以在水平轨道上左右移动的滑块（小车），目标是让竖立在滑块上的小棒尽量不倒下。上一课的规则与状态由我们自己编写的 Board 类给出（见 8-Reinforcement/1-QLearning/notebook.ipynb），本课则引入一个专门的模拟环境，由它来模拟小棒平衡背后的物理。

OpenAI Gym：安装与库导入

RL 领域最流行的模拟环境库是 OpenAI 维护的 Gym，可用它创建从 CartPole 到 Atari 游戏的各种环境。Gym 环境统一提供 reset、step、render 接口以及 观测空间（observation space） 与 动作空间（action space） 两套抽象，这是后文"用最少改动适配新环境"的基础。

运行前提：本课代码建议在本地 Python（例如 Visual Studio Code）中运行，render() 会在新窗口中打开模拟画面；在 Notebook 线上环境运行时可能需要额外配置。另外，从课程代码使用的 API 形式看（step 返回四元组、env.render() 无参数），内容对应旧版 gym 0.x；迁移到新版 Gymnasium 时需注意 step 返回五元组、render 需显式传入 mode 参数。

代码块 1——安装 gym 并导入所需库：

import sys
!{sys.executable} -m pip install gym

import gym
import matplotlib.pyplot as plt
import numpy as np
import random

实验一：初始化 CartPole 环境

每个 Gym 环境都关联两个空间定义：

• 观测空间（Observation space）：定义从环境收到的信息结构。对 CartPole 而言，我们收到小棒位置、速度等若干数值；
• 动作空间（Action space）：定义可能的动作。本例中动作空间是离散的，仅包含两个动作：向左和向右。

代码块 2——初始化环境并查看空间：

env = gym.make("CartPole-v1")
print(env.action_space)
print(env.observation_space)
print(env.action_space.sample())

跑 100 步随机模拟

为了观察环境如何工作，先做一次 100 步的短模拟：每一步从 action_space 中随机抽取一个动作执行。

代码块 3（✅ 建议在本地 Python 中运行，可以看到窗口动画）：

env.reset()

for i in range(100):
   env.render()
   env.step(env.action_space.sample())
env.close()

随机动作下小棒会很快倒下（见课程配图 cartpole-nobalance.gif），这正是未训练策略的预期表现。

step 返回值：观测、奖励与 done 标志

模拟中需要不断获取观测来决定如何行动。step 函数返回：当前观测 obs、奖励 rew、表示"是否该继续模拟"的标志 done，以及附加信息 info。

代码块 4：

env.reset()

done = False
while not done:
   env.render()
   obs, rew, done, info = env.step(env.action_space.sample())
   print(f"{obs} -> {rew}")
env.close()

运行输出形如：

[ 0.03403272 -0.24301182  0.02669811  0.2895829 ] -> 1.0
[ 0.02917248 -0.04828055  0.03248977  0.00543839] -> 1.0
[ 0.02820687  0.14636075  0.03259854 -0.27681916] -> 1.0
[ 0.03113408  0.34100283  0.02706215 -0.55904489] -> 1.0
[ 0.03795414  0.53573468  0.01588125 -0.84308041] -> 1.0
...
[ 0.17299878  0.15868546 -0.20754175 -0.55975453] -> 1.0
[ 0.17617249  0.35602306 -0.21873684 -0.90998894] -> 1.0

每一步返回的观测向量包含 4 个值：

序号	含义
0	小车位置（position of cart）
1	小车速度（velocity of cart）
2	小棒角度（angle of pole）
3	小棒旋转速率（rotation rate of pole）

代码块 5——查看这些数值的上下界：

print(env.observation_space.low)
print(env.observation_space.high)

仓库中已运行的解答 Notebook（8-Reinforcement/2-Gym/solution/notebook.ipynb）给出的实际输出是：

[-4.8000002e+00 -3.4028235e+38 -4.1887903e-01 -3.4028235e+38]
[ 4.8000002e+00  3.4028235e+38  4.1887903e-01  3.4028235e+38]

即小车位置界限为 ±4.8、小棒角度界限为 ±0.418879（约 24 度，超出即判失败），而两个速度分量的界限接近 float32 最大值（±3.4e38）——实际上是无界的。这一点直接影响后文离散化方案的取舍。

同时注意：每一步的奖励恒为 1.0。这是因为目标是"存活尽可能久"，即让小棒保持接近竖直越久越好。

✅ 正式判据：当连续 100 次试验的平均累计奖励达到 195 时，CartPole 模拟被视为已解决（solved）。

状态离散化：连续观测 → 有限状态集合

Q-Learning 需要构建一张定义"每个状态下该做什么"的 Q-Table，因此状态必须是离散的，即只能取有限个值。我们需要把连续观测离散化，映射到有限状态集。原文档给出了两种做法：

• 分箱（Divide into bins）：若已知某数值的取值区间，可将区间划分为若干 bin（箱），用"数值属于第几个箱"代替数值本身，可用 numpy 的 np.digitize 实现。此时状态空间大小精确可知，等于各维度 bin 数的乘积。
• 线性缩放 + 取整：用线性插值把数值带入某个有限区间（例如 -20 到 20），再四舍五入转为整数。这种方法对状态规模的控制力稍弱——如果不知道输入的确切范围就危险：本例 4 个观测值中有 2 个（两个速度）没有上下界，理论上可能产生无限多个状态。

本课选择第二种方案。后续训练会看到：尽管界未显式定义，这两个值极少跑出某些有限区间，极端值状态非常罕见。

代码块 6——核心离散化函数，把观测映射为 4 个整数的元组：

def discretize(x):
    return tuple((x/np.array([0.25, 0.25, 0.01, 0.1])).astype(np.int))

四个缩放因子（0.25、0.25、0.01、0.1）分别对应小车位置、小车速度、小棒角度、小棒角速度的"分辨率"：例如角度除以 0.01，意味着相差 0.01 弧度以内的小棒被视为同一离散状态。

代码块 7——再实现并对比分箱方案：

def create_bins(i,num):
    return np.arange(num+1)*(i[1]-i[0])/num+i[0]

print("Sample bins for interval (-5,5) with 10 bins\n",create_bins((-5,5),10))

ints = [(-5,5),(-2,2),(-0.5,0.5),(-2,2)] # Intervallo di valori per ogni parametro
nbins = [20,20,10,10] # numero di contenitori per ogni parametro
bins = [create_bins(ints[i],nbins[i]) for i in range(4)]

def discretize_bins(x):
    return tuple(np.digitize(x[i],bins[i]) for i in range(4))

代码块 8——跑一轮模拟，观察两种离散化的输出差异：

env.reset()

done = False
while not done:
   #env.render()
   obs, rew, done, info = env.step(env.action_space.sample())
   #print(discretize_bins(obs))
   print(discretize(obs))
env.close()

✅ 两种方案的索引语义不同：discretize_bins 返回的是 0 起始的箱编号，输入值在 0 附近时得到的是区间中部的编号（本例为 10）；而 discretize 不关心输出范围、允许负数，因此状态值不发生平移，0 仍然对应 0。若想看窗口动画就取消 env.render() 的注释，否则让它在后台静默运行会快得多——后续 Q-Learning 训练正使用这种"无头"模式。

Q-Table 数据结构：为什么用字典而不是 numpy 张量

前一课状态只是 0~8 的数字对，Q-Table 可以直接用形状 8x8x2 的 numpy 张量（2 为动作数）。如果用分箱离散化，状态向量尺寸已知，同样可以用 20x20x10x10x2 的数组（各维度分别等于对应观测参数的 bin 数）。

但本课的 discretize 函数无法保证状态永远落在某个界限内（速度无界），张量索引随时可能越界。因此改用字典表示 Q-Table：以 (state, action) 元组为键，Q 值为值。

代码块 9：

Q = {}
actions = (0,1)

def qvalues(state):
    return [Q.get((state,a),0) for a in actions]

qvalues(state) 返回给定状态下所有可能动作对应的 Q 值列表；若表项尚不存在，Q.get 以默认值 0 填充。字典方案以内存换取了"状态空间大小未知"的鲁棒性，这也正是它适用于任意 Gym 连续环境的通用写法。

启动 Q-Learning：超参数与训练主循环

代码块 10——设置超参数：

# iperparametri
alpha = 0.3
gamma = 0.9
epsilon = 0.90

三个超参数的作用：

超参数	名称	含义
alpha	学习率（learning rate）	定义每一步对 Q-Table 当前值调整的力度。前一课从 1 开始并随训练递减；本例为简单起见保持恒定，后续可自行实验调节
gamma	折扣因子（discount factor）	表示未来奖励相对当前奖励应被优先考虑的程度
epsilon	探索/利用因子	决定"按 Q-Table 值选动作"与"随机选动作"的比例。本算法中在 epsilon 比例的回合里依据 Q-Table 选择，剩余回合执行随机动作，从而探索从未见过的搜索空间区域

✅ 用平衡来理解：随机动作（探索）相当于朝错误方向乱打一拳，小棒必须学会从这些"错误"中恢复平衡。

对算法的两点改进

相对前一课，本课引入两项工程化改进：

• 计算平均累计奖励：每 5000 次迭代打印一次进度，并对该窗口内的累计奖励取平均。若平均值超过 195，就可以认为问题以高于正式判据的质量被解决（我们平均的样本远多于判据要求的 100 次）。
• 记录最优 Q-Table：跟踪累计平均结果的最大值 Qmax，并把对应时刻的 Q-Table 存为 Qbest。训练后期平均结果有时会开始回落——这意味着新写入的值正在"破坏"已学好的 Q 值，此时应回退到历史最优表。

代码块 11——完整训练循环（含 Bellman 更新）：

def probs(v,eps=1e-4):
    v = v-v.min()+eps
    v = v/v.sum()
    return v

Qmax = 0
cum_rewards = []
rewards = []
for epoch in range(100000):
    obs = env.reset()
    done = False
    cum_reward=0
    # == esegue la simulazione ==
    while not done:
        s = discretize(obs)
        if random.random()<epsilon:
            # sfruttamento - sceglie l'azione in accordo alle probabilità di Q-Table
            v = probs(np.array(qvalues(s)))
            a = random.choices(actions,weights=v)[0]
        else:
            # esplorazione - sceglie casualmente l'azione
            a = np.random.randint(env.action_space.n)

        obs, rew, done, info = env.step(a)
        cum_reward+=rew
        ns = discretize(obs)
        Q[(s,a)] = (1 - alpha) * Q.get((s,a),0) + alpha * (rew + gamma * max(qvalues(ns)))
    cum_rewards.append(cum_reward)
    rewards.append(cum_reward)
    # == Stampa periodicamente i risultati e calcola la ricompensa media ==
    if epoch%5000==0:
        print(f"{epoch}: {np.average(cum_rewards)}, alpha={alpha}, epsilon={epsilon}")
        if np.average(cum_rewards) > Qmax:
            Qmax = np.average(cum_rewards)
            Qbest = Q
        cum_rewards=[]

逐行拆解关键逻辑：

1. probs(v) 把一组 Q 值转换为概率分布：先减去最小值并加极小量 eps=1e-4（防零除），再归一化，用于"按 Q-Table 概率采样动作"（即前一课讲过的探索/利用折中）。
2. 动作选择：以 epsilon 概率走利用分支（按 Q 值概率采样 random.choices(actions, weights=v)），否则走探索分支（np.random.randint(env.action_space.n) 均匀随机）。
3. 核心更新即 Bellman 公式的课程实现：Q[(s,a)] = (1-alpha)·Q(s,a) + alpha·(r + gamma·max Q(s'))，其中 max(qvalues(ns)) 对下一离散状态的所有动作取最大值。
4. rewards 向量逐回合累计累计奖励，供后文绘图；每 5000 次迭代打印窗口平均，并用 Qmax/Qbest 保留历史最优。

✅ 训练结果的两个常见现象：接近目标——窗口平均可能达到 195 以上（或非常接近）；奖励开始回落——已学好的 Q 值可能被更差的新值覆盖。这两点用训练曲线可以看得更清楚。

绘制训练进度：原始曲线与移动平均

训练期间每个迭代的累计奖励已存入 rewards。直接 plt.plot(rewards) 画出的原始曲线几乎不可读——随机训练过程使每个回合的时长差异极大：

为让曲线有意义，可以计算一个长度（如 100）的移动平均（running average），用 np.convolve 一行即可实现：

代码块 12：

def running_average(x,window):
    return np.convolve(x,np.ones(window)/window,mode='valid')

plt.plot(running_average(rewards,100))

平滑后可以看到累计奖励的整体上升趋势与后期的回落段，为"何时该用 Qbest 回退"提供了直观依据。

调节超参数：让学习更稳定

为使学习更稳定，建议在训练过程中动态调整部分超参数：

• 学习率 alpha：可以以接近 1 的值起步，然后持续递减。随着 Q-Table 逐渐积累可靠的值，后续更新应只做小幅修正，而不是被新值完全覆盖。
• epsilon 递增：原文档建议让 epsilon 从较低值缓慢上升到接近 1——由于本代码中"按 Q-Table 概率选择"的概率正是 epsilon，递增 epsilon 即意味着越来越少地随机探索、越来越多地利用已有知识。

由此布置了课程中的前两个练习：

任务 1：自由调节超参数，看能否取得更高的累计奖励（超过 195？）。

任务 2：正式解题标准是连续 100 次运行平均 195 奖励。请在训练中直接度量这一指标，确认问题被正式解决，而不只是窗口平均值达标。

查看训练成果：按 Q-Table 概率推理

训练完成后，运行推理循环——注意它复用了与训练完全相同的动作选择策略（按 Q-Table 概率分布采样）：

代码块 13：

obs = env.reset()
done = False
while not done:
   s = discretize(obs)
   env.render()
   v = probs(np.array(qvalues(s)))
   a = random.choices(actions,weights=v)[0]
   obs,_,done,_ = env.step(a)
env.close()

运行后应能看到小棒长时间保持平衡（文首 GIF 即此效果）。仓库的解答 Notebook（8-Reinforcement/2-Gym/solution/notebook.ipynb）还额外演示了如何把推理过程存成动图：用 env.render(mode='rgb_array') 逐帧取图，再用 PIL 的 Image.save(..., save_all=True, append_images=...) 拼成 cartpole-balance.gif——这正是文首展示图的生成方式。

🚀 挑战任务

任务 3：上面的推理用的是训练结束时的 Q-Table，未必是最优的那份。记得历史最优表被保存在 Qbest 变量里！把 Qbest 复制给 Q 后重跑同样的推理示例，对比是否有差别。

任务 4：推理时我们并非每步选最优动作，而是按概率分布采样。是否应当总是选 Q 值最高的动作？可用 np.argmax 找到对应最高 Q 值的动作编号来实现该策略，实现后观察平衡效果是否改善。

⚠️ 一个值得注意的实现细节：从源码结构看，训练循环里的 Qbest = Q 在 Python 中只是引用赋值而非深拷贝，Q 字典后续仍会被原地更新，Qbest 与 Q 实际指向同一对象。若希望严格冻结历史最优表，应使用 copy.deepcopy(Q) 之类的深拷贝手段——这解释了任务 3 为何强调"把 Qbest 复制到 Q"，也提示读者在复现时自行处理这一点。

课后作业：训练一辆 Mountain Car

本课配套作业（意大利语版 translations/assignment.it.md，英文版 assignment.md）要求把同样的 Q-Learning 算法用最少改动迁移到 Gym 的 Mountain Car 环境，以验证 Gym 统一 API（相同的 reset/step/render 方法与观测/动作空间抽象）的可移植性：

• 动作空间：

值	含义
0	向左加速
1	不加速
2	向右加速

• 观测空间（2 个值）：

序号	观测	最小值	最大值
0	汽车位置	-1.2	0.6
1	汽车速度	-0.07	0.07

• 奖励规则：到达山巅旗标（位置 = 0.5）奖励 0，其余每步奖励 -1；
• 终止条件：汽车位置大于 0.5，或回合长度超过 200 步；
• 关键难点：发动机功率不足以一次爬坡，唯一成功策略是前后摆动积攒动量。

具体步骤：从 notebook.ipynb 出发，替换环境、改写状态离散化函数（注意本环境有 3 个动作，actions 需相应调整），最小化改动训练，并通过调节超参数优化结果。评分细则要求"算法成功从 CartPole 示例移植、能少于 200 步抓旗"为最高档。

结语：本课学到了什么

至此，我们演示了如何仅靠一个定义目标状态的奖励函数 + 智能探索搜索空间的机会，就能训练出表现良好的 Agent——Q-Learning 在离散环境与连续状态（动作仍离散）两类场景中均被成功应用。

后续还需研究两类更复杂的情形：动作空间也连续的环境，以及观测空间极其复杂的环境（如 Atari 游戏截图）。在这些问题上，通常需要神经网络等更强的机器学习技术才能取得好效果——这些进阶主题是更高级 AI 课程的内容。完整的 12 周课程结构与各模块入口见仓库根目录 README.md。