【深度学习】——深入Keras：从基础到实战的深度学习指南第11章共12章

文章目录 第 11 章：实战案例11.1 图像分类实战项目背景与目标选择合适的数据集（以 Oxford Flowers 102 为例）数据预处理、模型构建、训练与评估的完整流程模型优化与调优的实践 11.2 目标检测实战目标检测任务的介绍与常用算法原理使用 Keras 构建基于卷积神经网络的目标检测模型（简单的 YOLO - like 模型示例）数据集标注与处理模型训练、评估与应用 11.3 自然语言处理实战文本分类案例（以新闻分类为例）数据收集与预处理构建循环神经网络或卷积神经网络进行文本分类模型训练与性能优化 序列到序列模型（如机器翻译、文本摘要）原理与架构在 Keras 中实现序列到序列模型的方法 11.4 时间序列预测实战时间序列数据的特点与分析方法使用循环神经网络（LSTM、GRU）或 Transformer 进行时间序列预测以 LSTM 为例预测股票价格以 Transformer 为例预测电力负荷

---

第 11 章：实战案例 11.1 图像分类实战 项目背景与目标

在图像分类领域，我们希望构建一个能够准确识别图像所属类别的模型。本次实战选择经典的图像分类任务，目标是利用给定的图像数据集训练出一个性能良好的图像分类模型，实现对不同类别的图像进行准确分类。

选择合适的数据集（以 Oxford Flowers 102 为例）

Oxford Flowers 102 数据集包含 102 种不同种类的花卉图像，共 8189 张图像，分为训练集、验证集和测试集。

数据预处理、模型构建、训练与评估的完整流程 import numpy as np import os from keras.preprocessing.image import ImageDataGenerator, load_img, img_to_array from keras.models import Sequential from keras.layers import Conv2D, MaxPooling2D, Flatten, Dense from keras.utils import to_categorical import matplotlib.pyplot as plt # 数据路径 data_dir = 'oxford_flowers_102' train_dir = os.path.join(data_dir, 'train') val_dir = os.path.join(data_dir, 'val') test_dir = os.path.join(data_dir, 'test') # 图像尺寸和批量大小 img_width, img_height = 224, 224 batch_size = 32 num_classes = 102 # 数据预处理 train_datagen = ImageDataGenerator( rescale=1. / 255, rotation_range=40, width_shift_range=0.2, height_shift_range=0.2, shear_range=0.2, zoom_range=0.2, horizontal_flip=True, fill_mode='nearest' ) val_datagen = ImageDataGenerator(rescale=1. / 255) train_generator = train_datagen.flow_from_directory( train_dir, target_size=(img_width, img_height), batch_size=batch_size, class_mode='categorical' ) val_generator = val_datagen.flow_from_directory( val_dir, target_size=(img_width, img_height), batch_size=batch_size, class_mode='categorical' ) # 构建模型 model = Sequential() model.add(Conv2D(32, (3, 3), activation='relu', input_shape=(img_width, img_height, 3))) model.add(MaxPooling2D((2, 2))) model.add(Conv2D(64, (3, 3), activation='relu')) model.add(MaxPooling2D((2, 2))) model.add(Conv2D(128, (3, 3), activation='relu')) model.add(MaxPooling2D((2, 2))) model.add(Flatten()) model.add(Dense(128, activation='relu')) model.add(Dense(num_classes, activation='softmax')) # 编译模型 model pile(loss='categorical_crossentropy', optimizer='adam', metrics=['accuracy']) # 训练模型 history = model.fit_generator( train_generator, steps_per_epoch=train_generator.samples // batch_size, epochs=10, validation_data=val_generator, validation_steps=val_generator.samples // batch_size ) # 评估模型 test_datagen = ImageDataGenerator(rescale=1. / 255) test_generator = test_datagen.flow_from_directory( test_dir, target_size=(img_width, img_height), batch_size=batch_size, class_mode='categorical' ) test_loss, test_acc = model.evaluate_generator(test_generator, steps=test_generator.samples // batch_size) print(f"测试准确率: {test_acc}") # 绘制训练和验证的准确率和损失曲线 plt.plot(history.history['accuracy']) plt.plot(history.history['val_accuracy']) plt.title('模型准确率') plt.ylabel('准确率') plt.xlabel('Epoch') plt.legend(['训练', '验证'], loc='upper left') plt.show() plt.plot(history.history['loss']) plt.plot(history.history['val_loss']) plt.title('模型损失') plt.ylabel('损失') plt.xlabel('Epoch') plt.legend(['训练', '验证'], loc='upper left') plt.show() 模型优化与调优的实践 调整超参数：可以尝试调整学习率、隐藏层节点数、训练轮数等超参数。例如，将学习率从默认的 0.001 调整为 0.0001 或 0.01，观察模型性能的变化。增加数据增强：进一步丰富数据增强的方式，如添加亮度调整、对比度调整等操作，增加训练数据的多样性。使用预训练模型：可以使用如 VGG16、ResNet50 等预训练模型进行迁移学习，将预训练模型的权重迁移到当前任务中，提高模型的性能。 11.2 目标检测实战 目标检测任务的介绍与常用算法原理

目标检测任务旨在识别图像中存在的目标物体，并确定它们的位置（通常用边界框表示）和类别。常用的算法如 YOLO（You Only Look Once）系列算法，其原理是将图像划分为网格，每个网格预测多个边界框及其对应的类别概率，通过一次前向传播就能得到所有目标的检测结果，具有速度快的特点。

使用 Keras 构建基于卷积神经网络的目标检测模型（简单的 YOLO - like 模型示例） import numpy as np from keras.models import Sequential from keras.layers import Conv2D, MaxPooling2D, Flatten, Dense # 假设每个网格预测 2 个边界框，每个边界框有 5 个参数（x, y, w, h, confidence） # 加上 20 个类别概率（以 COCO 数据集为例，这里简化为 20 类） # 假设图像尺寸为 416x416，网格数量为 13x13 grid_size = 13 num_boxes = 2 num_classes = 20 output_size = grid_size * grid_size * (num_boxes * (4 + 1) + num_classes) model = Sequential() model.add(Conv2D(32, (3, 3), activation='relu', input_shape=(416, 416, 3))) model.add(MaxPooling2D((2, 2))) model.add(Conv2D(64, (3, 3), activation='relu')) model.add(MaxPooling2D((2, 2))) # 更多卷积层和池化层... model.add(Flatten()) model.add(Dense(512, activation='relu')) model.add(Dense(output_size, activation='sigmoid')) # 编译模型（这里简单使用均方误差损失，实际应用中需更复杂的损失函数） model pile(loss='mse', optimizer='adam', metrics=['accuracy']) 数据集标注与处理

目标检测数据集需要标注每个目标物体的边界框（通常用左上角和右下角坐标表示）和类别。可以使用工具如 LabelImg 进行标注。标注后的数据需要进行预处理，包括将图像调整为模型输入的尺寸，将标注信息转换为模型可接受的格式（如将边界框坐标归一化到 0 - 1 范围）。

# 假设已经有标注好的数据集，包含图像和对应的标注信息 # 这里简单模拟数据加载和预处理 import cv2 # 加载图像和标注信息 def load_data(image_path, label_path): image = cv2.imread(image_path) image = cv2.resize(image, (416, 416)) image = image / 255.0 # 读取标注信息并转换为模型可接受的格式 with open(label_path, 'r') as f: lines = f.readlines() labels = [] for line in lines: parts = line.strip().split() class_id = int(parts[0]) x_center = float(parts[1]) y_center = float(parts[2]) width = float(parts[3]) height = float(parts[4]) label = [x_center, y_center, width, height, class_id] labels.append(label) # 这里假设将所有标注信息整理为一个数组，根据实际情况调整 label_array = np.array(labels) return image, label_array # 数据生成器 def data_generator(image_paths, label_paths, batch_size): while True: batch_images = [] batch_labels = [] for _ in range(batch_size): index = np.random.randint(0, len(image_paths)) image, label = load_data(image_paths[index], label_paths[index]) batch_images.append(image) batch_labels.append(label) batch_images = np.array(batch_images) batch_labels = np.array(batch_labels) yield batch_images, batch_labels 模型训练、评估与应用 # 假设已经准备好训练集、验证集的图像路径和标注路径 train_image_paths = [] # 训练集图像路径列表 train_label_paths = [] # 训练集标注路径列表 val_image_paths = [] # 验证集图像路径列表 val_label_paths = [] # 验证集标注路径列表 # 训练模型 batch_size = 16 train_generator = data_generator(train_image_paths, train_label_paths, batch_size) val_generator = data_generator(val_image_paths, val_label_paths, batch_size) history = model.fit_generator( train_generator, steps_per_epoch=len(train_image_paths) // batch_size, epochs=10, validation_data=val_generator, validation_steps=len(val_image_paths) // batch_size ) # 评估模型 # 这里简单模拟评估，实际应用中需更复杂的评估指标如 mAP（平均精度均值） test_image_paths = [] # 测试集图像路径列表 test_label_paths = [] # 测试集标注路径列表 test_generator = data_generator(test_image_paths, test_label_paths, batch_size) test_loss, test_acc = model.evaluate_generator(test_generator, steps=len(test_image_paths) // batch_size) print(f"测试准确率: {test_acc}") # 应用模型进行预测 # 假设输入一张图像进行预测 test_image_path = 'test_image.jpg' test_image, _ = load_data(test_image_path, '') # 这里不需要标注信息 test_image = np.expand_dims(test_image, axis=0) prediction = model.predict(test_image) # 对预测结果进行解码，得到边界框和类别信息（这里省略具体解码过程） 11.3 自然语言处理实战 文本分类案例（以新闻分类为例） 数据收集与预处理

可以从新闻网站、新闻 API 等渠道收集新闻文本数据，并标注它们的类别（如政治、经济、娱乐等）。预处理步骤包括：

文本清洗：去除特殊字符、标点符号、数字等。分词：使用工具如 nltk 或 jieba 对文本进行分词。去除停用词：移除常见的无意义词汇，如 “的”、“了”、“在” 等。词干提取或词形还原：将单词还原为基本形式，如 “running” 还原为 “run”。文本向量化：使用 Keras 的 Tokenizer 将文本转换为数值序列。 import nltk from nltk.corpus import stopwords from nltk.stem import WordNetLemmatizer import string from keras.preprocessing.text import Tokenizer from keras.preprocessing.sequence import pad_sequences import numpy as np # 假设已经收集好新闻文本数据和对应的类别标签 news_texts = [] # 新闻文本列表 news_labels = [] # 新闻类别标签列表 # 文本清洗和预处理 def preprocess_text(text): text = text.lower() text = text.translate(str.maketrans('', '', string.punctuation)) tokens = nltk.word_tokenize(text) stop_words = set(stopwords.words('english')) tokens = [token for token in tokens if token not in stop_words] lemmatizer = WordNetLemmatizer() tokens = [lemmatizer.lemmatize(token) for token in tokens] return " ".join(tokens) preprocessed_texts = [preprocess_text(text) for text in news_texts] # 文本向量化 tokenizer = Tokenizer(num_words=10000) tokenizer.fit_on_texts(preprocessed_texts) sequences = tokenizer.texts_to_sequences(preprocessed_texts) padded_sequences = pad_sequences(sequences, maxlen=200) # 标签编码 news_labels = np.array(news_labels) from keras.utils import to_categorical encoded_labels = to_categorical(news_labels) 构建循环神经网络或卷积神经网络进行文本分类 from keras.models import Sequential from keras.layers import Embedding, LSTM, Dense, Conv1D, GlobalMaxPooling1D # 构建 LSTM 模型 model_lstm = Sequential() model_lstm.add(Embedding(input_dim=10000, output_dim=128, input_length=200)) model_lstm.add(LSTM(128)) model_lstm.add(Dense(10, activation='softmax')) # 假设 10 个新闻类别 # 构建 CNN 模型 model_cnn = Sequential() model_cnn.add(Embedding(input_dim=10000, output_dim=128, input_length=200)) model_cnn.add(Conv1D(128, 5, activation='relu')) model_cnn.add(GlobalMaxPooling1D()) model_cnn.add(Dense(10, activation='softmax')) # 编译模型 model_lstm pile(loss='categorical_crossentropy', optimizer='adam', metrics=['accuracy']) model_cnn pile(loss='categorical_crossentropy', optimizer='adam', metrics=['accuracy']) 模型训练与性能优化 # 划分训练集和验证集 from sklearn.model_selection import train_test_split x_train, x_val, y_train, y_val = train_test_split(padded_sequences, encoded_labels, test_size=0.2, random_state=42) # 训练 LSTM 模型 history_lstm = model_lstm.fit(x_train, y_train, epochs=10, batch_size=32, validation_data=(x_val, y_val)) # 训练 CNN 模型 history_cnn = model_cnn.fit(x_train, y_train, epochs=10, batch_size=32, validation_data=(x_val, y_val)) # 性能评估 from sklearn.metrics import accuracy_score # LSTM 模型评估 y_pred_lstm = model_lstm.predict(x_val) y_pred_lstm = np.argmax(y_pred_lstm, axis=1) y_true_lstm = np.argmax(y_val, axis=1) lstm_accuracy = accuracy_score(y_true_lstm, y_pred_lstm) print(f"LSTM 模型验证准确率: {lstm_accuracy}") # CNN 模型评估 y_pred_cnn = model_cnn.predict(x_val) y_pred_cnn = np.argmax(y_pred_cnn, axis=1) y_true_cnn = np.argmax(y_val, axis=1) cnn_accuracy = accuracy_score(y_true_cnn, y_pred_cnn) print(f"CNN 模型验证准确率: {cnn_accuracy}") 序列到序列模型（如机器翻译、文本摘要） 原理与架构

序列到序列模型（Seq2Seq）通常由编码器和解码器组成。编码器将输入序列（如源语言句子）编码为一个固定长度的向量表示，解码器则将这个向量表示解码为目标序列（如目标语言句子）。在训练过程中，通过最大化目标序列的概率来调整模型参数。常见的架构中会使用循环神经网络（如 LSTM、GRU）或 Transformer 来实现编码器和解码器。

在 Keras 中实现序列到序列模型的方法 from keras.models import Model from keras.layers import Input, LSTM, Dense import numpy as np # 假设已经准备好英语和法语的句子对 english_sentences = [] # 英语句子列表 french_sentences = [] # 法语句子列表 # 最大长度 max_english_len = max([len(sentence.split()) for sentence in english_sentences]) max_french_len = max([len(sentence.split()) for sentence in french_sentences]) # 词汇表大小 english_vocab_size = len(set(word for sentence in english_sentences for word in sentence.split())) french_vocab_size = len(set(word for sentence in french_sentences for word in sentence.split())) # 嵌入维度 embedding_dim = 256 latent_dim = 512 # 输入层 encoder_inputs = Input(shape=(max_english_len,)) encoder_embedding = Embedding(input_dim=english_vocab_size, output_dim=embedding_dim)(encoder_inputs) encoder_lstm = LSTM(latent_dim, return_state=True) encoder_outputs, state_h, state_c = encoder_lstm(encoder_embedding) encoder_states = [state_h, state_c] # 解码器输入 decoder_inputs = Input(shape=(max_french_len,)) decoder_embedding = Embedding(input_dim=french_vocab_size, output_dim=embedding_dim)(decoder_inputs) decoder_lstm = LSTM(latent_dim, return_sequences=True, return_state=True) decoder_outputs, _, _ = decoder_lstm(decoder_embedding, initial_state=encoder_states) # 输出层 decoder_dense = Dense(french_vocab_size, activation='softmax') decoder_outputs = decoder_dense(decoder_outputs) # 构建模型 model = Model([encoder_inputs, decoder_inputs], decoder_outputs) # 编译模型 model pile(optimizer='adam', loss='categorical_crossentropy') # 数据预处理 tokenizer_english = Tokenizer(num_words=english_vocab_size) tokenizer_english.fit_on_texts(english_sentences) input_sequences = tokenizer_english.texts_to_sequences(english_sentences) padded_input_sequences = pad_sequences(input_sequences, maxlen=max_english_len) tokenizer_french = Tokenizer(num_words=french_vocab_size) tokenizer_french.fit_on_texts(french_sentences) target_sequences = tokenizer_french.texts_to_sequences(french_sentences) padded_target_sequences = pad_sequences(target_sequences, maxlen=max_french_len) from keras.utils import to_categorical one_hot_targets = to_categorical(padded_target_sequences, num_classes=french_vocab_size) # 训练模型 model.fit([padded_input_sequences, padded_target_sequences], one_hot_targets, epochs=10, batch_size=32) # 构建推理模型 encoder_model = Model(encoder_inputs, encoder_states) decoder_state_input_h = Input(shape=(latent_dim,)) decoder_state_input_c = Input(shape=(latent_dim,)) decoder_states_inputs = [decoder_state_input_h, decoder_state_input_c] decoder_embedding2 = Embedding(input_dim=french_vocab_size, output_dim=embedding_dim)(decoder_inputs) decoder_outputs2, state_h2, state_c2 = decoder_lstm(decoder_embedding2, initial_state=decoder_states_inputs) decoder_states2 = [state_h2, state_c2] decoder_outputs2 = decoder_dense(decoder_outputs2) decoder_model = Model( [decoder_inputs] + decoder_states_inputs, [decoder_outputs2] + decoder_states2 ) # 翻译函数 def translate(sentence): input_seq = tokenizer_english.texts_to_sequences([sentence]) padded_input_seq = pad_sequences(input_seq, maxlen=max_english_len) states_value = encoder_model.predict(padded_input_seq) target_seq = np.zeros((1, max_french_len)) target_seq[0, 0] = tokenizer_french.word_index['<start>'] stop_condition = False decoded_sentence = '' while not stop_condition: output_tokens, h, c = decoder_model.predict([target_seq] + states_value) sampled_token_index = np.argmax(output_tokens[0, -1, :]) sampled_word = None for word, index in tokenizer_french.word_index.items(): if index == sampled_token_index: sampled_word = word break if sampled_word!= '<end>': decoded_sentence += sampled_word + ' ' target_seq[0, -1] = sampled_token_index else: stop_condition = True states_value = [h, c] return decoded_sentence.strip() # 测试翻译 test_sentence = "I love you" translation = translate(test_sentence) print(f"翻译结果: {translation}") 11.4 时间序列预测实战 时间序列数据的特点与分析方法

时间序列数据是按时间顺序排列的观测值序列，具有趋势性（长期上升或下降趋势）、季节性（周期性重复出现的模式）、周期性（不一定是季节性的固定周期波动）和随机性（不可预测的噪声部分）等特点。分析方法包括：

可视化：绘制时间序列图，直观观察趋势、季节性等特征，使用工具如 matplotlib。平稳性检验：如使用 ADF（Augmented Dickey - Fuller）检验判断序列是否平稳，若不平稳，可能需要进行差分等变换使其平稳。白噪声检验：确定序列是否为白噪声，若是白噪声则难以建模预测。 使用循环神经网络（LSTM、GRU）或 Transformer 进行时间序列预测 以 LSTM 为例预测股票价格 import numpy as np import pandas as pd from keras.models import Sequential from keras.layers import LSTM, Dense from sklearn.preprocessing import MinMaxScaler import matplotlib.pyplot as plt # 读取股票数据 stock_data = pd.read_csv('stock.csv') # 假设股票数据文件为 stock.csv，包含日期和收盘价字段 close_prices = stock_data['Close'].values.reshape(-1, 1) # 数据归一化 scaler = MinMaxScaler() scaled_close_prices = scaler.fit_transform(close_prices) # 构建训练数据 look_back = 30 # 利用过去 30 天的数据预测未来一天 X = [] y = [] for i in range(len(scaled_close_prices) - look_back): X.append(scaled_close_prices[i:i + look_back]) y.append(scaled_close_prices[i + look_back]) X = np.array(X) y = np.array(y) # 划分训练集和测试集 train_size = int(len(X) * 0.8) X_train, X_test = X[:train_size], X[train_size:] y_train, y_test = y[:train_size], y[train_size:] # 构建 LSTM 模型 model = Sequential() model.add(LSTM(units=50, return_sequences=True, input_shape=(look_back, 1))) model.add(LSTM(units=50)) model.add(Dense(units=1)) # 编译模型 model pile(optimizer='adam', loss='mean_squared_error') # 训练模型 model.fit(X_train, y_train, epochs=50, batch_size=32) # 预测 train_predict = model.predict(X_train) test_predict = model.predict(X_test) # 反归一化 train_predict = scaler.inverse_transform(train_predict) test_predict = scaler.inverse_transform(test_predict) y_train = scaler.inverse_transform(y_train) y_test = scaler.inverse_transform(y_test) # 绘制预测结果 plt.plot(stock_data['Date'].iloc[:train_size + look_back], y_train, label='实际训练集') plt.plot(stock_data['Date'].iloc[train_size + look_back:], y_test, label='实际测试集') plt.plot(stock_data['Date'].iloc[look_back:train_size + look_back], train_predict, label='预测训练集') plt.plot(stock_data['Date'].iloc[train_size + look_back:], test_predict, label='预测测试集') plt.legend() plt.show() 以 Transformer 为例预测电力负荷 import numpy as np import pandas as pd from keras.models import Model from keras.layers import Input, MultiHeadAttention, Dense, Dropout, LayerNormalization from sklearn.preprocessing import StandardScaler import matplotlib.pyplot as plt # 读取电力负荷数据 power_data = pd.read_csv('power.csv') # 假设电力负荷数据文件为 power.csv，包含时间和负荷值字段 load_values = power_data['Load'].values.reshape(-1, 1) # 数据标准化 scaler = StandardScaler() scaled_load_values = scaler.fit_transform(load_values) # 构建输入序列 seq_length = 24 # 利用过去 24 小时的数据预测未来一小时 X = [] y = [] for i in range(len(scaled_load_values) - seq_length): X.append(scaled_load_values[i:i + seq_length]) y.append(scaled_load_values[i + seq_length]) X = np.array(X) y = np.array(y) # 划分训练集和测试集 train_size = int(len(X) * 0.8) X_train, X_test = X[:train_size], X[train_size:] y_train, y_test = y[:train_size], y[train_size:] # 构建 Transformer 模型 input_layer = Input(shape=(seq_length, 1)) attn_layer = MultiHeadAttention(num_heads=4, key_dim=32)(input_layer, input_layer) attn_layer = Dropout(0.1)(attn_layer) attn_layer = LayerNormalization(epsilon=1e-6)(attn_layer) ffn_layer = Dense(64, activation='relu')(attn_layer) ffn_layer = Dense(1)(ffn_layer) model = Model(inputs=input_layer, outputs=ffn_layer) # 编译模型 model pile(optimizer='adam', loss='mean_squared_error') # 训练模型 model.fit(X_train, y_train, epochs=30, batch_size=32) # 预测 train_predict = model.predict(X_train) test_predict = model.predict(X_test) # 反标准化 train_predict = scaler.inverse_transform(train_predict) test_predict = scaler.inverse_transform(test_predict) y_train = scaler.inverse_transform(y_train) y_test = scaler.inverse_transform(y_test) # 绘制预测结果 plt.plot(power_data['Time'].iloc[:train_size + seq_length], y_train, label='实际训练集') plt.plot(power_data['Time'].iloc[train_size + seq_length:], y_test, label='实际测试集') plt.plot(power_data['Time'].iloc[seq_length:train_size + seq_length], train_predict, label='预测训练集') plt.plot(power_data['Time'].iloc[train_size + seq_length:], test_predict, label='预测测试集') plt.legend() plt.show()

请注意，上述代码中的数据文件路径（如 stock.csv、power.csv）需要根据实际情况替换为真实存在的数据集路径，并且数据集的格式应与代码中的假设相符。同时，在实际应用中，还可以进一步优化模型架构、超参数以及数据预处理方法，以提高预测性能。