从零构建Grid World一个Python强化学习环境的工程实践如果你刚开始接触强化学习可能会被那些复杂的数学公式和抽象概念搞得晕头转向。理论固然重要但没有什么比亲手搭建一个环境、看着智能体在里面探索学习更能让人理解其精髓了。今天我们就抛开那些厚重的教科书直接动手用Python从零开始构建一个经典的Grid World环境。这不仅仅是一个教学示例更是一个完整的、可交互的、能让你直观看到算法效果的工程实践项目。我们将使用NumPy来处理核心逻辑用Matplotlib来实现动态可视化。整个过程会像搭积木一样从定义网格、设定规则到实现状态转移和奖励机制最后我们会让两种经典的策略——随机策略和最优策略——在这个世界里运行并用清晰的图表展示它们学到的“价值地图”。无论你是想为后续的Q-learning、SARSA等算法准备一个测试床还是单纯想深入理解马尔可夫决策过程MDP的环境建模这篇文章都将提供一条清晰的路径和可直接运行的代码。1. 环境蓝图定义我们的网格世界在写第一行代码之前我们需要在脑子里把这个世界画出来。Grid World顾名思义就是一个网格化的世界。我们把它设计成一个5x5的棋盘每个格子代表一个状态。智能体就像一个棋子初始时可能被放在任意位置它的目标是通过行动获得尽可能多的累积奖励。这个世界有一些特殊的“传送门”格子这是经典Grid World问题的关键设定状态A (0,1)一旦踏入无论执行什么动作都会立刻获得10的奖励并被传送到A‘ (4,1)。状态B (0,3)与A类似会获得5的奖励并被传送到B‘ (2,3)。除了这些特殊规则智能体在每个普通格子有四个基本动作上U、下D、左L、右R。如果动作会导致智能体走出网格边界比如在最上面一格还想往上走那么它会被“墙”挡住留在原地并受到-1的小惩罚。其他在网格内的普通移动奖励为0。我们用Python字典和列表来刻画这个世界的所有规则。首先进行基础的初始化。import numpy as np import matplotlib.pyplot as plt from matplotlib.table import Table # 世界参数 WORLD_SIZE 5 A_POS [0, 1] A_PRIME_POS [4, 1] B_POS [0, 3] B_PRIME_POS [2, 3] DISCOUNT 0.9 # 未来奖励的折扣因子 # 动作空间 ACTIONS [L, U, R, D] ACTIONS_FIGS [←, ↑, →, ↓] # 用于后续可视化策略的箭头符号 ACTION_PROB 0.25 # 随机策略下每个动作的均匀选择概率接下来我们需要构建两个核心的查询表next_state和reward。它们的作用是给定当前状态s和动作a立刻能查出下一个状态s‘和即时奖励r。这是环境动力学模型的核心。def build_world_dynamics(): 构建世界的状态转移和奖励模型。 next_state [[{} for _ in range(WORLD_SIZE)] for _ in range(WORLD_SIZE)] reward [[{} for _ in range(WORLD_SIZE)] for _ in range(WORLD_SIZE)] for i in range(WORLD_SIZE): for j in range(WORLD_SIZE): state [i, j] # 处理特殊状态A和B if state A_POS: for a in ACTIONS: next_state[i][j][a] A_PRIME_POS reward[i][j][a] 10.0 continue if state B_POS: for a in ACTIONS: next_state[i][j][a] B_PRIME_POS reward[i][j][a] 5.0 continue # 处理普通状态的动作 for a in ACTIONS: if a U: next_i i - 1 if i 0 else i r -1.0 if i 0 else 0.0 elif a D: next_i i 1 if i WORLD_SIZE - 1 else i r -1.0 if i WORLD_SIZE - 1 else 0.0 elif a L: next_j j - 1 if j 0 else j r -1.0 if j 0 else 0.0 elif a R: next_j j 1 if j WORLD_SIZE - 1 else j r -1.0 if j WORLD_SIZE - 1 else 0.0 next_state[i][j][a] [next_i, next_j] reward[i][j][a] r return next_state, reward # 构建世界模型 NEXT_STATE, REWARD build_world_dynamics()为了方便后续使用我们把这些逻辑封装成一个step函数这是强化学习环境的标准接口之一。def step(state, action): 执行一步动作。 参数: state: 当前状态如 [i, j] action: 动作L, U, R, D 之一 返回: next_state: 下一个状态 reward: 即时奖励 i, j state return NEXT_STATE[i][j][action], REWARD[i][j][action]注意这里我们采用了确定性的状态转移。在更复杂的环境里step函数可以返回一个概率分布表示执行动作后可能到达的多个下一个状态及其概率。2. 可视化工具让价值与策略一目了然强化学习算法最终会输出两样东西状态价值函数V(s)每个状态有多“好”和策略π(a|s)在每个状态应该做什么。我们需要直观地看到它们。我们将创建两个绘图函数一个用于绘制网格中每个状态的价值数字另一个用于绘制最优策略箭头。首先是绘制价值函数的draw_value_image函数。它会把一个二维数组价值表画成一个美观的表格并高亮特殊状态A、B、A‘、B’。def draw_value_image(value_grid, titleNone): 将价值函数网格绘制成表格。 参数: value_grid: 二维numpy数组形状为(WORLD_SIZE, WORLD_SIZE) title: 可选的图表标题 fig, ax plt.subplots(figsize(6, 6)) ax.set_axis_off() if title: ax.set_title(title, fontsize16, pad20) tb Table(ax, bbox[0, 0, 1, 1]) nrows, ncols value_grid.shape width, height 1.0 / ncols, 1.0 / nrows # 添加单元格 for (i, j), val in np.ndenumerate(value_grid): cell_text f{val:.2f} # 标记特殊状态 if [i, j] A_POS: cell_text \n(A) facecolor lightcyan elif [i, j] A_PRIME_POS: cell_text \n(A) facecolor lightcyan elif [i, j] B_POS: cell_text \n(B) facecolor lightyellow elif [i, j] B_PRIME_POS: cell_text \n(B) facecolor lightyellow else: facecolor white tb.add_cell(i, j, width, height, textcell_text, loccenter, facecolorfacecolor, edgecolorblack) # 添加行号和列号 for i in range(nrows): tb.add_cell(i, -1, width, height, textstr(i1), locright, edgecolornone, facecolorwhitesmoke) for j in range(ncols): tb.add_cell(-1, j, width, height/2, textstr(j1), loccenter, edgecolornone, facecolorwhitesmoke) ax.add_table(tb) plt.tight_layout() return fig, ax其次是绘制策略的draw_policy函数。它会计算每个状态下哪个或哪些动作能带来最大的未来价值并用箭头表示出来。def draw_policy(value_grid): 根据价值函数推导并绘制最优策略用箭头表示。 参数: value_grid: 二维numpy数组最优价值函数 fig, ax plt.subplots(fsize(6, 6)) ax.set_axis_off() ax.set_title(Optimal Policy Derived from Value Function, fontsize16, pad20) tb Table(ax, bbox[0, 0, 1, 1]) nrows, ncols value_grid.shape width, height 1.0 / ncols, 1.0 / nrows for (i, j), _ in np.ndenumerate(value_grid): state [i, j] # 计算每个动作的Q值动作价值即时奖励 折扣 * 下一状态价值 action_values [] for action in ACTIONS: next_state, r step(state, action) next_i, next_j next_state q_value r DISCOUNT * value_grid[next_i, next_j] action_values.append(q_value) # 找出最大Q值对应的动作可能不止一个 max_q np.max(action_values) best_action_indices np.where(np.isclose(action_values, max_q))[0] # 将最优动作组合成字符串如“↑→”表示向上和向右都是最优 policy_symbols .join([ACTIONS_FIGS[idx] for idx in best_action_indices]) # 标记特殊状态 cell_text policy_symbols if state A_POS: cell_text \n(A) facecolor lightcyan elif state A_PRIME_POS: cell_text \n(A) facecolor lightcyan elif state B_POS: cell_text \n(B) facecolor lightyellow elif state B_PRIME_POS: cell_text \n(B) facecolor lightyellow else: facecolor white tb.add_cell(i, j, width, height, textcell_text, loccenter, facecolorfacecolor, edgecolorblack) # 添加行号和列号 for i in range(nrows): tb.add_cell(i, -1, width, height, textstr(i1), locright, edgecolornone, facecolorwhitesmoke) for j in range(ncols): tb.add_cell(-1, j, width, height/2, textstr(j1), loccenter, edgecolornone, facecolorwhitesmoke) ax.add_table(tb) plt.tight_layout() return fig, ax有了这两个工具我们就能将算法计算出的抽象数字和策略转化为一眼就能看懂的图形。3. 策略评估计算随机漫步的价值现在让我们把智能体放进这个世界并赋予它第一个策略随机策略。也就是说在任何一个普通状态它都以均等的概率各25%选择上、下、左、右四个动作。在A和B状态规则强制其传送。我们的目标是计算出在这个固定策略下每个状态的价值V(s)。价值定义为从该状态出发遵循当前策略所能获得的未来累积奖励的期望折扣后总和。这可以通过求解贝尔曼期望方程得到。我们采用迭代策略评估的方法。其核心思想是“动态规划”从一个初始的猜测价值表比如全零开始反复用贝尔曼方程进行更新直到价值表收敛。贝尔曼期望方程的更新规则为V_{k1}(s) Σ_{a} π(a|s) * Σ_{s, r} p(s, r | s, a) * [ r γ * V_k(s) ]在我们的确定性环境中可以简化为V_{k1}(s) Σ_{a} π(a|s) * [ r(s, a) γ * V_k(s) ]其中π(a|s)是策略随机策略下为0.25r(s, a)是即时奖励s‘是确定的下一个状态γ是折扣因子0.9。def evaluate_random_policy(max_iterations1000, theta1e-4): 评估随机策略均匀随机选择动作。 返回: value_table: 收敛后的状态价值函数 iterations: 实际迭代次数 # 初始化价值表为0 value_table np.zeros((WORLD_SIZE, WORLD_SIZE)) iterations 0 for _ in range(max_iterations): iterations 1 delta 0.0 # 记录本轮最大价值变化 new_value_table np.zeros_like(value_table) for i in range(WORLD_SIZE): for j in range(WORLD_SIZE): state [i, j] new_value 0.0 # 对每个可能的动作计算其贡献的期望 for action in ACTIONS: next_state, reward step(state, action) next_i, next_j next_state # 贝尔曼期望方程更新 new_value ACTION_PROB * (reward DISCOUNT * value_table[next_i, next_j]) # 记录本轮该状态的价值变化 delta max(delta, abs(new_value - value_table[i, j])) new_value_table[i, j] new_value value_table new_value_table # 如果价值表变化很小认为已收敛 if delta theta: print(f随机策略评估在 {iterations} 次迭代后收敛。) break else: print(f警告随机策略评估在 {max_iterations} 次迭代后未完全收敛。) return value_table, iterations让我们运行这个评估过程并可视化结果。# 评估随机策略 random_policy_values, iters evaluate_random_policy() print(f随机策略下的状态价值表迭代{iters}次后:\n, np.round(random_policy_values, 2)) # 可视化 fig1, _ draw_value_image(random_policy_values, State Values under Random Policy) plt.show()运行后你会看到一个5x5的表格。观察结果你会发现一些有趣的现象状态A的价值非常高接近22因为它能直接带来10奖励并传送到A‘。状态B的价值也显著高于周围格子。靠近A和B的格子价值也会被“拉高”因为智能体有机会走到这些高奖励格子。边角格子的价值通常较低因为容易撞墙受罚。这个价值表精确地刻画了在“无脑随机走”的策略下每个位置的“好坏”程度。4. 价值迭代寻找通往最优的路径随机策略显然不是最好的。一个聪明的智能体应该学会选择动作。价值迭代算法就是用来寻找最优策略的一种动态规划方法。它不评估某个固定策略而是直接迭代更新最优价值函数V*(s)其贝尔曼最优方程更新规则为V_{k1}(s) max_{a} [ r(s, a) γ * V_k(s) ]这个公式非常直观一个状态的最优价值等于所有可能动作中能带来的“即时奖励下一状态折扣价值”的最大值。通过不断迭代这个“最大化”操作价值函数会最终收敛到最优价值函数。收敛后我们可以通过“贪心”地选择那个能使上式最大的动作来提取出最优策略。def value_iteration(max_iterations1000, theta1e-4): 执行价值迭代算法寻找最优价值函数和策略。 返回: optimal_values: 最优状态价值函数 iterations: 实际迭代次数 optimal_values np.zeros((WORLD_SIZE, WORLD_SIZE)) iterations 0 for _ in range(max_iterations): iterations 1 delta 0.0 new_optimal_values np.zeros_like(optimal_values) for i in range(WORLD_SIZE): for j in range(WORLD_SIZE): state [i, j] # 计算每个动作的Q值 action_values [] for action in ACTIONS: next_state, reward step(state, action) next_i, next_j next_state q_value reward DISCOUNT * optimal_values[next_i, next_j] action_values.append(q_value) # 贝尔曼最优方程更新取最大值 best_value np.max(action_values) delta max(delta, abs(best_value - optimal_values[i, j])) new_optimal_values[i, j] best_value optimal_values new_optimal_values if delta theta: print(f价值迭代在 {iterations} 次迭代后收敛。) break else: print(f警告价值迭代在 {max_iterations} 次迭代后未完全收敛。) return optimal_values, iterations运行价值迭代并同时展示最优价值函数和从中推导出的最优策略。# 执行价值迭代 optimal_values, opt_iters value_iteration() print(f最优状态价值表迭代{opt_iters}次后:\n, np.round(optimal_values, 2)) # 可视化最优价值函数 fig2, _ draw_value_image(optimal_values, Optimal State Values (Value Iteration)) # 可视化最优策略 fig3, _ draw_policy(optimal_values) plt.show()对比随机策略和价值迭代的结果你会发现显著差异价值更高最优价值表中的数字普遍比随机策略下的高因为智能体学会了做出更好的选择。策略清晰从策略图中你可以看到清晰的“行动流”。例如在A’(4,1)下方和右侧的格子策略会指向A‘因为这是去往高奖励状态A的捷径。整个网格会呈现出一种向高价值区域A、B及其传送点汇聚的箭头流向。为了更清晰地对比两种结果我们可以将它们放在一个表格中状态 (i,j)随机策略价值 V_π(s)最优价值 V*(s)价值提升幅度(0,0)~1.5~22.0巨大(0,1) (A)~22.0~25.0显著(4,1) (A‘)~21.5~24.5显著(2,2) (中心)~3.5~18.0显著(4,4) (右下角)~0.5~14.5巨大提示价值迭代找到的是全局最优策略。在这个确定性的环境中它等价于另一种经典算法——策略迭代。在实际编码中策略迭代通常涉及交替进行的“策略评估”和“策略改进”步骤收敛可能更快但每次迭代的计算量稍大。5. 高级话题与工程化扩展我们构建的Grid World环境虽然经典但相对简单。在实际的强化学习项目中环境需要更健壮、更灵活。下面我们来探讨几个工程化的扩展方向让你的环境更具实用性。5.1 封装成Gymnasium风格的环境OpenAI Gym及其后继者Gymnasium是强化学习领域事实上的环境标准接口。将我们的环境封装成类似的格式可以无缝接入大量现有的算法库如Stable-Baselines3, Ray RLlib。一个最简化的Gymnasium环境需要实现reset和step方法并定义observation_space和action_space。import gymnasium as gym from gymnasium import spaces import numpy as np class GridWorldEnv(gym.Env): metadata {render_modes: [human, rgb_array]} def __init__(self, world_size5, render_modeNone): super().__init__() self.world_size world_size self.render_mode render_mode # 定义观察空间和动作空间 # 观察空间智能体所在位置的坐标 (i, j) self.observation_space spaces.Box(low0, highworld_size-1, shape(2,), dtypenp.int32) # 动作空间4个离散动作 self.action_space spaces.Discrete(4) # 0:左, 1:上, 2:右, 3:下 self.action_map {0: L, 1: U, 2: R, 3: D} # 初始化世界动力学复用之前的构建函数 self.next_state, self.reward self._build_dynamics() self.state None self.window None def _build_dynamics(self): # ... (此处省略具体实现与前面build_world_dynamics函数逻辑相同) # 返回NEXT_STATE, REWARD pass def reset(self, seedNone, optionsNone): super().reset(seedseed) # 随机初始化智能体位置但避开特殊状态A和B或者固定从(2,2)开始 # 这里选择从网格中心开始 self.state np.array([self.world_size // 2, self.world_size // 2], dtypenp.int32) info {} return self.state, info def step(self, action): action_str self.action_map[action] i, j self.state next_state_coords self.next_state[i][j][action_str] reward self.reward[i][j][action_str] self.state np.array(next_state_coords, dtypenp.int32) terminated False # 这个Grid World没有终止状态可以一直进行 truncated False # 可以设置一个最大步数限制作为截断条件 info {} return self.state, reward, terminated, truncated, info def render(self): if self.render_mode human: if self.window is None: # 初始化pygame或matplotlib窗口进行可视化 pass # 绘制当前网格和智能体位置 pass # 返回RGB数组用于录制 return np.zeros((100, 100, 3), dtypenp.uint8) def close(self): if self.window is not None: # 关闭渲染窗口 pass封装成类后你就可以像使用标准Gym环境一样使用它env GridWorldEnv() state, info env.reset() for _ in range(10): action env.action_space.sample() # 随机动作 next_state, reward, terminated, truncated, info env.step(action) print(fState: {state}, Action: {action}, Reward: {reward}, Next State: {next_state}) state next_state if terminated or truncated: break env.close()5.2 引入随机性与不确定性现实世界充满不确定性。我们可以修改环境的动力学使其不再是确定性的。例如可以让智能体执行的动作有10%的概率“滑向”相邻方向。def stochastic_step(state, action, slip_prob0.1): 带有随机滑动的step函数。 参数: slip_prob: 执行预定动作失败随机滑向其他方向的概率。 i, j state # 预定动作 intended_next, intended_reward step(state, action) if np.random.rand() slip_prob: # 成功执行预定动作 return intended_next, intended_reward else: # 滑动随机选择其他三个方向之一 other_actions [a for a in ACTIONS if a ! action] slip_action np.random.choice(other_actions) slip_next, slip_reward step(state, slip_action) return slip_next, slip_reward这种随机性会让策略评估和价值迭代的计算变得稍微复杂需要计算期望但更能模拟真实场景也是检验算法鲁棒性的好方法。5.3 性能优化与向量化计算当状态空间变大时双重for循环的迭代会变慢。我们可以利用NumPy的向量化操作来加速。例如价值迭代的更新可以改写为对整个价值矩阵的操作。虽然对于5x5的网格没必要但对于更大的网格如50x50或更复杂的值函数近似如神经网络性能优化至关重要。一个思路是将状态转移表示为一个三维张量transition_probabilities[s, a, s]和一个三维奖励张量rewards[s, a, s]。然后贝尔曼最优更新可以写成向量化的形式V_{k1} np.max(np.sum(transition_probabilities * (rewards DISCOUNT * V_k), axis2), axis1)这需要预先构建庞大的转移矩阵在状态空间离散且不大时是可行的。5.4 添加动态渲染与交互使用Matplotlib的动画模块FuncAnimation或更强大的游戏库如Pygame可以实现智能体在网格中移动的实时动画。这对于教学演示和调试策略非常直观。你可以用不同颜色表示状态价值的高低用箭头实时显示智能体采取的动作甚至允许用户用键盘手动控制智能体进行探索。构建这个Grid World环境的过程本身就是一个对强化学习基础概念的深刻理解过程。从定义MDP五元组状态、动作、转移概率、奖励、折扣因子到实现策略评估和优化算法再到结果的可视化每一步都对应着理论中的一个环节。当你看到代表最优策略的箭头从网格的各个角落指向高奖励区域时你看到的不仅仅是代码的输出更是智能体通过“思考”学会的生存之道。这个环境可以作为你后续探索更高级算法如蒙特卡洛方法、时序差分学习、深度Q网络的完美起点。试着修改奖励函数、增加障碍物、设计更复杂的任务目标你会发现这个简单的网格世界蕴藏着强化学习无限的可能性。