强化学习形状错误：ValueError输入维度不匹配问题求助

错误信息 #

ValueError: Error when checking input: expected flatten_1_input to have 2 dimensions, but got array with shape (1, 1, 2)

环境版本 #

• Python 3.11.7
• TensorFlow: 2.13.0
• Keras: 2.13.1

问题原因 #

keras-rl的DQNAgent会根据设置的window_length自动为观测值增加一个维度，你设置了window_length=1，所以输入形状被扩展为(1,1,2)，但模型中Flatten层的输入形状直接使用observation_space.shape（即(2,)），两者维度不匹配导致报错。

解决方案 #

1. 调整模型输入形状

修改build_model函数，将输入形状调整为包含window_length的维度：

def build_model(input_shape, nb_actions):
    model = Sequential()
    # 适配window_length=1，输入形状改为 (1, 2)
    model.add(Flatten(input_shape=(1,) + input_shape))  
    model.add(Dense(32, activation='relu'))
    model.add(Dense(32, activation='relu'))
    model.add(Dense(nb_actions, activation='linear'))
    return model

2. 修正动作空间设置

当前action_space = gym.spaces.Discrete(1)只有1个动作，智能体无法做出有效决策，需根据任务调整：

• 如果是离散任务：增加动作数量，比如Discrete(2)
• 如果是连续预测任务：改用连续动作空间，并替换为支持连续动作的算法（如DDPG）

# 示例：改为2个离散动作
self.action_space = gym.spaces.Discrete(2)

3. 优化环境step逻辑

原代码中到达最后一行时复用当前状态，可能导致数据重复，建议改为：

def step(self, action):
    self.current_step += 1
    done = self.current_step >= len(self.df)
    if done:
        next_state = np.zeros_like(self.state)  # 用零向量作为结束状态
    else:
        next_state = self.df.iloc[self.current_step, 1:3].values
    # 可选：添加合理的奖励函数（原代码奖励为0无法驱动训练）
    reward = -np.abs(action - self.df.iloc[self.current_step, 0]) if not done else 0
    info = {}
    return next_state, reward, done, info

完整修改后代码 #

import numpy as np
import pandas as pd
import gym
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense, Flatten
from rl.agents import DQNAgent
from rl.policy import BoltzmannQPolicy
from rl.memory import SequentialMemory
from tensorflow.keras.optimizers.legacy import Adam

class CustomEnv(gym.Env):
    def __init__(self, df):
        super(CustomEnv, self).__init__()
        self.df = df
        # 调整为2个离散动作，根据任务需求修改
        self.action_space = gym.spaces.Discrete(2)  
        self.observation_space = gym.spaces.Box(low=-np.inf, high=np.inf, shape=(2,), dtype=np.float32)
        self.current_step = 0

    def reset(self):
        self.current_step = 0
        self.state = self.df.iloc[self.current_step, 1:3].values
        return self.state

    def step(self, action):
        self.current_step += 1
        done = self.current_step >= len(self.df)
        if done:
            next_state = np.zeros_like(self.state)
        else:
            next_state = self.df.iloc[self.current_step, 1:3].values
        # 基于预测误差设置奖励，驱动智能体学习
        reward = -np.abs(action - self.df.iloc[self.current_step, 0]) if not done else 0
        info = {}
        return next_state, reward, done, info

df = pd.DataFrame({
    'F_1_d_returns': [-0.038076, 0.083333, 0.060577, -0.013599, -0.020221],
    '1_d_returns': [-0.062030, -0.038076, 0.083333, 0.060577, -0.013599],
    '2_d_returns': [-0.133681, -0.097744, 0.042084, 0.148958, 0.046154]
})

env = CustomEnv(df)
states = env.observation_space.shape
actions = env.action_space.n

def build_model(input_shape, nb_actions):
    model = Sequential()
    # 适配window_length=1的输入维度
    model.add(Flatten(input_shape=(1,) + input_shape))
    model.add(Dense(32, activation='relu'))
    model.add(Dense(32, activation='relu'))
    model.add(Dense(nb_actions, activation='linear'))
    return model

model = build_model(states, actions)

def build_agent(model, actions):
    policy = BoltzmannQPolicy()
    memory = SequentialMemory(limit=50000, window_length=1)
    dqn = DQNAgent(model=model, memory=memory, policy=policy, 
                  nb_actions=actions, nb_steps_warmup=10, target_model_update=1e-2)
    return dqn

dqn = build_agent(model, actions)
dqn.compile(Adam(lr=1e-3), metrics=['mae'])
dqn.fit(env, nb_steps=50000, visualize=False, verbose=1)

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