神经网络的Python实现详解

一、TensorFlow基本运算与实现方案

TensorFlow是一个基于数据流编程的符号数学系统,广泛应用于机器学习和深度学习领域。其核心概念是张量(Tensor)计算图(Graph)

1.1 基本运算实现

import tensorflow as tf
import numpy as np

# 1. 张量创建
# 标量(0维张量)
scalar = tf.constant(3.14)
# 向量(1维张量)
vector = tf.constant([1, 2, 3, 4, 5])
# 矩阵(2维张量)
matrix = tf.constant([[1, 2], [3, 4]])
# 3维张量
tensor_3d = tf.constant([[[1, 2], [3, 4]], [[5, 6], [7, 8]]])

# 2. 基本数学运算
a = tf.constant([[1, 2], [3, 4]], dtype=tf.float32)
b = tf.constant([[5, 6], [7, 8]], dtype=tf.float32)

# 加法
add_result = tf.add(a, b)
# 减法
sub_result = tf.subtract(a, b)
# 乘法(逐元素)
mul_result = tf.multiply(a, b)
# 矩阵乘法
matmul_result = tf.matmul(a, b)
# 除法
div_result = tf.divide(a, b)

# 3. 张量变形操作
# 改变形状
reshaped = tf.reshape(matrix, [4, 1])
# 转置
transposed = tf.transpose(matrix)
# 展平
flattened = tf.reshape(matrix, [-1])

# 4. 激活函数实现
def sigmoid(x):
    return 1 / (1 + tf.exp(-x))

def relu(x):
    return tf.maximum(0, x)

def tanh(x):
    return tf.tanh(x)

# 示例使用
x = tf.constant([-2.0, -1.0, 0.0, 1.0, 2.0])
print("Sigmoid:", sigmoid(x).numpy())
print("ReLU:", relu(x).numpy())
print("Tanh:", tanh(x).numpy())

1.2 TensorFlow实现方案

TensorFlow提供了多种实现神经网络的方案:

  1. 低级API:直接操作张量和计算图
  2. Keras API:高级神经网络API
  3. Estimator API:用于生产环境的API
  4. Eager Execution:即时执行模式
# 使用低级API构建简单神经网络
class SimpleNN:
    def __init__(self, input_size, hidden_size, output_size):
        # 初始化权重和偏置
        self.W1 = tf.Variable(tf.random.normal([input_size, hidden_size]))
        self.b1 = tf.Variable(tf.zeros([hidden_size]))
        self.W2 = tf.Variable(tf.random.normal([hidden_size, output_size]))
        self.b2 = tf.Variable(tf.zeros([output_size]))
    
    def __call__(self, x):
        # 前向传播
        hidden = tf.nn.relu(tf.matmul(x, self.W1) + self.b1)
        output = tf.matmul(hidden, self.W2) + self.b2
        return tf.nn.softmax(output)
    
    def train_step(self, x, y, learning_rate=0.01):
        with tf.GradientTape() as tape:
            predictions = self(x)
            loss = tf.reduce_mean(
                tf.keras.losses.categorical_crossentropy(y, predictions)
            )
        
        # 计算梯度
        gradients = tape.gradient(loss, [self.W1, self.b1, self.W2, self.b2])
        
        # 更新参数
        optimizer = tf.keras.optimizers.Adam(learning_rate)
        optimizer.apply_gradients(zip(gradients, [self.W1, self.b1, self.W2, self.b2]))
        
        return loss

二、Keras全神经网络实现

Keras是TensorFlow的高级API,提供了简洁的接口来构建和训练神经网络。

2.1 全连接神经网络实现

from tensorflow import keras
from tensorflow.keras import layers
import numpy as np

# 1. 创建全连接神经网络模型
def create_fully_connected_nn(input_shape, num_classes):
    """
    创建全连接神经网络
    
    参数:
    input_shape: 输入数据的形状
    num_classes: 分类数量
    """
    model = keras.Sequential([
        # 输入层
        layers.Input(shape=input_shape),
        
        # 第一个全连接层(隐藏层)
        layers.Dense(128, activation='relu', name='hidden_layer_1'),
        layers.BatchNormalization(),
        layers.Dropout(0.3),
        
        # 第二个全连接层(隐藏层)
        layers.Dense(64, activation='relu', name='hidden_layer_2'),
        layers.BatchNormalization(),
        layers.Dropout(0.3),
        
        # 第三个全连接层(隐藏层)
        layers.Dense(32, activation='relu', name='hidden_layer_3'),
        
        # 输出层
        layers.Dense(num_classes, activation='softmax', name='output_layer')
    ])
    
    return model

# 2. 模型编译与训练
def train_keras_model(model, train_data, train_labels, val_data, val_labels):
    """
    编译和训练Keras模型
    """
    # 编译模型
    model.compile(
        optimizer=keras.optimizers.Adam(learning_rate=0.001),
        loss='categorical_crossentropy',
        metrics=['accuracy']
    )
    
    # 定义回调函数
    callbacks = [
        keras.callbacks.EarlyStopping(
            monitor='val_loss',
            patience=10,
            restore_best_weights=True
        ),
        keras.callbacks.ReduceLROnPlateau(
            monitor='val_loss',
            factor=0.5,
            patience=5,
            min_lr=1e-6
        ),
        keras.callbacks.ModelCheckpoint(
            'best_model.h5',
            monitor='val_accuracy',
            save_best_only=True
        )
    ]
    
    # 训练模型
    history = model.fit(
        train_data, train_labels,
        validation_data=(val_data, val_labels),
        epochs=50,
        batch_size=32,
        callbacks=callbacks,
        verbose=1
    )
    
    return model, history

# 3. 示例:在MNIST数据集上训练
def mnist_example():
    # 加载MNIST数据集
    (x_train, y_train), (x_test, y_test) = keras.datasets.mnist.load_data()
    
    # 数据预处理
    x_train = x_train.reshape(-1, 28*28).astype('float32') / 255.0
    x_test = x_test.reshape(-1, 28*28).astype('float32') / 255.0
    
    # 标签one-hot编码
    y_train = keras.utils.to_categorical(y_train, 10)
    y_test = keras.utils.to_categorical(y_test, 10)
    
    # 创建模型
    model = create_fully_connected_nn(input_shape=(784,), num_classes=10)
    
    # 显示模型结构
    model.summary()
    
    # 训练模型
    trained_model, history = train_keras_model(
        model, x_train, y_train, x_test, y_test
    )
    
    # 评估模型
    test_loss, test_accuracy = trained_model.evaluate(x_test, y_test)
    print(f"测试准确率: {test_accuracy:.4f}")
    
    return trained_model, history

2.2 Keras自定义层和模型

# 自定义激活函数层
class CustomActivationLayer(layers.Layer):
    def __init__(self, activation_type='relu', **kwargs):
        super().__init__(**kwargs)
        self.activation_type = activation_type
    
    def call(self, inputs):
        if self.activation_type == 'relu':
            return tf.nn.relu(inputs)
        elif self.activation_type == 'sigmoid':
            return tf.nn.sigmoid(inputs)
        elif self.activation_type == 'tanh':
            return tf.nn.tanh(inputs)
        elif self.activation_type == 'leaky_relu':
            return tf.nn.leaky_relu(inputs, alpha=0.01)
        else:
            return inputs
    
    def get_config(self):
        config = super().get_config()
        config.update({"activation_type": self.activation_type})
        return config

# 自定义神经网络模型
class CustomNeuralNetwork(keras.Model):
    def __init__(self, input_dim, hidden_dims, output_dim, dropout_rate=0.3):
        super().__init__()
        
        # 创建隐藏层
        self.hidden_layers = []
        for i, hidden_dim in enumerate(hidden_dims):
            self.hidden_layers.append(
                layers.Dense(hidden_dim, name=f'hidden_layer_{i}')
            )
            self.hidden_layers.append(
                CustomActivationLayer('relu', name=f'activation_{i}')
            )
            self.hidden_layers.append(
                layers.Dropout(dropout_rate, name=f'dropout_{i}')
            )
        
        # 输出层
        self.output_layer = layers.Dense(output_dim, activation='softmax')
    
    def call(self, inputs, training=False):
        x = inputs
        for layer in self.hidden_layers:
            if isinstance(layer, layers.Dropout):
                x = layer(x, training=training)
            else:
                x = layer(x)
        return self.output_layer(x)

三、自定义程序实现

3.1 激活函数实现

import numpy as np

class ActivationFunctions:
    """自定义激活函数实现"""
    
    @staticmethod
    def sigmoid(x):
        """
        Sigmoid激活函数
        公式: σ(x) = 1 / (1 + exp(-x))
        """
        return 1 / (1 + np.exp(-x))
    
    @staticmethod
    def sigmoid_derivative(x):
        """Sigmoid函数的导数"""
        sig = ActivationFunctions.sigmoid(x)
        return sig * (1 - sig)
    
    @staticmethod
    def relu(x):
        """
        ReLU激活函数
        公式: f(x) = max(0, x)
        """
        return np.maximum(0, x)
    
    @staticmethod
    def relu_derivative(x):
        """ReLU函数的导数"""
        return np.where(x > 0, 1, 0)
    
    @staticmethod
    def tanh(x):
        """
        Tanh激活函数
        公式: tanh(x) = (exp(x) - exp(-x)) / (exp(x) + exp(-x))
        """
        return np.tanh(x)
    
    @staticmethod
    def tanh_derivative(x):
        """Tanh函数的导数"""
        return 1 - np.tanh(x) ** 2
    
    @staticmethod
    def leaky_relu(x, alpha=0.01):
        """
        Leaky ReLU激活函数
        公式: f(x) = x if x > 0 else alpha * x
        """
        return np.where(x > 0, x, alpha * x)
    
    @staticmethod
    def leaky_relu_derivative(x, alpha=0.01):
        """Leaky ReLU函数的导数"""
        return np.where(x > 0, 1, alpha)
    
    @staticmethod
    def softmax(x):
        """
        Softmax激活函数
        用于多分类问题的输出层
        """
        exp_x = np.exp(x - np.max(x, axis=-1, keepdims=True))
        return exp_x / np.sum(exp_x, axis=-1, keepdims=True)

3.2 损失函数实现

class LossFunctions:
    """自定义损失函数实现"""
    
    @staticmethod
    def mse_loss(y_true, y_pred):
        """
        均方误差损失函数
        用于回归问题
        """
        return np.mean((y_true - y_pred) ** 2)
    
    @staticmethod
    def mse_loss_derivative(y_true, y_pred):
        """MSE损失的导数"""
        return 2 * (y_pred - y_true) / len(y_true)
    
    @staticmethod
    def binary_crossentropy(y_true, y_pred, epsilon=1e-15):
        """
        二元交叉熵损失函数
        用于二分类问题
        """
        # 防止log(0)
        y_pred = np.clip(y_pred, epsilon, 1 - epsilon)
        return -np.mean(y_true * np.log(y_pred) + (1 - y_true) * np.log(1 - y_pred))
    
    @staticmethod
    def binary_crossentropy_derivative(y_true, y_pred, epsilon=1e-15):
        """二元交叉熵损失的导数"""
        y_pred = np.clip(y_pred, epsilon, 1 - epsilon)
        return (y_pred - y_true) / (y_pred * (1 - y_pred) * len(y_true))
    
    @staticmethod
    def categorical_crossentropy(y_true, y_pred, epsilon=1e-15):
        """
        分类交叉熵损失函数
        用于多分类问题
        """
        y_pred = np.clip(y_pred, epsilon, 1 - epsilon)
        return -np.sum(y_true * np.log(y_pred)) / len(y_true)
    
    @staticmethod
    def categorical_crossentropy_derivative(y_true, y_pred, epsilon=1e-15):
        """分类交叉熵损失的导数"""
        y_pred = np.clip(y_pred, epsilon, 1 - epsilon)
        return (y_pred - y_true) / len(y_true)

3.3 训练函数定义

class NeuralNetworkTrainer:
    """自定义神经网络训练器"""
    
    def __init__(self, model, learning_rate=0.01):
        self.model = model
        self.learning_rate = learning_rate
        self.loss_history = []
        self.accuracy_history = []
    
    def train(self, X_train, y_train, X_val, y_val, 
              epochs=100, batch_size=32, verbose=True):
        """
        训练神经网络
        
        参数:
        X_train, y_train: 训练数据和标签
        X_val, y_val: 验证数据和标签
        epochs: 训练轮数
        batch_size: 批次大小
        verbose: 是否显示训练进度
        """
        n_samples = X_train.shape[0]
        n_batches = int(np.ceil(n_samples / batch_size))
        
        for epoch in range(epochs):
            # 打乱数据
            indices = np.random.permutation(n_samples)
            X_shuffled = X_train[indices]
            y_shuffled = y_train[indices]
            
            epoch_loss = 0
            epoch_accuracy = 0
            
            for batch in range(n_batches):
                # 获取当前批次数据
                start = batch * batch_size
                end = min(start + batch_size, n_samples)
                X_batch = X_shuffled[start:end]
                y_batch = y_shuffled[start:end]
                
                # 前向传播
                predictions = self.model.forward(X_batch)
                
                # 计算损失
                batch_loss = self.model.compute_loss(y_batch, predictions)
                epoch_loss += batch_loss
                
                # 计算准确率
                if y_batch.ndim > 1:  # 多分类
                    pred_labels = np.argmax(predictions, axis=1)
                    true_labels = np.argmax(y_batch, axis=1)
                else:  # 二分类
                    pred_labels = (predictions > 0.5).astype(int)
                    true_labels = y_batch
                
                batch_accuracy = np.mean(pred_labels == true_labels)
                epoch_accuracy += batch_accuracy
                
                # 反向传播和参数更新
                self.model.backward(X_batch, y_batch, self.learning_rate)
            
            # 计算平均损失和准确率
            avg_loss = epoch_loss / n_batches
            avg_accuracy = epoch_accuracy / n_batches
            
            # 验证集评估
            val_predictions = self.model.forward(X_val)
            val_loss = self.model.compute_loss(y_val, val_predictions)
            
            if y_val.ndim > 1:
                val_pred_labels = np.argmax(val_predictions, axis=1)
                val_true_labels = np.argmax(y_val, axis=1)
            else:
                val_pred_labels = (val_predictions > 0.5).astype(int)
                val_true_labels = y_val
            
            val_accuracy = np.mean(val_pred_labels == val_true_labels)
            
            # 记录历史
            self.loss_history.append({
                'epoch': epoch + 1,
                'train_loss': avg_loss,
                'val_loss': val_loss,
                'train_accuracy': avg_accuracy,
                'val_accuracy': val_accuracy
            })
            
            if verbose and (epoch + 1) % 10 == 0:
                print(f"Epoch {epoch + 1}/{epochs}")
                print(f"  Train Loss: {avg_loss:.4f}, Train Acc: {avg_accuracy:.4f}")
                print(f"  Val Loss: {val_loss:.4f}, Val Acc: {val_accuracy:.4f}")
                print("-" * 50)
    
    def plot_training_history(self):
        """绘制训练历史"""
        import matplotlib.pyplot as plt
        
        epochs = [h['epoch'] for h in self.loss_history]
        train_loss = [h['train_loss'] for h in self.loss_history]
        val_loss = [h['val_loss'] for h in self.loss_history]
        train_acc = [h['train_accuracy'] for h in self.loss_history]
        val_acc = [h['val_accuracy'] for h in self.loss_history]
        
        fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(12, 4))
        
        # 损失曲线
        ax1.plot(epochs, train_loss, 'b-', label='Train Loss')
        ax1.plot(epochs, val_loss, 'r-', label='Val Loss')
        ax1.set_xlabel('Epoch')
        ax1.set_ylabel('Loss')
        ax1.set_title('Training and Validation Loss')
        ax1.legend()
        ax1.grid(True)
        
        # 准确率曲线
        ax2.plot(epochs, train_acc, 'b-', label='Train Accuracy')
        ax2.plot(epochs, val_acc, 'r-', label='Val Accuracy')
        ax2.set_xlabel('Epoch')
        ax2.set_ylabel('Accuracy')
        ax2.set_title('Training and Validation Accuracy')
        ax2.legend()
        ax2.grid(True)
        
        plt.tight_layout()
        plt.show()

3.4 完整自定义神经网络实现

class CustomNeuralNetwork:
    """从零实现的全连接神经网络"""
    
    def __init__(self, layer_sizes, activation='relu', 
                 weight_init='he', learning_rate=0.01):
        """
        初始化神经网络
        
        参数:
        layer_sizes: 每层神经元数量列表,如 [784, 128, 64, 10]
        activation: 激活函数类型 ('relu', 'sigmoid', 'tanh')
        weight_init: 权重初始化方法 ('he', 'xavier', 'random')
        learning_rate: 学习率
        """
        self.layer_sizes = layer_sizes
        self.activation_type = activation
        self.learning_rate = learning_rate
        self.weights = []
        self.biases = []
        self.activations = []
        
        # 初始化权重和偏置
        for i in range(len(layer_sizes) - 1):
            input_size = layer_sizes[i]
            output_size = layer_sizes[i + 1]
            
            # 权重初始化
            if weight_init == 'he':
                # He初始化,适合ReLU
                std = np.sqrt(2.0 / input_size)
                W = np.random.randn(input_size, output_size) * std
            elif weight_init == 'xavier':
                # Xavier初始化,适合Sigmoid/Tanh
                std = np.sqrt(1.0 / input_size)
                W = np.random.randn(input_size, output_size) * std
            else:
                # 随机初始化
                W = np.random.randn(input_size, output_size) * 0.01
            
            b = np.zeros((1, output_size))
            
            self.weights.append(W)
            self.biases.append(b)
    
    def _activate(self, x, derivative=False):
        """应用激活函数"""
        if self.activation_type == 'relu':
            if derivative:
                return np.where(x > 0, 1, 0)
            return np.maximum(0, x)
        elif self.activation_type == 'sigmoid':
            if derivative:
                sig = 1 / (1 + np.exp(-x))
                return sig * (1 - sig)
            return 1 / (1 + np.exp(-x))
        elif self.activation_type == 'tanh':
            if derivative:
                return 1 - np.tanh(x) ** 2
            return np.tanh(x)
        else:
            raise ValueError(f"不支持的激活函数: {self.activation_type}")
    
    def forward(self, X):
        """前向传播"""
        self.activations = [X]
        current_activation = X
        
        # 隐藏层的前向传播
        for i in range(len(self.weights) - 1):
            z = np.dot(current_activation, self.weights[i]) + self.biases[i]
            a = self._activate(z)
            self.activations.append(a)
            current_activation = a
        
        # 输出层(使用softmax)
        z_output = np.dot(current_activation, self.weights[-1]) + self.biases[-1]
        a_output = self._softmax(z_output)
        self.activations.append(a_output)
        
        return a_output
    
    def _softmax(self, x):
        """Softmax函数"""
        exp_x = np.exp(x - np.max(x, axis=1, keepdims=True))
        return exp_x / np.sum(exp_x, axis=1, keepdims=True)
    
    def compute_loss(self, y_true, y_pred):
        """计算交叉熵损失"""
        m = y_true.shape[0]
        log_likelihood = -np.sum(y_true * np.log(y_pred + 1e-15))
        loss = log_likelihood / m
        return loss
    
    def backward(self, X, y, learning_rate):
        """反向传播"""
        m = X.shape[0]
        gradients_w = [np.zeros_like(w) for w in self.weights]
        gradients_b = [np.zeros_like(b) for b in self.biases]
        
        # 输出层的误差
        delta = self.activations[-1] - y
        
        # 反向传播误差
        for l in range(len(self.weights) - 1, -1, -1):
            # 计算当前层的梯度
            a_prev = self.activations[l]
            gradients_w[l] = np.dot(a_prev.T, delta) / m
            gradients_b[l] = np.sum(delta, axis=0, keepdims=True) / m
            
            # 如果不是输入层,则传播误差到前一层
            if l > 0:
                delta = np.dot(delta, self.weights[l].T) * \
                        self._activate(self.activations[l], derivative=True)
        
        # 更新权重和偏置
        for l in range(len(self.weights)):
            self.weights[l] -= learning_rate * gradients_w[l]
            self.biases[l] -= learning_rate * gradients_b[l]
    
    def predict(self, X):
        """预测"""
        return self.forward(X)
    
    def predict_classes(self, X):
        """预测类别"""
        probabilities = self.predict(X)
        return np.argmax(probabilities, axis=1)
    
    def evaluate(self, X, y):
        """评估模型"""
        predictions = self.predict_classes(X)
        if y.ndim > 1:
            true_labels = np.argmax(y, axis=1)
        else:
            true_labels = y
        accuracy = np.mean(predictions == true_labels)
        return accuracy

# 使用示例
def custom_nn_example():
    # 创建数据集
    from sklearn.datasets import make_classification
    from sklearn.model_selection import train_test_split
    from sklearn.preprocessing import OneHotEncoder
    
    # 生成二分类数据集
    X, y = make_classification(
        n_samples=1000, n_features=20, n_informative=15,
        n_redundant=5, n_classes=3, random_state=42
    )
    
    # 对标签进行one-hot编码
    encoder = OneHotEncoder(sparse_output=False)
    y_onehot = encoder.fit_transform(y.reshape(-1, 1))
    
    # 划分训练集和测试集
    X_train, X_test, y_train, y_test = train_test_split(
        X, y_onehot, test_size=0.2, random_state=42
    )
    
    # 创建神经网络
    nn = CustomNeuralNetwork(
        layer_sizes=[20, 64, 32, 3],
        activation='relu',
        weight_init='he',
        learning_rate=0.01
    )
    
    # 创建训练器
    trainer = NeuralNetworkTrainer(nn, learning_rate=0.01)
    
    # 训练模型
    trainer.train(
        X_train, y_train,
        X_test, y_test,
        epochs=100,
        batch_size=32,
        verbose=True
    )
    
    # 绘制训练历史
    trainer.plot_training_history()
    
    # 评估模型
    test_accuracy = nn.evaluate(X_test, y_test)
    print(f"测试集准确率: {test_accuracy:.4f}")
    
    return nn, trainer

四、卷积神经网络实现

4.1 TensorFlow实现卷积神经网络

import tensorflow as tf
from tensorflow.keras import layers, models
import numpy as np

class CNNWithTensorFlow:
    """使用TensorFlow实现卷积神经网络"""
    
    @staticmethod
    def create_simple_cnn(input_shape, num_classes):
        """
        创建简单的CNN模型
        
        参数:
        input_shape: 输入图像形状 (height, width, channels)
        num_classes: 分类数量
        """
        model = models.Sequential([
            # 第一卷积层
            layers.Conv2D(32, (3, 3), activation='relu', 
                         input_shape=input_shape, padding='same'),
            layers.BatchNormalization(),
            layers.MaxPooling2D((2, 2)),
            layers.Dropout(0.25),
            
            # 第二卷积层
            layers.Conv2D(64, (3, 3), activation='relu', padding='same'),
            layers.BatchNormalization(),
            layers.MaxPooling2D((2, 2)),
            layers.Dropout(0.25),
            
            # 第三卷积层
            layers.Conv2D(128, (3, 3), activation='relu', padding='same'),
            layers.BatchNormalization(),
            layers.MaxPooling2D((2, 2)),
            layers.Dropout(0.25),
            
            # 展平层
            layers.Flatten(),
            
            # 全连接层
            layers.Dense(256, activation='relu'),
            layers.BatchNormalization(),
            layers.Dropout(0.5),
            
            # 输出层
            layers.Dense(num_classes, activation='softmax')
        ])
        
        return model
    
    @staticmethod
    def create_advanced_cnn(input_shape, num_classes):
        """
        创建更复杂的CNN模型
        """
        inputs = layers.Input(shape=input_shape)
        
        # 卷积块1
        x = layers.Conv2D(64, (3, 3), padding='same')(inputs)
        x = layers.BatchNormalization()(x)
        x = layers.Activation('relu')(x)
        x = layers.Conv2D(64, (3, 3), padding='same')(x)
        x = layers.BatchNormalization()(x)
        x = layers.Activation('relu')(x)
        x = layers.MaxPooling2D((2, 2))(x)
        x = layers.Dropout(0.25)(x)
        
        # 卷积块2
        x = layers.Conv2D(128, (3, 3), padding='same')(x)
        x = layers.BatchNormalization()(x)
        x = layers.Activation('relu')(x)
        x = layers.Conv2D(128, (3, 3), padding='same')(x)
        x = layers.BatchNormalization()(x)
        x = layers.Activation('relu')(x)
        x = layers.MaxPooling2D((2, 2))(x)
        x = layers.Dropout(0.25)(x)
        
        # 卷积块3
        x = layers.Conv2D(256, (3, 3), padding='same')(x)
        x = layers.BatchNormalization()(x)
        x = layers.Activation('relu')(x)
        x = layers.Conv2D(256, (3, 3), padding='same')(x)
        x = layers.BatchNormalization()(x)
        x = layers.Activation('relu')(x)
        x = layers.MaxPooling2D((2, 2))(x)
        x = layers.Dropout(0.25)(x)
        
        # 全局平均池化
        x = layers.GlobalAveragePooling2D()(x)
        
        # 全连接层
        x = layers.Dense(512, activation='relu')(x)
        x = layers.BatchNormalization()(x)
        x = layers.Dropout(0.5)(x)
        
        # 输出层
        outputs = layers.Dense(num_classes, activation='softmax')(x)
        
        model = models.Model(inputs=inputs, outputs=outputs)
        return model
    
    @staticmethod
    def train_cnn_model(model, train_data, train_labels, 
                       val_data, val_labels, epochs=50, batch_size=32):
        """
        训练CNN模型
        """
        # 编译模型
        model.compile(
            optimizer=tf.keras.optimizers.Adam(learning_rate=0.001),
            loss='categorical_crossentropy',
            metrics=['accuracy']
        )
        
        # 定义数据增强
        data_augmentation = tf.keras.Sequential([
            layers.RandomFlip("horizontal"),
            layers.RandomRotation(0.1),
            layers.RandomZoom(0.1),
        ])
        
        # 定义回调函数
        callbacks = [
            tf.keras.callbacks.EarlyStopping(
                monitor='val_loss',
                patience=15,
                restore_best_weights=True
            ),
            tf.keras.callbacks.ReduceLROnPlateau(
                monitor='val_loss',
                factor=0.5,
                patience=5,
                min_lr=1e-6
            ),
            tf.keras.callbacks.ModelCheckpoint(
                'best_cnn_model.h5',
                monitor='val_accuracy',
                save_best_only=True
            )
        ]
        
        # 训练模型
        history = model.fit(
            train_data, train_labels,
            validation_data=(val_data, val_labels),
            epochs=epochs,
            batch_size=batch_size,
            callbacks=callbacks,
            verbose=1
        )
        
        return model, history
    
    @staticmethod
    def cnn_mnist_example():
        """在MNIST数据集上训练CNN示例"""
        # 加载数据
        (x_train, y_train), (x_test, y_test) = tf.keras.datasets.mnist.load_data()
        
        # 数据预处理
        x_train = x_train.reshape(-1, 28, 28, 1).astype('float32') / 255.0
        x_test = x_test.reshape(-1, 28, 28, 1).astype('float32') / 255.0
        
        # 标签one-hot编码
        y_train = tf.keras.utils.to_categorical(y_train, 10)
        y_test = tf.keras.utils.to_categorical(y_test, 10)
        
        # 创建模型
        model = CNNWithTensorFlow.create_simple_cnn(
            input_shape=(28, 28, 1),
            num_classes=10
        )
        
        # 显示模型结构
        model.summary()
        
        # 训练模型
        trained_model, history = CNNWithTensorFlow.train_cnn_model(
            model, x_train, y_train, x_test, y_test,
            epochs=30, batch_size=64
        )
        
        # 评估模型
        test_loss, test_accuracy = trained_model.evaluate(x_test, y_test)
        print(f"测试集损失: {test_loss:.4f}")
        print(f"测试集准确率: {test_accuracy:.4f}")
        
        return trained_model, history

4.2 卷积神经网络原理与实现

class ConvolutionalLayer:
    """从零实现的卷积层"""
    
    def __init__(self, num_filters, filter_size, stride=1, padding=0):
        """
        初始化卷积层
        
        参数:
        num_filters: 卷积核数量
        filter_size: 卷积核大小
        stride: 步长
        padding: 填充大小
        """
        self.num_filters = num_filters
        self.filter_size = filter_size
        self.stride = stride
        self.padding = padding
        
        # 初始化卷积核和偏置
        self.filters = np.random.randn(
            num_filters, filter_size, filter_size
        ) * np.sqrt(2.0 / (filter_size * filter_size))
        self.biases = np.zeros((num_filters, 1))
        
        # 缓存用于反向传播
        self.input = None
        self.output = None
        
    def forward(self, input_data):
        """
        前向传播
        input_data: (batch_size, height, width, channels)
        """
        batch_size, input_height, input_width, input_channels = input_data.shape
        
        # 计算输出尺寸
        output_height = (input_height + 2*self.padding - self.filter_size) // self.stride + 1
        output_width = (input_width + 2*self.padding - self.filter_size) // self.stride + 1
        
        # 添加padding
        if self.padding > 0:
            padded_input = np.pad(
                input_data,
                ((0, 0), (self.padding, self.padding), 
                 (self.padding, self.padding), (0, 0)),
                mode='constant'
            )
        else:
            padded_input = input_data
        
        # 初始化输出
        output = np.zeros((batch_size, output_height, output_width, self.num_filters))
        
        # 执行卷积操作
        for b in range(batch_size):
            for f in range(self.num_filters):
                for h in range(output_height):
                    for w in range(output_width):
                        # 计算当前窗口位置
                        h_start = h * self.stride
                        h_end = h_start + self.filter_size
                        w_start = w * self.stride
                        w_end = w_start + self.filter_size
                        
                        # 提取当前窗口
                        window = padded_input[b, h_start:h_end, w_start:w_end, :]
                        
                        # 执行卷积运算
                        output[b, h, w, f] = np.sum(
                            window * self.filters[f]
                        ) + self.biases[f]
        
        # 缓存输入和输出
        self.input = input_data
        self.output = output
        
        return output
    
    def backward(self, d_output, learning_rate=0.01):
        """
        反向传播
        d_output: 输出梯度
        """
        batch_size, output_height, output_width, _ = d_output.shape
        _, input_height, input_width,

----

## 参考来源
- [AlexNet网络搭建(tensorflow,keras)](https://blog.csdn.net/qq_57780419/article/details/124047538)
- [27、基于Python平台二维卷积神经网络的X光图像肺炎检测](https://blog.csdn.net/ol789012/article/details/152069614)
- [paddle实现AlexNet](https://blog.csdn.net/nominior/article/details/104081576)
- [Python 人工智能实战:深度学习芯片](https://blog.csdn.net/universsky2015/article/details/134067509)
- [Python 深度学习实战:图像分类](https://blog.csdn.net/universsky2015/article/details/134658033)

 

Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐