ML-For-Beginners OpenAI Gym 实战：用 Q-Learning 训练 CartPole 平衡智能体

本篇指南基于 ML-For-Beginners 强化学习课程中的 OpenAI Gym 一课，讲解如何把上一课用于离散棋盘问题的 Q-Learning 算法迁移到连续状态环境：以 CartPole（小车杆平衡）问题为载体，完成环境初始化、状态离散化、Q-Table 结构设计、训练循环与超参数调优的完整流程。读完后，你将能够独立跑通一个基于 OpenAI Gym 的 Q-Learning 训练管线，并知道如何把它改造成 MountainCar 等其他 Gym 环境的解决方案。

从"玩具问题"到连续状态问题

上一课（Q-Learning）中我们解决的棋盘问题看起来像一个玩具问题，似乎与真实场景无关。但实际上许多真实世界问题与它结构相同——比如下国际象棋或围棋：都有一个给定规则的棋盘和离散状态。

本课把 Q-Learning 的相同原理应用到连续状态问题上，即状态由一个或多个实数给出的问题。课程给出的故事背景是：

问题：如果彼得（Peter）想要逃离狼群，他需要能够移动得更快。我们将看到彼得如何使用 Q-Learning 学习滑冰，特别是如何保持平衡。

为了教学目的，课程使用了一个简化的平衡问题——CartPole：世界中有一个可以左右移动的水平滑块（小车），目标是平衡滑块顶部的垂直杆。这个环境正好对应"滑冰保持平衡"的抽象：滑块位置对应滑冰者重心，杆的姿态对应身体倾斜度。

准备工作：OpenAI Gym 环境

上一课中，游戏规则和状态由我们自己定义的 Board 类给出；本课则使用一个能模拟平衡杆背后物理规则的模拟环境。OpenAI Gym 是训练强化学习算法最流行的模拟环境之一，通过它可以从 CartPole 这样的经典控制问题一直创建到 Atari 游戏环境（环境清单见 OpenAI Gym 官方文档的 classic control 分类）。

首先安装 gym 并导入所需库（代码块 1）：

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

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

✅ 本课代码最好在本地 Python 环境（如 Visual Studio Code）中运行，模拟窗口会在新窗口打开；在线 Notebook 环境中渲染 Gym 环境可能需要对代码做额外调整。

初始化 CartPole：观察空间与动作空间

每个 Gym 环境都关联两个核心抽象：

• 观察空间（Observation space）：定义从环境中接收到的信息结构。对 CartPole 问题，我们收到杆的位置、速度等数值；
• 动作空间（Action space）：定义可能的动作。本例中动作空间是离散的，由两个动作组成——left（0，向左推）和 right（1，向右推）。

要初始化环境，输入以下代码（代码块 2）：

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

gym.make("CartPole-v1") 返回一个环境实例，后续所有交互都通过它的 reset()、step()、render()、close() 四个方法完成。env.action_space.sample() 会随机采样一个合法动作，用来快速检验动作空间。

随机动作跑 100 步：看环境"摔倒"

为观察环境如何工作，可以先跑一个 100 步的短模拟——每一步都从 action_space 中随机选一个动作（代码块 3）：

env.reset()

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

随机策略下小车很快失稳，你会看到杆倒下的模拟动画（课程中对应的演示图 8-Reinforcement/2-Gym/images/cartpole-nobalance.gif）。这就是"未训练智能体"的基线行为。

step 函数：观察、奖励与终止标志

在模拟过程中，我们需要通过观察来决定如何行动。step 函数返回当前观察值、奖励值和指示"是否还有意义继续模拟"的终止标志 done（代码块 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()

Notebook 输出会看到类似这样的内容：

[ 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	推车位置
1	推车速度
2	杆的角度
3	杆的转速

可以查看这些数值的最小值和最大值（代码块 5）：

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

注意每步奖励值恒为 1.0：我们的目标是尽可能长时间"存活"，即在最长的时间内将杆保持在合理的垂直位置。

✅ 判据：如果能在 100 次连续试验中获得平均 195 的累积奖励，则认为 CartPole 问题已被解决。（由于每步奖励为 1，平均 195 步意味着杆在约 195 步内不倒下。）

状态离散化：把实数观察映射为有限状态

Q-Learning 需要构建 Q-Table 来定义在每个状态下做什么，因此状态必须是离散的、包含有限个离散值。我们需要把连续观察离散化，映射到一组有限状态。课程给出了两种方法：

1. 分箱（bins）。如果知道某个值的区间，就把区间分成若干 bin，用值所属的箱序号替换原值。可以用 NumPy 的 digitize 方法完成。此时状态大小是精确已知的，取决于所选箱子数量；
2. 线性缩放到有限区间后取整。例如把值缩放到 -20~20 之类的有限区间，然后四舍五入成整数。这对状态大小的控制稍弱——比如 4 个观察值中有 2 个（推车速度、杆转速）理论上无上下界，可能导致状态数量无限。但在实际训练中，这些值很少超出某些有限区间，极值状态非常罕见。

方法一：缩放取整

将观察向量除以一组步长再转成整数元组（代码块 6）：

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.25、小车速度精度 0.25、杆角度精度 0.01 弧度、杆转速精度 0.1 弧度/秒"。注意这里允许离散后的状态值为负数，因此 0 仍然对应 0，状态值不发生偏移。

方法二：分箱 digitize

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)] # 每个参数的值间隔
nbins = [20,20,10,10] # 每个参数的 bin 数量
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))

两种方法的关键差异：discretize_bins 返回的箱编号从 0 开始，因此在 0 附近的输入值会返回区间中间位置附近的编号（约 10），状态值整体发生了偏移；而 discretize 允许输出为负数，0 对应 0。可以运行一个短模拟观察两种离散化下的状态值（代码块 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()

✅ 想看环境实际执行时，取消 env.render() 一行的注释；不渲染（"隐形"执行）会快得多，Q-Learning 训练过程就采用这种后台方式。

Q-Table 结构：为什么用字典

上一课中状态是 0 到 8 的一对简单数字，用形状为 8x8x2 的 NumPy 张量表示 Q-Table 很方便。如果采用分箱离散化，状态向量大小也是已知的，同样可以用 20x20x10x10x2 的数组表示状态（末尾的 2 是动作空间维度，前四个维度对应观察空间各参数的箱数）。

但 discretize 这种缩放取整方案下，无法保证状态永远落在某个范围内（因为部分原始值无界），因此课程改用字典表示 Q-Table：以 (state, action) 二元组为键，值为 Q-Table 条目（代码块 9）：

Q = {}
actions = (0,1)

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

其中 qvalues() 返回给定状态在全部可能动作下的 Q 值列表；条目不存在时默认返回 0。字典方案对状态空间无界也天然安全——遇到新状态就插入新键即可。

Q-Learning 训练循环

现在开始教彼得"保持平衡"。

超参数

# hyperparameters
alpha = 0.3
gamma = 0.9
epsilon = 0.90

参数	含义	本课取值与说明
alpha	学习率，决定每一步在多大程度上调整 Q-Table 当前值	0.3，保持不变；上一课是从 1 开始逐步衰减
gamma	折扣因子，决定多大程度上优先考虑未来奖励而非当前奖励	0.9
epsilon	探索/开发因素	0.90：以该概率按 Q-Table 值选择动作，剩余情况执行随机动作，从而探索从未见过的搜索空间区域

⚠️ 注意本实现中 epsilon 的语义与部分文献相反：random.random() < epsilon 时走开发（exploitation）分支（按 Q 值概率采样），否则走探索（exploration）分支（完全随机）。所以这里"提高 epsilon"意味着更倾向于开发、更少探索，后文调参建议正是据此给出。

✅ 用平衡的直觉理解：选择随机动作（探索）相当于朝错误方向随机推一把，杆必须学会从这些"错误"中恢复平衡。

训练主循环（含两处算法改进）

在上一课算法基础上做了两项改进：

• 计算平均累积奖励：每 5000 次迭代打印一次进度，并对该窗口内的累积奖励取平均。若平均分超过 195，可认为问题已解决——甚至比官方判据（100 次运行）质量要求更高；
• 跟踪最大平均累积结果 Qmax 并保存对应 Q-Table（Qbest）：训练中平均结果有时会下降，说明 Q-Table 里已学好的值可能被"破坏"，因此保留训练期间观察到的最佳 Q-Table。

训练代码（代码块 11，与 solution 版 notebook 一致）：

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
    # == do the simulation ==
    while not done:
        s = discretize(obs)
        if random.random() < epsilon:
            # exploitation - chose the action according to Q-Table probabilities
            v = probs(np.array(qvalues(s)))
            a = random.choices(actions, weights=v)[0]
        else:
            # exploration - randomly chose the action
            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)
    # == Periodically print results and calculate average reward ==
    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 = []

几个值得注意的实现细节：

• probs() 把两个 Q 值平移为正值（减去最小值并加 eps=1e-4 防除零）再归一化，得到动作选择的概率分布；
• 更新规则 Q[(s,a)] = (1-alpha)*Q.get((s,a),0) + alpha*(rew + gamma*max(qvalues(ns))) 就是标准 Q-Learning 更新：当前 Q 值与贝尔曼目标值 rew + gamma * max Q(s', .) 的凸组合；
• rewards 向量收集每次模拟的累积奖励，用于后续绘图。

运行这些代码后通常能观察到：

• 接近目标：非常接近（甚至可能已经达到）"连续 100+ 次运行获得 195 累积奖励"的目标；即使数字更小也不能断定失败，因为我们平均的是 5000 次运行，而正式判据只要求 100 次；
• 奖励开始下降：有时平均奖励会下跌，意味着可能"破坏"了 Q-Table 中已经学好的值——这正是保存 Qbest 的原因。

绘制训练进度

把每次迭代的累积奖励直接画出来（plt.plot(rewards)）几乎读不出任何信息，因为随机训练过程中每次模拟的步长差异很大。更好的做法是计算一系列实验的滚动平均（running average），窗口取 100，用 np.convolve 即可（代码块 12）：

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

plt.plot(running_average(rewards, 100))

课程中给出的两张对照图（8-Reinforcement/2-Gym/images/train_progress_raw.png 与 8-Reinforcement/2-Gym/images/train_progress_runav.png）清楚地显示：原始曲线噪声极大，而滚动平均后能看出奖励先上升、在高位震荡并偶有回落的趋势。

超参数调优

为了让学习更稳定，训练期间调整超参数是有意义的：

• 学习率 alpha：可以从接近 1 的值开始，然后不断减小。随着训练推进，Q-Table 中会积累较好的 Q 值，此时应只做小幅调整，而不是用新值完全覆盖；
• 提高 epsilon：按本实现的语义，缓慢把 epsilon 从较低值提升到接近 1，意味着减少探索、增加开发。

课程布置的两个验证任务：

任务 1：调整超参数取值，看能否获得更高的累积奖励——超过 195 了吗？

任务 2：要正式解决问题，需要在 100 次连续运行中取得 195 的平均奖励。在训练期间做这个度量，确认你已正式解决问题。

在行动中看到结果

训练完成后，运行模拟并使用与训练相同的动作选择策略（按 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 还给出了一个扩展技巧：用 env.render(mode='rgb_array') 逐帧取图，再用 PIL 存成循环动画 GIF：

from PIL import Image
obs = env.reset()
done = False
i = 0
ims = []
while not done:
    s = discretize(obs)
    img = env.render(mode='rgb_array')
    ims.append(Image.fromarray(img))
    v = probs(np.array([Qbest.get((s,a),0) for a in actions]))
    a = random.choices(actions, weights=v)[0]
    obs, _, done, _ = env.step(a)
    i += 1
env.close()
ims[0].save('cartpole-balance.gif', save_all=True, append_images=ims[1::2], loop=0, duration=5)
print(i)

注意这里直接查询 Qbest，而不是训练结束时的 Q。

挑战任务

任务 3：上面的推理演示用的是 Q-Table 的最终副本，它未必是最好的。还记得我们已把表现最好的 Q-Table 存进 Qbest 了吗？把 Qbest 复制给 Q，用最佳 Q-Table 重跑一遍，观察是否有差异。

任务 4：这里每一步并不是选最优动作，而是按概率分布采样。是否应该始终选择 Q 值最高的动作？可以用 np.argmax 找出对应较高 Q-Table 值的动作编号，实现这一策略并验证是否改善平衡效果。

作业：迁移到 MountainCar 环境

OpenAI Gym 的设计让所有环境提供相同 API——相同的 reset、step、render 方法和相同的动作空间/观察空间抽象。因此应该只需最少量的代码修改，就能让同一套强化学习算法适应不同环境。本课的正式作业就是训练山地车（MountainCar）：

• 环境：一辆卡在山谷中的车，目标是在每一步执行"0 向左加速 / 1 不加速 / 2 向右加速"之一，驶出山谷到达旗帜；
• 核心难点：引擎功率不足以一次翻山，唯一成功策略是来回行驶积聚动量；
• 观察空间：车位置（-1.2 ~ 0.6）与车速度（-0.07 ~ 0.07）两个值；
• 奖励：到达旗杆位置（0.5）奖励 0，位置小于 0.5 每步奖励 -1；车位置超过 0.5 或步数超过 200 时终止。

做法是从 现有 notebook 代码 出发，替换为新环境、修改状态离散化函数，并以最小代码改动让现有 Q-Learning 算法在新环境上训练收敛（可能需要调整超参数）。评判标准是：算法能成功改编自 CartPole 示例、代码修改最少、且能在 200 步内到达旗帜。

小结与局限

本课展示的核心思路是：只需给智能体一个"定义期望状态"的奖励函数，并让它有机会智能地探索搜索空间，就能训练出良好策略。至此我们已经把 Q-Learning 成功应用到**离散环境和连续环境（离散动作）**两种情况。

但 Table 型 Q-Learning 有明显边界：

• 当动作空间也连续时，"每个状态枚举所有动作的 Q 值"不再可行；
• 当观察空间非常复杂（例如 Atari 游戏屏幕的像素图像）时，即使离散化，状态空间也会爆炸。

这些问题通常需要更强的机器学习技术——用神经网络近似 Q 函数（即深度 Q-Learning 一系的方法）——才能达到好效果，属于更进阶的 AI 课程内容。就本仓库这一课而言，掌握"环境 API → 状态离散化 → 字典式 Q-Table → ε 策略训练循环 → 滚动平均评估"这条完整管线，就是把经典 Q-Learning 从玩具棋盘推向真实物理环境的关键一步。