马尔可夫决策过程Q值实现求助:计算结果异常
Q值计算错误问题排查与修复
问题背景
实现Q值迭代时,大部分代码可运行,但Q值计算结果不正确。需求是:当智能体动作指向无状态区域或灰色格子时,需反弹回原状态。
Q值计算公式:
$Q(s,a) = R(s,a) + \gamma \sum_{s'} P(s'|s,a) V(s')$
其中:
- $R(s,a)$:执行动作$a$的即时奖励
- $\gamma$:折扣因子
- $P(s'|s,a)$:从状态$s$执行动作$a$后转移到$s'$的概率
- $V(s')$:状态$s'$的价值
当前代码
from cell import states import pygame import drawfn ACTION_EAST=0 ACTION_SOUTH=1 ACTION_WEST=2 ACTION_NORTH=3 TRANSITION_SUCCEED=0.8 #执行动作A后到达预期状态S'的概率 TRANSITION_FAIL=0.2 #执行动作A后到达非预期状态S'的概率,比如向东时可能向北或向南,各占0.1 GAMMA=0.9 #折扣因子 ACTION_REWARD=-0.1 #每个动作的即时奖励(四个动作奖励相同) CONVERGENCE=0.0000001 #收敛阈值 cur_convergence=100 #####需要实现的函数############################ #确保箭头指向空白或灰色格子时反弹回原状态 def computeQValue(s,action): print('Compute Q Values') #无返回值 #尝试所有动作 #s是每个单元格的状态 #动作取值0-3:0-东,1-南,2-西,3-北 #根据每个单元格的动作计算Q值 #更新状态数据中的Q值 # 计算给定动作和状态的Q值 global state_value global q_values global policy global transition transitions = {0:(1,0),1:(0,-1), 2:(-1,0),3:(0,1)} i = 0 #遍历每一行 for row in states: #遍历每行中的状态 next_state_value = s.state_value for state in states: i += 1 print('loop' +str(i)) #计算给定动作和状态的Q值 if action == ACTION_EAST: s.q_values[0] = ACTION_REWARD + GAMMA * (TRANSITION_SUCCEED * next_state_value + (TRANSITION_FAIL * (s.q_values[1] + s.q_values[3]))/2) print(s.q_values[0]) elif action == ACTION_SOUTH: s.q_values[1] = ACTION_REWARD + GAMMA * (TRANSITION_SUCCEED * next_state_value + (TRANSITION_FAIL * (s.q_values[2] + s.q_values[0]))/2) print(s.q_values[1]) elif action == ACTION_WEST: s.q_values[2] = ACTION_REWARD + GAMMA * (TRANSITION_SUCCEED * next_state_value + (TRANSITION_FAIL * (s.q_values[1] + s.q_values[3]))/2) print(s.q_values[2]) else: s.q_values[3] = ACTION_REWARD + GAMMA * (TRANSITION_SUCCEED * next_state_value + (TRANSITION_FAIL * (s.q_values[0] + s.q_values[2]))/2) print(s.q_values[3]) def valueIteration(): print('Value Iteration.') #无返回值 #循环调用 #使用computeQValue更新每个单元格的状态值 #理想情况下应在100次迭代内得到策略 #使用cur_convergence和CONVERGENCE判断收敛 global CONVERGENCE global cur_convergence global conv_gap if (cur_convergence >= CONVERGENCE): print('at line 62') for r in states: for s in r: old_qval= s.state_value print('cell location' + str(s.location)) for i in range(4): computeQValue(s, i) for r in states: for s in r: if(s.location == (1,1) or s.location == (3,0) or s.location == (3,1)): pass else: s.state_value = max(s.q_values) new_qval = s.state_value cur_convergence = new_qval-old_qval else: print('at line 78') CONVERGENCE = 101
当前与期望输出
- 当前输出:所有单元格状态值均为-0.1,无梯度变化
- 期望输出:单元格状态值呈现梯度变化,目标区域附近值更高,符合Q值迭代的收敛结果
问题根源与修复方案
1. computeQValue函数循环逻辑错误
原函数嵌套遍历所有states,导致单个状态的Q值被重复覆盖,完全偏离“计算单个状态s对应动作action的Q值”的目标,需删除多余循环。
2. 未实现状态反弹逻辑
原代码未判断动作指向的目标状态是否有效(无状态区域或灰色格子),需添加逻辑:若目标状态无效,则转移回原状态s。
3. next_state_value计算错误
原代码直接用s.state_value作为转移后的状态值,未根据动作计算实际目标状态的价值,违反Q值公式要求。
4. 收敛判断逻辑错误
原代码仅更新最后一个状态的收敛差值,应计算所有状态中最大的差值作为当前收敛指标,否则会提前终止或无效迭代。
5. 错误引用Q值代替状态值
原代码在计算转移概率时错误使用s.q_values,根据公式应使用转移后状态的状态值V(s'),而非Q值。
修正后的代码
from cell import states import pygame import drawfn ACTION_EAST = 0 ACTION_SOUTH = 1 ACTION_WEST = 2 ACTION_NORTH = 3 TRANSITION_SUCCEED = 0.8 TRANSITION_FAIL = 0.2 GAMMA = 0.9 ACTION_REWARD = -0.1 CONVERGENCE = 0.0000001 cur_convergence = 100 # 定义状态转移方向 transitions = { ACTION_EAST: (1, 0), ACTION_SOUTH: (0, -1), ACTION_WEST: (-1, 0), ACTION_NORTH: (0, 1) } # 获取指定坐标的状态,不存在则返回None def get_state_by_pos(x, y): for row in states: for state in row: if state.location == (x, y): return state return None # 计算单个状态s对应动作action的Q值 def computeQValue(s, action): # 灰色格子(障碍)不计算Q值 if s.location in [(1,1), (3,0), (3,1)]: return total_value = 0.0 # 1. 处理成功转移的情况 dx, dy = transitions[action] target_x = s.location[0] + dx target_y = s.location[1] + dy target_state = get_state_by_pos(target_x, target_y) # 若目标状态无效(无状态区域或障碍),则反弹回原状态 if target_state is None or target_state.location in [(1,1), (3,0), (3,1)]: succeed_value = s.state_value else: succeed_value = target_state.state_value total_value += TRANSITION_SUCCEED * succeed_value # 2. 处理失败转移的情况(两个垂直方向) fail_actions = [] if action in [ACTION_EAST, ACTION_WEST]: fail_actions = [ACTION_NORTH, ACTION_SOUTH] else: fail_actions = [ACTION_EAST, ACTION_WEST] fail_sum = 0.0 for fail_act in fail_actions: dx_fail, dy_fail = transitions[fail_act] fail_target_x = s.location[0] + dx_fail fail_target_y = s.location[1] + dy_fail fail_target_state = get_state_by_pos(fail_target_x, fail_target_y) if fail_target_state is None or fail_target_state.location in [(1,1), (3,0), (3,1)]: fail_sum += s.state_value else: fail_sum += fail_target_state.state_value total_value += (TRANSITION_FAIL / 2) * fail_sum # 更新当前动作的Q值 s.q_values[action] = ACTION_REWARD + GAMMA * total_value def valueIteration(): global cur_convergence if cur_convergence >= CONVERGENCE: max_diff = 0.0 # 1. 先计算所有状态的Q值 for row in states: for s in row: if s.location in [(1,1), (3,0), (3,1)]: continue for action in range(4): computeQValue(s, action) # 2. 更新状态值并计算最大收敛差值 for row in states: for s in row: if s.location in [(1,1), (3,0), (3,1)]: continue old_value = s.state_value s.state_value = max(s.q_values) diff = abs(s.state_value - old_value) if diff > max_diff: max_diff = diff cur_convergence = max_diff else: # 收敛后停止迭代 pass
关键说明
- 添加
get_state_by_pos函数用于根据坐标查找状态,方便判断目标状态是否有效。 - 严格遵循Q值公式:先计算成功/失败转移的状态值加权和,再结合奖励与折扣因子得到Q值。
- 实现反弹逻辑:当动作指向无效状态时,转移回原状态并使用原状态值计算。
- 修正收敛判断:计算所有状态值变化的最大差值作为收敛指标,确保迭代至真正收敛。
内容的提问来源于stack exchange,提问作者jwolf
相关产品推荐
相关产品推荐

