ML-For-Beginners 强化学习实战:基于 OpenAI Gym 的 CartPole Q-Learning 训练全流程
【免费下载链接】ML-For-Beginners12 weeks, 26 lessons, 52 quizzes, classic Machine Learning for all项目地址: https://gitcode.com/GitHub_Trending/ml/ML-For-Beginners
本篇基于 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=[]逐行拆解关键逻辑:
probs(v)把一组 Q 值转换为概率分布:先减去最小值并加极小量eps=1e-4(防零除),再归一化,用于"按 Q-Table 概率采样动作"(即前一课讲过的探索/利用折中)。- 动作选择:以
epsilon概率走利用分支(按 Q 值概率采样random.choices(actions, weights=v)),否则走探索分支(np.random.randint(env.action_space.n)均匀随机)。 - 核心更新即 Bellman 公式的课程实现:
Q[(s,a)] = (1-alpha)·Q(s,a) + alpha·(r + gamma·max Q(s')),其中max(qvalues(ns))对下一离散状态的所有动作取最大值。 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。
【免费下载链接】ML-For-Beginners12 weeks, 26 lessons, 52 quizzes, classic Machine Learning for all项目地址: https://gitcode.com/GitHub_Trending/ml/ML-For-Beginners
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考