当前位置：首页 > article >正文

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

article 2025/3/7 0:20:58

文章目录

第 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.compile(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.compile(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.compile(loss='categorical_crossentropy',
                   optimizer='adam',
                   metrics=['accuracy'])
model_cnn.compile(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.compile(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.compile(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.compile(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()