别再死记硬背公式了！用Python手把手带你实现Model-based强化学习（附值迭代/策略迭代代码）

用Python实战Model-based强化学习从零实现值迭代与策略迭代在传统强化学习教材中数学公式和算法伪代码常常让学习者望而生畏。本文将通过Python代码实现一个完整的网格世界Grid World环境并逐步构建值迭代和策略迭代算法。我们将用NumPy处理矩阵运算Matplotlib可视化策略更新过程最后对比两种算法的收敛特性与适用场景。1. 搭建网格世界环境任何强化学习实践都需要一个明确的环境模型。我们首先构建一个5x5的网格世界import numpy as np import matplotlib.pyplot as plt class GridWorld: def __init__(self, size5): self.size size self.walls [(1,1), (2,3), (3,1)] # 障碍物坐标 self.goal (4,4) # 目标位置 self.trap (0,4) # 陷阱位置 self.actions [up, down, left, right] def step(self, state, action): x, y state if action up: x max(0, x-1) elif action down: x min(self.size-1, x1) elif action left: y max(0, y-1) elif action right: y min(self.size-1, y1) next_state (x, y) if next_state in self.walls: # 撞墙保持原位 next_state state if next_state self.goal: reward 10 done True elif next_state self.trap: reward -10 done True else: reward -1 # 每步有小惩罚 done False return next_state, reward, done这个环境包含以下关键元素状态空间5x5网格共25个状态动作空间上、下、左、右四个基本动作转移规则碰到边界或障碍物保持原位奖励机制到达目标点10掉入陷阱-10每移动一步-1鼓励高效路径2. 值迭代算法实现值迭代的核心思想是通过不断更新状态价值函数来逼近最优值函数。以下是完整实现def value_iteration(env, gamma0.9, theta1e-4): V np.zeros((env.size, env.size)) policy np.empty((env.size, env.size), dtypeobject) while True: delta 0 for i in range(env.size): for j in range(env.size): if (i,j) env.goal or (i,j) env.trap: continue v_old V[i,j] q_values [] for action in env.actions: (next_i, next_j), reward, _ env.step((i,j), action) q reward gamma * V[next_i, next_j] q_values.append(q) V[i,j] max(q_values) delta max(delta, abs(v_old - V[i,j])) if delta theta: break # 提取最优策略 for i in range(env.size): for j in range(env.size): if (i,j) env.goal: policy[i,j] goal elif (i,j) env.trap: policy[i,j] trap else: q_values [] for action in env.actions: (next_i, next_j), reward, _ env.step((i,j), action) q reward gamma * V[next_i, next_j] q_values.append(q) best_action env.actions[np.argmax(q_values)] policy[i,j] best_action return V, policy关键实现细节初始化价值函数V初始化为全零矩阵循环终止条件当最大价值变化小于阈值θ时停止状态价值更新使用贝尔曼最优方程更新每个状态价值策略提取在所有状态价值收敛后根据最优价值函数提取策略可视化收敛过程def plot_value_convergence(V_history): plt.figure(figsize(10,6)) for i in range(V_history[0].shape[0]): for j in range(V_history[0].shape[1]): plt.plot([v[i,j] for v in V_history], labelf({i},{j})) plt.xlabel(Iterations) plt.ylabel(State Value) plt.title(Value Iteration Convergence) plt.legend() plt.show()3. 策略迭代算法实现策略迭代包含策略评估和策略提升两个交替进行的阶段def policy_iteration(env, gamma0.9, theta1e-4): # 初始化随机策略 policy np.random.choice(env.actions, size(env.size, env.size)) policy[env.goal] goal policy[env.trap] trap V np.zeros((env.size, env.size)) while True: # 策略评估 while True: delta 0 for i in range(env.size): for j in range(env.size): if (i,j) env.goal or (i,j) env.trap: continue v_old V[i,j] action policy[i,j] (next_i, next_j), reward, _ env.step((i,j), action) V[i,j] reward gamma * V[next_i, next_j] delta max(delta, abs(v_old - V[i,j])) if delta theta: break # 策略提升 policy_stable True for i in range(env.size): for j in range(env.size): if (i,j) env.goal or (i,j) env.trap: continue old_action policy[i,j] q_values [] for action in env.actions: (next_i, next_j), reward, _ env.step((i,j), action) q reward gamma * V[next_i, next_j] q_values.append(q) best_action env.actions[np.argmax(q_values)] policy[i,j] best_action if old_action ! best_action: policy_stable False if policy_stable: break return V, policy算法特点对比特性值迭代策略迭代计算复杂度O(S²A) per iterationO(S³) per evaluation收敛速度线性收敛二次收敛内存需求只需存储V需存储V和π适用场景大状态空间小状态空间策略更新频率最后一次性提取每次迭代都更新4. 算法对比与实战技巧在实际运行中我们发现两种算法各有优劣值迭代特点每次迭代更新所有状态价值不需要完整策略评估适合状态空间较大的问题代码实现相对简单策略迭代特点策略评估阶段需要完全收敛通常需要更少的迭代次数适合可以承受每次迭代高计算量的场景策略质量在每次迭代都有提升实用调试技巧收敛监控# 在值迭代中添加 print(fIteration {k}, Delta: {delta:.4f}) # 在策略迭代中添加 print(fPolicy improved {np.sum(policy ! new_policy)} states)可视化工具def plot_policy(policy, env): arrows {up:↑, down:↓, left:←, right:→, goal:G, trap:X} grid np.empty((env.size, env.size), dtypeU2) for i in range(env.size): for j in range(env.size): grid[i,j] arrows[policy[i,j]] print(grid)超参数调优γ (折扣因子)通常在0.9-0.99之间θ (收敛阈值)一般设为1e-4到1e-6最大迭代次数防止无限循环实际项目中我通常会先用策略迭代在小规模问题上验证思路再对大规模问题采用值迭代。当状态空间特别大时可以考虑异步更新策略只随机选择部分状态进行更新。