本文用一个 4×4 的 GridWorld讲清楚蒙特卡洛方法如何从完整回合中估计动作价值以及 ε-greedy 如何帮助智能体在探索与利用之间做出选择。实验包含 Q 表、训练曲线、策略箭头和最终路径适合已经了解 Python 循环、字典和函数的读者。1. 从一个问题开始智能体怎样知道往哪走假设我们把智能体放在一个 4×4 的网格中要求它从左上角走到右下角。S □ □ □ □ □ □ □ □ □ □ □ □ □ □ G其中S 是起点 (0,0)G 是终点 (3,3)。智能体每次可以向上、下、左、右移动一格碰到边界就停留在原地。人很容易看出向右走 3 步再向下走 3 步就能到达终点。但我们不会把这条路线直接写进智能体的策略而是让它通过反复尝试学习。为此环境规定**每执行一次动作奖励为 −1到达终点后回合结束。** 进入终点的那一步也获得 −1并没有额外的终点正奖励。撞墙虽然没有改变位置也会消耗一步并得到 −1。这就把“尽快到达终点”变成了一个可计算的目标对于本例中每步奖励相同的轨迹走得越久累计折扣回报越低。本文采用的学习思路是先走完一个回合再回头判断其中每个动作的长期结果。这就是蒙特卡洛方法的核心直觉。2. 把网格问题表示成强化学习环境2.1 状态、动作、奖励与回合概念在本文中的定义代码中的表示状态智能体所在的格子(row, col)动作向右、向下、向左、向上ACTIONS奖励每次执行动作获得 −1reward -1终止条件进入 (3,3)done next_state GOAL回合从起点开始的一次尝试episode一步交互执行动作并获得环境反馈step(state, action)网格共有 16 个状态其中 15 个是非终止状态。每个非终止状态都有 4 个动作因此 Q 表中需要保存 60 个动作价值。这里的 step 函数是环境提供的交互接口。智能体通过调用它获得样本而不是枚举所有转移来计算价值。这也是本实现采用无模型学习方式的原因学习更新不需要预先给出状态转移概率表。2.2 环境如何处理边界new_row max(0, min(row dr, ROWS - 1)) new_col max(0, min(col dc, COLS - 1))以 (0,0) 执行 up 为例行坐标原本会变成 −1但经过边界限制后仍然是 0所以智能体留在原地。注意“状态没有改变”并不意味着“没有发生交互”。这一步仍然会被记录进轨迹也会影响回报。3. Q 表把动作的长期结果记录下来只看即时奖励四个动作都得到 −1无法比较好坏。真正需要比较的是**在当前状态执行某个动作之后后续整段行程的回报如何**动作价值函数描述的正是这个量其中s 是状态a 是动作π 是后续遵循的策略G_t 是从当前时刻开始的折扣回报。期望表示同样的状态和动作可能对应多种后续轨迹需要综合多次经验判断。代码使用 Q[state][action] 保存估计值Q { state: {action: 0.0 for action in ACTIONS} for state in STATES if state ! GOAL }终点不需要选择动作因此没有为它建立动作条目。Q 值全部初始化为 0是因为开始时没有经验并不表示动作的真实价值为 0。本例回报为负因此 0 还带有乐观初始化的意味尚未采样的动作可能暂时比已经得到负回报的动作更有吸引力。Q 值越大动作越好。对于负数−2 大于 −5所以 −2 对应的动作更值得选择。4. ε-greedy既利用经验也保留探索如果智能体永远只选择当前 Q 值最大的动作它可能过早相信不准确的估计。ε-greedy 用一个简单规则解决这个问题- 以 ε 的概率从所有动作中随机选择- 以 1−ε 的概率从当前 Q 值最大的动作中选择。本例设置 EPSILON 0.1即 10% 进入随机探索分支90% 进入利用分支。def choose_action(state, epsilon): if random.random() epsilon: return random.choice(ACTIONS) max_q max(Q[state].values()) best_actions [ action for action in ACTIONS if abs(Q[state][action] - max_q) 1e-12 ] return random.choice(best_actions)这里有两个容易忽略的细节。第一随机探索也可能选中最优动作。如果当前只有一个最优动作它被选择的总概率是第二如果有多个动作并列最优利用分支会在它们之间随机选择。设并列最优动作有 k 个则每个最优动作的概率为5. 蒙特卡洛方法等回合结束再计算回报5.1 为什么要保存整条轨迹代码使用一个列表保存当前回合的经历trajectory.append((state, action, reward))每条记录包含当前状态、所选动作和该动作产生的即时奖励。这里的状态是执行动作之前的状态。trajectory 在每个回合开始时重新创建因此它保存的是当前回合而不是所有训练历史。蒙特卡洛更新需要知道一个动作之后实际经历了什么所以要先获得完整回合再计算每个时刻的回报。5.2 即时奖励与回报有什么区别即时奖励是当前一步得到的反馈回报则把当前一步及后续步骤的奖励合并起来。设回合在时刻 T 终止则γ 是折扣因子。本文取 GAMMA 0.9距离当前越远的奖励权重越小。公式中的 R_{t1}就是代码中 trajectory[t] 保存的 reward_t它是执行第 t 个动作后获得的奖励。两种下标写法描述的是同一个交互结果。假设智能体从 (1,2) 出发依次向下、向下、向右经过 3 步到达终点。每步奖励为 −1因此第一步对应的回报是第二步对应的回报为 −1.9第三步对应的回报为 −1。如果从起点用 6 步到达终点这条轨迹的初始回报为这说明两件事6 步轨迹的未折扣奖励和是 −6但其折扣回报约为 −4.686最终学到的起点 Q 值也不必恰好等于这个数因为训练过程中存在探索和不同长度的轨迹。5.3 为什么从后往前计算回报满足递推关系终点之后不再产生奖励因此从 0 开始沿轨迹倒着计算即可returns [0.0] * len(trajectory) G 0.0 for t in range(len(trajectory) - 1, -1, -1): state_t, action_t, reward_t trajectory[t] G reward_t GAMMA * G returns[t] G对刚才的 3 步例子计算顺序为最后一步 −1倒数第二步 −1.9第一步 −2.71。这里的 G 是沿已经采样的轨迹计算出来的后续回报没有使用下一状态的 Q 估计因此这个递推计算仍然是 MC 回报计算。6. 首次访问一个回合内同一个状态—动作对只更新一次智能体可能在同一个回合中反复经过某个格子甚至重复执行相同动作。首次访问方法只使用这个状态—动作对在该回合第一次出现时的回报visited set() for t, (state_t, action_t, _) in enumerate(trajectory): pair (state_t, action_t) if pair in visited: continue visited.add(pair) N[state_t][action_t] 1 n N[state_t][action_t] Q[state_t][action_t] ( returns[t] - Q[state_t][action_t] ) / n计算回报时倒序遍历判断首次访问时正序遍历。** 如果直接倒序遍历并把第一次遇到的状态—动作对拿来更新选中的其实是它在原始轨迹中最后一次出现的位置。visited 每个回合都会清空所以同一个状态—动作对可以在下一个回合再次更新。7. 增量平均不保存所有回报也能计算均值假设某个状态—动作对已经获得 n 个首次访问回报样本那么 Q 的样本平均为将前 n−1 个样本的平均值代入这就是代码中的更新式old_q Q[state_t][action_t] Q[state_t][action_t] old_q (returns[t] - old_q) / n例如前三次采样回报为 −4、−6、−5则 Q 值依次变为 −4、−5、−5与直接计算平均值一致。这里还需要修正一个常见理解**N 记录的是被用于更新的首次访问回报样本数而不是训练中的所有逐步访问次数。** 同一个回合内重复访问不会增加 N被超时丢弃的回合也不会增加 N。把 N 初始化为整数 0语义会更清楚。8. 把整个训练过程串起来每个训练回合执行以下流程1. 从 START 出发用 ε-greedy 选择动作。2. 与环境交互保存 (state, action, reward)。3. 到达终点后倒序计算每个时刻的回报。4. 正序遍历轨迹更新每个状态—动作对的首次访问样本。5. 下一回合继续依据更新后的 Q 表选择动作。Q 表改变以后ε-greedy 的利用分支也随之改变。因此代码不需要额外维护一张策略表也已经完成了策略的逐步改进。这段程序属于**同策略首次访问蒙特卡洛控制**采样使用当前 ε-greedy 策略价值更新用于改进同一套策略。训练结束后再设置 ε0只是测试学到的贪心行为。原始代码代码中的注释会更清楚# 导入随机数库用于epsilon-greedy探索 import random # 导入matplotlib绘图库用于可视化训练曲线、策略网格、行走路径 import matplotlib.pyplot as plt # 设置随机种子保证每次运行结果固定方便复现实验 random.seed(42) # 1. 创建 GridWorld 网格世界环境 # 网格行数、列数4×4网格 ROWS 4 COLS 4 # 起点坐标 (行,列) START (0, 0) # 目标终点坐标 GOAL (3, 3) # 定义4个可选动作 ACTIONS [right, down, left, up] # 动作映射每个动作对应的坐标增量 (dr, dc) # dr:行变化dc:列变化 MOVES { up: (-1, 0), # 向上行-1列不变 down: (1, 0), # 向下行1列不变 left: (0, -1), # 向左列-1 right: (0, 1) # 向右列1 } # 生成网格世界所有状态所有单元格坐标 STATES [ (row, col) for row in range(ROWS) for col in range(COLS) ] def step(state, action): 环境一步交互函数 :param state: 当前状态 (row, col) :param action: 选择的动作 :return: next_state下一状态, reward即时奖励, done是否到达终点 # 如果当前已经在终点不再移动奖励0回合结束 if state GOAL: return GOAL, 0, True row, col state dr, dc MOVES[action] # 计算新坐标max/min实现边界阻挡碰到网格边界不会走出地图 new_row max(0, min(row dr, ROWS - 1)) new_col max(0, min(col dc, COLS - 1)) next_state (new_row, new_col) # 每走一步奖励固定为-1鼓励智能体尽快到达终点步数越少总回报越高 reward -1 # 判断是否到达终点 done next_state GOAL return next_state, reward, done # 2.初始化Q值和访问次数 # Q表Q[state][action]存储状态- Q {} # N表N[state][action]记录该[state][action]在训练中被访问的次数用于增量平均更新 N {} # 遍历所有状态初始化Q和N终点状态不存储到达即停止 for state in STATES: if state GOAL: continue # 每个状态下4个动作初始Q值全部置0 Q[state] { action: 0.0 for action in ACTIONS } # 访问次数全部初始化为0 N[state] { action: 0.0 for action in ACTIONS } # 3.设置强化学习超参数 GAMMA 0.9 # 折扣因子未来奖励的衰减系数0~1之间,决定未来获得的奖励在计算当前回报时应该占多大权重 EPSILON 0.1 # epsilon-greedy的探索概率10%概率随机探索90%选当前最优动作 EPISODES 15000 # 总训练回合数 智能体最多尝试进行多少次完整的训练回合MC需要较多的样本数量可以改善采样估计 MAX_STEPS 500 # 单个episode最大步数防止智能体无限原地循环 episode_steps [] # 保存每一个成功到达终点回合的步数用于后续的绘图 skipped_episodes 0 # 记录因为超时未到达终点而跳过更新的回合数量 # 4. epsilon-greedy 动作选择函数 def choose_action(state, epsilon): ε-贪心策略以epsilon概率探索1-epsilon概率利用选Q最大动作 :param state: 当前状态 :param epsilon: 探索概率 :return: 选中的动作字符串 # 探索分支随机选动作避免过早陷入次优策略 if random.random() epsilon: return random.choice(ACTIONS) # 利用分支选取当前状态下Q值最大的动作 max_q max(Q[state].values()) # 收集所有Q值等于最大值的动作多个动作Q相同时随机挑一个 best_actions [ action for action in ACTIONS if abs(Q[state][action] - max_q) 1e-12 ] return random.choice(best_actions) # 5.蒙特卡洛首次访问Frist—Visit MC)训练主循环 for episode_number in range(EPISODES): # Episode(回合:从起点开始的一整次尝试直到到达终点或被步数上限截断。 # step(步智能体在某个状态下执行一次动作并获得一次环境反馈 # 一个Episode可以包含很多个Step state START # 每个回合从起点开始 trajectory [] # 轨迹缓存存储本幕所有状态动作即时奖励是单个Episode的历史记录不是所有Episode的历史记录 reached_goal False # 标记本回合是否成功抵达终点 # 第一步执行完整Episode,收集轨迹 for t in range(MAX_STEPS): # 根据ε-贪心选动作 action choose_action(state, EPSILON) # 和环境交互得到下一状态、奖励、结束标记 next_state, reward, done step(state, action) # 将当前步骤存入轨迹 trajectory.append( (state, action, reward) ) # 状态向前转移 state next_state # 到达终点终止本回合采样 if done: reached_goal True break # 本回合步数耗尽仍未到达终点丢弃这条轨迹不更新Q表 if not reached_goal: skipped_episodes 1 continue # 保存本回合到达终点的总步数 episode_steps.append(len(trajectory)) # 收集轨迹trajectory,计算完整回报returns # 第二步从后往前计算每一步的回报G(return) returns [0.0] * len(trajectory) # 创建回报列表 G 0.0 # 设置初始回报 # 逆序遍历轨迹从最后一步向前递推回报 # 因为当前回报需要利用后续回报而终点的回报已经确定为0所以我们从终点向前计算就能一步一步求出整段轨迹中每个时刻的完整回报 for t in range(len(trajectory) - 1, -1, -1): # 从len(trajectory) - 1开始到-1之前结束每次减1 state_t, action_t, reward_t trajectory[t] G reward_t GAMMA * G returns[t] G # 将计算出的回报放入对应的位置 # 第三步首次访问MC更新规则 # 同一个episode里,一个(state,action)对只在*第一次出现*时更新 # 集合中的元素不能重复 visited set() # 记录本episode已经更新过的(state,action)对 # 遍历轨迹并且同时获取每条记录的位置和内容 for t, (state_t, action_t, reward_t) in enumerate(trajectory): pair (state_t, action_t) # 表示一个状态——动作对因为在RL中同一个状态下的不同动作具有不同的价值 # 我们后续需要进行首次访问判断 if pair in visited: continue # 如果当前状态——动作对已经在本次Episode中处理过就跳过这次循环。 visited.add(pair) N[state_t][action_t] 1 # 访问次数1 n N[state_t][action_t] # 方便后面计算1/n # 增量平均公式Q_new Q_old (G - Q_old)/n old_q Q[state_t][action_t] Q[state_t][action_t] ( old_q (returns[t] - old_q) / n ) # MC的Q值更新 # 每训练3000个回合就打印一次训练信息监控收敛 if (episode_number 1) % 3000 0: recent episode_steps[-200:] # 取最近200个成功回合的步数 print( fEpisode {episode_number 1} | f最近平均步数: {sum(recent) / len(recent):.2f} ) # 6. 打印最终训练完成后的Q表 print(\n 最终 Q 表 ) # 格式化表头输出 print( f{State:10} f{Up:10} f{Down:10} f{Left:10} f{Right:10} f{Best:10} ) # 遍历所有状态打印每个动作Q值和最优动作 for state in STATES: if state GOAL: continue # 获取该状态Q值最大的动作贪心最优动作 best_action max( Q[state], keyQ[state].get ) print( f{str(state):10} f{Q[state][up]:10.3f} f{Q[state][down]:10.3f} f{Q[state][left]:10.3f} f{Q[state][right]:10.3f} f{best_action:10} ) # 7. 使用纯贪心策略(ε0)测试最终学到的策略输出路径 state START path [state] # 保存测试行走路径 success False # 最多走30步防止完全卡死 for _ in range(30): action choose_action(state, epsilon0.0) next_state, reward, done step(state, action) path.append(next_state) state next_state if done: success True break print(\n最终路径, path) print(路径步数, len(path) - 1) print(是否到达终点, success) print(未完成训练回合, skipped_episodes) # 8. matplotlib可视化绘图3张子图 # 子图1训练步数移动平均曲线 # 子图2网格上的贪心策略箭头 # 子图3智能体测试走出来的最终路径 # # 创建一行3个子图画布大小14×4.5英寸 fig, axes plt.subplots( 1, 3, figsize(14, 4.5) ) # ---------- 图1训练步数移动平均 ---------- window 100 # 滑动窗口大小取最近100个episode做平均 moving_average [] for i in range(len(episode_steps)): # 窗口左边界 start_index max(0, i - window 1) # 取出窗口内步数 recent_steps episode_steps[start_index:i 1] moving_average.append(sum(recent_steps) / len(recent_steps)) axes[0].plot(moving_average) # 绘制最短路径参考虚线4×4网格从(0,0)到(3,3)最少6步 axes[0].axhline( y6, colorgray, linestyle--, labelShortest path: 6 ) axes[0].set_title(MC Training Progress) axes[0].set_xlabel(Completed Episodes) axes[0].set_ylabel(Average Steps) axes[0].legend() axes[0].grid(True) # ---------- 图2学习得到的贪心策略方向箭头 ---------- # 动作转字符箭头 arrows { up: ↑, down: ↓, left: ←, right: → } # 在网格每个单元格绘制最优动作符号 for row in range(ROWS): for col in range(COLS): state (row, col) if state GOAL: symbol G else: # 选取贪心最优动作 action max(Q[state], keyQ[state].get) symbol arrows[action] # 在子图2对应位置添加文本 axes[1].text( col, row, symbol, hacenter, vacenter, fontsize22 ) axes[1].set_title(Learned Greedy Policy) # ---------- 图3测试得到的最终行走路径 ---------- # 循环绘制相邻状态之间的箭头 for i in range(len(path) - 1): row1, col1 path[i] row2, col2 path[i 1] axes[2].annotate( , xy(col2, row2), xytext(col1, row1), arrowpropsdict( arrowstyle-, lw2, colortab:blue ) ) # 标注起点S、终点G axes[2].text( 0, 0, S, fontsize18, hacenter, vacenter ) axes[2].text( 3, 3, G, fontsize18, hacenter, vacenter ) axes[2].set_title(Final Path) # 统一设置子图2、子图3的坐标轴和网格 for ax in axes[1:]: ax.set_xlim(-0.5, COLS - 0.5) ax.set_ylim(ROWS - 0.5, -0.5) ax.set_xticks(range(COLS)) ax.set_yticks(range(ROWS)) # minor刻度用于画单元格之间的分隔线 ax.set_xticks( [i 0.5 for i in range(COLS - 1)], minorTrue ) ax.set_yticks( [i 0.5 for i in range(ROWS - 1)], minorTrue ) ax.grid( whichminor, colorgray, linewidth1 ) # 自动调整子图间距防止文字重叠 plt.tight_layout() plt.show()运行结果Episode 3000 | 最近平均步数: 6.82 Episode 6000 | 最近平均步数: 6.49 Episode 9000 | 最近平均步数: 6.63 Episode 12000 | 最近平均步数: 6.64 Episode 15000 | 最近平均步数: 6.72 最终 Q 表 State Up Down Left Right Best (0, 0) -5.501 -4.969 -5.477 -4.993 down (0, 1) -5.523 -4.385 -5.794 -5.326 down (0, 2) -8.351 -10.000 -10.000 -4.767 right (0, 3) -9.657 -3.857 -10.000 -10.000 down (1, 0) -5.545 -4.355 -4.952 -4.394 down (1, 1) -5.058 -3.680 -5.001 -4.376 down (1, 2) -10.000 -2.943 -4.514 -4.252 down (1, 3) -6.719 -2.279 -5.445 -4.407 down (2, 0) -4.979 -3.781 -4.396 -3.674 right (2, 1) -4.329 -2.902 -4.396 -2.919 down (2, 2) -4.048 -2.406 -4.100 -2.012 right (2, 3) -3.398 -1.000 -3.246 -2.249 down (3, 0) -4.459 -4.048 -4.015 -2.939 right (3, 1) -3.689 -2.910 -3.690 -2.007 right (3, 2) -2.872 -2.050 -2.940 -1.000 right 最终路径 [(0, 0), (1, 0), (2, 0), (2, 1), (3, 1), (3, 2), (3, 3)] 路径步数 6 是否到达终点 True 未完成训练回合 2左图训练步数移动平均。** 初期 Q 值没有可靠依据智能体会绕路或撞墙随着经验增加平均步数明显下降。由于训练时 ε 始终是 0.1仍然会随机探索因此后期平均步数在 6 以上波动是正常现象。横轴是成功完成回合的索引超时回合没有画进去。不能只看这条曲线判断全部训练尝试的表现还应同时检查超时数量或成功率。中图各个格子的贪心动作。** 每个箭头展示该格子 Q 值最大的动作。它是对 Q 表的贪心提取训练采样实际使用的是带探索的 ε-greedy。绘图用 max(Q[state], keyQ[state].get) 选一个最大值而 choose_action 会在容差内并列最优动作之间随机选择。因此出现并列时策略图不一定呈现测试会采用的全部选择。右图从指定起点执行贪心策略得到的轨迹。** 它验证的是这次起点测试不是全部状态的最优性。