# 神经网络入门实战：PyTorch与Keras实现手写数字识别

手写数字识别是深度学习入门的经典案例，能够帮助初学者理解神经网络的基本原理。本文将通过PyTorch和Keras两种框架，构建神经网络模型识别MNIST手写数字数据集。

## 环境准备与数据加载

```python

# PyTorch版本

import torch

import torch.nn as nn

import torch.optim as optim

from torchvision import datasets, transforms

import matplotlib.pyplot as plt

# Keras版本

import tensorflow as tf

from tensorflow import keras

from tensorflow.keras import layers

import numpy as np

# 数据加载和预处理

def load_data_pytorch():

    """PyTorch数据加载"""

    transform = transforms.Compose([

        transforms.ToTensor(),

        transforms.Normalize((0.1307,), (0.3081,))

    ])

    

    train_dataset = datasets.MNIST(

        './data', train=True, download=True, transform=transform

    )

    test_dataset = datasets.MNIST(

        './data', train=False, transform=transform

    )

    

    train_loader = torch.utils.data.DataLoader(

        train_dataset, batch_size=64, shuffle=True

    )

    test_loader = torch.utils.data.DataLoader(

        test_dataset, batch_size=1000, shuffle=False

    )

    

    return train_loader, test_loader

def load_data_keras():

    """Keras数据加载"""

    (x_train, y_train), (x_test, y_test) = 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 = keras.utils.to_categorical(y_train, 10)

    y_test = keras.utils.to_categorical(y_test, 10)

    

    return (x_train, y_train), (x_test, y_test)

# 数据可视化

def visualize_samples(images, labels, num_samples=10):

    """可视化样本数据"""

    plt.figure(figsize=(15, 3))

    for i in range(num_samples):

        plt.subplot(1, num_samples, i+1)

        plt.imshow(images[i].reshape(28, 28), cmap='gray')

        plt.title(f'Label: {labels[i]}')

        plt.axis('off')

    plt.show()

```

## PyTorch神经网络实现

```python

class PyTorchNeuralNetwork(nn.Module):

    """PyTorch神经网络模型"""

    def __init__(self):

        super(PyTorchNeuralNetwork, self).__init__()

        

        # 定义网络结构

        self.flatten = nn.Flatten()

        self.fc1 = nn.Linear(28*28, 128)

        self.relu = nn.ReLU()

        self.dropout = nn.Dropout(0.3)

        self.fc2 = nn.Linear(128, 64)

        self.fc3 = nn.Linear(64, 10)

        self.softmax = nn.Softmax(dim=1)

        

    def forward(self, x):

        # 前向传播

        x = self.flatten(x)

        x = self.fc1(x)

        x = self.relu(x)

        x = self.dropout(x)

        x = self.fc2(x)

        x = self.relu(x)

        x = self.fc3(x)

        return self.softmax(x)

    

    def predict(self, x):

        """预测方法"""

        self.eval()

        with torch.no_grad():

            outputs = self.forward(x)

            _, predicted = torch.max(outputs, 1)

        return predicted

class PyTorchTrainer:

    """PyTorch模型训练器"""

    def __init__(self, model, learning_rate=0.001):

        self.model = model

        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

        self.model.to(self.device)

        

        self.criterion = nn.CrossEntropyLoss()

        self.optimizer = optim.Adam(model.parameters(), lr=learning_rate)

        

    def train_epoch(self, train_loader):

        """训练一个epoch"""

        self.model.train()

        running_loss = 0.0

        correct = 0

        total = 0

        

        for batch_idx, (data, target) in enumerate(train_loader):

            data, target = data.to(self.device), target.to(self.device)

            

            self.optimizer.zero_grad()

            outputs = self.model(data)

            loss = self.criterion(outputs, target)

            loss.backward()

            self.optimizer.step()

            

            running_loss += loss.item()

            _, predicted = torch.max(outputs.data, 1)

            total += target.size(0)

            correct += (predicted == target).sum().item()

            

        return running_loss/len(train_loader), 100*correct/total

    

    def evaluate(self, test_loader):

        """评估模型"""

        self.model.eval()

        test_loss = 0

        correct = 0

        total = 0

        

        with torch.no_grad():

            for data, target in test_loader:

                data, target = data.to(self.device), target.to(self.device)

                outputs = self.model(data)

                test_loss += self.criterion(outputs, target).item()

                _, predicted = torch.max(outputs.data, 1)

                total += target.size(0)

                correct += (predicted == target).sum().item()

                

        return test_loss/len(test_loader), 100*correct/total

    

    def train(self, train_loader, test_loader, epochs=10):

        """完整训练过程"""

        train_losses, train_accs = [], []

        test_losses, test_accs = [], []

        

        for epoch in range(epochs):

            train_loss, train_acc = self.train_epoch(train_loader)

            test_loss, test_acc = self.evaluate(test_loader)

            

            train_losses.append(train_loss)

            train_accs.append(train_acc)

            test_losses.append(test_loss)

            test_accs.append(test_acc)

            

            print(f'Epoch {epoch+1}/{epochs}: '

                  f'Train Loss: {train_loss:.4f}, Train Acc: {train_acc:.2f}%, '

                  f'Test Loss: {test_loss:.4f}, Test Acc: {test_acc:.2f}%')

            

        return train_losses, train_accs, test_losses, test_accs

```

## Keras神经网络实现

```python

class KerasNeuralNetwork:

    """Keras神经网络模型"""

    def __init__(self, input_shape=(28, 28, 1)):

        self.model = self.build_model(input_shape)

        

    def build_model(self, input_shape):

        """构建Keras模型"""

        model = keras.Sequential([

            layers.Input(shape=input_shape),

            layers.Flatten(),

            layers.Dense(128, activation='relu'),

            layers.Dropout(0.3),

            layers.Dense(64, activation='relu'),

            layers.Dense(10, activation='softmax')

        ])

        

        model.compile(

            optimizer='adam',

            loss='categorical_crossentropy',

            metrics=['accuracy']

        )

        <"n3.j9k5.org.cn"><"w7.j9k5.org.cn"><"g9.j9k5.org.cn">

        return model

    

    def train(self, x_train, y_train, x_test, y_test, epochs=10, batch_size=64):

        """训练模型"""

        history = self.model.fit(

            x_train, y_train,

            batch_size=batch_size,

            epochs=epochs,

            validation_data=(x_test, y_test),

            verbose=1

        )

        

        return history

    

    def evaluate(self, x_test, y_test):

        """评估模型"""

        test_loss, test_acc = self.model.evaluate(x_test, y_test, verbose=0)

        return test_loss, test_acc

    

    def predict(self, x):

        """预测方法"""

        predictions = self.model.predict(x)

        return np.argmax(predictions, axis=1)

    

    def save_model(self, filepath):

        """保存模型"""

        self.model.save(filepath)

        print(f"模型已保存到: {filepath}")

class KerasCallbackHandler:

    """Keras回调函数处理器"""

    def __init__(self):

        self.callbacks = []

        

    def add_early_stopping(self, patience=5):

        """添加早停回调"""

        early_stopping = keras.callbacks.EarlyStopping(

            monitor='val_loss',

            patience=patience,

            restore_best_weights=True

        )

        self.callbacks.append(early_stopping)

        return self

    

    def add_checkpoint(self, filepath):

        """添加模型检查点"""

        checkpoint = keras.callbacks.ModelCheckpoint(

            filepath=filepath,

            monitor='val_accuracy',

            save_best_only=True

        )

        self.callbacks.append(checkpoint)

        return self

    

    def add_reduce_lr(self, factor=0.5, patience=3):

        """添加学习率衰减"""

        reduce_lr = keras.callbacks.ReduceLROnPlateau(

            monitor='val_loss',

            factor=factor,

            patience=patience

        )

        self.callbacks.append(reduce_lr)

        return self

    

    def get_callbacks(self):

        """获取所有回调函数"""

        return self.callbacks

```

## 模型训练与比较

```python

def compare_frameworks():

    """比较PyTorch和Keras实现"""

    

    print("="*60)

    print("PyTorch实现")

    print("="*60)

    

    # PyTorch训练

    train_loader, test_loader = load_data_pytorch()

    

    pytorch_model = PyTorchNeuralNetwork()

    pytorch_trainer = PyTorchTrainer(pytorch_model, learning_rate=0.001)

    

    pytorch_losses = pytorch_trainer.train(train_loader, test_loader, epochs=5)

    

    print("\n" + "="*60)

    print("Keras实现")

    print("="*60)

    

    # Keras训练

    (x_train, y_train), (x_test, y_test) = load_data_keras()

    

    keras_model = KerasNeuralNetwork()

    callback_handler = KerasCallbackHandler()

    callback_handler.add_early_stopping(patience=3)

    

    history = keras_model.train(

        x_train, y_train, x_test, y_test,

        epochs=5,

        batch_size=64

    )

    

    # 评估结果

    keras_loss, keras_acc = keras_model.evaluate(x_test, y_test)

    print(f"Keras测试准确率: {keras_acc:.4f}")

    

    return pytorch_model, keras_model

def visualize_training_results(history, framework_name):

    """可视化训练结果"""

    plt.figure(figsize=(12, 4))

    

    # 绘制损失曲线

    plt.subplot(1, 2, 1)

    plt.plot(history.history['loss'], label='训练损失')

    plt.plot(history.history['val_loss'], label='验证损失')

    plt.title(f'{framework_name} - 损失曲线')

    plt.xlabel('Epoch')

    plt.ylabel('损失值')

    plt.legend()

    plt.grid(True)

    

    # 绘制准确率曲线

    plt.subplot(1, 2, 2)

    plt.plot(history.history['accuracy'], label='训练准确率')

    plt.plot(history.history['val_accuracy'], label='验证准确率')

    plt.title(f'{framework_name} - 准确率曲线')

    plt.xlabel('Epoch')

    plt.ylabel('准确率')

    plt.legend()

    plt.grid(True)

    

    plt.tight_layout()

    plt.show()

```

## 预测与结果分析

```python

class PredictionAnalyzer:

    """预测结果分析器"""

    def __init__(self, model, framework='pytorch'):

        self.model = model

        self.framework = framework

        

    def analyze_predictions(self, test_data, test_labels, num_samples=5):

        """分析预测结果"""

        if self.framework == 'pytorch':

            predictions = self.model.predict(test_data)

            test_labels = test_labels.cpu().numpy()

        else:

            predictions = self.model.predict(test_data)

            test_labels = np.argmax(test_labels, axis=1)

        

        # 计算准确率

        correct = (predictions == test_labels).sum()

        total = len(test_labels)

        accuracy = 100 * correct / total

        

        print(f"总体准确率: {accuracy:.2f}%")

        

        # 显示错误预测示例

        self.show_misclassifications(

            test_data, test_labels, predictions, num_samples

        )

        

        return accuracy

    

    def show_misclassifications(self, data, true_labels, predictions, num_samples=5):

        """显示错误分类的样本"""

        incorrect_indices = np.where(predictions != true_labels)[0]

        

        if len(incorrect_indices) == 0:

            print("所有预测都正确！")

            return

        

        print(f"\n显示 {min(num_samples, len(incorrect_indices))} 个错误预测:")

        

        plt.figure(figsize=(15, 3))

        for i, idx in enumerate(incorrect_indices[:num_samples]):

            plt.subplot(1, num_samples, i+1)

            

            if self.framework == 'pytorch':

                img = data[idx].cpu().numpy().squeeze()

            else:

                img = data[idx].squeeze()

                

            plt.imshow(img, cmap='gray')

            plt.title(f'真实: {true_labels[idx]}\n预测: {predictitions[idx]}')

            plt.axis('off')

        

        plt.show()

    

    def create_confusion_matrix(self, true_labels, predictions):

        """创建混淆矩阵"""

        from sklearn.metrics import confusion_matrix

        import seaborn as sns

        

        cm = confusion_matrix(true_labels, predictions)

        

        plt.figure(figsize=(10, 8))

        sns.heatmap(cm, annot=True, fmt='d', cmap='Blues')

        plt.title('混淆矩阵')

        plt.xlabel('预测标签')

        plt.ylabel('真实标签')

        plt.show()

        

        return cm

```

## 模型优化与改进

```python

class AdvancedNeuralNetwork(nn.Module):

    """改进的神经网络模型"""

    def __init__(self):

        super(AdvancedNeuralNetwork, self).__init__()

        <"q2.j9k5.org.cn"><"t5.j9k5.org.cn"><"i1.j9k5.org.cn">

        # 卷积神经网络结构

        self.conv1 = nn.Conv2d(1, 32, kernel_size=3, padding=1)

        self.conv2 = nn.Conv2d(32, 64, kernel_size=3, padding=1)

        self.pool = nn.MaxPool2d(2, 2)

        self.dropout1 = nn.Dropout2d(0.25)

        self.dropout2 = nn.Dropout(0.5)

        

        # 计算全连接层输入尺寸

        self.flatten = nn.Flatten()

        self.fc1 = nn.Linear(64 * 7 * 7, 128)

        self.fc2 = nn.Linear(128, 10)

        

        self.relu = nn.ReLU()

        self.softmax = nn.Softmax(dim=1)

        

    def forward(self, x):

        # 卷积层

        x = self.relu(self.conv1(x))

        x = self.pool(x)

        x = self.relu(self.conv2(x))

        x = self.pool(x)

        x = self.dropout1(x)

        

        # 全连接层

        x = self.flatten(x)

        x = self.relu(self.fc1(x))

        x = self.dropout2(x)

        x = self.fc2(x)

        

        return self.softmax(x)

def train_advanced_model():

    """训练改进模型"""

    train_loader, test_loader = load_data_pytorch()

    

    advanced_model = AdvancedNeuralNetwork()

    trainer = PyTorchTrainer(advanced_model, learning_rate=0.001)

    

    print("训练改进的卷积神经网络...")

    trainer.train(train_loader, test_loader, epochs=5)

    

    return advanced_model

```

## 完整工作流程示例

```python

def complete_workflow(framework='pytorch'):

    """完整的神经网络工作流程"""

    

    print("1. 加载和预处理数据...")

    if framework == 'pytorch':

        train_loader, test_loader = load_data_pytorch()

        

        print("2. 构建PyTorch神经网络...")

        model = PyTorchNeuralNetwork()

        

        print("3. 训练模型...")

        trainer = PyTorchTrainer(model)

        train_losses, train_accs, test_losses, test_accs = trainer.train(

            train_loader, test_loader, epochs=5

        )

        

        print("4. 评估模型...")

        test_loss, test_acc = trainer.evaluate(test_loader)

        print(f"最终测试准确率: {test_acc:.2f}%")

        

        print("5. 预测分析...")

        analyzer = PredictionAnalyzer(model, framework='pytorch')

        

        # 获取测试数据用于分析

        test_data, test_labels = next(iter(test_loader))

        accuracy = analyzer.analyze_predictions(test_data, test_labels)

        

    else:

        (x_train, y_train), (x_test, y_test) = load_data_keras()

        

        print("2. 构建Keras神经网络...")

        model = KerasNeuralNetwork()

        

        print("3. 训练模型...")

        history = model.train(x_train, y_train, x_test, y_test, epochs=5)

        

        print("4. 评估模型...")

        test_loss, test_acc = model.evaluate(x_test, y_test)

        print(f"最终测试准确率: {test_acc:.4f}")

        

        print("5. 预测分析...")

        analyzer = PredictionAnalyzer(model, framework='keras')

        

        # 使用部分测试数据进行预测

        sample_size = 1000

        sample_data = x_test[:sample_size]

        sample_labels = y_test[:sample_size]

        

        accuracy = analyzer.analyze_predictions(sample_data, sample_labels)

    

    return model, accuracy

if __name__ == "__main__":

    # 选择框架

    framework_choice = input("选择框架 (pytorch/keras): ").lower()

    

    if framework_choice in ['pytorch', 'keras']:

        model, accuracy = complete_workflow(framework_choice)

        print(f"\n使用{framework_choice}框架完成手写数字识别任务")

        print(f"模型准确率: {accuracy:.2f}%")

    else:

        print("请输入有效的框架选择: pytorch 或 keras")

```

## 学习建议与总结

通过本文的实践，可以掌握以下关键技能：

1. **数据预处理**：掌握MNIST数据集的加载和预处理方法

2. **模型构建**：理解全连接神经网络的基本结构

3. **训练流程**：掌握模型的训练、验证和评估方法

4. **框架对比**：了解PyTorch和Keras的异同点

进一步学习建议：

- 尝试调整网络结构（层数、神经元数量）

- 实验不同的优化器和学习率

- 添加批归一化层改进模型性能

- 尝试卷积神经网络获得更高准确率

- 将模型部署为Web服务

手写数字识别作为深度学习入门项目，为后续学习图像分类、目标检测等复杂任务奠定基础。建议在掌握基本方法后，尝试在更多数据集上应用所学知识。