共享特征提取器多头部神经网络独立训练TensorFlow报错求助
你遇到的核心问题是多个共享底层的Keras Model对象在同一个TensorFlow计算图中冲突:每个Model在compile时会生成对应输出的目标占位符(比如报错里的activation_1_target就是model1的输出目标张量)。当你切换到model2训练时,TensorFlow的计算图里仍然保留着model1的目标占位符,但你训练model2时并没有给这个占位符喂数据,所以触发了InvalidArgumentError。
方案一:使用单多输出模型,动态调整损失权重(推荐)
这个方案改动最小,完全利用Keras原生功能,完美适配你的强化学习场景(不同头部用不同批次训练):
构建包含所有输出的完整模型
不再单独创建model1/model2/model3,而是把所有头部输出整合到一个模型里:alphaLeaky=0.3 inputs = Input(shape=(state_shape[0],state_shape[1],state_shape[2])) # 共享特征提取器部分 x = ZeroPadding2D(padding=(1,1))(inputs) x = LocallyConnected2D(1, (6,6), activation='linear', padding='valid')(x) x = Flatten()(x) x = Dense(768,kernel_initializer='lecun_uniform',bias_initializer='zeros')(x) x = advanced_activations.LeakyReLU(alpha=alphaLeaky)(x) x = Dense(512,kernel_initializer='lecun_uniform',bias_initializer='zeros')(x) x = advanced_activations.LeakyReLU(alpha=alphaLeaky)(x) # 各个头部 head1 = Dense(256,kernel_initializer='lecun_uniform',bias_initializer='zeros')(x) head1 = advanced_activations.LeakyReLU(alpha=alphaLeaky)(head1) head1 = Dense(action_number,kernel_initializer='lecun_uniform',bias_initializer='zeros')(head1) head1 = Activation('linear')(head1) head2 = Dense(256,kernel_initializer='lecun_uniform',bias_initializer='zeros')(x) head2 = advanced_activations.LeakyReLU(alpha=alphaLeaky)(head2) head2 = Dense(action_number,kernel_initializer='lecun_uniform',bias_initializer='zeros')(head2) head2 = Activation('linear')(head2) head3 = Dense(256,kernel_initializer='lecun_uniform',bias_initializer='zeros')(x) head3 = advanced_activations.LeakyReLU(alpha=alphaLeaky)(head3) head3 = Dense(action_number,kernel_initializer='lecun_uniform',bias_initializer='zeros')(head3) head3 = Activation('linear')(head3) # 创建多输出模型 full_model = Model(inputs=inputs, outputs=[head1, head2, head3])训练时动态切换目标头部
编译模型时,通过loss_weights控制哪个头部的损失会被计算(只有权重非0的头部会贡献损失,进而更新权重),训练时只传入对应头部的目标数据,其他头部传None:# 训练头部1 full_model.compile( loss=['mse', 'mse', 'mse'], optimizer=Adamax(lr=PAS_INITIAL, beta_1=0.9, beta_2=0.999, epsilon=1e-08, decay=0.0), loss_weights=[1.0, 0.0, 0.0] # 只计算头部1的损失 ) full_model.fit( x=your_head1_input_data, y=[head1_targets, None, None], batch_size=your_batch_size, epochs=your_epochs ) # 切换训练头部2 full_model.compile( loss=['mse', 'mse', 'mse'], optimizer=Adamax(lr=PAS_INITIAL, beta_1=0.9, beta_2=0.999, epsilon=1e-08, decay=0.0), loss_weights=[0.0, 1.0, 0.0] # 只计算头部2的损失 ) full_model.fit( x=your_head2_input_data, y=[None, head2_targets, None], batch_size=your_batch_size, epochs=your_epochs )这个方案的优势:
- 共享特征提取器的权重会被所有头部的训练共同更新,符合你的需求
- 完全避免了多Model导致的占位符冲突
- 每个头部可以用独立的批次数据训练,完美适配强化学习场景
方案二:自定义训练循环(更灵活)
如果你需要更精细的控制,可以使用TensorFlow的GradientTape手动记录梯度,只更新当前头部和共享层的权重:
定义共享层和头部
先把共享特征提取器封装成一个独立的Model,方便后续获取可训练变量:# 共享特征提取器 inputs = Input(shape=(state_shape[0],state_shape[1],state_shape[2])) x = ZeroPadding2D(padding=(1,1))(inputs) x = LocallyConnected2D(1, (6,6), activation='linear', padding='valid')(x) x = Flatten()(x) x = Dense(768,kernel_initializer='lecun_uniform',bias_initializer='zeros')(x) x = advanced_activations.LeakyReLU(alpha=alphaLeaky)(x) x = Dense(512,kernel_initializer='lecun_uniform',bias_initializer='zeros')(x) x = advanced_activations.LeakyReLU(alpha=alphaLeaky)(x) shared_model = Model(inputs=inputs, outputs=x) # 头部1 head1_input = Input(shape=(512,)) # 对应shared_model的输出维度 head1 = Dense(256,kernel_initializer='lecun_uniform',bias_initializer='zeros')(head1_input) head1 = advanced_activations.LeakyReLU(alpha=alphaLeaky)(head1) head1 = Dense(action_number,kernel_initializer='lecun_uniform',bias_initializer='zeros')(head1) head1 = Activation('linear')(head1) head1_model = Model(inputs=head1_input, outputs=head1) # 头部2、3同理,重复上述代码创建head2_model、head3_model自定义训练函数
使用GradientTape记录梯度,只更新共享层和当前头部的权重:optimizer = Adamax(lr=PAS_INITIAL, beta_1=0.9, beta_2=0.999, epsilon=1e-08, decay=0.0) loss_fn = tf.keras.losses.MeanSquaredError() def train_head(shared_model, head_model, input_data, targets): with tf.GradientTape() as tape: # 先通过共享层提取特征 features = shared_model(input_data, training=True) # 再通过头部生成预测 predictions = head_model(features, training=True) # 计算损失 loss = loss_fn(targets, predictions) # 获取需要更新的权重:共享层 + 当前头部 trainable_vars = shared_model.trainable_variables + head_model.trainable_variables # 计算梯度并更新 gradients = tape.gradient(loss, trainable_vars) optimizer.apply_gradients(zip(gradients, trainable_vars)) return loss.numpy()调用训练函数
训练不同头部时,传入对应的模型和数据即可:# 训练头部1 loss = train_head(shared_model, head1_model, head1_input_data, head1_targets) # 训练头部2 loss = train_head(shared_model, head2_model, head2_input_data, head2_targets)这个方案的优势是完全手动控制训练流程,适合复杂的强化学习场景(比如需要自定义梯度裁剪、多步更新等)。
优先选择方案一,它代码改动小,利用Keras原生功能就能解决问题,并且完美适配你的需求。如果需要更灵活的训练控制,再考虑方案二。
内容的提问来源于stack exchange,提问作者Adakor

