简介本资源是一套面向机器学习与脑电EEG信号分析初学者及研究者的SEED数据集实践代码合集聚焦情绪识别这一典型应用场景特别适合高校学生、科研入门者开展EEG特征提取、模型训练与性能对比实验。压缩包共200个文件以86个Python脚本为核心含CNN、RGNN、DANN等主流模型实现、50个CSV格式的原始/预处理脑电样本数据、8个Jupyter Notebook实验记录及6个Markdown说明文档为主干辅以MATLAB数据文件、模型权重.pkl/.mat和可视化图片整体78.93MB结构清晰便于模块化学习与复现。已有4352人学习下载涵盖从4D-CNN94%准确率到CNNSVM73%、RGNN67.7%等多算法实现包含完整训练日志、示例矩阵及跨被试数据划分逻辑可直接用于基线复现、模型改进或课程设计参考。1. “seed数据集相关代码整理”不是在打包下载链接而是在构建可复现的数据加载基线很多刚接触模型训练的工程师看到“seed数据集相关代码整理”这个标题第一反应是找现成的.zip下载包或git clone地址——但实际场景中真正的“整理”发生在代码层如何用确定性方式加载、划分、增强、缓存 seed 数据集使其在 PyTorch/TensorFlow 环境下跨机器、跨会话保持完全一致的样本顺序与变换结果。这直接决定模型对比实验是否可信同一组超参在不同 GPU 上跑出的 loss 曲线若因数据 pipeline 随机性偏差超过 ±0.02就无法判断是算法改进还是数据抖动。本篇聚焦于以 seed 为控制锚点的数据集代码工程实践覆盖从原始文件读取、索引生成、采样器配置到 DataLoader 并发安全的全链路。适用对象包括需要提交可复现实验的论文作者、部署多节点训练任务的 MLOps 工程师、以及正在调试数据泄露问题的算法研究员。文中所有代码均基于 Python 3.8 PyTorch 2.0不依赖任何第三方数据集管理库如torchvision.datasets的封装逻辑直击底层可控点。2. 用torch.Generator和random.seed()双控种子确保数据加载全流程确定性2.1 为什么单设torch.manual_seed(42)不够——三类随机源必须分别初始化PyTorch 数据加载流程中存在三个独立随机源各自影响不同环节Python 内置random模块控制Dataset.__getitem__中的随机增强如random.choice,random.shuffleNumPy 的np.random常用于自定义 transform 中的几何变换如cv2.warpAffine的参数生成PyTorch 的torch.Generator驱动DataLoader的sampler如RandomSampler和worker_init_fn中的 tensor 创建。若只调用torch.manual_seed(42)random.shuffle()仍会使用系统时间生成新种子导致每次运行train_dataset[0]返回不同样本。必须显式同步三者import random import numpy as np import torch def set_seed(seed: int 42) - None: 统一设置三类随机源种子确保数据加载确定性 random.seed(seed) # 控制 random 模块 np.random.seed(seed) # 控制 numpy 随机 torch.manual_seed(seed) # 控制 torch CPU 张量 if torch.cuda.is_available(): torch.cuda.manual_seed_all(seed) # 控制所有 GPU 设备 torch.backends.cudnn.deterministic True # 关闭 cuDNN 非确定性算法 torch.backends.cudnn.benchmark False # 禁用自动优化避免不同输入触发不同 kernel set_seed(42)提示torch.backends.cudnn.deterministic True是关键开关。cuDNN 默认启用非确定性卷积算法如CUDNN_CONVOLUTION_FWD_ALGO_IMPLICIT_PRECOMP_GEMM即使输入 tensor 完全相同也可能因硬件调度差异返回微小数值偏差。该设置强制使用确定性算法代价是约 5–10% 推理速度下降但对实验复现不可或缺。2.2 构建可复现的Subset划分用torch.Generator生成固定索引常见误区是用sklearn.model_selection.train_test_split划分数据集索引——它依赖np.random但未绑定到DataLoader的 worker 初始化流程。正确做法是用torch.Generator生成划分索引并保存为.npy文件供后续加载import torch from torch.utils.data import Subset, Dataset class CustomDataset(Dataset): def __init__(self, data_dir: str): self.file_list [f for f in os.listdir(data_dir) if f.endswith(.jpg)] self.labels [int(f.split(_)[0]) for f in self.file_list] # 示例标签 def __len__(self): return len(self.file_list) def __getitem__(self, idx): img_path os.path.join(data_dir, self.file_list[idx]) label self.labels[idx] # 此处可加入 random-based augmentations已受 set_seed() 控制 return torch.randn(3, 224, 224), label # 占位返回 # 生成固定划分索引仅执行一次结果持久化 full_dataset CustomDataset(data/seed_dataset) indices torch.randperm(len(full_dataset), generatortorch.Generator().manual_seed(42)).tolist() train_indices indices[:int(0.7 * len(full_dataset))] val_indices indices[int(0.7 * len(full_dataset)):int(0.9 * len(full_dataset))] test_indices indices[int(0.9 * len(full_dataset)):] # 保存索引供复用避免每次重新 shuffle np.save(data/seed_dataset/train_indices.npy, train_indices) np.save(data/seed_dataset/val_indices.npy, val_indices) np.save(data/seed_dataset/test_indices.npy, test_indices) # 后续加载时直接读取不依赖随机生成 train_subset Subset(full_dataset, np.load(data/seed_dataset/train_indices.npy)) val_subset Subset(full_dataset, np.load(data/seed_dataset/val_indices.npy)) test_subset Subset(full_dataset, np.load(data/seed_dataset/test_indices.npy))2.2.1 为什么不用torch.utils.data.random_split()random_split()内部调用torch.randperm()但其generator参数默认为None即使用全局torch.default_generator。若在DataLoader多 worker 场景下未显式重置各 worker 可能继承不同状态的 generator导致子集内容错乱。而手动保存索引文件彻底解耦划分逻辑与加载逻辑是工业级项目首选。2.3DataLoader的 worker 初始化每个子进程必须拥有独立且确定的随机状态当num_workers 0时PyTorch 启动多个子进程加载数据。若不干预各 worker 继承父进程的随机状态但torch.Generator在 fork 后不会自动重置导致所有 worker 生成完全相同的随机数序列例如所有 worker 都从random.shuffle()返回同一顺序。解决方案是在worker_init_fn中为每个 worker 设置唯一 seeddef worker_init_fn(worker_id: int) - None: 为每个 DataLoader worker 设置独立随机种子 # 基于全局 seed 和 worker_id 生成唯一 seed避免冲突 worker_seed torch.initial_seed() % 2**32 # 获取当前 worker 的初始 seed np.random.seed(worker_seed) random.seed(worker_seed) # 构建 DataLoader关键参数已标注 train_loader torch.utils.data.DataLoader( train_subset, batch_size32, shuffleTrue, # 启用 shuffle 才需 RandomSampler num_workers4, pin_memoryTrue, drop_lastTrue, # 必须指定 generator否则 RandomSampler 使用全局 default_generator generatortorch.Generator().manual_seed(42), # 关键初始化每个 worker 的随机状态 worker_init_fnworker_init_fn )注意generatortorch.Generator().manual_seed(42)这行不可省略。它确保RandomSampler在每个 epoch 开始时按相同顺序打乱索引。若缺失即使worker_init_fn正确shuffleTrue也会因 sampler 随机性失效。3. 解析 seed 数据集结构从文件命名规则到标签映射表的标准化处理3.1 识别典型 seed 数据集目录结构及元数据文件“seed 数据集”并非特指某公开数据集如 MNIST 或 COCO而是泛指以确定性种子为核心设计原则的自建数据集。其常见物理结构有三类结构类型目录示例特点代码解析要点扁平文件夹seed_dataset/class_name_id.jpg文件名编码类别与 ID无子目录需正则提取class_name构建class_to_idx映射分层目录seed_dataset/train/class_A/001.jpg类别为子目录名天然提供标签用torchvision.datasets.ImageFolder可直接加载但需覆写loader保证确定性带元数据文件seed_dataset/images/,seed_dataset/labels.csv标签与路径分离支持复杂标注如 bbox、mask必须解析 CSV按seed排序后切分避免 pandas 默认随机以最典型的labels.csv结构为例含 seed 列filename,label,seed img_001.jpg,cat,12345 img_002.jpg,dog,67890 img_003.jpg,cat,24680 ...解析代码需确保按seed列排序后划分而非默认行序import pandas as pd def load_seed_csv(csv_path: str, seed_col: str seed) - pd.DataFrame: 加载并按 seed 列升序排序保证划分可复现 df pd.read_csv(csv_path) # 关键按 seed 排序而非读取顺序 df df.sort_values(byseed_col).reset_index(dropTrue) return df # 使用示例 df load_seed_csv(data/seed_dataset/labels.csv) # 后续划分 train/val/test 时直接按 df.index 切片无需 shuffle train_df df.iloc[:int(0.7 * len(df))] val_df df.iloc[int(0.7 * len(df)):int(0.9 * len(df))] test_df df.iloc[int(0.9 * len(df)):]3.2 构建确定性标签映射避免os.listdir()的文件系统顺序干扰当数据集采用分层目录如train/cat/,train/dog/时os.listdir()返回顺序依赖文件系统ext4 vs NTFS不可靠。正确做法是显式排序后构建class_to_idximport os from typing import Dict, List def build_class_mapping(root_dir: str, sort_key: str name) - Dict[str, int]: 构建确定性类别映射规避文件系统顺序差异 classes [d for d in os.listdir(root_dir) if os.path.isdir(os.path.join(root_dir, d))] if sort_key name: classes sorted(classes) # 字典序排序跨平台一致 elif sort_key seed: # 若目录名含 seed如 cat_12345按数字部分排序 classes sorted(classes, keylambda x: int(x.split(_)[-1])) class_to_idx {cls: idx for idx, cls in enumerate(classes)} return class_to_idx # 示例root_dir data/seed_dataset/train class_to_idx build_class_mapping(data/seed_dataset/train) print(class_to_idx) # {cat: 0, dog: 1} —— 顺序确定3.2.1 自定义Dataset中的安全文件读取在__getitem__中必须用sorted()包裹os.listdir()否则image_files[0]在不同机器上可能指向不同图片class SeedImageDataset(Dataset): def __init__(self, root_dir: str, class_to_idx: Dict[str, int], transformNone): self.class_to_idx class_to_idx self.transform transform self.samples [] for class_name in class_to_idx: class_path os.path.join(root_dir, class_name) # 关键显式排序消除文件系统差异 image_files sorted([f for f in os.listdir(class_path) if f.lower().endswith((.jpg, .png))]) for img_file in image_files: self.samples.append((os.path.join(class_path, img_file), class_to_idx[class_name])) def __len__(self): return len(self.samples) def __getitem__(self, idx): img_path, label self.samples[idx] # PIL Image.open() 本身确定性无需额外 seed image Image.open(img_path).convert(RGB) if self.transform: image self.transform(image) return image, label4. 验证 seed 数据集代码的确定性三步交叉检查法4.1 检查 DataLoader 输出张量的哈希一致性最直接的验证是比对两个独立运行的DataLoader迭代器首 N 个 batch 的 SHA256 哈希值。以下脚本生成batch_hash.txt供比对import hashlib import torch def compute_batch_hash(dataloader, num_batches: int 5) - List[str]: 计算前 num_batches 个 batch 的 SHA256 哈希验证确定性 hashes [] for i, (images, labels) in enumerate(dataloader): if i num_batches: break # 将 batch tensor 转为 bytes注意 dtype 和 layout tensor_bytes images.numpy().tobytes() labels.numpy().tobytes() hash_obj hashlib.sha256(tensor_bytes) hashes.append(hash_obj.hexdigest()[:16]) # 取前 16 位简化显示 return hashes # 运行两次比对结果 train_loader create_dataloader() # 使用前述确定性配置 hashes_run1 compute_batch_hash(train_loader) hashes_run2 compute_batch_hash(train_loader) # 第二次运行 print(Run 1:, hashes_run1) print(Run 2:, hashes_run2) assert hashes_run1 hashes_run2, Batch hashes differ! Determinism broken.提示images.numpy().tobytes()比images.cpu().numpy().data.tobytes()更安全避免因 tensor deviceCPU/GPU或 memory layoutcontiguous/non-contiguous引入差异。4.2 监控DataLoaderworker 的随机状态漂移当num_workers 0时可通过日志确认每个 worker 是否正确初始化def worker_init_fn_debug(worker_id: int) - None: worker_seed torch.initial_seed() % 2**32 print(f[Worker {worker_id}] Initial seed: {worker_seed}) np.random.seed(worker_seed) random.seed(worker_seed) # 在 DataLoader 中启用 train_loader torch.utils.data.DataLoader( ..., worker_init_fnworker_init_fn_debug # 临时替换为 debug 版本 )正常输出应为[Worker 0] Initial seed: 123456789 [Worker 1] Initial seed: 123456790 [Worker 2] Initial seed: 123456791 [Worker 3] Initial seed: 123456792种子值呈等差递增torch.initial_seed()在 fork 后为 worker 分配连续 seed证明worker_init_fn生效。4.3 对比不同 PyTorch 版本下的Generator行为边界PyTorch 1.13 对torch.Generator的manual_seed()实现更严格但旧版本如 1.9存在Generator在DataLoader中被意外 reset 的 bug。必须在目标环境验证# 测试 Generator 状态是否跨 epoch 持久 gen torch.Generator() gen.manual_seed(42) print(Epoch 1, step 0:, gen.initial_seed()) # 应为 42 for _ in range(100): _ torch.rand(1, generatorgen) # 消耗部分状态 print(After 100 rand calls:, gen.initial_seed()) # 仍为 42证明状态未重置 # 若此处输出变化则说明 Generator 实现有缺陷需升级 PyTorch若initial_seed()在消耗后改变表明该版本Generator未正确维护内部状态必须升级至 PyTorch 2.0。5. 进阶技巧用torchdata的DataPipe替代传统Dataset实现声明式 seed 控制5.1DataPipe如何简化 seed 管理从 imperative 到 declarativePyTorch 2.0 引入的torchdata需pip install torchdata提供函数式数据流水线其shuffle()、sharding_filter()等操作原生支持seed参数无需手动管理Generatorfrom torchdata.datapipes.iter import IterableWrapper, Shuffler # 声明式构建 pipeline无需自定义 Dataset file_list [data/seed_dataset/img_001.jpg, data/seed_dataset/img_002.jpg, ...] dp IterableWrapper(file_list) # 直接指定 seedshuffle 操作自动创建确定性 Generator dp dp.shuffle(buffer_size1000, seed42) # 解析文件并加载 def load_and_label(x): label int(x.split(_)[1].split(.)[0]) # 从文件名提取标签 return torch.randn(3, 224, 224), label dp dp.map(load_and_label) # 分批 dp dp.batch(32, drop_lastTrue) # 转为 DataLoader仍需 worker_init_fn但 pipeline 内部已确定 train_loader DataLoader(dp, num_workers4, worker_init_fnworker_init_fn)优势在于shuffle(buffer_size1000, seed42)的语义比RandomSampler更清晰且DataPipe的seed参数在fork后自动适配 worker减少出错概率。5.2 构建 seed-aware 缓存机制避免重复解码开销对于大尺寸图像数据集每次__getitem__解码 JPEG 是性能瓶颈。torchdata提供in_memory_cache但需确保缓存键包含 seed 信息from torchdata.datapipes.iter import InMemoryCacheHolder def cache_key_func(sample): 缓存键必须包含 seed否则不同 seed 的 pipeline 共享缓存 # sample 是 (image_path, label)需将 seed 注入键 return (sample[0], 42) # 固定 seed 42或从上下文传入 dp dp.in_memory_cache(cache_key_funccache_key_func)若忽略seed当切换seed123重新运行时DataPipe可能复用seed42的缓存导致数据错乱。此细节是seed 数据集相关代码整理中易被忽视的隐性风险点。5.3 用torch.compile()加速确定性 pipelinePyTorch 2.3torch.compile()默认启用torch._dynamo.config.cache_size_limit 64但若 pipeline 中有random调用编译器可能因副作用拒绝优化。解决方案是将随机操作移出编译区域# ❌ 错误在 torch.compile 函数内调用 random torch.compile def forward_with_aug(x): if random.random() 0.5: # 编译器无法追踪此随机性 x transforms.RandomHorizontalFlip()(x) return model(x) # ✅ 正确预生成 mask编译区只做确定性运算 def generate_aug_mask(batch_size: int, seed: int) - torch.Tensor: torch.manual_seed(seed) return torch.rand(batch_size) 0.5 torch.compile def compiled_forward(x, aug_mask): x torch.where(aug_mask.unsqueeze(-1).unsqueeze(-1), transforms.functional.hflip(x), x) return model(x) # 使用 aug_mask generate_aug_mask(32, 42) # 在 compile 区外生成 output compiled_forward(images, aug_mask)此模式将随机性隔离在编译器感知范围外既享受torch.compile的 20–30% 速度提升又不破坏 seed 确定性。本文还有配套的精品资源点击获取