【Keras】Keras使用进阶

def GenModel(img_width = 512, img_height = 512 , model_name = 'AlexNet'):
    # 指定输入图像大小
    image_input_shape = (img_width, img_height, 3)
    # 定义网络（AlexNet）
    if model_name == 'AlexNet':
        print('\n--------Start build model ', model_name, '--------\n')
        
        # 定义网络输入
        image_input = Input(shape=image_input_shape, name='image_input')

        # 定义主卷积（图像）网络卷积->激活->池化
        conv_image = Conv2D(96, (11, 11), strides = (4, 4), padding = 'valid', activation = 'relu')(image_input)
        conv_image = MaxPooling2D(pool_size = (3, 3), strides = (2, 2))(conv_image)

        conv_image = Conv2D(256, (5, 5), strides = (1, 1), padding = 'same', activation = 'relu')(conv_image)
        conv_image = MaxPooling2D(pool_size = (3, 3), strides = (2, 2))(conv_image)

        conv_image = Conv2D(384, (3,3), strides = (1, 1), padding = 'same', activation = 'relu')(conv_image)
        
        conv_image = Conv2D(384, (3,3), strides = (1, 1), padding = 'same', activation = 'relu')(conv_image)

        conv_image = Conv2D(384, (3,3), strides = (1, 1), padding = 'same', activation = 'relu')(conv_image)
        
        conv_image = MaxPooling2D(pool_size = (3, 3), strides = (2, 2))(conv_image)

        conv_image = Flatten()(conv_image)

        # 定义2个输出，分别用于预测衣服的颜色和类型
        out_color = Dense(4096, activation='relu',)(conv_image)        
        out_color = Dense(512, activation='relu',)(out_color)
        out_color = Dense(3, activation='sigmoid', name='out_age')(out_color) 
                                                           
        out_type = Dense(4096, activation='relu',)(conv_image)
        out_type = Dense(512, activation='relu',)(out_sex)
        out_type = Dense(3, activation='sigmoid', name='out_sex')(out_sex)

        # 定义整个单输入，2输出的模型
        model=Model(inputs = image_input, outputs = [out_color, out_type]) 
        return model

# 编译模型，定义损失函数
    opt=Adadelta()
    # opt=SGD()
    print('\n-------- optimizer: %s --------\n'%(opt.__class__.__name__) )
    model.compile(  optimizer = opt, 
                    loss = {'out_color': 'categorical_crossentropy', 'out_type': 'categorical_crossentropy'}, 
                    loss_weights = {'out_color': out_color_weight, 'out_type': out_type_weight, metrics = ['accuracy']) # 这里loss_weights需要自己手动设定下

# grab the image paths and randomly shuffle them
print("[INFO] loading images...")
imagePaths = sorted(list(paths.list_images(args["dataset"])))
random.seed(42)
random.shuffle(imagePaths)

# initialize the data and labels
data = []
labels = []

# loop over the input images
for imagePath in imagePaths:
    # load the image, pre-process it, and store it in the data list
    image = cv2.imread(imagePath)
    image = cv2.resize(image, (IMAGE_DIMS[1], IMAGE_DIMS[0]))
    image = img_to_array(image)
    data.append(image)

    # extract set of class labels from the image path and update the
    # labels list
    l = label = imagePath.split(os.path.sep)[-2].split("_")
    labels.append(l)

# scale the raw pixel intensities to the range [0, 1]
data = np.array(data, dtype="float") / 255.0
labels = np.array(labels)
print("[INFO] data matrix: {} images ({:.2f}MB)".format(
    len(imagePaths), data.nbytes / (1024 * 1000.0)))
print(labels)
# binarize the labels using scikit-learn's special multi-label
# binarizer implementation
print("[INFO] class labels:")
mlb = MultiLabelBinarizer()
labels = mlb.fit_transform(labels)
print(labels)
# loop over each of the possible class labels and show them
for (i, label) in enumerate(mlb.classes_):
    print("{}. {}".format(i + 1, label))

# partition the data into training and testing splits using 80% of
# the data for training and the remaining 20% for testing
(trainX, testX, trainY, testY) = train_test_split(data,
    labels, test_size=0.2, random_state=42)

# construct the image generator for data augmentation
aug = ImageDataGenerator(rotation_range=25, width_shift_range=0.1,
    height_shift_range=0.1, shear_range=0.2, zoom_range=0.2,
    horizontal_flip=True, fill_mode="nearest")

# train the network
print("[INFO] training network...")
H = model.fit_generator(
    aug.flow(trainX, trainY, batch_size=BS),
    validation_data=(testX, testY),
    steps_per_epoch=len(trainX) // BS,
    epochs=EPOCHS, verbose=1)

from keras.optimizers import RMSprop, Adam, Adadelta

opt=Adam()
opt_name=opt.__class__.__name__
print(opt_name)

checkpoint = ModelCheckpoint(filepath='./best_model.weights',
                             monitor='val_acc',
                             verbose=1,
                             save_best_only=True)

model.fit(x_train,
          y_train,
          epochs=10,
          validation_data=(x_test, y_test),
          callbacks=[checkpoint])

# checkpoint
# https://keras.io/zh/callbacks/
# 如果验证损失下降， 那么在每个训练轮之后保存模型
ArgName=',epo='+str(epoch)+',bsize='+str(batch_size)+',lr='+str(LearningRate)+',DropRate='+str(DropoutRate)
FileNamePy=os.path.basename(__file__).split('.')[-2]
checkpoint_filepath = FileNamePy+ArgName


checkpointer_val_best = ModelCheckpoint(filepath=checkpoint_filepath, monitor='val_acc',
    verbose=1, save_best_only=True, mode='max', save_weights_only=True)

callbacks_list = [checkpointer_val_best]

hist=model.fit_generator(
    train_generator,
    steps_per_epoch=nb_train_samples,
    epochs=epoch,
    validation_data=validation_generator,
    validation_steps=nb_validation_samples,
    callbacks = callbacks_list)

from keras.callbacks import Callback

def evaluate(): # 评测函数
    pred = model.predict(x_test)
    return np.mean(pred.argmax(axis=1) == y_test) # 爱算啥就算啥


# 定义Callback器，计算验证集的acc，并保存最优模型
class Evaluate(Callback):

    def __init__(self):
        self.accs = []
        self.highest = 0.

    def on_epoch_end(self, epoch, logs=None):
        acc = evaluate()
        self.accs.append(acc)
        if acc >= self.highest: # 保存最优模型权重
            self.highest = acc
            model.save_weights('best_model.weights')

        # 爱运行什么就运行什么
        print 'acc: %s, highest: %s' % (acc, self.highest)


evaluator = Evaluate()
model.fit(x_train,
          y_train,
          epochs=10,
          callbacks=[evaluator])

from keras.callbacks import LearningRateScheduler

def lr_schedule(epoch):
    # 根据epoch返回不同的学习率
    if epoch < 50:
        lr = 1e-2
    elif epoch < 80:
        lr = 1e-3
    else:
        lr = 1e-4
    return lr

lr_scheduler = LearningRateScheduler(lr_schedule)

model.fit(x_train,
          y_train,
          epochs=10,
          callbacks=[evaluator, lr_scheduler])

# add a x -> x^2 layer
model.add(Lambda(lambda x: x ** 2))

# add a layer that returns the concatenation
# of the positive part of the input and
# the opposite of the negative part

def antirectifier(x):
    x -= K.mean(x, axis=1, keepdims=True)
    x = K.l2_normalize(x, axis=1)
    pos = K.relu(x)
    neg = K.relu(-x)
    return K.concatenate([pos, neg], axis=1)

def antirectifier_output_shape(input_shape):
    shape = list(input_shape)
    assert len(shape) == 2  # only valid for 2D tensors
    shape[-1] *= 2
    return tuple(shape)

model.add(Lambda(antirectifier,
         output_shape=antirectifier_output_shape))

# 对于简单的定制操作，可以通过使用layers.core.Lambda层来完成。
# 该方法的适用情况：仅对流经该层的数据做个变换，而这个变换本身没有需要学习的参数

# 切片后再分别进行embedding和average pooling
import numpy as np  
from keras.models import Sequential  
from keras.layers import Dense, Activation,Reshape  
from keras.layers import merge  
from keras.utils import plot_model
from keras.layers import *
from keras.models import Model  

def get_slice(x, index):
    return x[:, index]

keep_num = 3 
field_lens = 90
input_field = Input(shape=(keep_num, field_lens))
avg_pools = []
for n in range(keep_num):
    block = Lambda(get_slice,output_shape=(1,field_lens),arguments={'index':n})(input_field)
    x_emb = Embedding(input_dim=100, output_dim=200, input_length=field_lens)(block)
    x_avg = GlobalAveragePooling1D()(x_emb)
    avg_pools.append(x_avg)  
output = concatenate([p for p in avg_pools])
model = Model(input_field, output) 
plot_model(model, to_file='model/lambda.png',show_shapes=True)  

plt.figure(figsize=(21, 12))
im = plt.imread('model/lambda.png')
plt.imshow(im)

# 对于具有可训练权重的定制层，需要自己来实现。 

from keras import backend as K
from keras.engine.topology import Layer
import numpy as np

class MyLayer(Layer):

    def __init__(self, output_dim, **kwargs):
        self.output_dim = output_dim
        super(MyLayer, self).__init__(**kwargs)

    def build(self, input_shape):
        # Create a trainable weight variable for this layer.
        self.kernel = self.add_weight(name='kernel', 
                                      shape=(input_shape[1], self.output_dim),
                                      initializer='uniform',
                                      trainable=True)
        super(MyLayer, self).build(input_shape)  # Be sure to call this somewhere!

    def call(self, x):
        return K.dot(x, self.kernel)

    def compute_output_shape(self, input_shape):
        return (input_shape[0], self.output_dim)