Gym.spaces.Box场景中DQN处理连续动作空间的报错解决

问题背景 #

自定义Gym环境的动作空间和观测空间均为连续型（基于gym.spaces.Box实现），随机迭代运行正常，但使用DQNAgent训练时出现两个问题：

1. 输出的动作仅为0或1，不符合连续动作空间的要求
2. 触发报错：TypeError: 'int' object is not subscriptable

核心原因 #

1. DQNAgent的设计局限：DQNAgent是为离散动作空间设计的，它会输出离散的动作索引（而非连续动作值）。你的动作空间维度为2，所以它输出0或1这两个整数索引，导致代码中action[0]访问时触发类型错误（整数无法下标访问）。
2. 输出层激活与动作空间不匹配：原模型最后一层用tanh激活，输出范围是[-1,1]，没有映射到你定义的动作空间范围[0,10]和[0,100]，即使输出连续值也无法满足环境要求。

解决方案 #

改用支持连续动作空间的DDPGAgent（来自keras-rl库），并调整模型和Agent配置：

关键修改点 #

• 替换DQNAgent为DDPGAgent
• 调整模型输出层为线性激活，配合处理器自动映射到动作空间范围
• 添加NormalizerProcessor处理动作的归一化与反归一化
• 修正render方法的参数缺失问题（需添加self）

修改后的完整代码 #

import numpy as np
import gym 
from gym import spaces
from tensorflow.keras.models import Sequential 
from tensorflow.keras.layers import Dense, Activation, Flatten
from tensorflow.keras.optimizers import Adam
from rl.agents.ddpg import DDPGAgent
from rl.memory import SequentialMemory  
from rl.processors import NormalizerProcessor

class Example3(gym.Env):
    def __init__(self):
        # 连续动作空间：[0,10] 和 [0,100]
        self.action_space = gym.spaces.Box(
            low=np.array([0, 0], dtype=np.float32),
            high=np.array([10, 100], dtype=np.float32),
            dtype=np.float32
        )
        # 连续观测空间
        lower_bound = np.array([-4, -1, 0], dtype=np.float32)
        upper_bound = np.array([2, 1, 10], dtype=np.float32)        
        self.observation_space = spaces.Box(lower_bound, upper_bound, dtype=np.float32)
    
        self.time = 100
        self.state = None

        # 报告数组初始化
        self.E1report = np.array([])
        self.actionreport = np.array([])
        self.E2report = np.array([])
        self.E3report = np.array([])
        self.Costreport = np.array([])

    def step(self, action):
        # 记录动作
        self.actionreport = np.append(self.actionreport, action)
        np.savetxt('ActionReport.txt', self.actionreport)

        E1, E2, E3 = self.state
        self.time -= 1

        # 计算theta与各指标，避免除以0
        theta = action[0] + action[1]
        E1 = (-4 * np.sin(theta)) / theta if theta != 0 else 0
        E2 = np.sin(theta)
        E3 = theta ** 2 / 10
        cost = np.sin(theta) + theta + theta ** 2 / 2 + 2022

        # 记录指标
        self.E1report = np.append(self.E1report, E1)
        np.savetxt('E1Report.txt', self.E1report)
        self.E2report = np.append(self.E2report, E2)
        np.savetxt('E2Report.txt', self.E2report)
        self.E3report = np.append(self.E3report, E3)
        np.savetxt('E3Report.txt', self.E3report)
        self.Costreport = np.append(self.Costreport, cost)
        np.savetxt('CostReport.txt', self.Costreport)

        # 更新状态
        self.state = (E1, E2, E3)

        # 计算奖励
        reward = 1 if (-cost < 2025) else 0
        # 判断是否结束
        done = self.time == 0
        info = {}        

        return np.array(self.state, dtype=np.float32), reward, done, info

    def reset(self):
        E1 = np.random.uniform(-4, 2)
        E2 = np.random.uniform(-1, 1)
        E3 = np.random.uniform(0, 10)
        self.state = (E1, E2, E3)
        self.time = 100
        return np.array(self.state, dtype=np.float32)

    def render(self, mode='human'):  # 添加self参数和默认mode
        pass 

# 初始化环境
env = Example3()
nb_actions = env.action_space.shape[0]

# 构建DDPG Actor模型（输出连续动作）
actor = Sequential()
actor.add(Flatten(input_shape=(1,) + env.observation_space.shape))
actor.add(Dense(16))
actor.add(Activation('relu'))
actor.add(Dense(16))
actor.add(Activation('relu'))
actor.add(Dense(nb_actions))
actor.add(Activation('linear'))  # 线性激活，由处理器映射到动作空间
print(actor.summary())

# 构建DDPG Critic模型（评估动作价值）
critic = Sequential()
critic.add(Flatten(input_shape=(1,) + env.observation_space.shape + (nb_actions,)))
critic.add(Dense(32))
critic.add(Activation('relu'))
critic.add(Dense(32))
critic.add(Activation('relu'))
critic.add(Dense(1))
critic.add(Activation('linear'))
print(critic.summary())

# 配置DDPGAgent
memory = SequentialMemory(limit=20000, window_length=1)
processor = NormalizerProcessor()  # 自动处理动作的归一化/反归一化

ddpg = DDPGAgent(
    actor=actor,
    critic=critic,
    nb_actions=nb_actions,
    memory=memory,
    nb_steps_warmup_critic=100,
    nb_steps_warmup_actor=100,
    target_model_update=1e-3,
    processor=processor
)
ddpg.compile(Adam(lr=1e-4, clipnorm=1.), metrics=['mae'])

# 开始训练
ddpg.fit(env, nb_steps=10000, visualize=False, verbose=1)

额外说明 #

• DDPG是Actor-Critic架构，需要分别定义Actor（输出连续动作）和Critic（评估动作价值）两个模型
• NormalizerProcessor会自动将模型输出映射到你的动作空间范围，无需手动处理
• 代码中添加了theta != 0的判断，避免出现除以0的错误
• 调整了学习率并添加梯度裁剪，提升训练稳定性

内容的提问来源于stack exchange，提问作者Hamed HSNY