{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.11.13","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"gpu","dataSources":[{"sourceId":114675,"databundleVersionId":13685133,"sourceType":"competition"}],"dockerImageVersionId":31090,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"# This Python 3 environment comes with many helpful analytics libraries installed\n# It is defined by the kaggle/python Docker image: https://github.com/kaggle/docker-python\n# For example, here's several helpful packages to load\n\nimport numpy as np # linear algebra\nimport pandas as pd # data processing, CSV file I/O (e.g. pd.read_csv)\n\n# Input data files are available in the read-only \"../input/\" directory\n# For example, running this (by clicking run or pressing Shift+Enter) will list all files under the input directory\n\nimport os\nfor dirname, _, filenames in os.walk('/kaggle/input'):\n    for filename in filenames:\n        print(os.path.join(dirname, filename))\n\n# You can write up to 20GB to the current directory (/kaggle/working/) that gets preserved as output when you create a version using \"Save & Run All\" \n# You can also write temporary files to /kaggle/temp/, but they won't be saved outside of the current session","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true,"execution":{"iopub.status.busy":"2025-09-10T02:40:43.09252Z","iopub.execute_input":"2025-09-10T02:40:43.092802Z","iopub.status.idle":"2025-09-10T02:40:43.447603Z","shell.execute_reply.started":"2025-09-10T02:40:43.092773Z","shell.execute_reply":"2025-09-10T02:40:43.446865Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# 下载 CIFAR10 数据集\nimport torchvision.datasets as datasets\nimport torchvision.transforms as transforms\nimport torch\nimport torch.nn as nn\nimport torch.nn.functional as F\nimport torch.optim as optim\nfrom torch.utils.data import DataLoader, Dataset\nimport numpy as np\nimport pandas as pd\nfrom pathlib import Path\n\n# 定义数据转换\ntransform = transforms.Compose([\n    transforms.ToTensor(),\n    transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))\n])\n# 下载训练集和测试集\ntrain_dataset = datasets.CIFAR10(root='/kaggle/input/cnn-cifar-10', train=True, transform=transform)\ntest_dataset = datasets.CIFAR10(root='/kaggle/input/cnn-cifar-10', train=False, transform=transform)\n\nclass CIFAR10Dataset(Dataset):\n    \"\"\"\n    CIFAR10数据集加载器\n    \"\"\"\n    def __init__(self, x, y):\n        self._x = x\n        self._y = y\n\n    def __getitem__(self, idx):\n        return {\n            \"images\": torch.tensor(self._x[idx], dtype=torch.float32),\n            \"label\": torch.tensor(self._y[idx], dtype=torch.long),\n        }\n\n    def __len__(self):\n        return len(self._x)\n\ndef prepare_data_loader(\n    path: str,\n    ratio: float,\n    train_batch_size: int,\n    num_workers: int,\n) -> dict:\n    \"\"\"\n    参数:\n        path (str): .npz格式的数据集文件路径\n        ratio (float): 训练集比例\n        train_batch_size (int): 批次大小\n        num_workers (int): 数据加载的工作进程数\n    返回:\n        dict: 包含训练和测试数据加载器的字典\n    \"\"\"\n    print(\"开始加载数据...\")  # 添加调试信息\n    train_dataset = datasets.CIFAR10(root='/kaggle/input/cnn-cifar-10', train=True, transform=transform)\n    test_dataset = datasets.CIFAR10(root='/kaggle/input/cnn-cifar-10', train=False, transform=transform)\n    train_loader = torch.utils.data.DataLoader(\n                                                dataset=train_dataset, \n                                               batch_size=train_batch_size, \n                                               shuffle=True,\n                                               drop_last=True,\n                                               pin_memory=True  \n                                              )\n    test_loader = torch.utils.data.DataLoader(\n                                              dataset=test_dataset, \n                                              batch_size=train_batch_size, \n                                              shuffle=False,\n                                               pin_memory=True  \n    )\n\n    # train_loader = DataLoader(\n    #     dataset=train_dataset,\n    #     batch_size=train_batch_size,\n    #     shuffle=True,\n    #     num_workers=1,  \n    #     drop_last=True,\n    #     pin_memory=True  \n    # )\n\n    # test_loader = DataLoader(\n    #     dataset=test_dataset,\n    #     batch_size=test_batch_size,\n    #     shuffle=False,\n    #     num_workers=1,  \n    #     pin_memory=True  \n    # )\n    print(\"数据加载器创建完成\")  # 添加调试信息\n\n    return {\"train\": train_loader, \"test\": test_loader}\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"class SimpleCNN(nn.Module):\n    \"\"\"\n    可配置的CNN模型\n    \n    参数:\n        in_channels (int): 输入通道数\n        num_classes (int): 分类类别数\n        conv_layers (list): 每个卷积层的输出通道数列表\n        fc_layers (list): 每个全连接层的输出维度列表\n        kernel_size (int): 卷积核大小\n        dropout_rate (float): Dropout比率\n    \"\"\"\n    def __init__(\n        self,\n        in_channels: int = 3,\n        num_classes: int = 10,\n        conv_layers: list = [32, 64],  # 默认两层卷积\n        fc_layers: list = [128, 64],   # 默认两层全连接\n        kernel_size: int = 3,\n        dropout_rate: float = 0.1\n    ):\n        super().__init__()\n        \n        # 构建卷积层\n        self.conv_blocks = nn.ModuleList()\n        current_channels = in_channels\n        \n        for i, out_channels in enumerate(conv_layers):\n            conv_block = nn.Sequential(\n                nn.Conv2d(\n                    in_channels=current_channels,\n                    out_channels=out_channels,\n                    kernel_size=kernel_size,\n                    stride=1,\n                    padding='same'\n                ),\n                # nn.BatchNorm2d(out_channels), 可选择是否添加BatchNorm层\n                nn.ReLU(),\n                nn.MaxPool2d(kernel_size=2)\n            )\n            self.conv_blocks.append(conv_block)\n            current_channels = out_channels\n           \n            \n        # 计算展平后的特征维度\n        # 每经过一次MaxPool2d或AvgPool2d，特征图尺寸减半\n        feature_size = current_channels * (32 // (2 ** len(conv_layers))) ** 2\n        \n        # 构建全连接层\n        self.fc_blocks = nn.ModuleList()\n        current_dim = feature_size\n        \n        for fc_dim in fc_layers:\n            fc_block = nn.Sequential(\n                nn.Linear(current_dim, fc_dim),\n                nn.ReLU(),\n                nn.Dropout(dropout_rate)\n            )\n            self.fc_blocks.append(fc_block)\n            current_dim = fc_dim\n            \n        # 输出层\n        self.output_layer = nn.Linear(current_dim, num_classes)\n\n    def forward(self, x):\n         # 通过所有卷积层\n         for conv_block in self.conv_blocks:\n             x = conv_block(x)\n    \n         # 展平\n         x = torch.flatten(x, start_dim=1)\n    \n         # 通过所有全连接层\n         for fc_block in self.fc_blocks:\n              x = fc_block(x)\n        \n         # 输出层\n         x = self.output_layer(x)\n         return x \n        \n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"class SimpleMLP(nn.Module):\n    \"\"\"\n    可配置的多层感知机模型\n    \n    参数:\n        input_size (int): 输入特征维度 (对于CIFAR10是3*32*32=3072)\n        num_classes (int): 分类类别数\n        hidden_layers (list): 每个隐藏层的输出维度列表\n        dropout_rate (float): Dropout比率\n        activation (str): 激活函数类型 ('relu', 'tanh', 'sigmoid')\n    \"\"\"\n    def __init__(\n        self,\n        input_size: int = 3072,  # 3*32*32 for CIFAR10\n        num_classes: int = 10,\n        hidden_layers: list = [512, 256, 128],  # 默认三层隐藏层\n        dropout_rate: float = 0.1,\n        activation: str = 'relu'\n    ):\n        super(SimpleMLP, self).__init__()\n        \n        # 选择激活函数\n        if activation == 'relu':\n            self.activation = nn.ReLU()\n        elif activation == 'tanh':\n            self.activation = nn.Tanh()\n        elif activation == 'sigmoid':\n            self.activation = nn.Sigmoid()\n        else:\n            raise ValueError(f\"Unsupported activation function: {activation}\")\n            \n        # 构建隐藏层\n        self.fc_blocks = nn.ModuleList()\n        current_dim = input_size\n        \n        for hidden_dim in hidden_layers:\n            fc_block = nn.Sequential(\n                nn.Linear(current_dim, hidden_dim),\n                self.activation,\n                nn.BatchNorm1d(hidden_dim),  # 添加批归一化\n                nn.Dropout(dropout_rate)\n            )\n            self.fc_blocks.append(fc_block)\n            current_dim = hidden_dim\n            \n        # 输出层\n        self.output_layer = nn.Linear(current_dim, num_classes)\n\n    def forward(self, x):\n        # 展平输入\n        batch_size = x.size(0)\n        x = x.view(batch_size, -1)\n        \n        # 通过所有隐藏层\n        for fc_block in self.fc_blocks:\n            x = fc_block(x)\n            \n        # 输出层\n        x = self.output_layer(x)\n        return x\n        ","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import torch.optim as optim\n\nloss_fn = nn.CrossEntropyLoss()\n\ndef train_step(\n    model: nn.Module, optimizer, batch: dict, device: torch.device\n):\n    \"\"\"\n    单步训练\n    \"\"\"\n    batch_images, labels = batch\n    batch_images = batch_images.to(device)\n    labels = labels.to(device)\n        \n    optimizer.zero_grad()\n\n    logits = model(batch_images) # 模型正向过程\n        \n    loss = loss_fn(logits, labels) # 计算总损失\n        \n    loss.backward() # 反向传播\n        \n    torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) # 添加梯度裁剪，防止梯度爆炸\n        \n    optimizer.step()\n\n    return loss.item(), logits, labels\n        \n\ndef eval_step(model: nn.Module, batch: dict, device: torch.device):\n    # 单步评估\n    model.eval()\n    with torch.no_grad():\n        batch_images, labels = batch\n        batch_images = batch_images.to(device)\n        labels = labels.to(device) \n        logits = model(batch_images) # 模型正向过程\n        loss = loss_fn(logits, labels) # 计算总损失\n        return loss.item(), logits, labels\n    \ndef train_per_epoch(\n    model: nn.Module,\n    optimizer: optim.Optimizer,\n    batch_size: int,\n    train_loader: DataLoader,\n    device: torch.device,\n):\n    model.train()\n    num_data = len(train_loader.dataset)\n    num_batches = len(train_loader)\n    correct = 0\n    total_loss = 0\n    print(f\"开始训练 - 总样本数: {num_data}\")\n    print(f\"总批次数: {len(train_loader)}\")\n    \n    for batch_idx, batch in enumerate(train_loader):\n        loss, logits, labels = train_step(model, optimizer, batch, device)\n        total_loss += loss\n        _, predicted = torch.max(logits, 1)\n        correct += (predicted == labels).sum().item()    \n        if batch_idx % 50 == 0:  # 改为每50个批次打印一次\n            current = batch_idx * batch_size + len(batch[0])\n            print(f\"批次 {batch_idx}: Loss: {loss:>6.4f}, 进度: {current:>5d}/{num_data:>5d}\")\n    accuracy = correct / num_data\n    avg_loss = total_loss / num_batches\n    print(f\"Train Error: \\n Accuracy: {(100*accuracy):>0.1f}%, Avg loss: {avg_loss:>.8f} \\n\")\n        \ndef test_per_epoch(\n    model: nn.Module,\n    test_loader: DataLoader,\n    device: torch.device,\n):\n    \"\"\"\n    每轮测试\n    \"\"\"\n    model.eval()\n    total_loss = 0.0\n    correct = 0\n    num_batches = len(test_loader)\n    num_data = len(test_loader.dataset)\n    with torch.no_grad():\n        for batch in test_loader:\n            loss, logits, labels = eval_step(model, batch, device)\n            total_loss += loss\n            _, predicted = torch.max(logits, 1)\n            correct += (predicted == labels).sum().item()\n\n    avg_loss = total_loss / num_batches\n    accuracy = correct / num_data\n    print(f\"Test Error: \\n Accuracy: {(100*accuracy):>0.1f}%, Avg loss: {avg_loss:>.8f} \\n\")\n    return 100*accuracy\n\n\ndef controller(\n    seed: int,\n    model_type: str,  # 'cnn' 或 'mlp'\n    # CNN特有参数\n    in_channels: int = None,\n    conv_layers: list = None,\n    kernel_size: int = None,\n    fc_layers  : list = None,\n    # MLP特有参数\n    input_size: int = None,\n    hidden_layers: list = None,\n    activation: str = None,\n    # 通用参数\n    num_classes: int = 10,\n    dropout_rate: float = 0.1,\n    data_path: Path = None,\n    ratio: float = 0.8,\n    train_batch_size: int = 64,\n    num_workers: int = 4,\n    epochs: int = 10,\n    learning_rate: float = 0.001,\n    weight_decay: float = 0.004\n):\n    \"\"\"\n    训练控制器\n    \n    参数:\n        model_type: 选择模型类型 ('cnn' 或 'mlp')\n        其他参数见各模型的文档\n    \"\"\"\n    device = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n    torch.manual_seed(seed)\n    \n    if model_type == 'cnn':\n        model = SimpleCNN(\n            in_channels=in_channels,\n            num_classes=num_classes,\n            conv_layers=conv_layers,\n            fc_layers=fc_layers,  \n            kernel_size=kernel_size,\n            dropout_rate=dropout_rate\n        ).to(device)\n    elif model_type == 'mlp':\n        model = SimpleMLP(\n            input_size=input_size,\n            num_classes=num_classes,\n            hidden_layers=hidden_layers,\n            dropout_rate=dropout_rate,\n            activation=activation\n        ).to(device)\n    else:\n        raise ValueError(f\"Unsupported model type: {model_type}\")\n\n    optimizer = optim.Adam(model.parameters(), lr=learning_rate)  #选择Adam优化器\n    scheduler = optim.lr_scheduler.CosineAnnealingLR(optimizer,T_max=epochs) # 选择余弦退火模型\n    loader_dict = prepare_data_loader(path=data_path, ratio=ratio, train_batch_size=train_batch_size, num_workers=2 )\n    train_loader = loader_dict[\"train\"]\n    test_loader = loader_dict[\"test\"]\n    print(f\"Using device: {device}\") \n    best_acc = 0.0\n    for epoch in range(epochs):\n        print(f\"Epoch {epoch + 1} \\n--------------------------------\")\n        train_per_epoch(model, optimizer, train_batch_size, train_loader, device)\n        acc = test_per_epoch(model, test_loader, device)\n        scheduler.step()\n        if acc > best_acc:\n            best_acc = acc\n            torch.save(model.state_dict(), 'best_model.pth')\n    print(f\"Training completed! Best accuracy: {best_acc:.2f}%\")\n    return model","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# 随机数种子：确保每次运行代码得到相同的结果\nseed = 42\n\n# 模型类型：选择使用CNN（卷积神经网络）或MLP（多层感知机）\nmodel_type = 'cnn'\n\n# 卷积层配置：定义CNN模型中卷积层的结构\n# [32, 64]表示有两层卷积层，第一层输出32个特征图，第二层输出64个特征图\n# [32, 64, 128]则表示三层卷积层\nconv_layers = [32, 64]\n\n# 卷积核大小：定义每次卷积操作时，窗口的大小\n# kernel_size=3表示使用3×3的滑动窗口\nkernel_size = 3\n\n# 全连接层配置：定义CNN末端全连接层的结构\n# [128, 64]表示有两层全连接层，第一层有128个神经元，第二层有64个神经元\nfc_layers = [256, 128]\n\n# Dropout比率：随机\"关闭\"一部分神经元，防止模型过度依赖某些特征\n# 比率0.2表示每次训练时随机关闭20%的神经元\ndropout_rate = 0.3\n\n#ratio比率：训练集中用于训练的数据量/总数据量\n#比率0.8表示50000张图片中有40000张用于训练，10000用于测试\nratio = 0.8\n\n# 训练轮数：整个训练数据集要被训练的次数\nepochs = 20\n\n# 批次大小：每次训练选取的图片数量\nbatch_size = 128\n\n# 学习率：模型在训练过程中调整参数的步长\nlearning_rate = 0.001\n\n#  CNN模型示例\nmodel = controller(\n    seed=seed,\n    model_type=model_type,\n    in_channels=3,\n    conv_layers=conv_layers,  \n    kernel_size=kernel_size,\n    fc_layers=fc_layers,\n    dropout_rate=dropout_rate,\n    data_path=Path(\"/kaggle/input/cnns-cifar-10/train_data.npz\"),\n    ratio = ratio,\n    epochs=epochs,  \n    train_batch_size=batch_size,  \n    learning_rate=learning_rate,\n    weight_decay = 0.004 \n)\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')\n\ndef evaluater_with_dataloader_and_save(model, test_loader, device, solution_path=None):\n    \"\"\"\n    使用 test_dataloader 计算准确率并可选择保存预测结果\n    \n    参数:\n    - model: 需要评估的模型\n    - test_loader: 测试数据加载器\n    - device: 计算设备 (CPU 或 GPU)\n    - solution_path: 保存预测结果的路径(可选)\n    \n    返回:\n    - accuracy: 模型在测试集上的准确率\n    \"\"\"\n    model.eval()  # 设置为评估模式\n    correct = 0\n    total = 0\n    all_ids = []\n    all_predictions = []\n    \n    with torch.no_grad():  # 不计算梯度\n        for images, labels in test_loader:\n            # 如果 test_loader 包含ID信息，需要相应调整这里的解包方式\n            images, labels = images.to(device), labels.to(device)\n            outputs = model(images)\n            _, predicted = torch.max(outputs.data, 1)\n            total += labels.size(0)\n            correct += (predicted == labels).sum().item()\n            \n            # 如果需要保存预测结果\n            if solution_path is not None:\n                # 假设我们能够获取ID信息\n                batch_ids = range(len(all_predictions), len(all_predictions) + len(predicted))\n                all_ids.extend(batch_ids)\n                all_predictions.extend(predicted.cpu().numpy())\n    \n    accuracy = correct / total\n    print(f'测试集准确率: {accuracy:.4f}')\n    \n    # 保存预测结果\n    if solution_path is not None:\n        predictions_df = pd.DataFrame({\"Id\": all_ids, \"label\": all_predictions})\n        predictions_df.to_csv(solution_path, index=False)\n        print(f'预测结果已保存至 {solution_path}')\n    \n    return accuracy\nloader_dict = prepare_data_loader(path=Path(\"/kaggle/input/cnns-cifar-10/train_data.npz\"),train_batch_size=128, num_workers=2,ratio=0.8 )\ntrain_loader = loader_dict[\"train\"]\ntest_loader = loader_dict[\"test\"]\nevaluater_with_dataloader_and_save(\n    model,\n    test_loader=test_loader,\n    solution_path=Path(\"/kaggle/working/submission.csv\"),\n    device = device\n    )","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}