You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

马尔可夫决策过程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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.29 10:37:01