1. 项目整体设计与思路拆解
1.1 这个项目到底在做什么
先说结论:这是一个用纯Python实现的强化学习入门级完整实例,核心由两块拼成——一块是Q-Learning算法本体,另一块是围绕它搭建的gym自定义可视化环境。整个项目跑通之后,你能在图形界面里亲眼看到智能体从"四处乱撞"到"熟练走通"的完整过程,而不是面对黑乎乎的终端数字去脑补训练效果。
为什么我强调"可视化"这三个字?因为我自己就是从纯命令行版本走过来的。最早我写过一版只打印Q表和每轮回报的控制台程序,代码能跑,收敛曲线也画得出来,但有个致命问题:你根本不知道智能体每一步到底在干嘛。它是不是卡在某个角落反复横跳?是不是绕了远路?这些信息在数字里很难直观暴露,但一旦把场景渲染出来,所有问题都清清楚楚。这个项目本质上解决的是"强化学习入门者看不到、摸不着训练过程"的痛点。
1.2 为什么选Q-Learning + gym这套组合
选Q-Learning,是因为它是强化学习里最经典的表格型算法,没有复杂的神经网络,不用装PyTorch或TensorFlow,一张二维Q表就能解释清楚完整的学习逻辑。对入门者来说,Q-Learning就像学编程时先学C语言而不是直接上框架——你不该在还没理解奖励、状态、动作这些基础概念之前,就被深度强化学习的各种技巧淹没。
选gym自定义环境,是因为gym定义了一套非常标准的交互接口:reset()重置场景、step(action)执行一步并返回状态/奖励/结束标志,这套接口几乎成了强化学习实验的事实标准。自己动手写一个满足这套接口的环境类,意味着你不仅能跑通这个例子,以后想把自己工作里的实际问题改造成强化学习场景,也完全知道该从哪儿下手。
还有一个很现实的原因:gym自带的环境(比如经典的迷宫、倒立摆之类)虽然好用,但参数都是别人定好的,改起来总有点隔靴搔痒。自定义环境的好处在于——你想设几个终点、障碍怎么摆、奖励怎么给,全部由你自己说了算,这恰恰是把知识变成能力的关键一步。
1.3 整体架构一览
这个项目的代码组织我建议分成三个文件:
env.py:自定义gym环境类,负责状态定义、动作逻辑、奖励计算和可视化渲染agent.py:Q-Learning智能体,维护Q表、实现ε-greedy策略和Q值更新train.py:训练主程序,负责多轮迭代、参数设置、结果统计和动态可视化
分文件的原因很简单:环境和算法是两个独立的关注点,混在一个文件里后面想改奖励函数或换算法会非常痛苦。我在实际动手时先写环境,因为环境定义了问题的边界——智能体能看到什么、能做什么、做错了怎么罚,这些不明确,算法根本无从谈起。
::: tip 这套架构也方便你以后换算法:把Q-Learning换成Sarsa或DQN,只需要替换agent.py,环境完全不用动。 :::
2. 前置准备:吃透gym自定义环境的接口规范
2.1 环境类必备的四个方法
在动手写代码之前,必须先搞清楚gym自定义环境的接口约定。一个能被训练循环正常驱动的环境类,至少要实现这几个方法:
| 方法 | 作用 | 关键点 |
|---|---|---|
__init__ | 初始化环境,定义状态空间和动作空间 | 需要先调用super().__init__() |
reset() | 重置到初始状态,返回观测值 | 每个episode开始时调用一次 |
step(action) | 执行动作,返回观测、奖励、终止标志、额外信息 | 必须处理非法动作 |
render() | 渲染当前场景 | 支持human和rgb_array两种模式 |
其中最容易踩坑的是step()的返回值。它必须是一个四元组:observation(当前状态)、reward(本轮奖励)、terminated(是否结束)、truncated(是否因为步数上限结束)、info(额外信息字典)。注意,新版gym把终止原因拆成了terminated和truncated两个标志,如果只返回三个值,训练循环会直接报错。
2.2 状态空间与动作空间的设计
我做的这个示例是一个8×8的网格世界,左上角是起点,右下角是终点,中间随机摆了三块障碍物。智能体每一步可以选择向上、下、左、右四个方向移动一格。
状态空间用离散坐标表示,也就是(row, col)二元组。这里有个关键选择:为什么不直接用一个整数编号?很简单,坐标表示直观,调试的时候打印状态一眼就能看出智能体在哪个位置。Q表则用字典嵌套来维护,外层key是状态坐标,内层key是动作index,这样不需要预先知道状态总数,状态空间多大都装得下。
动作空间用gym.spaces.Discrete(4),对应四个方向的动作映射表:
ACTION_MEANING = { 0: "up", 1: "down", 2: "left", 3: "right" }这里补充一个实际经验:动作越少,学习越快。四个方向是最基础的选择,你完全可以扩展成八方向甚至带斜向移动,但Q表的收敛速度会明显变慢,因为动作空间变大意味着每个状态下要尝试的可能性变多。
2.3 奖励函数怎么定
奖励设计是整个环境搭建里最需要动脑子的部分。我用的方案:
- 到达终点:+10
- 撞到障碍物:-5
- 正常走一步:-0.1(轻微的步数惩罚)
- 走出边界:-5,并留在原地
为什么正常走一步要设置负的微小惩罚?因为如果没有这一步惩罚,智能体在到达终点前绕多少路都无所谓,学习效率会非常低。这个-0.1逼着智能体去找最短路径,同时又不会大到让小Q值被惩罚主导、导致智能体宁可停在原地也不动。
走出边界的处理也需要留意。很多入门实现直接把出界当作终止,但这样会导致智能体频繁地被迫结束episode,学到的是"边界附近很危险"而非"怎么走到终点"。我采用的方法是:出界就回退到原位置并给负奖励,让episode继续跑下去。这样既惩罚了出界行为,又保证智能体有足够多的探索机会。
还有一个细节:如果连续步数超过200还没到终点,我设置了truncated=True强制结束这一轮。这个步数上限非常重要,不然智能体可能在某个角落卡死,进入死循环,训练永远进行不下去。
3. 实操过程:从零手写自定义可视化环境
3.1 环境类骨架代码
直接上代码。这个环境类我用纯Python和matplotlib实现渲染,不依赖pygame,减少环境依赖的麻烦。
import gym from gym import spaces import numpy as np import matplotlib.pyplot as plt import matplotlib.patches as patches class GridWorldEnv(gym.Env): def __init__(self, size=8, obstacles=None, render_mode="human"): super().__init__() self.size = size self.action_space = spaces.Discrete(4) # 状态空间是 8*8 的网格,用 Discrete 表示每个格子的编号 self.observation_space = spaces.Discrete(size * size) if obstacles is None: # 默认三个障碍物位置 obstacles = [(3, 3), (5, 2), (6, 5)] self.obstacles = set(obstacles) self.start = (0, 0) self.goal = (size - 1, size - 1) self.agent_pos = self.start self.render_mode = render_mode # 方向映射 self.action_delta = { 0: (-1, 0), # up 1: (1, 0), # down 2: (0, -1), # left 3: (0, 1), # right } def reset(self, seed=None, options=None): super().reset(seed=seed) self.agent_pos = self.start return self._get_obs(), {} def _get_obs(self): # 将二维坐标转换为0~63的整数状态编号 row, col = self.agent_pos return row * self.size + col def _decode_obs(self, obs): row = obs // self.size col = obs % self.size return row, col这里注意我用了Discrete(size * size)作为观测空间,而不是Tuple(Discrete(size), Discrete(size))。原因是Q-Learning需要把观测值作为Q表的key,一个整数比一个元组更省内存、查找更快。如果你想在项目里用MultiDiscrete或者Box类型的状态空间,Q表就不能用字典存储了,得换成数组或者换用函数逼近方法,这一点后面会细说。
3.2 step逻辑与奖励计算
接下来是核心的step()方法:
def step(self, action): assert self.action_space.contains(action), f"非法动作: {action}" row, col = self.agent_pos d_row, d_col = self.action_delta[action] new_row, new_col = row + d_row, col + d_col # 边界检查:出界则留在原地并给负奖励 if new_row < 0 or new_row >= self.size or new_col < 0 or new_col >= self.size: reward = -5.0 terminated = False # 位置不变,视为撞墙 elif (new_row, new_col) in self.obstacles: reward = -5.0 terminated = False # 撞障碍物,留在原地 else: self.agent_pos = (new_row, new_col) if self.agent_pos == self.goal: reward = 10.0 terminated = True else: reward = -0.1 terminated = False # 步数上限截断 self.step_count += 1 truncated = self.step_count >= self.max_steps if self.render_mode == "human": self.render() return self._get_obs(), reward, terminated, truncated, {}这里有一个我踩过的坑:初始化的时候一定要记得设置self.step_count = 0和self.max_steps = 200,并且reset()时要重置计数。我曾经在一次重构中忘了在reset()里重置步数计数器,结果训练到第二个episode就开始触发truncated,整个训练过程一团糟,排查了半天才找到问题。
3.3 可视化渲染实现
可视化是gym自定义环境最有成就感的部分。我用matplotlib实现两级渲染:
def render(self): if self.render_mode is None: return if self.fig is None: fig, ax = plt.subplots(figsize=(6, 6)) self.fig, self.ax = fig, ax self.ax.clear() # 绘制网格 for i in range(self.size + 1): self.ax.axhline(i, color='black', linewidth=0.8) self.ax.axvline(i, color='black', linewidth=0.8) # 绘制障碍物 for (r, c) in self.obstacles: rect = patches.Rectangle((c, self.size - 1 - r), 1, 1, linewidth=1, edgecolor='black', facecolor='gray') self.ax.add_patch(rect) # 绘制终点 goal_r, goal_c = self.goal rect = patches.Rectangle((goal_c, self.size - 1 - goal_r), 1, 1, linewidth=1, edgecolor='black', facecolor='green', alpha=0.7) self.ax.add_patch(rect) # 绘制智能体 agent_r, agent_c = self.agent_pos circle = patches.Circle((agent_c + 0.5, self.size - 1 - agent_r + 0.5), 0.35, facecolor='red') self.ax.add_patch(circle) self.ax.set_xlim(0, self.size) self.ax.set_ylim(0, self.size) self.ax.set_aspect('equal') self.ax.set_xticks([]) self.ax.set_yticks([]) plt.pause(0.01)这里有一个数学转换容易绕晕的地方:matplotlib的坐标轴原点在左下角,而网格世界的坐标原点在左上角。所以渲染时要把行坐标翻转,公式就是y = size - 1 - row。我最初忽略了这个转换,结果智能体看起来在往上走实际是在往下走,特别迷惑人。
训练模式下我建议把渲染放在step()里,配合plt.pause(0.01)实现动画效果。但注意:渲染开销很大,如果每个episode都逐帧渲染,训练速度会慢得很明显。我一般是这样处理的:
- 探索调试阶段:
render_mode="human",看得清楚最重要 - 正式批量训练:
render_mode=None,只统计回报数据 - 训练结束后:单独用训练好的Q表跑一段带渲染的演示
这个分阶段策略能帮你兼顾调试体验和训练速度。
4. Q-Learning算法核心与实现细节
4.1 算法原理:一张表和一条更新公式
Q-Learning的核心是一个Q表,它记录了在每一种状态下采取每一个动作所对应的长期回报期望。训练的过程就是不断根据实际获得的奖励来修正这张表。
更新公式是:
Q(s, a) ← Q(s, a) + α * [r + γ * max_a' Q(s', a') - Q(s, a)]翻译成大白话就是:新估计 = 旧估计 + 学习率 ×(实际奖励 + 对未来最优情况的折扣估计 - 旧估计)。括号里的部分叫做"时序差分误差"(TD error),它衡量的是这次决策带来的惊喜程度——如果实际结果比预期好,Q值就上调;比预期差,Q值就下调。
我在实现时直接用numpy数组来存储Q表:
class QLearningAgent: def __init__(self, observation_size, action_size, learning_rate=0.1, gamma=0.9, epsilon=0.1): # Q表:行是状态,列是动作 self.q_table = np.zeros((observation_size, action_size)) self.lr = learning_rate self.gamma = gamma self.epsilon = epsilon self.action_size = action_size def choose_action(self, obs): # ε-greedy策略:以ε概率随机探索,否则选Q值最大的动作 if np.random.random() < self.epsilon: return np.random.randint(self.action_size) else: return int(np.argmax(self.q_table[obs])) def update(self, obs, action, reward, next_obs, terminated): best_next_q = 0 if terminated else np.max(self.q_table[next_obs]) td_error = reward + self.gamma * best_next_q - self.q_table[obs, action] self.q_table[obs, action] += self.lr * td_error有个容易写错的细节:当terminated=True时,公式里的max_a' Q(s', a')必须是0,因为游戏已经结束,后面没有未来回报了。如果忘了这个条件,Q值会无限膨胀,训练彻底发散。我一开始就犯过这个错,表现为训练了很久Q表里全是数值特别离谱的大数,一看就是因为终止状态处理错了。
4.2 三个超参数的选择逻辑
Q-Learning里三个超参数是绕不开的,我分别说一下选择逻辑:
学习率α:决定每次更新Q值时迈多大步子。取0.1是我推荐的起点,太大(比如0.5以上)会让Q值来回震荡,太小(比如0.01)训练会很慢。0.1的经验取值来自实际观感:它能在一两百个episode内看到明显收敛趋势,同时不会出现过冲。
折扣因子γ:决定未来奖励的重要程度。0.9表示智能体相对有远见,愿意为了最终到达终点而忍受前面几步的负奖励。如果你设成0.5,智能体会变得很"短视",盯着眼前的奖励,在一些需要绕路才能到达终点的情况下很可能学不会。
探索率ε:决定随机探索的概率。我习惯用固定值0.1,也就是90%的时间选最优动作,10%的时间随机尝试。更讲究一点的做法是让ε随训练进度衰减,从1.0降到0.01,前期充分探索、后期专注利用。但在这个小例子里,固定0.1已经能稳定收敛,不引入衰减反而让代码更简单,也方便初学者理解。
4.3 训练主循环与日志统计
训练循环没有太多玄学,就是一个多轮迭代的过程:
def train(env, agent, episodes=500): episode_rewards = [] episode_steps = [] for episode in range(episodes): obs, _ = env.reset() total_reward = 0 step_count = 0 while True: action = agent.choose_action(obs) next_obs, reward, terminated, truncated, _ = env.step(action) agent.update(obs, action, reward, next_obs, terminated) obs = next_obs total_reward += reward step_count += 1 if terminated or truncated: break episode_rewards.append(total_reward) episode_steps.append(step_count) # 每50个episode打印一次训练进度 if (episode + 1) % 50 == 0: avg_reward = np.mean(episode_rewards[-50:]) avg_steps = np.mean(episode_steps[-50:]) print(f"Episode {episode + 1}: avg_reward={avg_reward:.2f}, " f"avg_steps={avg_steps:.2f}, " f"成功到达次数={sum(1 for r in episode_rewards[-50:] if r > 0)}") return episode_rewards, episode_steps这里有个小细节:我在判断是否成功到达时用的是total_reward > 0而不是reward == 10,因为到达终点那一轮的总回报不止10,还有前面若干步的-0.1惩罚,所以只要大于0基本就能判定是成功到达。如果你想精确统计成功率,更稳妥的做法是在step()返回的info字段里加一个success标志。
5. 训练结果分析:从震荡到收敛
5.1 三个维度的收敛判断
训练完成后,不能只看总回报够不够大,我习惯从三个维度验证模型是否真正学会了:
第一是回报趋势。正常训练会看到前几十个episode回报在负数区间剧烈震荡(智能体在乱撞),随后逐渐爬升到正数,最后稳定在一个区间。如果始终震荡得很厉害,大概率是探索率太高或者学习率不合适。
第二是成功到达率。统计最近100个episode里成功到达终点的比例,这个指标比回报更直观。训练充分后,成功率应该在95%以上。偶尔失败是因为ε的随机探索导致撞到了障碍物,属于正常现象。
第三是路径合理性。用训练好的Q表(把ε设成0)跑一局,看智能体走的路径是不是最短路径。8×8的网格理论最短路径长度是14步(需要绕开障碍物),如果智能体走出一个18步甚至是绕圈的路径,说明Q值虽然收敛了,但收敛到了局部次优解。这时候可以试着调大学习率或者延长训练轮数。
5.2 可视化效果演示
训练完成后,我用一段独立的演示脚本来展示智能体的行为:
def demo(env, agent, max_steps=100): obs, _ = env.reset() env.render_mode = "human" for step in range(max_steps): action = np.argmax(agent.q_table[obs]) obs, reward, terminated, truncated, _ = env.step(action) if terminated: print(f"成功到达终点,用了 {step + 1} 步") break if truncated: print("步数超限,任务失败") break演示时把ε设成0,因为此时我们不再需要探索,只想要最优策略。你会看到红色小球沿着一条笔直的路线绕过障碍物、直达绿色终点,非常解压。
5.3 一组可复现的参数与预期结果
我在本地实测过一组参数,直接贴出来供参考:
| 参数 | 取值 | 说明 |
|---|---|---|
| 网格大小 | 8×8 | 不算大,但足够演示 |
| 障碍物 | (3,3), (5,2), (6,5) | 三个位置,避开起点终点 |
| 学习率α | 0.1 | 更新步长 |
| 折扣因子γ | 0.9 | 远期回报折损 |
| 探索率ε | 0.1 | 固定值 |
| 训练轮数 | 500 | 实测250轮左右开始稳定 |
| 每轮步数上限 | 200 | 防止死循环 |
在这个配置下,大概跑到第150~200个episode时成功率就开始明显上升,到300个episode后基本稳定在95%以上,最优路径约14步。刷新一下随机数种子,结果会有小幅波动,但整体趋势一致。
6. 常见问题与排查技巧实录
6.1 高频问题速查表
我在实现过程中遇到了不少问题,也帮周围的同学排查过类似的报错,整理成了下面的速查表:
| 现象 | 可能原因 | 解决方案 |
|---|---|---|
step()返回元素数量不对 | 新版gym要求返回5个值 | 检查返回值是否是(obs, reward, terminated, truncated, info)五元组 |
| 训练不收敛,回报一直在-5~-2徘徊 | 探索率过高或学习率过低 | 把ε降到0.05~0.1,α调到0.1~0.3 |
| Q表数值爆炸 | 终止状态未来Q值没置0 | 更新公式中terminated=True时max Q(s')取0 |
| 可视化窗口没反应 | plt.pause()间隔太短或没有调用 | 在render()里加plt.pause(0.01~0.05),确认调用了render() |
| 智能体反复撞同一面墙 | 奖励设计不合理或出界处理不当 | 给撞墙一个明显的负奖励,检查出界是否回退原地 |
| 训练速度特别慢 | render_mode="human"一直开着 | 批量训练时关闭渲染,只在演示阶段打开 |
6.2 三个值得记录的独家经验
第一个经验:Q表别用Python的list嵌套,直接用numpy二维数组。我的第一版代码用的是defaultdict存Q表,状态key用坐标元组。在小场景里跑没问题,但一旦状态空间变大,字典查找的开销和内存碎片会让你很不舒服。用numpy数组之后,不仅更新操作变成纯数值运算,速度至少快了一个量级,而且代码更简洁。如果状态空间的维度特别高导致数组放不下,那时候才考虑用函数逼近,而不是硬撑表格法。
第二个经验:环境里加一个"成功到达终点"的统计标志,会省掉你大量调试时间。只靠回报值判断训练效果其实挺模糊的,因为负回报可能是"没到终点"也可能是"到了但绕了远路",两种情况的优化方向完全不同。我在info字典里加了一个success字段,每轮结束直接查终点是否被到达,训练是否有效就一目了然了。
第三个经验:matplotlib渲染一定要设置ax.set_aspect('equal')。如果不加,图形窗口缩放时格子会被拉成矩形,视觉上很别扭。另外渲染密集调用时记得加plt.pause(),否则窗口会卡死,看起来就像程序死循环了一样。
调试过程中我的习惯是:先在小网格(比如4×4)上验证算法的正确性,确认逻辑无误后再放大到8×8。小场景训练快,几秒钟就能跑完几十轮,非常适合快速验证。等你把超参数调好、代码调通了,再换大场景也不迟。
7. 写在最后的几个实操建议
这个项目做完之后,我自己最大的感受是:Q-Learning的代码就那么几十行,难的不是敲出来,而是理解环境、奖励、观测、动作这几者之间如何咬合。如果你也是刚接触强化学习,建议按我给的顺序动手:先跑通环境,再理解Q表更新,最后再调参看效果。
另外提供一个扩展方向:把这个示例里的Q-Learning替换成Sarsa算法,对比两者在同一个环境下的学习曲线差异。Sarsa和Q-Learning的更新公式几乎一样,只是把更新目标从max Q(s', a')换成了Q(s', a')(a'是实际采取的下一步动作),两者在探索策略上的行为差异非常值得探究。用同一个可视化环境跑两个算法做对比,你对强化学习"离线学习"和"在线学习"的区别会理解得比看十篇文章都深。
还有个小技巧送给想继续玩的朋友:把障碍物位置随机化,每次训练前重新生成一批障碍物,测试算法在不同布局下的泛化能力。你会发现,表格型Q-Learning换一个布局就得重新训练,这正是它跟深度强化学习的本质差距所在——也是你继续深入下去的一个好理由。