使用Stable Baselines 3自定义Gymnasium环境时RL模型无法学习的问题

我正尝试用强化学习（RL）解决发电机调度问题：目标是在电价最高时段运行发电机，但燃料（允许运行时长）有限，且未来每小时电价已知。

训练时模型性能始终不优于随机策略，怀疑观测空间定义有问题。当前观测空间包含：

• 当前状态：发电机启停状态（与上一步动作一致，用于设置初始状态及后续添加状态切换惩罚）
• 当前小时数：每步递增1
• 剩余允许运行时长：发电机每运行1小时递减1
• 电价列表：单轮episode内保持不变

奖励机制：

• 若当前小时运行发电机，奖励为对应时段电价；否则奖励为0
• 若运行时长超出限制，奖励设为-1000并终止episode

已尝试PPO、A2C、RecurrentPPO等算法，调整了learning_rate和ent_coef，但模型无学习提升，TensorBoard曲线无明显改善。自定义Gymnasium环境代码如下：

def test():
    price_list = [-10, -10, -10, -10, 100, 100, 100, 0, 0, 0]
    env = OneEngine(allowed_h_to_run=6, price_list=price_list)
    env = DummyVecEnv([lambda: env])
    model = RecurrentPPO("MlpLstmPolicy",
                         env,
                         verbose=1,
                         tensorboard_log="./a2c_test_tensorboard/")
    model.learn(episodes)

class OneEngine(gym.Env):
    """
    Custom Environment that follows gym interface.
    One engine that can be on or off
    List of prices
    Nr of hours allowed to run
    """
    # Render in console
    metadata = {"render_modes": ["console"]}
    # Reward for when motor is run for more than the allowed hours
    OUT_OF_BOUNDS_REWARD = -1000.0
    # Define constants for clearer code
    class CurrentState(Enum):
        OFF = 0
        ON = 1

    def __init__(self, price_list=None, allowed_h_to_run=8,
                 render_mode="console"):
        super(OneEngine, self).__init__()
        if price_list is None:
            price_list = [10, 80, 30, 20, 140, 152, 188, 30, 50, 130]
        self.render_mode = render_mode
        # Save initial state for reset
        self.initial_hours_to_run = allowed_h_to_run
        self.price_list = price_list.copy()
        # Position in price list
        self.current_hour = 0
        self.price_list_len = len(price_list)
        smallest_price = min(price_list)
        largest_price = max(price_list)
        # How much hours of gas does the engine have
        self.allowed_h_to_run = allowed_h_to_run
        self.current_state = self.CurrentState.OFF
        # 2 actions - engine OFF, engine ON
        n_actions = 2
        self.action_space = spaces.Discrete(n_actions)
        # The observation is:
        # Current state - motor on or off
        # Current hour
        # Hours left to run on  - how many hours(steps is the motor allowed to run)
        # Price list
        self.observation_space = spaces.Box(
            low=np.array([self.CurrentState.OFF.value, 0, 0] + [smallest_price] * self.price_list_len),
            high=np.array([self.CurrentState.ON.value, self.price_list_len + 1, allowed_h_to_run + 1] +
                          [largest_price] * self.price_list_len),
            shape=(3 + self.price_list_len,),
            dtype=np.float32)

    def reset(self, seed=None, options=None):
        """
        :return: (np.array)
        """
        super().reset(seed=seed, options=options)
        # Initialize the agent with off state, initial hours to run and initial price list
        self.current_state = self.CurrentState.OFF
        self.allowed_h_to_run = self.initial_hours_to_run
        self.current_hour = 0
        initial_observation = np.concatenate((np.array([self.current_state.value]).astype(np.float32),
                                              np.array([self.current_hour]).astype(np.float32),
                                              np.array([self.initial_hours_to_run]).astype(np.float32),
                                              np.array(self.price_list).astype(np.float32)))
        return initial_observation, {}

    def step(self, action):
        if action == self.CurrentState.ON.value:
            self.allowed_h_to_run -= 1
            self.current_state = self.CurrentState.ON
        elif action == self.CurrentState.OFF.value:
            self.current_state = self.CurrentState.OFF
        else:
            raise ValueError(
                f"Received invalid action={action} which is not part of the action space"
            )
        info = {}
        terminated = False
        # Reward depending on if engine was on
        reward = float(self.price_list[self.current_hour] * action)
        if self.allowed_h_to_run < 0:
            # Too many hours used , not allowed
            reward = self.OUT_OF_BOUNDS_REWARD
            terminated = True
        # Increment current hour
        self.current_hour += 1
        if self.current_hour >= self.price_list_len:
            # we reached end of price list
            terminated = True
        truncated = False  # we do not limit the number of steps here

        observation = np.concatenate((np.array([self.current_state.value]).astype(np.float32),
                                      np.array([self.current_hour]).astype(np.float32),
                                      np.array([self.allowed_h_to_run]).astype(np.float32),
                                      np.array(self.price_list).astype(np.float32)))
        return (
            observation,
            reward,
            terminated,
            truncated,
            info,
        )

    def render(self):
        if self.render_mode == "console":
            if self.current_hour > 0:
                print(f"Hour nr.: {self.current_hour} \tPrice:{self.price_list[self.current_hour - 1]}\t\t\t{self.CurrentState(self.current_state).name}")
                print(f"Hours allowed to run {self.allowed_h_to_run}")
            else:
                # TODO:
                print("Initial state...")
            if self.allowed_h_to_run < 0:
                print(f"Attempted to use more hours than allowed. Penalty: {self.OUT_OF_BOUNDS_REWARD}")

    def close(self):
        pass


if __name__ == "__main__":
    test()

排查思路与改进建议 #

观测空间优化 #

• 移除冗余的完整电价列表：当前观测传入整个电价列表，实际只需要当前及未来时段的电价，完整列表会引入大量冗余信息，增加模型学习难度。可改为传入self.price_list[self.current_hour:]，若要固定维度，可用0填充已过去的时段。
• 标准化观测值：观测里的小时数、剩余运行时长、电价数值范围差异大，建议对每个维度做标准化（比如小时数除以总时长，剩余运行时长除以初始允许时长，电价做归一化），帮助模型更快收敛。
• 修正观测空间边界：当前观测空间的high值存在不合理之处：current_hour最大为price_list_len - 1，所以high应设为price_list_len - 1；剩余运行时长的high应为初始允许时长，避免边界定义错误干扰模型判断。

奖励机制调整 #

• 优化超量惩罚逻辑：当前只要剩余运行时长<0就直接给-1000并终止，可提前给出惩罚信号，比如剩余时长不足时，运行的奖励扣减部分数值，或给小负奖励，而非直接触发极端惩罚终止，给模型留调整空间。
• 添加状态切换惩罚（可选）：如果发电机启停有实际成本，可给切换动作（OFF→ON或ON→OFF）添加小负奖励，贴合实际场景的同时引导模型减少不必要的启停。
• 奖励缩放：当前奖励范围从-1000到100，差异过大，建议对奖励做缩放（比如除以最大电价绝对值），让奖励数值更平稳，避免极端值干扰模型学习。

算法与训练设置优化 #

• 优先使用普通PPO：当前是有限时域的确定性任务（电价已知），普通PPO配合MlpPolicy足够，RecurrentPPO适合序列依赖极强的场景，反而会增加复杂度。
• 扩大超参数调整范围：尝试更大的ent_coef（0.1-0.5）提升探索性，或更小的learning_rate（1e-4到1e-5）；同时增加训练episode数量（至少几千轮），确保模型有足够学习时间。
• 添加基线对比：手动实现最优策略的奖励值，对比随机策略和模型的奖励，确认模型是否有接近最优的可能，排除任务本身的问题。

环境逻辑验证 #

• 检查step函数奖励计算：当前奖励为self.price_list[self.current_hour] * action，current_hour在执行动作后才递增，逻辑正确，但可打印几轮episode的动作、奖励、状态，确认环境行为符合预期。
• 验证重置函数：确认reset后current_hour、allowed_h_to_run、current_state都正确初始化，避免状态泄漏影响训练。

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