Keras 允许您从全局范围获取任何张量。实际上,y_true 和 y_pred 甚至可能不会被使用,as here。
您的模型可以有多个输入(您可以在推理时将此输入设为虚拟,或将权重加载到具有单个输入的模型中)。请注意,您仍然需要它进行验证。
import keras
from keras.layers import *
from keras import backend as K
import numpy as np
inputs_x = Input(shape=(10,))
inputs_w = Input(shape=(10,))
y = Dense(10,kernel_initializer='glorot_uniform' )(inputs_x)
model = keras.Model(inputs=[inputs_x, inputs_w], outputs=[y])
def my_loss(y_true, y_pred):
return K.abs((y_true-y_pred)*inputs_w)
def my_metrics(y_true, y_pred):
# just to output something
return K.mean(inputs_w)
model.compile(optimizer='adam', loss=[my_loss], metrics=[my_metrics])
data = np.random.normal(size=(50000, 10))
labels = np.random.normal(size=(50000, 10))
weights = np.random.normal(size=(50000, 10))
model.fit([data, weights], labels, batch_size=256, validation_data=([data[:100], weights[:100]], labels[:100]), epochs=100)
要在没有权重的情况下进行验证,您需要编译另一个具有不同损失且不使用权重的模型版本。
UPD:还要注意,如果 Keras 返回数组而不是标量,Keras 将汇总您损失的所有元素
UPD:Tor tensorflow 2.1.0 似乎变得更加复杂。前进的方向是@marco-cerliani 指出的方向(标签、重量和数据被输入模型,并通过.add_loss() 添加自定义损失张量),但是他的解决方案对我来说开箱即用.第一件事是该模型不想使用无损失,拒绝同时接受输入和输出。所以,我引入了额外的虚拟损失函数。当数据集大小不能被批量大小整除时,就会出现第二个问题。在 keras 和 tf 1.x 中,最后一批问题通常由 steps_per_epoch 和 validation_steps 参数解决,但这里如果在第一批 Epoch 2 上开始失败。所以我需要制作简单的自定义数据生成器。
import tensorflow.keras as keras
from tensorflow.keras.layers import *
from tensorflow.keras import backend as K
import numpy as np
inputs_x = Input(shape=(10,))
inputs_w = Input(shape=(10,))
inputs_l = Input(shape=(10,))
y = Dense(10,kernel_initializer='glorot_uniform' )(inputs_x)
model = keras.Model(inputs=[inputs_x, inputs_w, inputs_l], outputs=[y])
def my_loss(y_true, y_pred):
return K.abs((y_true-y_pred)*inputs_w)
def my_metrics():
# just to output something
return K.mean(inputs_w)
def dummy_loss(y_true, y_pred):
return 0.
loss = my_loss(y, inputs_l)
metric = my_metrics()
model.add_loss(loss)
model.add_metric(metric, name='my_metric', aggregation='mean')
model.compile(optimizer='adam', loss=dummy_loss)
data = np.random.normal(size=(50000, 10))
labels = np.random.normal(size=(50000, 10))
weights = np.random.normal(size=(50000, 10))
dummy = np.zeros(shape=(50000, 10)) # or in can be labels, no matter now
# looks like it does not like when len(data) % batch_size != 0
# If I set steps_per_epoch, it fails on the second epoch.
# So, I proceded with data generator
class DataGenerator(keras.utils.Sequence):
'Generates data for Keras'
def __init__(self, x, w, y, y2, batch_size, shuffle=True):
'Initialization'
self.x = x
self.w = w
self.y = y
self.y2 = y2
self.indices = list(range(len(self.x)))
self.shuffle = shuffle
self.batch_size = batch_size
self.on_epoch_end()
def __len__(self):
'Denotes the number of batches per epoch'
return len(self.indices) // self.batch_size
def __getitem__(self, index):
'Generate one batch of data'
# Generate indexes of the batch
ids = self.indices[index*self.batch_size:(index+1)*self.batch_size]
# the last None to remove weird warning
# https://stackoverflow.com/questions/59317919
return [self.x[ids], self.w[ids], self.y[ids]], self.y2[ids], [None]
def on_epoch_end(self):
'Updates indexes after each epoch'
if self.shuffle == True:
np.random.shuffle(self.indices)
batch_size = 256
train_generator = DataGenerator(data,weights,labels, dummy, batch_size=batch_size, shuffle=True)
val_generator = DataGenerator(data[:2*batch_size],weights[:2*batch_size],labels[:2*batch_size], dummy[:2*batch_size], batch_size=batch_size, shuffle=True)
model.fit(x=train_generator, validation_data=val_generator,epochs=100)